Intégration de la pile ELK : Synchronisation de Logstash, Elasticsearch et Kibana

Lorsque Logstash, Elasticsearch et Kibana sont désynchronisés, les journaux disparaissent, les tableaux de bord semblent vides ou les champs apparaissent avec un mauvais type. L'intégration de la pile ELK ne consiste pas seulement à démarrer trois services, mais à s'assurer que les noms d'index, les mappings, les horodatages, les identifiants et les vues de données sont tous en accord.

Ce guide présente un pipeline de journalisation pratique : Logstash reçoit les événements, les analyse, les envoie à Elasticsearch, et Kibana lit les index ou les flux de données résultants. Les exemples utilisent la syntaxe classique du pipeline Logstash et les API Elasticsearch que vous pouvez exécuter depuis Kibana Dev Tools.

Comprendre le flux de données

Tracez un événement à travers la pile :

1. Logstash reçoit les données de Beats, TCP, syslog, fichiers, files d'attente ou d'une autre entrée.
2. Les filtres Logstash analysent, enrichissent, renomment et normalisent les champs.
3. Elasticsearch indexe l'événement à l'aide de modèles, de mappings et de politiques de cycle de vie.
4. Kibana interroge Elasticsearch via une vue de données et affiche l'événement dans Discover, les tableaux de bord, Lens ou les alertes.

La plupart des bugs d'intégration se produisent aux limites. Logstash ne peut pas se connecter, Elasticsearch rejette le document, ou Kibana regarde la mauvaise vue de données ou la mauvaise plage horaire.

Configuration de Logstash pour un flux de données propre

Les pipelines Logstash ont trois blocs principaux : input, filter et output. Gardez chaque bloc simple et testable.

Plugins d'entrée

Les plugins d'entrée courants incluent :

• beats : Reçoit les événements de Filebeat, Metricbeat et autres Beats.
• tcp / udp : Reçoit les événements via des sockets réseau.
• file : Lit les fichiers locaux. Utile pour les petits déploiements et les tests, mais les agents sont généralement meilleurs pour les hôtes de production distribués.
• syslog : Reçoit les messages syslog.

Exemple d'entrée Beats avec TLS :

input {
  beats {
    port => 5044
    ssl_enabled => true
    ssl_certificate => "/etc/pki/tls/certs/logstash.crt"
    ssl_key => "/etc/pki/tls/private/logstash.key"
  }
}

Assurez-vous que le port est ouvert, que le certificat correspond à la façon dont les clients se connectent et que les noms des options correspondent à la version du plugin installé. Les versions récentes du plugin d'entrée Beats utilisent ssl_enabled.

Plugins de filtre

Les filtres transforment les événements bruts en champs utiles. L'ordre est important car Logstash exécute les filtres séquentiellement.

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }

  date {
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
  }

  mutate {
    remove_field => [ "message" ]
  }
}

Utilisez grok pour le texte non structuré, date pour définir @timestamp, mutate pour le nettoyage des champs, et geoip lorsque vous avez besoin d'enrichissement basé sur la localisation IP. Testez les motifs grok sur des lignes de journal réelles avant de les mettre en production. Une petite erreur d'analyse peut envoyer des milliers d'événements dans Elasticsearch avec des champs manquants.

Plugin de sortie

Pour la pile ELK, la sortie Elasticsearch est la destination habituelle.

output {
  elasticsearch {
    hosts => ["https://elasticsearch-node1:9200", "https://elasticsearch-node2:9200"]
    index => "my-logs-%{+YYYY.MM.dd}"
    user => "logstash_writer"
    password => "${LOGSTASH_ES_PASSWORD}"
    ssl_enabled => true
    cacert => "/etc/logstash/certs/http_ca.crt"
  }
}

La valeur index est le contrat avec les modèles Elasticsearch et les vues de données Kibana. Si Logstash écrit my-logs-2026.05.23, votre modèle et votre vue de données doivent correspondre à my-logs-*.

Pour les environnements plus vastes, envisagez les flux de données et la gestion du cycle de vie des index au lieu d'index quotidiens gérés manuellement. Si vous utilisez des flux de données, suivez les conseils actuels d'Elastic pour la sortie Logstash concernant les paramètres data_stream plutôt que de mélanger les options de flux de données et d'index classiques.

