Logstash : plugins Input, Filter, Output pour Elasticsearch et Kafka

Table des matières :

Logstash est souvent le chaînon manquant entre des sources “bruitées” (logs applicatifs, événements métier, métriques semi-structurées) et des backends adaptés (Elasticsearch pour la recherche/observabilité, Kafka pour le transport et le replay). Le point clé, côté architecture, c’est qu’on ne “branche” pas Logstash au hasard : on choisit précisément les plugins Input, Filter et Output pour maîtriser la latence, la résilience et la compatibilité de schéma.

Dans un contexte e-commerce (PrestaShop, Symfony/Monolog, Nginx/HAProxy), Logstash joue souvent 3 rôles en même temps :

Normaliser (réduire les variations de format / typage),
Rendre la donnée rejouable (Kafka, persistent queue, idempotence),
Rendre la donnée requêtable (ECS, data streams, ILM, mapping maîtrisé).

L’objectif n’est pas “d’ingérer tout” : c’est d’ingérer ce qui sert (debug, performance, sécurité, audit), avec une rétention et un niveau de détail compatibles RGPD.

Pipeline Logstash : modèle Input → Filter → Output et contraintes réelles

Elastic résume correctement le rôle de Logstash : « Logstash is a free and open server-side data processing pipeline that ingests data from a multitude of sources simultaneously, transforms it, and then sends it to your favorite “stash.” » (documentation officielle Elastic, Logstash). Cette phrase est moins marketing qu’elle n’en a l’air : Logstash est un moteur d’exécution de plugins (Ruby/Java) avec une pipeline d’événements typée “hash” (un event), sur laquelle vous greffez des transformations.

Le modèle mental à garder est strict : Input produit des événements, Filter modifie/enrichit ces événements, Output les publie. C’est trivial sur un schéma, beaucoup moins dans un SI où vous devez gérer : (1) des formats multiples (JSON, plain text, CSV, syslog), (2) des timestamps incohérents (timezone, drift), (3) des champs non normalisés (naming, types), (4) des contraintes d’indexation Elasticsearch (mapping, dynamic templates), et (5) des contraintes Kafka (offsets, consumer groups, ordering par partition).

Une façon simple de “poser” les contraintes réelles est de les mapper à des leviers Logstash concrets :

Contrainte terrain	Symptôme	Levier Logstash (souvent)	Risque si ignoré
Timestamps incohérents (UTC vs Europe/Paris, formats mixtes)	graphes faux, corrélation impossible	`date {}` + fallback + tags d’erreur	alerting inutilisable
Variantes de formats (Nginx, PHP-FPM, JSON partiel)	`_grokparsefailure`, champs manquants	`dissect` quand stable, `grok` sinon, route “échec parsing”	bruit + coûts d’indexation
Explosion de champs (payloads JSON non maîtrisés)	mappings qui “partent en vrille”	`prune`, `remove_field`, whitelist	conflits de mapping / ingestion bloquée
PII dans les logs (email, IP non nécessaire, téléphone)	surface RGPD inutile	pseudonymisation + minimisation (`mutate`)	risque conformité
Backpressure (ES lent / cluster en tension)	latence, retries, blocage	PQ + tuning batch + ILM/data streams	perte ou saturation

En production, deux mécanismes de Logstash deviennent rapidement non négociables : Persistent Queues (PQ) pour absorber les pics et amortir les indisponibilités d’outputs, et Dead Letter Queue (DLQ) pour isoler les événements “non indexables” (typiquement un conflit de mapping). Ce n’est pas un détail : sans PQ/DLQ, une erreur de mapping sur Elasticsearch peut se transformer en perte silencieuse d’événements ou en backpressure qui remonte jusqu’à vos producteurs.

Points pratiques (souvent découverts “trop tard”) :

PQ ≠ Kafka : la PQ protège Logstash contre un output indisponible, mais ne remplace pas un bus (pas de fan-out, pas de replay multi-consommateurs).
DLQ : très utile quand vous déployez de nouveaux champs côté application (ou module PrestaShop) et que le mapping ES n’est pas prêt. Le DLQ évite de bloquer tout un pipeline pour une minorité d’événements.
Séparation par flux : plutôt qu’un seul gros pipeline, des pipelines distincts (via pipelines.yml) limitent les effets de bord (un grok lourd sur Nginx ne doit pas impacter vos événements métier).

Exemple minimal de paramètres PQ/DLQ (à adapter à votre contexte disque et volumétrie), typiquement dans logstash.yml :

