Прием данных с помощью Fluent Bit в Azure Data Explorer

2025-05-27

Fluent Bit — это агент с открытым исходным кодом, который собирает журналы, метрики и трассировки из различных источников. Он позволяет фильтровать, изменять и агрегировать данные событий перед отправкой в хранилище. В этой статье описано, как использовать Fluent Bit для отправки данных в базу данных KQL.

В этой статье описывается, как загружать данные с помощью Fluent Bit.

Полный список соединителей данных см. в обзоре соединителей данных.

Необходимые компоненты

Fluent Bit.
Кластер и база данных Azure Data Explorer. Создайте кластер и базу данных.
Среда запроса. Для получения дополнительной информации см. обзор интеграции запросов .
URI кластера Kusto для параметра Ingestion_endpoint в формате https://ingest-<cluster>.<region>.kusto.windows.net. Дополнительные сведения см. в разделе Подключение к кластеру.

Создание представителя службы Microsoft Entra

Субъект-служба Microsoft Entra можно создать с помощью портала Azure или программно, как показано в следующем примере.

Эта учетная запись службы — это удостоверение, используемое соединителем для записи данных в вашу таблицу в Kusto. Вы предоставляете доступ этому служебному субъекту для доступа к ресурсам Kusto.

Войдите в подписку Azure с помощью Azure CLI. Затем авторизуйтесь в браузере.
```
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 скопируйте appIdpasswordданные и 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 (динамическаяtag) и (timestamp).

Вы можете создать таблицу со столбцами для каждого из этих свойств. Кроме того, если у вас есть структурированные журналы, можно создать таблицу со свойствами журнала, сопоставленными с настраиваемыми столбцами. Чтобы узнать больше, выберите соответствующую вкладку.

Схема по умолчанию
Настраиваемая схема

Чтобы создать таблицу для входящих журналов из 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 разрешениями роли ingestor для работы с базой данных. Дополнительные сведения см. в примерах . Замените заполнитель DatabaseName именем целевой базы данных, а ApplicationID значением AppId, сохраненным при создании учетной записи службы Microsoft Entra.

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

Настройка Fluent Bit для отправки журналов в таблицу

Чтобы настроить Fluent Bit для отправки журналов в таблицу Kusto, создайте файл конфигурации в классическом режиме или в режиме YAML с указанными ниже выходными свойствами:

Поле	Описание:	Обязательно	По умолчанию
Имя.	Имя конвейера.		`azure_kusto`
идентификатор арендатора	Идентификатор арендатора из Создание субъекта-службы Microsoft Entra.	✔️
идентификатор_клиента	Идентификатор приложения из Создание учетной записи службы Microsoft Entra.	✔️
секрет_клиента	Значение ключа секрета клиента (пароль) из создание субъекта-службы Microsoft Entra.	✔️
managed_identity_client_id	Идентификатор клиента управляемого удостоверения, используемого для проверки подлинности.	✔️
конечная точка приема данных	Введите значение, как указано для Ingestion_Endpoint.	✔️
имя_базы_данных	Имя базы данных, содержащей таблицу журналов.	✔️
имя_таблицы	Имя таблицы из Создание целевой таблицы.	✔️
ссылка_на_схему_импорта	Имя сопоставления приема из Создание целевой таблицы. Если вы не создали сопоставление приема, удалите свойство из файла конфигурации.
log_key	Имя ключа содержимого журнала. Например, `log`.		`log`
include_tag_key	Если этот параметр включен, тег добавляется к выходным данным.		`On`
тег_ключ	Ключевое название тега. Игнорируется, если `include_tag_key` имеет значение false.		`tag`
include_time_key	Метка времени добавляется к выходным данным, если он включен. Использует свойство `time_key`.		`On`
ключ_времени	Имя ключа метки времени в записях журнала. Игнорируется, если `include_time_key` false.		`timestamp`
тайм-аут подключения конечной точки загрузки	Время ожидания подключения различных конечных точек Kusto в секундах.		`60`
сжатие_включено	При включенной функции отправляет сжатые (gzip) HTTP-данные в Kusto.		`true`
интервал_обновления_ресурсов_ввода	Интервал обновления ресурсов точки подключения Kusto в секундах.
Работники	Количество рабочих для выполнения операций очистки для этих выходных данных.		`0`
буферизация_включена	Если включено, сначала буферизирует данные на диск перед их приемом в Kusto.		`Off`
путь_буфера	Указывает расположение каталога, в котором буферные данные будут храниться, если `buffering_enabled` есть `On`.		`/tmp/fluent-bit/azure-kusto/`
время ожидания загрузки	Указывает время ожидания отправки, если `buffering_enabled` имеет значение `On`. Файлы, старше указанного срока, обрабатываются даже если их размер меньше установленного предела.		`30m`
размер_загружаемого_файла	Указывает максимальный размер файла, который нужно отправить, если `buffering_enabled` это `On`.		`200MB`
azure_kusto_buffer_key	Ключ буфера Azure Kusto для идентификации экземпляров подключаемых модулей, когда `buffering_enabled` является `On`. Требуется для нескольких выходов Azure Kusto с управляемой буферизацией.		`key`
лимит_размера_директории_хранилища	Максимальный размер каталога, в котором хранятся буферные данные, если `buffering_enabled` это `On`.		`8GB`
раннее удаление файла буфера	Когда `buffering_enabled` равно `On`, следует ли удалить буферный файл сразу после успешного создания блоба.		`Off`
unify_tag	Создает один файл буфера, когда `buffering_enabled` есть `On`.		`On`
длина blob_uri	Установите длину создаваемого URI BLOB перед приемом в Kusto.		`64`
планировщик_макс_попытки	Если `buffering_enabled` это `On`так, задайте максимальное количество повторных попыток приема с помощью планировщика.		`3`
удалить_при_ошибке_макс_загрузки	Когда `buffering_enabled` является `On`, следует ли удалить файл буфера при максимальном количестве ошибок загрузки.		`Off`
IO_timeout (тайм-аут ввода-вывода)	Настройте тайм-аут ввода-вывода HTTP для загрузки.		`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
```

Поделиться через