Fluent Bit を使用して Azure Data Explorer にデータを取り込む

[アーティクル]
01/23/2025

Fluent Bit は、さまざまなソースからログ、メトリック、トレースを収集するオープンソースエージェントです。これにより、イベントデータをストレージに送信する前に、フィルター処理、変更、集計することができます。この記事では、Fluent Bit を使用して KQL データベースにデータを送信するプロセスについて説明します。

この記事では、Fluent Bit を使用してデータを取り込む方法について説明します。

データコネクタの完全な一覧については、データコネクタの概要に関する記事を参照してください。

前提条件

Fluent Bit。
Azure Data Explorer クラスターとデータベース。クラスターとデータベースを作成します。
クエリ環境。詳細については、「クエリ統合の概要」を参照してください。
https://ingest-<cluster>.<region>.kusto.windows.net のフォーマットで Ingestion_endpoint 値に Kusto クラスター URI を指定します。詳細については、「クラスター接続を追加する」を参照してください。

Microsoft Entra サービスプリンシパルを作成する

Microsoft Entra サービスプリンシパルは、次の例に示すように、Azure portal またはプログラムを使用して作成できます。

このサービスプリンシパルは、Kusto のテーブルにデータを書き込むコネクタによって使用される ID です。このサービスプリンシパルに対して、Kusto リソースにアクセスするためのアクセス許可を付与します。

Azure CLI 経由で Azure サブスクリプションにサインインします。次に、ブラウザーで認証します。
```
az login
```
プリンシパルをホストするサブスクリプションを選択します。この手順は、複数のサブスクリプションがある場合に必要です。
```
az account set --subscription YOUR_SUBSCRIPTION_GUID
```
サービスプリンシパルを作成します。この例では、サービスプリンシパルを my-service-principal と呼びます。
```
az ad sp create-for-rbac -n "my-service-principal" --role Contributor --scopes /subscriptions/{SubID}
```

返された JSON データから、appId、password、および tenant を後で使用のためにコピーします。

{
  "appId": "00001111-aaaa-2222-bbbb-3333cccc4444",
  "displayName": "my-service-principal",
  "name": "my-service-principal",
  "password": "00001111-aaaa-2222-bbbb-3333cccc4444",
  "tenant": "00001111-aaaa-2222-bbbb-3333cccc4444"
}

Microsoft Entra アプリケーションとサービスプリンシパルが作成されました。

ターゲットテーブルを作成する

Fluent Bit は、log (dynamic)、tag (string)、timestamp (datetime) の 3 つのプロパティを使用して、JSON 形式でログを転送します。

これらの各プロパティの列を含むテーブルを作成できます。または、構造化されたログがある場合は、カスタム列にマップされたログプロパティを含むテーブルを作成できます。詳細については、関連するタブを選択してください。

既定のスキーマ
カスタムスキーマ

Fluent Bit から受信したログのテーブルを作成するには:

クエリ環境を参照します。
テーブルを作成するデータベースを選択します。
次の .create table コマンドを実行します。
```
.create table FluentBitLogs (log:dynamic, tag:string, timestamp:datetime)
```
受信した JSON プロパティは、正しい列に自動的にマップされます。

Fluent Bit から受信した構造化ログのテーブルを作成するには:

クエリ環境を参照します。
テーブルを作成するデータベースを選択します。
.create table コマンドを実行します。たとえば、myString、myInteger、myDynamic という名前の 3 つのフィールドがログに含まれている場合は、次のスキーマを使用してテーブルを作成できます。
```
.create table FluentBitLogs (myString:string, myInteger:int, myDynamic: dynamic, timestamp:datetime)
```

JSON マッピングを作成し、ログプロパティを適切な列にマップします。次のコマンドは、前の手順の例に基づいてマッピングを作成します。

.create-or-alter table FluentBitLogs ingestion json mapping "LogMapping" 
    ```[
    {"column" : "myString", "datatype" : "string", "Properties":{"Path":"$.log.myString"}},
    {"column" : "myInteger", "datatype" : "int", "Properties":{"Path":"$.log.myInteger"}}, 
    {"column" : "myDynamic", "datatype" : "dynamic", "Properties":{"Path":"$.log.myInteger"}}, 
    {"column" : "timestamp", "datatype" : "datetime", "Properties":{"Path":"$.timestamp"}} 
    ]```

サービスプリンシパルにアクセス許可を付与する

「Microsoft Entra サービスプリンシパルの作成」からサービスプリンシパルにデータベースを操作するためのデータベースインジェスターロールアクセス許可を付与します。詳細については、「例」を参照してください。ターゲットデータベースの名前で DatabaseName プレースホルダーを置き換え、Microsoft Entra サービスプリンシパル作成時に保存した AppId 値で ApplicationID を置き換えます。

.add database <DatabaseName> ingestors ('aadapp=<ApplicationID>;<TenantID>')

テーブルにログを送信するように Fluent Bit を構成する

Kusto でテーブルにログを送信するように Fluent Bit を構成するには、次の出力プロパティを使用して、クラシックモードまたは YAML モード構成ファイルを作成します。

