Telegraf から Azure Data Explorer にデータを取り込む

Azure Data Explorer (ADX) では、Telegraf からのデータインジェストがサポートされています。 Telegraf は、オープンソースで軽量で最小限のメモリフットプリントエージェントです。 Telegraf は、ログ、メトリック、IoT データなどのテレメトリデータの収集、処理、書き込みに使用されます。

Telegraf では、数百の入力および出力プラグインがサポートされています。これは広く使用されており、オープンソースコミュニティがそれをサポートしています。

Azure Data Explorer ADX 出力プラグインは、Telegraf からのコネクタとして機能し、さまざまな種類の入力プラグインから Azure Data Explorer へのデータの取り込みをサポートします。

Prerequisites

Azure サブスクリプション。無料の Azure アカウントを作成します。
Azure Data Explorer クラスターとデータベース。クラスターとデータベースを作成します。
Telegraf. 仮想マシン (VM) またはコンテナーで Telegraf をホストします。 Telegraf は、監視対象のアプリまたはサービスがデプロイされている場所でローカルにホストすることも、専用の監視コンピューティング/コンテナー上でリモートでホストすることもできます。

サポートされている認証方法

このプラグインでは、次の認証方法がサポートされています。

アプリキーまたは証明書を備えた Microsoft Entra アプリケーション。
- Microsoft Entra ID でアプリを作成して登録する方法については、「アプリケーションを登録する」を参照してください。
- サービスプリンシパルについては、「Microsoft Entra ID のアプリケーションとサービスプリンシパルオブジェクト」を参照してください。
Microsoft Entra ユーザートークン
- プラグインをユーザーのように認証できます。このメソッドは、開発にのみ使用します。
Azure マネージドサービス ID (MSI) トークン
- サポートされている Azure 環境 (Azure Virtual Machines など) で Telegraf を実行している場合に推奨される認証方法。

どちらの方法を使用する場合でも、指定されたプリンシパルに Azure Data Explorer で データベースユーザー ロールを割り当てる必要があります。このロールを使用すると、プラグインでデータの取り込みに必要なテーブルを作成できます。プラグインが create_tables=false で構成されている場合、指定されたプリンシパルには、少なくとも Database Ingestor ロールが必要です。

認証方法を構成する

このプラグインでは、環境変数の特定の構成をチェックして、使用する認証方法を決定します。構成は指定した順序で評価され、検出された最初の構成が使用されます。有効な構成が検出されない場合、プラグインは認証に失敗します。

プラグインの認証を構成するには、選択した認証方法に適した環境変数を設定します。

クライアント資格情報 (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 マネージドサービス ID: 資格情報の管理をプラットフォームに委任します。 VM 上など、Azure でコードを実行します。 Azure では、すべての構成が処理されます。詳細については、Azure マネージドサービス ID に関する記事をご覧ください。この方法が使用できるのは、Azure Resource Manager 使用する場合に限られます。

Telegraf の構成

Telergraf は、構成駆動型のエージェントです。作業を開始するには、Telegraf をインストールして、必要な入力および出力プラグインを構成する必要があります。構成ファイルの既定の場所は次のとおりです。

Windows の場合: C:\Program Files\Telegraf\telegraf.conf
Linux の場合: etc/telegraf/telegraf.conf

Azure Data Explorer 出力プラグインを有効にするには、自動生成された構成ファイルで次のセクションのコメントを解除する必要があります。

[[outputs.azure_data_explorer]]
  ## The URI property of the Azure Data Explorer resource on Azure
  ## ex: 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, 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

サポートされるインジェストの種類

プラグインは、マネージド (ストリーミング) およびキュー (バッチ処理) インジェストをサポートします。既定のインジェストの種類は キューに登録されます。

Important

マネージドインジェストを使用するには、クラスターでストリーミングインジェストを有効にする必要があります。

プラグインのインジェストの種類を構成するには、自動生成された構成ファイルを次のように変更します。

  ##  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"

取り込まれたデータに対してクエリを実行する

Azure Data Explorer 出力プラグインと共に SQL および Syslog 入力プラグインを使用して収集されたデータの例を次に示します。 Azure Data Explorer でデータ変換とクエリを使用する方法の例を、入力方法ごとに示します。

SQL 入力プラグイン

次の表に、SQL 入力プラグインによって収集されたメトリックデータの例を示します。

name	tags	timestamp	fields
sqlserver_database_io	`{"database_name":"azure-sql-db2","file_type":"DATA","host":"adx-vm","logical_filename":"tempdev","measurement_db_type":"AzureSQLDB","physical_filename":"tempdb.mdf","replica_updateability":"READ_WRITE","sql_instance":"adx-sql-server"}`	2021-09-09T13:51:20Z	`{"current_size_mb":16,"database_id":2,"file_id":1,"read_bytes":2965504,"read_latency_ms":68,"reads":47,"rg_read_stall_ms":42,"rg_write_stall_ms":0,"space_used_mb":0,"write_bytes":1220608,"write_latency_ms":103,"writes":149}`
sqlserver_waitstats	`{"database_name":"azure-sql-db2","host":"adx-vm","measurement_db_type":"AzureSQLDB","replica_updateability":"READ_WRITE","sql_instance":"adx-sql-server","wait_category":"Worker Thread","wait_type":"THREADPOOL"}`	2021-09-09T13:51:20Z	`{"max_wait_time_ms":15,"resource_wait_ms":4469,"signal_wait_time_ms":0,"wait_time_ms":4469,"waiting_tasks_count":1464}`

収集されたメトリックオブジェクトは複合型であるため、 フィールド とタグの列は動的データ型として格納されます。このデータに対してクエリを実行するには、さまざまな方法があります。次に例を示します。

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)
```
Note

この方法は、大量のデータのパフォーマンスに影響する可能性があります。このような場合は、更新ポリシーのアプローチを使用します。

更新ポリシーを使用する: 更新ポリシーを使用して動的なデータ型の列を変換します。大量のデータに対してクエリを実行する場合は、この方法をお勧めします。

// 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 入力プラグインによって収集されたメトリックデータの例を示します。

name	tags	timestamp	fields
syslog	`{"appname":"azsecmond","facility":"user","host":"adx-linux-vm","hostname":"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: dial unix /var/run/mdsd/default_djson.socket: connect: no such file or directory","procid":"2184","severity_code":6,"timestamp":"1632148604890477000","version":1}`
syslog	`{"appname":"CRON","facility":"authpriv","host":"adx-linux-vm","hostname":"adx-linux-vm","severity":"info"}`	2021-09-20T14:37:01Z	`{"facility_code":10,"message":" pam_unix(cron:session): session opened for user root by (uid=0)","procid":"26446","severity_code":6,"timestamp":"1632148621120781000","version":1}`

拡張演算子または 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')
```

フィードバック

このページはお役に立ちましたか?

Last updated on 2025-08-24