Ingerir dados com Fluent Bit no Azure Data Explorer

Fluent Bit é um agente de código aberto que coleta logs, métricas e rastreamentos de várias fontes. Ele permite filtrar, modificar e agregar dados de eventos antes de enviá-los para o armazenamento. Este artigo orienta você através do processo de uso do Fluent Bit para enviar dados para seu banco de dados KQL.

Este artigo mostra como ingerir dados com o Fluent Bit.

Para obter uma lista completa de conectores de dados, consulte Visão geral dos conectores de dados.

Pré-requisitos

Bit fluente.
Um cluster e um banco de dados do Azure Data Explorer. Criar um cluster e um banco de dados.
Um ambiente de consulta. Para obter mais informações, consulte Visão geral das integrações de consulta.
O seu URI de cluster Kusto para o valor Ingestion_endpoint no formato https://ingest-<cluster>.<region>.kusto.windows.net. Para obter mais informações, consulte Adicionar uma conexão de cluster.

Criar um principal de serviço Microsoft Entra

A entidade de serviço do Microsoft Entra pode ser criada por meio do portal do Azure ou programaticamente, como no exemplo a seguir.

Esta entidade de serviço é a identidade usada pelo conector para gravar dados em sua tabela no Kusto. Você concede permissões para esta entidade de serviço acessar recursos do Kusto.

Inicie sessão na sua subscrição do Azure através da CLI do Azure. Em seguida, autentique-se no navegador.
```
az login
```
Escolha a assinatura para hospedar o principal. Esta etapa é necessária quando você tem várias assinaturas.
```
az account set --subscription YOUR_SUBSCRIPTION_GUID
```

Crie a entidade de serviço. Neste exemplo, a entidade de serviço é chamada de my-service-principal.

az ad sp create-for-rbac -n "my-service-principal" --role Contributor --scopes /subscriptions/{SubID}

Dos dados JSON retornados, copie os appId, passworde tenant para uso futuro.

{
  "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"
}

Você criou o seu aplicativo e entidade de serviço do Microsoft Entra.

Criar uma tabela de destino

O Fluent Bit encaminha logs no formato JSON com três propriedades: log (dinâmico), tag (string) e timestamp (data e hora).

Você pode criar uma tabela com colunas para cada uma dessas propriedades. Como alternativa, se você tiver logs estruturados, poderá criar uma tabela com propriedades de log mapeadas para colunas personalizadas. Para saber mais, selecione a guia relevante.

Esquema padrão
Esquema personalizado

Para criar uma tabela para os registos de entrada provenientes do Fluent Bit:

Navegue até o ambiente de consulta.
Selecione o banco de dados onde deseja criar a tabela.
Execute o seguinte comando .create table:
```
.create table FluentBitLogs (log:dynamic, tag:string, timestamp:datetime)
```
As propriedades JSON de entrada são automaticamente mapeadas na coluna correta.

Para criar uma tabela para logs estruturados de entrada do Fluent Bit:

Navegue até o ambiente de consulta.
Selecione o banco de dados onde deseja criar a tabela.
Execute o comando .create table. Por exemplo, se os logs contiverem três campos chamados myString, myIntegere myDynamic, você poderá criar uma tabela com o seguinte esquema:
```
.create table FluentBitLogs (myString:string, myInteger:int, myDynamic: dynamic, timestamp:datetime)
```

Crie um de mapeamento JSON para mapear as propriedades de log para as colunas apropriadas. O comando a seguir cria um mapeamento com base no exemplo da etapa anterior:

.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"}} 
    ]```

Conceder permissões ao principal do serviço

Conceda à entidade de serviço de Criar uma entidade de serviço do Microsoft Entra ingestor de banco de dados permissões de função para trabalhar com o banco de dados. Para obter mais informações, consulte Exemplos. Substitua o espaço reservado DatabaseName pelo nome do banco de dados de destino e ApplicationID pelo valor AppId que guardou ao criar um principal de serviço do Microsoft Entra.

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

Configurar o Fluent Bit para enviar logs para sua tabela

Para configurar o Fluent Bit para enviar logs para sua tabela no Kusto, crie um de modo clássico ou arquivo de configuração do modo YAML com as seguintes propriedades de saída:

Campo	Descrição	Necessário	Padrão
Nome	O nome do pipeline.		`azure_kusto`
identificador_de_inquilino	A ID do locatário do Criar uma entidade de serviço do Microsoft Entra.	✔️
ID do cliente	O ID da aplicação do Criar um principal de serviço Microsoft Entra.	✔️
segredo_do_cliente	O valor da chave secreta do cliente (palavra-passe) da Criar uma entidade de serviço do Microsoft Entra.	✔️
ID_de_cliente_de_identidade_gerida	A ID do cliente da identidade gerenciada a ser usada para autenticação.	✔️
ponto_de_ingestão	Insira o valor conforme descrito para Ingestion_Endpoint.	✔️
nome_da_base_de_dados	O nome do banco de dados que contém a tabela de logs.	✔️
nome_da_tabela	O nome da tabela de Criar uma tabela de destino.	✔️
referência_de_mapeamento_de_ingestão	O nome do mapeamento de ingestão a partir de Criar uma tabela de destino. Se você não criou um mapeamento de ingestão, remova a propriedade do arquivo de configuração.
chave_de_registro	Nome da chave do conteúdo do log. Por exemplo, `log`.		`log`
incluir_chave_etiqueta	Se habilitada, uma tag é anexada à saída.		`On`
chave_de_etiqueta	O nome da chave da tag. Ignorado se `include_tag_key` for falso.		`tag`
incluir_chave_de_tempo	Um carimbo de data/hora é anexado à saída, se habilitado. Usa a propriedade `time_key`.		`On`
chave_de_tempo	O nome da chave para o carimbo de data/hora nos registros de log. Ignorado se `include_time_key` falso.		`timestamp`
tempo_limite_de_conexão_do_ponto_de_ingestão	O tempo limite de conexão de vários pontos finais Kusto em segundos.		`60`
compressao_ativada	Envia carga HTTP compactada (gzip) para Kusto, se habilitado.		`true`
intervalo_de_atualização_de_recursos_de_ingestão	Os recursos de ingestão atualizam o intervalo do ponto de extremidade Kusto em segundos.
trabalhadores	O número de trabalhadores para executar operações de descarga para esta saída.		`0`
bufferização_ativada	Se ativado, armazena os dados em buffer no disco antes de ingerir no Kusto.		`Off`
buffer_path	Especifica o local do diretório onde os dados armazenados em buffer serão armazenados se `buffering_enabled` for `On`.		`/tmp/fluent-bit/azure-kusto/`
tempo limite de carregamento	Especifica o tempo limite para os carregamentos se `buffering_enabled` estiver `On`. Os ficheiros mais antigos do que este são ingeridos mesmo que abaixo do limite de tamanho.		`30m`
tamanho_do_ficheiro_de_carregamento	Especifica o tamanho máximo de um arquivo a ser carregado se `buffering_enabled` for `On`.		`200MB`
azure_kusto_buffer_key (chave de buffer do Kusto da Azure)	Chave de buffer do Kusto do Azure para identificar instâncias do plug-in quando `buffering_enabled` é `On`. Necessário para várias saídas do Azure Kusto com bufferização.		`key`
limite_de_tamanho_do_diretório_de_armazenamento	O tamanho máximo do diretório onde os dados em buffer são guardados caso `buffering_enabled` seja `On`.		`8GB`
buffer_eliminar_ficheiro_antecipadamente	Quando `buffering_enabled` é `On`, se o arquivo armazenado em buffer deve ser excluído antecipadamente após a criação bem-sucedida do blob.		`Off`
unify_tag	Cria um único arquivo de buffer quando `buffering_enabled` é `On`.		`On`
comprimento_do_blob_uri	Defina o comprimento do URI de blob gerado antes de ingerir para Kusto.		`64`
repeticoes_maximas_do_agendador	Quando `buffering_enabled` é `On`, defina o número máximo de novas tentativas para ingestão usando o agendador.		`3`
eliminar_em_erro_de_carregamento_máximo	Quando `buffering_enabled` é `On`, deve excluir o ficheiro de buffer no caso de erros máximos de upload.		`Off`
IO_tempo limite	Configure o tempo limite de E/S HTTP para uploads.		`60s`

Para ver um exemplo de arquivo de configuração, selecione a guia relevante:

modo clássico
modo 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

Confirmar a ingestão de dados

Assim que os dados chegarem à tabela, confirme a transferência de dados, verificando a contagem de linhas:
```
FluentBitLogs
| count
```
Para exibir um exemplo de dados de log, execute a seguinte consulta:
```
FluentBitLogs
| take 100
```

Visão geral das integrações de dados
Visão geral do Kusto Query Language (KQL)
Escrever consultas

Feedback

Esta página foi útil?

Last updated on 2025-05-27