フィールド	内容	必須	既定値
Name	パイプライン名。		`azure_kusto`
テナントID	「Microsoft Entra サービスプリンシパルを作成する」のテナント ID。	✔️
クライアントID	「Microsoft Entra サービスプリンシパルを作成する」のアプリケーション ID。	✔️
client_secret	「Microsoft Entra サービスプリンシパルを作成する」のクライアントシークレットのキー値 (パスワード)。	✔️
ingestion_endpoint	Ingestion_Endpointの説明に従って値を入力します。	✔️
データベースネーム	ログテーブルが含まれるデータベースの名前。	✔️
table_name	「ターゲットテーブルを作成する」のテーブルの名前。	✔️
ingestion_mapping_reference	「ターゲットテーブルを作成する」のインジェストマッピングの名前。インジェストマッピングを作成しなかった場合は、構成ファイルからプロパティを削除します。
log_key	ログコンテンツのキー名。たとえば、`log`します。		`log`
tag_key	タグのキー名。 `include_tag_key` が false の場合は無視されます。		`tag`
include_time_key	有効になっている場合は、タイムスタンプが出力に追加されます。 `time_key` プロパティを使用します。		`true`
time_key	ログレコード内のタイムスタンプのキー名。 false の場合 `include_time_key` 無視されます。		`timestamp`
ingestion_endpoint_connect_timeout	さまざまな Kusto エンドポイントの接続タイムアウト (秒単位)。		`60s`
圧縮_有効	有効になっている場合は、圧縮 HTTP ペイロード (gzip) を Kusto に送信します。		`true`
ingestion_resources_refresh_interval	インジェストリソースの Kusto エンドポイントの更新間隔 (秒単位)。		`3600`
労働者	この出力のフラッシュ操作を実行するためのワーカーの数。		`0`

構成ファイルの例を表示するには、関連するタブを選択します。

クラシックモード
YAML モード

[SERVICE]
    Daemon Off
    Flush 1
    Log_Level trace
    HTTP_Server On
    HTTP_Listen 0.0.0.0
    HTTP_Port 2020
    Health_Check On

[INPUT]
    Name tail
    Path /var/log/containers/*.log
    Tag kube.*
    Mem_Buf_Limit 1MB
    Skip_Long_Lines On
    Refresh_Interval 10

[OUTPUT]
    match *
    name azure_kusto
    tenant_id <TenantId>
    client_id <ClientId>
    client_secret <AppSecret>
    ingestion_endpoint <IngestionEndpoint>
    database_name <DatabaseName>
    table_name <TableName>
    ingestion_mapping_reference <MappingName>
    ingestion_endpoint_connect_timeout <IngestionEndpointConnectTimeout>
    compression_enabled <CompressionEnabled>
    ingestion_resources_refresh_interval <IngestionResourcesRefreshInterval>

config:
  service: |
    [SERVICE]
        Daemon Off
        Flush 1
        Log_Level trace
        HTTP_Server On
        HTTP_Listen 0.0.0.0
        HTTP_Port 2020
        Health_Check On
        
  inputs: |
    [INPUT]
        Name tail
        Path /var/log/containers/*.log
        multiline.parser docker, cri
        Tag kube.*
        Mem_Buf_Limit 1MB
        Skip_Long_Lines On
        Refresh_Interval 10

  filters: |
    [FILTER]
        Name kubernetes
        Match kube.*
        Merge_Log On
        Merge_Log_key log_processed
        K8S-Logging.Parser On
        K8S-Logging.Exclude Off


  outputs: |
    [OUTPUT]
        match *
        name azure_kusto
        tenant_id <TenantId>
        client_id <ClientId>
        client_secret <AppSecret>
        ingestion_endpoint <IngestionEndpoint>
        database_name <DatabaseName>
        table_name <TableName>
        ingestion_mapping_reference <MappingName>
        ingestion_endpoint_connect_timeout <IngestionEndpointConnectTimeout>
        compression_enabled <CompressionEnabled>
        ingestion_resources_refresh_interval <IngestionResourcesRefreshInterval>

データインジェストを確認する

テーブルにデータが到着したら、行数を確認して、データの転送を確認します。
```
FluentBitLogs
| count
```
ログデータのサンプルを表示するには、次のクエリを実行します。
```
FluentBitLogs
| take 100
```

次の方法で共有

Fluent Bit を使用して Azure Data Explorer にデータを取り込む

前提条件

Microsoft Entra サービスプリンシパルを作成する

ターゲットテーブルを作成する

サービスプリンシパルにアクセス許可を付与する

テーブルにログを送信するように Fluent Bit を構成する

データインジェストを確認する

フィードバック

その他のリソース

次の方法で共有

Fluent Bit を使用して Azure Data Explorer にデータを取り込む

前提条件

Microsoft Entra サービス プリンシパルを作成する

ターゲット テーブルを作成する

サービス プリンシパルにアクセス許可を付与する

テーブルにログを送信するように Fluent Bit を構成する

データ インジェストを確認する

関連するコンテンツ

フィードバック

その他のリソース

Microsoft Entra サービスプリンシパルを作成する

ターゲットテーブルを作成する

サービスプリンシパルにアクセス許可を付与する

データインジェストを確認する