Fluent Bit 是一种开源代理，用于从各种源收集日志、指标和跟踪。 使用它可以在将事件数据发送到存储之前对事件数据进行筛选、修改和聚合。 本文指导你完成使用 Fluent Bit 将数据发送到 KQL 数据库的过程。
本文介绍如何使用 Fluent Bit 引入数据。
有关数据连接器的完整列表，请参阅数据连接器概述。
先决条件
- Fluent Bit。
- 具有支持 Microsoft Fabric 的容量的工作区。
- 具有引入权限的 KQL 数据库。
- KQL 查询集，稍后将称为查询环境。
- 用作 TargetURI 值的数据引入 URI。 有关详细信息，请参阅复制 URI。
创建 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 属性会自动映射到正确的列中。
向服务主体授予权限
向通过创建 Microsoft Entra 服务主体得到的服务主体授予数据库引入者角色权限以使用数据库。 有关详细信息，请参阅示例。 将占位符 DatabaseName 替换为目标数据库的名称，将 ApplicationID 替换为创建 Microsoft Entra 服务主体时保存的 值AppId。
.add database <DatabaseName> ingestors ('aadapp=<ApplicationID>;<TenantID>')
配置 Fluent Bit 以将日志发送到表
若要配置 Fluent Bit 以将日志发送到 Kusto 中的表，请创建经典模式或 YAML 模式配置文件，其中包含以下输出属性：
| 字段 | 描述 | 必需 | 默认 |
|---|---|---|---|
| 名称 | 管道名称。 | azure_kusto |
|
| 租户ID | 创建 Microsoft Entra 服务主体时使用的租户 ID。 | ✔️ | |
| 客户端ID | 创建 Microsoft Entra 服务主体时使用的应用程序 ID。 | ✔️ | |
| 客户端密钥 | 创建Microsoft Entra 服务主体时使用的客户端密钥值（密码）。 | ✔️ | |
| managed_identity_client_id (托管身份客户端ID) | 用于身份验证的托管标识的客户端 ID。 | ✔️ | |
| 数据摄取终端 | 输入 Ingestion_Endpoint 中所述的值。 | ✔️ | |
| 数据库名称 | 包含日志表的数据库的名称。 | ✔️ | |
| 表名 | 创建目标表时使用的表的名称。 | ✔️ | |
| 摄入映射参考 | 创建目标表时使用的引入映射的名称。 如果未创建引入映射，请从配置文件中删除该属性。 | ||
| 日志键 | 日志内容的键名称。 例如：log。 |
log |
|
| include_tag_key | 如果启用，则会将标记追加到输出中。 | On |
|
| 标签键 | 标签的键名称。 如果 include_tag_key 为 false，则忽略。 |
tag |
|
| include_time_key | 时间戳将追加到输出（如果已启用）。 使用 time_key 属性。 |
On |
|
| 时间键 | 日志记录中时间戳的键名称。 如果 include_time_key 为 false，则忽略。 |
timestamp |
|
| 数据摄取端点连接超时 | 各种 Kusto 终结点的连接超时（以秒为单位）。 | 60 |
|
| compression_enabled | 如果启用，则向 Kusto 发送压缩的 HTTP 有效负载 (gzip)。 | true |
|
| ingestion_resources_refresh_interval | Kusto 终结点的引入资源刷新间隔（以秒为单位）。 | ||
| 辅助角色 | 执行此输出刷新操作的工作线程数量。 | 0 |
|
| 缓冲已启用 | 如果启用，在引入 Kusto 之前，将数据缓冲到磁盘中。 | Off |
|
| 缓冲路径 | 指定将存储缓冲数据的目录的位置（如果 buffering_enabled 为 On）。 |
/tmp/fluent-bit/azure-kusto/ |
|
| upload_timeout | 指定上传超时（如果 buffering_enabled 为 On）。 即使文件大小低于限制，早于此日期的文件也会被处理。 |
30m |
|
| 上传文件大小 | 指定要上传的文件的最大大小（如果 buffering_enabled 为 On）。 |
200MB |
|
| azure_kusto_buffer_key | Azure Kusto 缓冲区的密钥用于在buffering_enabledOn时标识插件实例。 使用缓冲的多个 Azure Kusto 输出是必需的。 |
key |
|
| store_dir_limit_size | 存储缓冲数据的目录的最大大小（如果 buffering_enabled 为 On）。 |
8GB |
|
| 缓冲文件提前删除 | 当 buffering_enabled 为 On 时，是否在成功创建 Blob 后提前删除缓冲文件。 |
Off |
|
| 统一_标签 | 创建单个缓冲区文件（如果 buffering_enabled 为 On. ） |
On |
|
| blob_uri_length | 在引入到 Kusto 之前，请设置生成的 Blob URI 的长度。 | 64 |
|
| 计划程序最大重试次数 | 当buffering_enabled是On时，请设置使用调度程序进行传输的最大重试次数。 |
3 |
|
| 在最大上传错误时删除 | 当buffering_enabled为On时，是否在最大上传错误的情况下删除缓冲文件。 |
Off |
|
| IO_timeout | 为上传配置 HTTP IO 超时。 | 60s |
若要查看示例配置文件，请选择相关选项卡：
[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
确认数据引入
数据到达表后，通过检查行计数来确认数据传输：
FluentBitLogs | count若要查看日志数据示例，请运行以下查询：
FluentBitLogs | take 100