Ingérer des données avec Fluent Bit dans Azure Data Explorer

Fluent Bit est un agent open source qui collecte des journaux d’activité, des métriques et des traces à partir de différentes sources. Il permet de filtrer, de modifier et d’agréger les données relatives aux événements avant de les envoyer au stockage. Cet article vous guide tout au long du processus d’utilisation de Fluent Bit pour envoyer des données à votre base de données KQL.

Cet article explique comment ingérer des données avec Fluent Bit.

Pour obtenir la liste complète des connecteurs de données, veuillez consulter la rubrique Vue d’ensemble des connecteurs de données.

Prerequisites

Fluent Bit.
Un cluster et une base de données Azure Data Explorer. Créer un cluster et une base de données.
Environnement de requête. Pour plus d’informations, consultez vue d’ensemble des intégrations de requêtes.
L’URI de votre cluster Kusto pour la valeur Ingestion_endpoint au format https://ingest-<cluster>.<région>.kusto.windows.net. Pour plus d’informations, consultez Ajouter une connexion de cluster.

Créer un principal de service Microsoft Entra

Le principal du service Microsoft Entra peut être créé via le portail Azure ou par programmation, comme dans l’exemple suivant.

Ce principal de service est l’identité utilisée par le connecteur pour écrire des données dans la table en Kusto. Vous accordez des autorisations pour que ce principal de service puisse accéder aux ressources Kusto.

Connectez-vous à votre abonnement Azure via Azure CLI. Authentifiez-vous ensuite dans le navigateur.
```
az login
```
Choisissez l’abonnement qui hébergera le principal. Cette étape est nécessaire quand vous avez plusieurs abonnements.
```
az account set --subscription YOUR_SUBSCRIPTION_GUID
```
Créez le principal de service. Dans cet exemple, le principal de service est appelé my-service-principal.
```
az ad sp create-for-rbac -n "my-service-principal" --role Contributor --scopes /subscriptions/{SubID}
```

À partir des données JSON renvoyées, copiez les informations appId, password et tenant pour une utilisation ultérieure.

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

Vous avez créé votre application Microsoft Entra et votre principal de service.

Créer une table cible

Fluent Bit transmet les journaux d’activité au format JSON avec trois propriétés : log (dynamic), tag (string) et timestamp (datetime).

Vous pouvez créer une table avec des colonnes pour chacune de ces propriétés. Si vous disposez de journaux d’activité structurés, vous pouvez également créer une table dont les propriétés de journalisation sont associées à des colonnes personnalisées. Pour en savoir plus, sélectionnez l’onglet correspondant.

Schéma par défaut
Schéma personnalisé

Pour créer une table pour les journaux d’activité entrants de Fluent Bit :

Accédez à votre environnement de requête.
Sélectionnez la base de données dans laquelle vous souhaitez créer la table.
Exécutez la commande .create tableci-dessous :
```
.create table FluentBitLogs (log:dynamic, tag:string, timestamp:datetime)
```
Les propriétés JSON entrantes sont automatiquement mappées à la colonne appropriée.

Pour créer une table pour les journaux d’activité structurés entrants de Fluent Bit :

Accédez à votre environnement de requête.
Sélectionnez la base de données dans laquelle vous souhaitez créer la table.
Exécutez la commande .create table. Par exemple, si vos journaux d’activité contiennent trois champs nommés myString, myInteger et myDynamic, vous pouvez créer une table avec le schéma suivant :
```
.create table FluentBitLogs (myString:string, myInteger:int, myDynamic: dynamic, timestamp:datetime)
```

Créez un mappage JSON pour mapper les propriétés de journalisation aux colonnes appropriées. La commande suivante crée un mappage basé sur l’exemple de l’étape précédente :

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

Accorder des autorisations au principal de service

Accordez au principal de service créé à l’étape Créer un principal de service Microsoft Entra les autorisations du rôle d’ingesteur de base de données pour qu’il puisse travailler avec la base de données. Pour plus d’informations, consultez Exemples. Remplacez l’espace réservé DatabaseName par le nom de la base de données cible et ApplicationID par la valeur AppId que vous avez enregistrée lors de la création d’un principal de service Microsoft Entra.

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

Configurer Fluent Bit pour envoyer les journaux d’activité à la table

Pour configurer Fluent Bit afin d’envoyer les journaux d’activité à la table en Kusto, créez un fichier de configuration en mode classique ou en mode YAML avec les propriétés de sortie suivantes :

