使用 Fluent Bit 将数据引入到 Azure 数据资源管理器

项目
12/28/2024

Fluent Bit 是一种开源代理，用于从各种源收集日志、指标和跟踪。它让你可以在将事件数据发送到存储之前对事件数据进行筛选、修改和聚合。本文指导你完成使用 Fluent Bit 将数据发送到 KQL 数据库的过程。

本文介绍了如何使用 Fluent Bit 引入数据。

有关数据连接器的完整列表，请参阅数据连接器概述。

先决条件

Fluent Bit。
Azure 数据资源管理器群集和数据库。创建群集和数据库。
查询环境。有关详细信息，请参阅查询集成概览。
Kusto 群集 URI，用于表示 Ingestion_endpoint 值，格式为 https://ingest-<cluster>.<region>.kusto.chinacloudapi.cn。有关详细信息，请参阅添加群集连接。

创建 Microsoft Entra 服务主体

Microsoft Entra 服务主体可以通过 Azure 门户或以编程方式进行创建，如下例所示。

此服务主体是连接器用来将数据写入 Kusto 中的表的标识。授予此服务主体访问 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 以 JSON 格式将日志随以下三个属性转发：log (dynamic)、tag (string) 和 timestamp (datetime)。

可以创建一个表，其中包含上述每个属性的列。或者，如果你有结构化日志，则可以创建一个表，其中包含映射到自定义列的日志属性。若要了解详细信息，请选择相关选项卡。

默认架构
自定义架构

若要为来自 Fluent Bit 的传入日志创建表，请执行以下操作：

浏览至查询环境。
选择要在其中创建表的数据库。
运行以下 .create table 命令：
```
.create table FluentBitLogs (log:dynamic, tag:string, timestamp:datetime)
```
传入的 JSON 属性会自动映射到正确的列中。

若要为来自 Fluent Bit 的传入结构化日志创建表，请执行以下操作：

浏览至查询环境。
选择要在其中创建表的数据库。
运行 .create table 命令。例如，如果日志包含名为 myString、myInteger 和 myDynamic 的三个字段，则可以创建具有以下架构的表：
```
.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 以将日志发送到表

若要配置 Fluent Bit 以将日志发送到 Kusto 中的表格，请创建经典模式或 YAML 模式配置文件，其中包含以下输出属性：

字段	说明	需要	默认
名称	管道名称。		`azure_kusto`
tenant_id	创建 Microsoft Entra 服务主体中的租户 ID。	✔️
client_id	创建 Microsoft Entra 服务主体中的应用程序 ID。	✔️
client_secret	创建 Microsoft Entra 服务主体中的客户端密钥值（密码）。	✔️
ingestion_endpoint	输入 Ingestion_Endpoint 所述的值。	✔️
database_name	包含日志表的数据库的名称。	✔️
table_name	创建目标表中的表的名称。	✔️
ingestion_mapping_reference	创建目标表中的引入映射的名称。如果未创建引入映射，请从配置文件中删除该属性。
log_key	日志内容的密钥名称。例如：`log`。		`log`
tag_key	标记的密钥名称。如果 `include_tag_key` 为 false，则忽略。		`tag`
include_time_key	时间戳将追加到输出（如果已启用）。使用 `time_key` 属性。		`true`
time_key	日志记录中时间戳的密钥名称。如果 `include_time_key` 为 false，则忽略。		`timestamp`
ingestion_endpoint_connect_timeout	各种 Kusto 终结点的连接超时（以秒为单位）。		`60s`
compression_enabled	如果已启用，则向 Kusto 发送压缩的 HTTP 有效负载 (gzip)。		`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
```

通过