queue.type: persisted
queue.max_bytes: 4gb
dead_letter_queue.enable: true
path.dead_letter_queue: /var/lib/logstash/dead_letter_queue

Plugins Input : ingestion fiable depuis Kafka, HTTP et fichiers

Un plugin Input Logstash définit le contrat d’ingestion : protocole, stratégie d’ack, persistance d’état (offsets, sincedb), et parfois décodage. Dans un stack e-commerce, les sources courantes sont : logs Nginx/HAProxy, logs PHP-FPM, logs applicatifs PrestaShop/Symfony, événements métiers (commande créée, paiement accepté) publiés sur un bus, et exports “batch” (CSV/JDBC). L’erreur classique est de tout faire entrer par file {} “parce que c’est simple” : vous perdez le contrôle fin (replay, partitions, garanties) dès que vous passez à l’échelle.

Un bon réflexe est de choisir l’input en fonction de la capacité de replay et du mode de fonctionnement (stream vs batch) :

Source	Plugin input recommandé	Pourquoi	Piège fréquent
Logs systèmes/app sur VM	`beats {}` (Filebeat/Winlogbeat)	shipping fiable + backpressure + TLS	parser en multiline côté Logstash au lieu de Filebeat
Flux événementiel	`kafka {}`	replay, partitions, découplage	mauvais `group_id` → consommation inattendue
Webhooks / callbacks	`http {}`	simple à intégrer	absence d’auth/rate-limit → point d’entrée exposé
Fichiers “batch” (exports)	`file {}`	pragmatique	rotation, `sincedb` mal géré, doublons
Base (catalogue/stocks)	`jdbc {}`	extraction incrémentale	full-scan faute de “tracking”

Pour Kafka, le plugin kafka en input est le bon choix quand vous voulez rejouer un flux, partitionner par clé (ex. shop_id ou order_id) et découpler producteurs/consommateurs. Kafka le rappelle lui-même : « Apache Kafka is an open-source distributed event streaming platform used by thousands of companies for high-performance data pipelines, streaming analytics, data integration, and mission-critical applications. » (site officiel Apache Kafka). Côté Logstash, vous devez être explicite sur group_id, topics, la stratégie de commit et la sécurité (SASL_SSL en entreprise).

Exemple minimal (testé avec Logstash 8.x et Kafka 3.x), avec décodage JSON en entrée et gestion TLS :

input {
  kafka {
    bootstrap_servers => "kafka-1:9093,kafka-2:9093"
    topics => ["prestashop.events"]
    group_id => "logstash-observability"
    auto_offset_reset => "latest"  # ou "earliest" pour bootstrap
    security_protocol => "SASL_SSL"
    sasl_mechanism => "SCRAM-SHA-512"
    sasl_jaas_config => "org.apache.kafka.common.security.scram.ScramLoginModule required username='logstash' password='***';"
    ssl_truststore_location => "/etc/logstash/kafka.truststore.jks"
    ssl_truststore_password => "***"
    codec => json
  }
}

Pour les sources HTTP (webhooks, callbacks PSP, logs push depuis edge), http {} est pratique mais dangereux si vous ne cadrez pas la volumétrie et l’auth. Deux patterns “propres” en production :

mettre l’input HTTP derrière un reverse proxy (Nginx/Traefik) qui gère TLS, auth, rate limiting ;
ou activer TLS + auth côté plugin et limiter strictement l’exposition réseau.

Exemple (simplifié) d’input HTTP avec TLS côté Logstash :

input {
  http {
    host => "0.0.0.0"
    port => 8443
    ssl => true
    ssl_certificate => "/etc/logstash/server.crt"
    ssl_key => "/etc/logstash/server.key"
    # Selon votre usage, ajouter une authent (ex. reverse proxy) reste souvent préférable.
    codec => json
  }
}

Le plugin beats {} reste la voie la plus robuste pour les logs systèmes via Filebeat/Winlogbeat (reliable shipping, backpressure, TLS). Enfin, pour des imports structurés (catalogue, stocks), jdbc {} est viable mais doit être traité comme un batch (fenêtrage, tracking column), sinon vous allez provoquer des lectures full-scan et vous tirer une balle dans le pied côté base.

Plugins Filter : parsing, normalisation ECS, enrichissement, déduplication

Le “vrai” boulot Logstash se passe dans les filters, et c’est là que les pipelines deviennent soit maintenables, soit ingérables. Pour parser du texte, grok est puissant mais coûteux (regex) ; dissect est nettement plus rapide si votre format est stable (délimiteurs). Exemple typique : logs Nginx. Avec dissect, vous mappez sans regex ; avec grok, vous gérez des variantes mais vous payez le CPU. Sur des flux >10k events/s, la différence se voit immédiatement sur la latence et les GC JVM.

