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


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

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.

  1. Войдите в подписку Azure с помощью Azure CLI. Затем авторизуйтесь в браузере.

    az login
    
  2. Выберите подписку для размещения субъекта. Этот шаг необходим, если у вас несколько подписок.

    az account set --subscription YOUR_SUBSCRIPTION_GUID
    
  3. Создайте субъект-службу. В этом примере принципал службы называется my-service-principal.

    az ad sp create-for-rbac -n "my-service-principal" --role Contributor --scopes /subscriptions/{SubID}
    
  4. Из возвращаемых данных 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, выполните приведенные ниже действия.

  1. Перейдите в среду запроса.

  2. Выберите базу данных, в которой вы хотите создать таблицу.

  3. Выполните следующую .create table команду:

    .create table FluentBitLogs (log:dynamic, tag:string, timestamp:datetime)
    

    Входящие свойства JSON автоматически сопоставляются с правильным столбцом.

Предоставление разрешений субъекту-службе

Предоставьте субъекту-службе из создание субъекта-службы 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

Чтобы просмотреть пример файла конфигурации, выберите соответствующую вкладку:

[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

Подтверждение приема данных

  1. После поступления данных в таблицу подтвердите передачу данных, проверив количество строк:

    FluentBitLogs
    | count
    
  2. Чтобы просмотреть пример данных журнала, выполните следующий запрос:

    FluentBitLogs
    | take 100