Synchroniser des données Azure Event Hubs vers Elasticsearch avec Logstash - Elasticsearch

Procédure

Étape 1 : Préparer l'environnement et les instances
Étape 2 : Créer et configurer un pipeline Logstash
Étape 3 : Vérifier le résultat

Étape 1 : Préparer l'environnement et les instances

Créez une instance Alibaba Cloud Elasticsearch et activez l'indexation automatique. Cet exemple utilise une instance version 7.10.

Pour plus d'informations, consultez

Créer une instance Alibaba Cloud Elasticsearch

et

Configurer les paramètres YML

.
Créez une instance Alibaba Cloud Logstash et configurez la transmission de données via Internet. Cet exemple utilise une instance version 7.4.

Pour plus d'informations, consultez

Créer une instance Alibaba Cloud Logstash

.

Les instances Alibaba Cloud Logstash sont déployées dans un VPC. L'instance Logstash doit accéder à Internet pour communiquer avec Azure Event Hubs. Pour activer l'accès Internet, configurez une passerelle NAT. Pour plus d'informations, consultez

Configurer la transmission de données Internet via une passerelle NAT

.

Remarque
Pour une installation Logstash autogérée, utilisez une instance ECS située dans le même VPC que l'instance Alibaba Cloud Elasticsearch. Si vous disposez déjà d'une instance ECS appropriée, il est inutile d'en créer une nouvelle. Associez une adresse IP élastique (EIP) à cette instance ECS.
Préparez votre environnement Azure Event Hubs.

Pour plus d'informations, consultez la

documentation officielle Azure Event Hubs

.

Étape 2 : Créer et configurer un pipeline Logstash

Accédez à la page Logstash Clusters.
Accédez au cluster cible.
1. Dans la barre de navigation supérieure, sélectionnez la région où se trouve le cluster.
2. Sur la page Logstash Clusters, recherchez le cluster et cliquez sur son ID.
Dans le volet de navigation de gauche, cliquez sur Pipelines.
Cliquez sur Create Pipeline.

Sur la page Create, saisissez un Pipeline ID et configurez le pipeline.

Cet exemple utilise la configuration de pipeline suivante :

input {
  azure_event_hubs {
     event_hub_connections => ["Endpoint=sb://abc-****.****.cn/;SharedAccessKeyName=gem-****-es-consumer;SharedAccessKey=******;EntityPath=xxxxxx"]
     initial_position => "beginning"
     threads => 2
     decorate_events => true
     consumer_group => "group-kl"
     storage_connection => "DefaultEndpointsProtocol=https;AccountName=xxxxx;AccountKey=*******;EndpointSuffix=core.****.cn"
     storage_container => "lettie_container"
   }
}
filter {

}
output {
  elasticsearch {
    hosts => ["es-cn-tl****5r50005adob.elasticsearch.aliyuncs.com:9200"]
    index => "test-log"
    password => "xxxxxx"
    user => "elastic"
  }
}

Tableau 1. Paramètres d'entrée
Paramètre	Description
event_hub_connections	Liste des chaînes de connexion des hubs d'événements source. La chaîne de connexion inclut le paramètre EntityPath du hub d'événements. Pour plus d'informations, consultez event_hub_connections. Remarque Chaque hub d'événements définit son propre paramètre event_hub_connections. Les autres paramètres sont partagés entre tous les hubs d'événements.
initial_position	Position dans le hub d'événements à partir de laquelle la lecture des données commence. Valeurs valides : beginning (par défaut), end et look_back. Pour plus d'informations, consultez initial position .
threads	Nombre total de threads dédiés au traitement des événements. Pour plus d'informations, consultez threads.
decorate_events	Indique s'il faut synchroniser les métadonnées du hub d'événements. Ces métadonnées comprennent le nom du hub d'événements, le groupe de consommateurs, l'hôte du processeur, la partition, le décalage, le numéro de séquence, l'horodatage et la taille de l'événement. Pour plus d'informations, consultez decorate events.
consumer_group	Groupe de consommateurs utilisé pour lire les données du hub d'événements. Créez un groupe de consommateurs dédié pour Logstash et assurez-vous que tous les nœuds Logstash utilisent ce groupe afin de garantir une coordination correcte. Pour plus d'informations, consultez consumer group.
storage_connection	Chaîne de connexion du compte de stockage Blob. Ce compte conserve les décalages entre les redémarrages et permet à plusieurs nœuds Logstash de traiter différentes partitions. Si ce paramètre est défini, le traitement reprend au dernier point de contrôle après un redémarrage. Sinon, le traitement démarre à la position spécifiée par le paramètre initial_position. Pour plus d'informations, consultez storage connection.
storage_container	Nom du conteneur de stockage utilisé pour conserver les décalages et coordonner plusieurs nœuds Logstash. Pour plus d'informations, consultez storage container. Remarque Pour éviter d'écraser les décalages, utilisez un nom storage_container différent. Si vous écrivez les mêmes données vers différents services, définissez ce paramètre avec un nom distinct pour chaque service.