Un pattern efficace en observabilité consiste à :

décoder au plus tôt (JSON → champs),
parser uniquement les lignes qui en ont besoin (conditionnels),
normaliser les champs (ECS / typage),
taguer proprement les erreurs (et router ces événements),
réduire la surface (suppression/whitelist des champs).

Exemple de parsing Nginx orienté ECS avec dissect, puis typage et enrichissement (à adapter à votre format de log exact) :

filter {
  if [event][dataset] == "nginx.access" or "nginx" in [tags] {
    dissect {
      mapping => {
        "message" => '%{client.ip} - %{user.name} [%{nginx.time}] "%{http.request.method} %{url.original} HTTP/%{http.version}" %{http.response.status_code} %{http.response.body.bytes} "%{http.request.referrer}" "%{user_agent.original}"'
      }
    }

    mutate {
      convert => {
        "[http][response][status_code]" => "integer"
        "[http][response][body][bytes]" => "integer"
      }
    }

    date {
      match => ["nginx.time", "dd/MMM/YYYY:HH:mm:ss Z"]
      target => "@timestamp"
      tag_on_failure => ["_dateparsefailure_nginx"]
    }

    useragent {
      source => "[user_agent][original]"
      target => "[user_agent]"
    }

    geoip {
      source => "[client][ip]"
      target => "[geoip]"
    }
  }
}

Normaliser vers un schéma cohérent évite les mappings “dynamiques” qui explosent. Dans l’écosystème Elastic, l’ECS est le standard de facto : « Elastic Common Schema (ECS) is a shared specification for field names and data types » (Elastic, ECS). Concrètement : client.ip, http.request.method, url.path, event.dataset, etc. Le filtre mutate (rename/convert), date (timestamp source → @timestamp), et useragent/geoip (enrichissement) sont les briques habituelles.

Deux points qui font une différence nette sur la maintenabilité :

Contrôler l’explosion de champs : si vos événements métier incluent des payloads JSON variables (ex. retour PSP, contenu panier), il est souvent préférable d’indexer un sous-ensemble (champs stables) et de stocker le reste ailleurs (ou dans un champ “raw” non analysé), plutôt que de laisser Elasticsearch créer des milliers de champs.
Gérer les erreurs de parsing comme un flux normal : un tag _grokparsefailure ou _dateparsefailure_* doit déclencher une stratégie (index “parsing-failed”, topic Kafka dédié, ou DLQ), pas être ignoré.

Côté conformité, vous devez décider où traiter la donnée personnelle (PII). Si vous indexez des logs applicatifs contenant email/téléphone, vous créez une surface RGPD inutile. Le minimum est de pseudonymiser en pipeline (ex. SHA-256) et de supprimer les champs non indispensables (remove_field). Si vous avez un doute sur vos obligations et la granularité des contrôles, la checklist RGPD PrestaShop du site est un bon point d’entrée : Audit RGPD PrestaShop : 88 points de contrôle pour boutiques.

Pour la déduplication et l’idempotence, fingerprint est sous-utilisé. Dans un flux Kafka rejouable, vous verrez des doublons (retries producteurs, rebalances, replays). Calculez un document_id déterministe à partir d’un ensemble de champs (ex. order_id + event_type + created_at) puis utilisez-le en output Elasticsearch (cf. section suivante). Vous évitez ainsi des indexations multiples qui faussent vos métriques et gonflent vos coûts.

Conseil pragmatique : utilisez un “sel” (salt) stable, stocké en secret (keystore), si vous fingerprint des identifiants sensibles. Vous gardez l’idempotence sans exposer des valeurs en clair.

Plugins Output : Elasticsearch (data streams, ILM) et Kafka (fan-out, replay)

L’output elasticsearch {} n’est pas “juste un endpoint”. Il implique un choix : index statique par jour (logs-YYYY.MM.DD) vs data streams (plus propre pour des logs/metrics avec ILM/rollover). Les data streams sont particulièrement adaptés quand vous avez :

une volumétrie variable,
une politique de rétention (ex. 7/30/90 jours),
et une logique “append-only” typique des logs/événements.

Si vous êtes déjà sur Elasticsearch pour la recherche produit, vous connaissez les impacts de mapping et de perf ; sinon, l’article interne PrestaShop ElasticSearch : accélérer la recherche produit sur grands catalogues donne un aperçu des contraintes côté cluster (shards, refresh, templates), qui se retrouvent aussi en observabilité.