Modèles et mappings Elasticsearch

Elasticsearch a besoin de mappings cohérents avant l'arrivée des documents. Sinon, le premier document peut définir un type de champ qui casse les événements ultérieurs. Un code d'état qui arrive d'abord comme "200" peut devenir du texte ou un mot-clé au lieu d'un nombre.

Exemple de modèle d'index composite :

PUT _index_template/my_log_template
{
  "index_patterns": ["my-logs-*"],
  "template": {
    "settings": {
      "number_of_shards": 1,
      "number_of_replicas": 1
    },
    "mappings": {
      "properties": {
        "@timestamp": {"type": "date"},
        "message": {"type": "text"},
        "host.name": {"type": "keyword"},
        "log.level": {"type": "keyword"},
        "http.response.status_code": {"type": "integer"}
      }
    }
  }
}

Utilisez keyword pour la correspondance exacte et les agrégations, text pour la recherche en texte intégral, les types numériques pour les métriques et les codes d'état, et date pour les champs temporels. Gardez un nombre modeste de fragments sauf si vous avez une raison mesurée d'en ajouter. Trop de petits fragments peuvent nuire aux performances du cluster.

Vues de données Kibana

L'interface utilisateur actuelle de Kibana utilise des vues de données. Les versions plus anciennes les appelaient modèles d'index. Créez une vue de données qui correspond aux noms d'index ou aux flux de données qu'Elasticsearch possède réellement.

Configuration typique :

1. Allez dans Stack Management -> Kibana -> Data Views.
2. Créez une vue de données telle que my-logs-*.
3. Choisissez @timestamp comme champ temporel.
4. Ouvrez Discover et élargissez le sélecteur de temps lors des tests.

Si Discover est vide, ne supposez pas que Logstash a échoué. Vérifiez la plage horaire, le motif de la vue de données et si @timestamp a été correctement analysé.

Résolution des problèmes d'intégration courants

Les données n'apparaissent pas dans Kibana

Vérifiez chaque étape :

GET _cat/indices/my-logs-*?v
GET my-logs-*/_search?size=1&sort=@timestamp:desc

Ensuite, vérifiez :

• Les journaux Logstash pour les erreurs de connexion, d'authentification, TLS ou de mapping.
• Les journaux Elasticsearch pour les documents rejetés et les échecs de sécurité.
• Le motif de la vue de données Kibana et la plage horaire sélectionnée.
• Si l'horodatage de l'événement est dans le futur, dans le passé ou manquant.

Les documents sont rejetés par Elasticsearch

Les conflits de mapping sont courants. Par exemple, un événement envoie http.response.status_code comme 200, tandis qu'un autre envoie "OK". Elasticsearch ne peut pas stocker les deux dans un champ integer.

Corrigez le filtre Logstash pour que le champ soit systématiquement typé, ou acheminez les mauvais événements vers un index séparé pour examen. Ne supprimez pas et ne recréez pas constamment des index sans corriger le pipeline qui crée les mauvais documents.

Logstash utilise trop de CPU

Des motifs grok coûteux, un volume d'événements élevé et des événements multilignes volumineux peuvent rapidement augmenter l'utilisation du CPU de Logstash. Commencez par mesurer quel pipeline est occupé, puis simplifiez les motifs, ancrez les expressions régulières et déplacez l'analyse simple vers les pipelines Beats ou Elasticsearch ingest lorsque cela est plus facile à gérer.

Les requêtes Kibana sont lentes

Les tableaux de bord lents proviennent souvent de plages horaires larges, d'agrégations à haute cardinalité, de trop de fragments ou de champs mappés comme text alors que Kibana a besoin de keyword. Utilisez des valeurs par défaut de tableau de bord plus étroites, le roulement ILM et des mappings de champ qui correspondent à vos visualisations.

À retenir

Considérez l'intégration de la pile ELK comme un contrat entre trois couches. Logstash doit émettre des champs prévisibles, Elasticsearch doit les mapper et les stocker correctement, et Kibana doit interroger la bonne vue de données sur la bonne plage horaire. Lorsque quelque chose se casse, suivez un exemple d'événement de l'entrée à l'index jusqu'au tableau de bord.