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

Azure Data Explorer では、Telegraf からのデータ インジェストがサポートされています。 Telegraf は、ログ、メトリック、IoT データを含むテレメトリ データの収集、処理、書き込みを行うための、オープンソース、軽量、最小限のメモリ フットプリント エージェントです。 Telegraf では、数百の入力および出力プラグインがサポートされています。 これは、オープン ソース コミュニティで広く使用され、適切にサポートされています。 Azure Data Explorer 出力プラグインは、Telegraf からのコネクタとして機能し、多くの種類の入力プラグインから Azure Data Explorer へのデータの取り込みをサポートします。

前提条件

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

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

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

• アプリ キーまたは証明書を含む Microsoft Entra アプリケーション。

  • Microsoft Entra ID でアプリを作成して登録する方法については、「アプリケーションを登録する」を参照してください。
  • サービス プリンシパルについては、「Microsoft Entra ID のアプリケーション オブジェクトとサービス プリンシパル オブジェクト」を参照してください。

• Microsoft Entra ユーザー トークン

  • プラグインをユーザーのように認証できます。 この方法は開発目的でのみ使用することをお勧めします。

• Azure マネージド サービス ID (MSI) トークン

  • これは、Azure Virtual Machines などの補助的な Azure 環境で Telegraf を実行する場合に推奨される認証方法です。

どちらの方法を使用する場合でも、指定されたプリンシパルに Azure Data Explorer の "データベース ユーザー" ロールを割り当てる必要があります。 このロールを使用すると、プラグインでデータの取り込みに必要なテーブルを作成できます。 プラグインが 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 マネージド サービス ID: 資格情報の管理をプラットフォームに委任します。 この方法では、Azure (たとえば、VM) でコードを実行する必要があります。 すべての構成が 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

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

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

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

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

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

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

extend 演算子または bag_unpack() プラグインを使用して動的な列をフラット化するには、複数の方法があります。 これらのいずれかを更新ポリシーの Transform_TargetTableName() 関数で使用できます。

• extend 演算子を使用する: 高速で堅牢なため、この方法を使用することをお勧めします。 スキーマが変更された場合でも、クエリやダッシュボードが破損することはありません。

  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')