Fluent Bit を使用してデータを取得する

2025-05-27

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

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

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

前提条件

Fluent Bit。
Microsoft Fabric 対応の容量を持つワークスペース。
インジェストアクセス許可を持つ KQL データベース。
KQL クエリセット。これは後でクエリ環境と呼ばれます。
TargetURI 値として使用するデータベースインジェスト URI。詳細については、「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 プレースホルダーをターゲットデータベースの名前に置き換え、ApplicationID プレースホルダーを、Microsoft Entra サービスプリンシパルの作成時に保存した AppId 値に置き換えます。

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

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

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

フィールド	内容	必須	既定値
名前	パイプライン名。		`azure_kusto`
テナントID	「Microsoft Entra サービスプリンシパルを作成する」のテナント ID。	✔️
クライアントID	「Microsoft Entra サービスプリンシパルを作成する」のアプリケーション ID。	✔️
クライアントシークレット	「Microsoft Entra サービスプリンシパルを作成する」のクライアントシークレットのキー値 (パスワード)。	✔️
managed_identity_client_id	認証に使用するマネージド ID のクライアント ID。	✔️
データ導入エンドポイント	Ingestion_Endpointの説明に従って値を入力します。	✔️
データベース名	ログテーブルが含まれるデータベースの名前。	✔️
テーブル名	「ターゲットテーブルを作成する」のテーブルの名前。	✔️
データ取り込みマッピング参照	「ターゲットテーブルを作成する」のインジェストマッピングの名前。インジェストマッピングを作成しなかった場合は、構成ファイルからプロパティを削除します。
ログキー	ログコンテンツのキー名。たとえば、`log`します。		`log`
タグキーを含める	有効にすると、タグが出力に追加されます。		`On`
tag_key	タグのキー名。 `include_tag_key` が false の場合は無視されます。		`tag`
include_time_key	有効になっている場合は、タイムスタンプが出力に追加されます。 `time_key` プロパティを使用します。		`On`
時間キー	ログレコード内のタイムスタンプのキー名。 false の場合 `include_time_key` 無視されます。		`timestamp`
インジェスチョンエンドポイント接続タイムアウト	さまざまな Kusto エンドポイントの接続タイムアウト (秒単位)。		`60`
圧縮_有効	有効になっている場合は、圧縮 HTTP ペイロード (gzip) を Kusto に送信します。		`true`
ingestion_resources_refresh_interval	インジェストリソースの Kusto エンドポイントの更新間隔 (秒単位)。
労働者	この出力のフラッシュ操作を実行するためのワーカーの数。		`0`
バッファリング有効	有効になっている場合は、Kusto に取り込む前にデータをディスクにバッファーします。		`Off`
バッファーパス	`buffering_enabled`が`On`されている場合にバッファー内のデータを格納するディレクトリの場所を指定します。		`/tmp/fluent-bit/azure-kusto/`
アップロードタイムアウト	`buffering_enabled`が`On`場合のアップロードのタイムアウトを指定します。これより古いファイルは、サイズ制限を下回っても取り込まれます。		`30m`
アップロードファイルサイズ	`buffering_enabled`が`On`場合にアップロードするファイルの最大サイズを指定します。		`200MB`
azure_kusto_buffer_key	`buffering_enabled`が`On`されたときにプラグインインスタンスを識別するための Azure Kusto バッファーキー。バッファリングを使用する複数の Azure Kusto 出力に必要です。		`key`
ストアディレクトリ制限サイズ	`buffering_enabled`が`On`されている場合にバッファーされたデータが格納されるディレクトリの最大サイズ。		`8GB`
バッファーファイル削除早期	`buffering_enabled`が`On`されている場合、BLOB の作成が成功した後、バッファーに格納されたファイルを早期に削除するかどうかを指定します。		`Off`
unify_tag	`buffering_enabled`が`On`されたときに、1 つのバッファーファイルを作成します。		`On`
blobのURIの長さ	Kusto に取り込む前に、生成された BLOB URI の長さを設定します。		`64`
scheduler_max_retries	`buffering_enabled`が`On`場合は、スケジューラを使用してインジェストの最大再試行回数を設定します。		`3`
アップロードエラーの上限で削除する	`buffering_enabled`が`On`場合、最大アップロードエラー時にバッファーファイルを削除するかどうか。		`Off`
IOタイムアウト	アップロードの HTTP IO タイムアウトを構成します。		`60s`

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

クラシックモード
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]
    [OUTPUT]
    Match *
    Name azure_kusto
    Tenant_Id <app_tenant_id>
    Client_Id <app_client_id>
    Client_Secret <app_secret>
    Ingestion_Endpoint https://ingest-<cluster>.<region>.kusto.windows.net
    Database_Name <database_name>
    Table_Name <table_name>
    Ingestion_Mapping_Reference <mapping_name>
    ingestion_endpoint_connect_timeout <ingestion_endpoint_connect_timeout>
    compression_enabled <compression_enabled>
    ingestion_resources_refresh_interval <ingestion_resources_refresh_interval>
    buffering_enabled On
    upload_timeout 2m
    upload_file_size 125M
    azure_kusto_buffer_key kusto1
    buffer_file_delete_early Off
    unify_tag On
    buffer_dir /var/log/
    store_dir_limit_size 16GB
    blob_uri_length 128
    scheduler_max_retries 3
    delete_on_max_upload_error Off
    io_timeout 60s

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]
        [OUTPUT]
    Match *
    Name azure_kusto
    Tenant_Id <app_tenant_id>
    Client_Id <app_client_id>
    Client_Secret <app_secret>
    Ingestion_Endpoint https://ingest-<cluster>.<region>.kusto.windows.net
    Database_Name <database_name>
    Table_Name <table_name>
    Ingestion_Mapping_Reference <mapping_name>
    ingestion_endpoint_connect_timeout <ingestion_endpoint_connect_timeout>
    compression_enabled <compression_enabled>
    ingestion_resources_refresh_interval <ingestion_resources_refresh_interval>
    buffering_enabled On
    upload_timeout 2m
    upload_file_size 125M
    azure_kusto_buffer_key kusto1
    buffer_file_delete_early Off
    unify_tag On
    buffer_dir /var/log/
    store_dir_limit_size 16GB
    blob_uri_length 128
    scheduler_max_retries 3
    delete_on_max_upload_error Off
    io_timeout 60s

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

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

KQL クエリセット内のデータのクエリを実行する

次の方法で共有

Fluent Bit を使用してデータを取得する

前提条件

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

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

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

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

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

関連するコンテンツ

フィードバック

その他のリソース

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

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

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

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