Tableau 2. Paramètres de sortie
Paramètre	Description
hosts	Endpoint du service Alibaba Cloud Elasticsearch. Définissez la valeur sur `http://<Alibaba Cloud Elasticsearch instance ID>.elasticsearch.aliyuncs.com:9200`.
index	Nom de l'index de destination.
user	Nom d'utilisateur pour accéder au service Elasticsearch. La valeur par défaut est elastic.
password	Mot de passe de l'utilisateur spécifié. Pour Alibaba Cloud Elasticsearch, le mot de passe de l'utilisateur elastic est défini lors de la création de l'instance. En cas d'oubli, réinitialisez-le. Pour plus d'informations, consultez Réinitialiser le mot de passe d'accès d'une instance.

Pour plus d'informations sur les paramètres Config, consultez Fichiers de configuration Logstash.

Cliquez sur Next step et configurez les paramètres du pipeline.

Paramètre	Description
Pipeline Workers	Nombre de threads de travail pour les étapes de filtrage et de sortie. Augmentez cette valeur si des événements s'accumulent ou si le CPU est sous-utilisé. Valeur par défaut : nombre de cœurs CPU.
Pipeline Batch Size	Nombre maximal d'événements qu'un worker collecte avant d'exécuter les filtres et les sorties. Des lots plus importants augmentent l'utilisation de la mémoire et peuvent nécessiter une taille de tas JVM plus grande (LS_HEAP_SIZE). Valeur par défaut : 125.
Pipeline Batch Delay	Temps d'attente en millisecondes avant de dispatcher un lot incomplet vers un thread de travail. Valeur par défaut : 50 ms.
Queue Type	Modèle de file d'attente interne pour la mise en mémoire tampon des événements entre les étapes. Valeurs valides : MEMORY : Valeur par défaut. Utilise une file d'attente en mémoire. PERSISTED : File d'attente persistante sur disque.
Queue Max Bytes	Quantité maximale de données que la file d'attente peut stocker, en `MB`. La valeur doit être un entier compris entre `1` et `2⁵³-1`. Valeur par défaut : `1024`. Remarque Assurez-vous que cette valeur est inférieure à la capacité totale de votre disque.
Queue Checkpoint Writes	Nombre maximal d'événements écrits avant qu'un point de contrôle ne soit forcé (file d'attente persistante uniquement). 0 signifie aucune limite. Valeur par défaut : 1024.

Avertissement

L'enregistrement et le déploiement de la configuration déclenchent un redémarrage de l'instance. Ne procédez à cette opération que si cela n'affecte pas votre activité.

Cliquez sur Save ou Save and Deploy.
- Save : Enregistre la configuration du pipeline sans l'appliquer. Après l'enregistrement, vous revenez à la page Pipelines. Dans la section Pipelines, cliquez sur Deploy Now dans la colonne Actions pour redémarrer l'instance et appliquer la configuration.
- Save and Deploy : Enregistre et déploie la configuration, puis redémarre l'instance pour appliquer les modifications.

Étape 3 : Vérifier le résultat

Connectez-vous à la console Kibana de votre cluster Elasticsearch et accédez à la page d'accueil Kibana.
Dans le menu de navigation de gauche, cliquez sur Dev tools.
Dans la Console, exécutez la commande suivante pour consulter les données synchronisées.
```
GET test-log3/_search
{
  "query":{
    "match":{
      "message":"L23"
     }
   }
}
```
Le résultat attendu est le suivant.