Telegraf azure_data_explorer 输出插件将指标写入 Azure Data Explorer、Synapse 与 Fabric 的完整实战指南【免费下载链接】telegrafAgent for collecting, processing, aggregating, and writing metrics, logs, and other arbitrary data.项目地址: https://gitcode.com/GitHub_Trending/te/telegrafTelegraf 的azure_data_explorer输出插件可将采集到的指标写入 Azure Data ExplorerADX、Azure Synapse Data Explorer 以及 Fabric 实时分析服务。本文基于插件官方文档plugins/outputs/azure_data_explorer/README.md并结合仓库源码完整讲解其配置参数、指标分组策略、表结构自动生成、两种摄取方式与认证机制并深入剖析插件的序列化与 Kusto 命令生成实现读完即可独立完成该插件的接入、调参与数据查询。插件定位与前置条件Azure Data Explorer 是一个分布式列式存储专为各类日志、指标与时序数据设计。⭐ 该插件自 Telegraf v1.20.0 引入标签为cloud、datastore支持全部平台 all。接入前需要完成两项准备工作在 Azure 上创建 Azure Data Explorer 集群和数据库可参考 Azure 门户中的 create-cluster-database-portal 文档流程准备一台 VM/计算节点或容器来承载 Telegraf——它可以部署在被监控应用所在的本地机器也可以部署在专门的监控计算节点上。全局配置与插件配置与 Telegraf 所有输出插件一致本插件也支持指标/标签/字段的过滤改写、插件别名与执行顺序等全局配置项详见 docs/CONFIGURATION.md。插件自身的完整配置样例即 plugins/outputs/azure_data_explorer/sample.conf如下# Sends metrics to Azure Data Explorer [[outputs.azure_data_explorer]] ## The URI property of the Azure Data Explorer resource on Azure ## ex: endpoint_url https://myadxresource.australiasoutheast.kusto.windows.net endpoint_url ## The Azure Data Explorer database that the metrics will be ingested into. ## The plugin will NOT generate this database automatically, its expected that this database already exists before ingestion. ## ex: exampledatabase database ## Timeout for Azure Data Explorer operations # timeout 20s ## Type of metrics grouping used when pushing to Azure Data Explorer. ## Default is TablePerMetric for one table per different metric. ## For more information, please check the plugin README. # metrics_grouping_type TablePerMetric ## Name of the single table to store all the metrics (Only needed if metrics_grouping_type is SingleTable). # table_name ## Creates tables and relevant mapping if set to true(default). ## Skips table and mapping creation if set to false, this is useful for running Telegraf with the lowest possible permissions (table ingestor role). # create_tables true ## Ingestion method to use. ## Available options are ## - managed -- streaming ingestion with fallback to batched ingestion or the queued method below ## - queued -- queue up metrics data and process sequentially # ingestion_type queued各参数在源码中的定义与默认值可参考 plugins/common/adx/adx.go 中的Config结构体配置项说明默认值endpoint_urlADX 资源的 URI如https://myadxresource.australiasoutheast.kusto.windows.net必填无database目标数据库名称插件不会自动创建数据库必须预先存在无必填timeoutADX 操作超时时间20smetrics_grouping_type指标分组方式TablePerMetric或SingleTableTablePerMetrictable_nameSingleTable模式下的目标表名无create_tables是否自动建表并创建 ingestion mappingtrueingestion_type摄取方式managed或queuedqueued在 插件初始化代码 中可以看到CreateTables默认true、Timeout默认 20 秒的注册逻辑连接建立时 plugins/common/adx/adx.go 的NewClient会做一系列校验endpoint/database 不能为空、metrics_grouping_type会被转小写并与合法值比对大小写不敏感、SingleTable模式必须提供table_name、未知ingestion_type直接报错timeout为 0 时回退为 20 秒。这些校验也都有对应的测试覆盖见 plugins/common/adx/adx_test.go 与 plugins/outputs/azure_data_explorer/azure_data_explorer_test.go。指标分组TablePerMetric 与 SingleTable通过metrics_grouping_type指定分组方式缺省为TablePerMetric。源码中对应常量为tablepermetric/singletableplugins/common/adx/adx.go写入入口 Write 方法 按该值分流到writeTablePerMetric或writeSingleTable。TablePerMetric默认插件按指标名metric name分组把每组指标写入一张同名的 ADX 表表不存在时自动创建表已存在时尝试将 Telegraf 指标 schema 合并进现有表即 Kusto 的.create-merge语义。由于表名与指标name一致如果计划给指标名加前缀必须确保最终名称满足 ADX 的表命名约束。从源码看writeTablePerMetric 的实现是以m.Name()为 key 建立tableName - 序列化字节流的 map逐条追加序列化结果后按表逐个调用PushMetrics推送。SingleTable所有指标统一写入一张表表名由table_name指定建表与 schema 合并逻辑同上。源码实现 writeSingleTable 则是把一批指标全部序列化拼接成一个字节流一次性推送到adx.TableName。表结构与自动建表ADX 表的 schema 与 TelegrafMetric对象结构一一对应共四个列fieldsdynamic、namestring、tagsdynamic、timestampdatetime。插件生成的建表命令形如.create-merge table [table-name] ([fields]:dynamic, [name]:string, [tags]:dynamic, [timestamp]:datetime)对应的 ingestion JSON mapping.create-or-alter table [table-name] ingestion json mapping table-name_mapping [{column:fields,Properties:{Path:$[\fields\]}},{column:name,Properties:{Path:$[\name\]}},{column:tags,Properties:{Path:$[\tags\]}},{column:timestamp,Properties:{Path:$[\timestamp\]}}]这两条命令由 plugins/common/adx/adx.go 中的createTableCommand/createTableMappingCommand用 KQL 构建器动态拼接表名安全注入并在 getMetricIngestor 中按CreateTables开关执行——即每个表仅在首次建立 ingestor 时创建一次表和 mapping随后 ingestor 会缓存在ingestorsmap 中复用。TestQueryConstruction 用断言精确验证了生成命令与上述文本完全一致。注意只有create_tablestrue默认时插件才执行上述建表动作置为false可跳过建表从而以最低权限运行。指标如何被序列化成 JSONInit阶段插件固定创建一个 JSON 序列化器azure_data_explorer.go时间戳单位为纳秒、格式为 RFC3339Nano。结合 plugins/serializers/json/json.go 的createObject可以看到每条指标被序列化为形如下面的单行 JSON每条以换行结尾{fields: {value: 1}, name: test1, tags: {tag1: value1}, timestamp: 2009-11-10T23:00:00Z}plugins/common/adx/adx_test.go 的TestPushMetrics正是用这一格式的样本验证推送链路而TestPushMetricsOutputs则覆盖了 TablePerMetric/SingleTable、建表开关、managed 摄取等组合场景。摄取方式managed 与 queuedingestion_type提供两种摄取路径managed流式摄取streaming ingestion失败时回退到批量摄取或queued方式。前提是 ADX 侧已开启 streaming ingestion可用以下 KQL 查询确认.show database DB-Name policy streamingingestionqueued默认将指标数据排队后顺序处理。源码中 queued 模式通过azkustoingest.New创建客户端并附加WithStaticBuffer(bufferSize, maxBuffers)静态缓冲选项1 MiB × 5 块见 adx.go 中的bufferSize/maxBuffers常量managed 模式则走azkustoingest.NewManaged。两种方式最终都通过 PushMetrics 发送以timeout配置限制请求上下文并按表名_mapping的 JSON mapping 调用FromReader完成摄取。认证机制与权限支持的认证方式插件通过检查一组约定环境变量自动选择认证方法AAD Application TokenService Principal支持 secret 或证书——生产环境推荐AAD User Token——以用户身份认证主要用于开发调试Managed Service IdentityMSI——在 Azure VM 或 Azure 基础设施上运行 Telegraf 时的首选方式。从源码结构看NewClient 使用WithDefaultAzureCredential()构建连接字符串即由 Azure SDK 的默认凭据链读取上文的环境变量完成认证连接中还通过SetConnectorDetails(Telegraf, ...)注入了 Telegraf 的客户端标识。无论采用哪种方式被指定的 Principal 都需在数据库层级被授予Database User角色以便插件建表并写入数据如果设置create_tablesfalse则最低只需Database Ingestor角色。按优先级排列的认证配置插件按以下顺序评估并选用第一个可用的认证配置Client CredentialsAAD 应用 ID Secret环境变量AZURE_TENANT_ID认证目标 TenantAZURE_CLIENT_ID应用客户端 IDAZURE_CLIENT_SECRET应用 Secret。Client CertificateAAD 应用 ID X.509 证书AZURE_TENANT_ID、AZURE_CLIENT_IDAZURE_CERTIFICATE_PATH证书路径AZURE_CERTIFICATE_PASSWORD证书密码。Resource Owner PasswordAAD 用户 密码。此 grant 类型不推荐需要交互式登录时请改用 device loginAZURE_TENANT_ID、AZURE_CLIENT_IDAZURE_USERNAME用户名AZURE_PASSWORD密码。Azure Managed Service Identity凭据管理完全委托给平台要求代码运行在 Azure 环境如 VM内所有配置由 Azure 侧完成仅在使用 Azure Resource Manager 时可用。在 ADX 中查询已采集的数据由于fields与tags以 dynamic 类型存储文档给出两类典型输入插件的查询方案。SQL Server 输入场景以 sqlserver 指标为例入湖数据形如nametagstimestampfieldssqlserver_database_io{database_name:azure-sql-db2,file_type:DATA,host:adx-vm, ...}2021-09-09T13:51:20Z{current_size_mb:16,read_bytes:2965504,reads:47, ...}sqlserver_waitstats{wait_category:Worker Thread,wait_type:THREADPOOL, ...}2021-09-09T13:51:20Z{max_wait_time_ms:15,wait_time_ms:4469, ...}方式一直接查询 JSON 属性——ADX 支持不解析直接查询 dynamic 列中的 JSON 属性Tablename | where name sqlserver_azure_db_resource_stats and todouble(fields.avg_cpu_percent) 7Tablename | distinct tostring(tags.database_name)大数据量下此方式有性能影响生产环境建议采用方式二。方式二Update Policy推荐——用更新策略把 dynamic 列转成目标表列// Function to transform data .create-or-alter function Transform_TargetTableName() { SourceTableName | mv-apply fields on (extend key tostring(bag_keys(fields)[0])) | project fieldnamekey, valuetodouble(fields[key]), name, tags, timestamp } // Create destination table with above querys results schema (if it doesnt exist already) .set-or-append TargetTableName | Transform_TargetTableName() | limit 0 // Apply update policy on destination table .alter table TargetTableName policy update [{IsEnabled: true, Source: SourceTableName, Query: Transform_TargetTableName(), IsTransactional: true, PropagateIngestionProperties: false}]Syslog 输入场景syslog 数据样例nametagstimestampfieldssyslog{appname:azsecmond,facility:user,host:adx-linux-vm,severity:info}2021-09-20T14:36:44Z{facility_code:1,message: 2021-09-20 14:36:44.890110 Failed to connect to mdsd: ...,procid:2184, ...}展平 dynamic 列有两种手段都可放进上面的Transform_TargetTableName()更新策略函数中extend操作符推荐——比bag_unpack更快且更稳健schema 变化时不会破坏查询或看板Tablename | extend facility_codetoint(fields.facility_code), messagetostring(fields.message), procid tolong(fields.procid), severity_codetoint(fields.severity_code), SysLogTimestampunixtime_nanoseconds_todatetime(tolong(fields.timestamp)), version todouble(fields.version), appname tostring(tags.appname), facility tostring(tags.facility), host tostring(tags.host), hostnametostring(tags.hostname), severitytostring(tags.severity) | project-away fields, tagsbag_unpack插件——自动展开 dynamic 列但源 schema 变化时动态扩展列可能引发问题Tablename | evaluate bag_unpack(tags, columnsConflictreplace_source) | evaluate bag_unpack(fields, columnsConflictreplace_source)小结azure_data_explorer插件以极少的配置项实现了“开箱即用”的 ADX 写入链路默认TablePerMetric分组 自动建表建 mapping queued 摄取配合 Default Azure Credential 认证即可完成接入需要更细的权限管控或更高吞吐时可通过create_tablesfalse降级权限、通过ingestion_typemanaged启用流式摄取。所有关键行为参数校验、KQL 命令生成、分组写入、假 ingestor 推送验证均有 测试用例 支撑可作为二次开发或行为排障的可靠参照。【免费下载链接】telegrafAgent for collecting, processing, aggregating, and writing metrics, logs, and other arbitrary data.项目地址: https://gitcode.com/GitHub_Trending/te/telegraf创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考