Exemple d’output vers Elasticsearch avec idempotence (document_id) et compat ECS. Adaptez selon votre version Elastic (8.x) et votre stratégie (index vs data stream) :

filter {
  fingerprint {
    source => ["order_id", "event_type", "created_at"]
    method => "SHA256"
    target => "[@metadata][doc_id]"
  }
}

output {
  elasticsearch {
    hosts => ["https://es-1:9200","https://es-2:9200"]
    user => "logstash_writer"
    password => "***"
    ssl_enabled => true
    cacert => "/etc/logstash/http_ca.crt"

    # Option A: index classique
    index => "prestashop-events-%{+YYYY.MM.dd}"

    # Idempotence
    document_id => "%{[@metadata][doc_id]}"
    action => "index"

    # Limiter les surprises de schéma
    ecs_compatibility => "v8"
  }
}

Variante si vous partez sur des data streams (souvent plus propre pour logs-* / events-*) : vous réduisez le bricolage “index par jour” et vous vous alignez sur une gestion ILM/rollover standardisée. Exemple (à adapter à vos conventions) :

output {
  elasticsearch {
    hosts => ["https://es-1:9200"]
    user => "logstash_writer"
    password => "***"
    ssl_enabled => true
    cacert => "/etc/logstash/http_ca.crt"

    data_stream => true
    data_stream_type => "logs"
    data_stream_dataset => "prestashop"
    data_stream_namespace => "prod"

    ecs_compatibility => "v8"
  }
}

L’output kafka {} sert surtout à re-publier après normalisation (ex. vers un topic observability.normalized) ou à implémenter une stratégie “DLQ maison” quand vous ne voulez pas dépendre du DLQ Logstash. Attention : Kafka n’offre pas magiquement de l’“exactly-once” si votre chaîne entière ne le supporte pas. Logstash, de son côté, ne garantit pas une sémantique end-to-end exactly-once : vous visez généralement du at-least-once et vous rendez les consumers idempotents (d’où l’intérêt du fingerprint et de document_id).

Exemple typique de fan-out : republier un événement normalisé en JSON, en conservant une clé de partition (utile pour l’ordre par commande) :

output {
  kafka {
    bootstrap_servers => "kafka-1:9093,kafka-2:9093"
    topic_id => "observability.normalized"
    codec => json
    message_key => "%{order_id}"
    security_protocol => "SASL_SSL"
    sasl_mechanism => "SCRAM-SHA-512"
    sasl_jaas_config => "org.apache.kafka.common.security.scram.ScramLoginModule required username='logstash' password='***';"
  }
}

Chaînes d’observabilité pour PrestaShop : logs applicatifs, SIEM, recherche

PrestaShop 8/9 (surtout 9, stack Symfony) s’intègre bien dans une stratégie d’observabilité “Elastic-friendly” si vous disciplinez vos logs. Dans les modules et le core Symfony, vous pouvez produire du JSON structuré (Monolog + formatter JSON) plutôt que du texte, ce qui réduit la dépendance à grok. Si votre priorité est la résolution d’incidents, cherchez d’abord à standardiser : request_id, cart_id, order_id, customer_id (pseudonymisé), shop_id, environment, git_sha. Sans ces corrélations, vous indexez du bruit.

Checklist minimaliste “événement métier exploitable” (utile pour éviter les logs impossibles à corréler) :

@timestamp fiable (UTC idéalement) + timezone maîtrisée côté affichage,
identifiant de corrélation (request_id / trace_id si vous avez de l’APM),
event.action / event.type (ex. payment.authorized, order.created),
order_id / cart_id / shop_id selon le périmètre,
service.name (web, worker, cron) + environment (prod/preprod),
un champ error.message / error.stack_trace seulement quand nécessaire (et attention aux PII).

Architecture qui marche dans la vraie vie (latence raisonnable, replay possible) :

Filebeat sur les nœuds web (Nginx/PHP-FPM) → Logstash (filters ECS) → Elasticsearch (logs)
PrestaShop events (hooks/modules) → Kafka (prestashop.events) → Logstash (normalisation + enrichissement) → Elasticsearch (events)
En parallèle : Logstash → Kafka (observability.normalized) pour alimenter d’autres consumers (alerting externe, data warehouse)

Si vous êtes en démarche “sécurité + auditabilité”, vous retombez sur les mêmes primitives : journalisation, corrélation, alerting. Le lien interne pertinent côté hardening/traçabilité est : Sécurité PrestaShop : protéger API backoffice, WAF et journalisation SIEM. Logstash n’est pas un SIEM, mais c’est fréquemment l’outil qui formate et enrichit les événements avant ingestion SIEM.

