将数据从 Telegraf 引入到 Azure 数据资源管理器

Azure 数据资源管理器（ADX）支持从 Telegraf引入数据。 Telegraf 是一种开源的轻型内存脚印代理。 Telegraf 用于收集、处理和写入遥测数据，包括日志、指标和 IoT 数据。

Telegraf 支持数百个输入和输出插件。 它广泛使用，开源社区支持它。

Azure 数据资源管理器 ADX 输出插件 充当 Telegraf 的连接器，并支持从许多类型的 输入插件 引入 Azure 数据资源管理器中的数据。

先决条件

• Azure 订阅。 创建 Azure 帐户。
• Azure 数据资源管理器群集和数据库。 创建群集和数据库。
• Telegraf。 在虚拟机 (VM) 或容器中托管 Telegraf。 Telegraf 可以在部署受监视应用或服务的本地托管，也可以在专用监视计算/容器上远程托管。

支持的身份验证方法

插件支持以下身份验证方法：

• 具有应用密钥或证书的 Microsoft Entra 应用程序。

  • 有关如何在 Microsoft Entra ID 中创建和注册应用的信息，请参阅注册应用程序。
  • 有关服务主体的信息，请参阅 Microsoft Entra ID 中的应用程序和服务主体对象。

• Microsoft Entra 用户令牌

  • 允许插件像用户一样进行身份验证。 仅使用此方法进行开发。

• Azure 托管服务标识 (MSI) 令牌

  • 如果在支持 Azure 环境中（例如 Azure 虚拟机）中运行 Telegraf，则首选身份验证方法。

无论使用哪种方法，都必须在 Azure 数据资源管理器中为指定主体分配“数据库用户”角色。 此角色允许插件创建引入数据所需的表。 如果插件已配置 create_tables=false，则指定的主体必须至少具有“数据库引入者”角色。

配置身份验证方法

插件检查环境变量的特定配置，以确定要使用的身份验证方法。 按指定顺序评估配置，并使用检测到的第一个配置。 如果未检测到有效的配置，插件将无法进行身份验证。

若要为插件配置身份验证，请为所选的身份验证方法设置适当的环境变量：

• 客户端凭据（Microsoft Entra 应用程序令牌）：Microsoft Entra 应用程序 ID 和机密。

  • AZURE_TENANT_ID：用于身份验证的 Microsoft Entra 租户 ID。
  • AZURE_CLIENT_ID：租户中应用注册的客户端 ID。
  • AZURE_CLIENT_SECRET：专为应用注册生成的客户端密码。

• 客户端证书（Microsoft Entra 应用程序令牌）：Microsoft Entra 应用程序 ID 和 X.509 证书。

  • AZURE_TENANT_ID：用于身份验证的 Microsoft Entra 租户 ID。
  • AZURE_CERTIFICATE_PATH：PEM 或 PFX 格式的证书和私钥对路径，可对应用注册进行身份验证。
  • AZURE_CERTIFICATE_PASSWORD：为证书设置的密码。

• 资源所有者密码（Microsoft Entra 用户令牌）：Microsoft Entra 用户和密码。 我们不建议使用此授权类型。 如果需要交互式登录，请使用设备登录。

  • AZURE_TENANT_ID：用于身份验证的 Microsoft Entra 租户 ID。
  • AZURE_CLIENT_ID：租户中应用注册的客户端 ID。
  • AZURE_USERNAME：Microsoft Entra 用户帐户的用户名（也称为 upn）。
  • AZURE_PASSWORD：Microsoft Entra 用户帐户的密码。 注意：此功能不支持启用了多重身份验证（MFA）的帐户。

• Azure 托管服务标识：将凭据管理委托给平台。 在 Azure 中运行代码，例如在 VM 上。 Azure 处理所有配置。 此方法仅在使用 Azure 资源管理器时才适用。

配置 Telegraf

Telergraf 是一个配置驱动的代理。 若要开始，必须安装 Telegraf 并配置所需的输入和输出插件。 配置文件的默认位置如下所示：

• 在 Windows 中：C:\Program Files\Telegraf\telegraf.conf
• 在 Linux 中：etc/telegraf/telegraf.conf

若要启用 Azure 数据资源管理器输出插件，必须在自动生成的配置文件中取消注释以下部分：

[[outputs.azure_data_explorer]]
  ## The URI property of the Azure Data Explorer resource on Azure
  ## ex: https://myadxresource.chinaeast2.kusto.chinacloudapi.cn
  # endpoint_url = ""

  ## The Azure Data Explorer database that the metrics will be ingested into.
  ## The plugin will NOT generate this database automatically, it's expected that this database already exists before ingestion.
  ## ex: "exampledatabase"
  # database = ""

  ## Timeout for Azure Data Explorer operations, default value is 20 seconds
  # timeout = "20s"

  ## Type of metrics grouping used when ingesting to Azure Data Explorer
  ## Default value is "TablePerMetric" which means there will be one table for each metric
  # 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 least possible access permissions i.e. 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"

查询引入数据

下面举例说明了使用 SQL 和 syslog 输入插件以及 Azure 数据资源管理器输出插件收集的数据。 对于每个输入方法，有一个示例演示了如何在 Azure 数据资源管理器中使用数据转换和查询。

SQL 输入插件

下表展示了 SQL 输入插件收集的示例指标数据：

由于收集的指标对象是一种复杂类型，因此“字段”和“标记”列存储为动态数据类型。 可通过多种方式查询此数据，例如：

• 直接查询 JSON 属性：可以采用原始格式查询 JSON 数据，而无需对其进行分析。

  示例 1

  Tablename
  | where name == "sqlserver_azure_db_resource_stats" and todouble(fields.avg_cpu_percent) > 7
  

  示例 2

  Tablename
  | distinct tostring(tags.database_name)
  

  注意

  此方法可能会影响大量数据的性能。 在这些情况下，请使用更新策略方法。

• 使用更新策略：使用更新策略转换动态数据类型列。 建议使用此方法查询大量数据。

  // Function to transform data
  .create-or-alter function Transform_TargetTableName() {
    SourceTableName
    | mv-apply fields on (extend key = tostring(bag_keys(fields)[0]))
    | project fieldname=key, value=todouble(fields[key]), name, tags, timestamp
  }
  
  // Create destination table with above query's results schema (if it doesn't exist already)
  .set-or-append TargetTableName <| Transform_TargetTableName() | take 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 输入插件收集的示例指标数据：

有多种方法可以使用 extended 运算符或 bag_unpack() 插件来展平动态列。 在更新策略 Transform_TargetTableName() 函数中可以使用其中任一方式。

• 使用扩展运算符：使用此方法，因为它速度更快且可靠。 即使架构发生更改，也不会中断查询或仪表板。

  Tablename
  
  | extend facility_code=toint(fields.facility_code), message=tostring(fields.message), procid= tolong(fields.procid), severity_code=toint(fields.severity_code),
  SysLogTimestamp=unixtime_nanoseconds_todatetime(tolong(fields.timestamp)), version= todouble(fields.version),
  appname= tostring(tags.appname), facility= tostring(tags.facility),host= tostring(tags.host), hostname=tostring(tags.hostname), severity=tostring(tags.severity)
  | project-away fields, tags
  

• 使用 bag_unpack() 插件：此方法会将动态类型列解压缩。 更改源架构可能会在动态扩展列时导致问题。

  Tablename
  | evaluate bag_unpack(tags, columnsConflict='replace_source')
  | evaluate bag_unpack(fields, columnsConflict='replace_source')