Champ	Descriptif	Obligatoire	Par défaut
Name	Nom du pipeline.		`azure_kusto`
identifiant_de_locataire	ID de locataire généré à l’étape Créer un principal de service Microsoft Entra.	✔️
client_id	ID d’application généré à l’étape Créer un principal de service Microsoft Entra.	✔️
secret du client	Valeur de clé secrète client (mot de passe) générée à l’étape Créer un principal de service Microsoft Entra.	✔️
managed_identity_client_id	ID client de l’identité managée à utiliser pour l’authentification.	✔️
ingestion_endpoint	Entrez la valeur décrite pour Ingestion_Endpoint.	✔️
nom_de_base_de_données	Nom de la base de données qui contient la table des journaux d’activité.	✔️
nom_de_table (table_name)	Nom de la table générée à l’étape Créer une table cible.	✔️
ingestion_mapping_reference	Nom du mappage d’ingestion généré à l’étape Créer une table cible. Si vous n’avez pas créé de mappage d’ingestion, supprimez la propriété du fichier de configuration.
clé_de_journalisation	Nom de clé du contenu du journal. Par exemple, `log`.		`log`
include_tag_key	Si elle est activée, une balise est ajoutée à la sortie.		`On`
tag_key	Nom de clé de la balise. Ignoré si `include_tag_key` est false.		`tag`
include_time_key	Un horodatage est ajouté à la sortie, s’il est activé. Utilise la propriété `time_key`.		`On`
time_key	Nom de clé pour l’horodatage dans les enregistrements du journal. Ignoré si `include_time_key` est faux.		`timestamp`
ingestion_endpoint_connect_timeout	Délai d’expiration de la connexion de différents points de terminaison Kusto en secondes.		`60`
compression_enabled	Envoie la charge utile HTTP compressée (gzip) à Kusto, si elle est activée.		`true`
ingestion_resources_refresh_interval	Intervalle d’actualisation des ressources d’ingestion du point de terminaison Kusto en secondes.
travailleur	Nombre de workers permettant d’effectuer des opérations de vidage pour cette sortie.		`0`
buffering_enabled	Si cette option est activée, les données sont mises en mémoire tampon sur le disque avant l'ingestion dans Kusto.		`Off`
chemin_du_buffer	Spécifie l’emplacement du répertoire dans lequel les données mises en mémoire tampon seront stockées si `buffering_enabled` correspond à `On`.		`/tmp/fluent-bit/azure-kusto/`
upload_timeout	Spécifie le délai d'expiration des chargements si `buffering_enabled` est `On`. Les fichiers plus anciens que cela sont ingérés même s’ils sont en dessous de la limite de taille.		`30m`
taille_du_fichier_téléchargé	Spécifie la taille maximale d’un fichier à téléverser si `buffering_enabled` est `On`.		`200MB`
azure_kusto_buffer_key	Clé de mémoire tampon Azure Kusto permettant d’identifier les instances de plug-in où `buffering_enabled` correspond à `On`. Requis pour plusieurs sorties Azure Kusto avec mise en mémoire tampon.		`key`
store_dir_limit_size	Taille maximale du répertoire dans lequel les données mises en mémoire tampon sont stockées si `buffering_enabled` est `On`.		`8GB`
buffer_file_delete_early	Lorsque `buffering_enabled` correspond à `On`, cela indique s’il faut supprimer le fichier mis en mémoire tampon rapidement après la création de l’objet blob.		`Off`
unify_tag	Crée un fichier de mémoire tampon unique quand `buffering_enabled` est `On`.		`On`
blob_uri_length	Définissez la longueur de l’URI d’objet blob généré avant de l’ingérer dans Kusto.		`64`
scheduler_max_retries	Quand `buffering_enabled` est `On`, définissez le nombre maximal de nouvelles tentatives d’ingestion à l’aide du planificateur.		`3`
delete_on_max_upload_error	Quand `buffering_enabled` est `On`, déterminer s'il faut supprimer le fichier tampon en cas d'erreurs de chargement maximales.		`Off`
IO_timeout	Configurez le délai d’expiration des entrées/sorties (E/S) HTTP pour les chargements.		`60s`

Pour voir un exemple de fichier de configuration, sélectionnez l’onglet correspondant :

Mode classique
Mode 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

Confirmer l’ingestion des données

Une fois que les données arrivent dans la table, confirmez le transfert de données en vérifiant le nombre de lignes :
```
FluentBitLogs
| count
```
Pour afficher un échantillon des données de journal, exécutez la requête suivante :
```
FluentBitLogs
| take 100
```

Vue d’ensemble des intégrations de données
Vue d’ensemble du langage de requête Kusto (KQL)
Écrire des requête

Commentaires

Cette page a-t-elle été utile ?

Last updated on 2025-07-17