Cas concret : exploitation d’un incident checkout. Vous corrélez un pic de 500 sur /order (Nginx) avec une hausse d’exceptions applicatives (PrestaShop) et des timeouts PSP (webhooks). Indexés proprement, ces événements deviennent requêtables (Kibana) et actionnables (alerting). En sécurité, la même logique vous sert à vérifier un rollout de patch ou un comportement anormal après une CVE ; l’article CVE-2025-61922 ps_checkout : mise à jour 5.0.5 et mesures illustre précisément le besoin de traces exploitables post-déploiement.

Exploitation en production : tuning JVM, tests, et sécurité TLS

Logstash est un process JVM. Donc oui : le tuning basique compte. Fixez un heap stable (ex. -Xms2g -Xmx2g), évitez le swap, surveillez la pression GC, et ajustez pipeline.workers (souvent = vCPU) et pipeline.batch.size (trade-off throughput/latence). Si vos filters sont lourds (grok, geoip), le CPU sera le bottleneck ; si votre output Elasticsearch est lent (refresh, cluster sous-dimensionné), vous verrez du backpressure. Dans ce cas, activez/validez les Persistent Queues pour absorber la dette temporaire plutôt que de perdre des événements.

Quelques réglages “terrain” à vérifier lors d’une mise en prod (sans tomber dans l’over-tuning) :

Disk : PQ + DLQ consomment du disque ; mettez des alertes (taux de remplissage, IOPS).
Concurrence : augmentez pipeline.workers si vos filters sont CPU-bound, mais gardez en tête que certains traitements (regex lourdes) scalent mal.
Batch : montez pipeline.batch.size si vous visez du débit (bulk ES), baissez si vous visez de la latence.
Observabilité de Logstash : exploitez les métriques de pipeline (events in/out, duration, queue) pour distinguer “parsing lent” vs “output lent”.

Pour sécuriser vos changements, utilisez systématiquement bin/logstash -t (validation de config) et versionnez vos pipelines (Git) comme du code. Les pipelines multi-fichiers finissent vite en spaghetti ; préférez pipelines.yml avec une séparation claire par flux (ex. nginx.conf, prestashop_events.conf). Si vous industrialisez, vous pouvez tester des patterns grok et des transformations via des tests de pipeline (RSpec Logstash). C’est la même philosophie que côté PHP : qualité et CI/CD d’abord, cf. PHPStan et Rector : industrialiser la qualité du code PHP.

Enfin, sécurité : chiffrez les flux (TLS), ne mettez jamais de credentials en clair dans les confs (utilisez keystore Logstash), et restreignez les droits Elasticsearch (user dédié, rôle d’écriture sur des index ciblés). Côté données, appliquez le principe de minimisation (PII) et des politiques de rétention (ILM). Et côté “surface d’attaque”, rappelez-vous que Logstash expose potentiellement des inputs réseau (http, tcp, beats) : derrière un réseau privé, avec auth/TLS, sinon vous ouvrez une porte d’entrée directe vers votre backend d’indexation.

Astuce pratique (souvent oubliée) : si vous activez le rechargement automatique des confs (--config.reload.automatic), faites-le avec discipline (déploiement atomique, validation -t en CI), car un pipeline invalide peut interrompre l’ingestion au pire moment.

Références externes (docs) :

Logstash (Elastic) : https://www.elastic.co/guide/en/logstash/current/index.html
Plugins Kafka input/output (Elastic) : https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html et https://www.elastic.co/guide/en/logstash/current/plugins-outputs-kafka.html
Output Elasticsearch (Elastic) : https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html
Apache Kafka (projet officiel) : https://kafka.apache.org/intro

Logstash : plugins Input, Filter, Output pour Elasticsearch et Kafka

Pipeline Logstash : modèle Input → Filter → Output et contraintes réelles

Plugins Input : ingestion fiable depuis Kafka, HTTP et fichiers

Plugins Filter : parsing, normalisation ECS, enrichissement, déduplication

Plugins Output : Elasticsearch (data streams, ILM) et Kafka (fan-out, replay)

Chaînes d’observabilité pour PrestaShop : logs applicatifs, SIEM, recherche

Exploitation en production : tuning JVM, tests, et sécurité TLS

À lire aussi

Vulnérabilités PHP RCE : corriger CVE-2026-109618 et CVE-2026-109619

CVE noyau Linux : stratégie de live patching pour hébergement mutualisé

Azure CLI : notes de version 2.89.1/2.89.0 et changements clés