Bonnes pratiques Logstash

1. Concevoir ses pipelines #

Séparer, simplifier, tester.

Plusieurs pipelines #

# config/pipelines.yml : un pipeline par flux, avec ses propres réglages
- pipeline.id: nginx
  path.config: "/etc/logstash/conf.d/nginx.conf"
  pipeline.workers: 4
  pipeline.batch.size: 250
  queue.type: persisted
  queue.max_bytes: 4gb
- pipeline.id: app
  path.config: "/etc/logstash/conf.d/app.conf"
  queue.type: persisted
  dead_letter_queue.enable: true

• Un pipeline par type de flux : pannes et lenteurs ne se propagent pas d'un flux à l'autre.
• Chaînez des pipelines avec les plugins pipeline input/output pour mutualiser des traitements.
• Versionnez les fichiers et validez en CI avec --config.test_and_exit.

2. Filtres performants #

dissect, conditionnels et date.

Préférer dissect à grok #

filter {
  if [fields][type] == "nginx" {
    dissect {
      mapping => { "message" => '%{client_ip} - - [%{ts}] "%{method} %{uri} HTTP/%{http_ver}" %{status} %{bytes}' }
    }
    date   { match => ["ts", "dd/MMM/yyyy:HH:mm:ss Z"]  target => "@timestamp" }
    mutate { convert => { "status" => "integer"  "bytes" => "integer" }  remove_field => ["ts", "message"] }
  } else if [fields][type] == "app" {
    json { source => "message"  skip_on_invalid_json => true }
  }
  # grok seulement pour les formats variables, avec ancrage et motifs précis
  grok { match => { "message" => "^%{TIMESTAMP_ISO8601:ts} %{LOGLEVEL:level} %{GREEDYDATA:msg}$" }  tag_on_failure => ["_grokfail"] }
}

Bonne Pratique : dissect est nettement plus rapide que grok quand le format est fixe. Réservez grok aux formats irréguliers et ancrez toujours vos motifs (^...$).

3. Fiabilité et sécurité #

Ne rien perdre, ne rien exposer.

Sorties et DLQ #

output {
  elasticsearch {
    hosts => ["https://es01:9200", "https://es02:9200"]
    api_key => "${ES_API_KEY}"                       # secret dans le keystore Logstash
    ssl_certificate_authorities => ["/etc/logstash/ca.crt"]
    data_stream => "true"
    data_stream_type => "logs"  data_stream_dataset => "nginx"  data_stream_namespace => "prod"
  }
}
# Secrets : bin/logstash-keystore create && bin/logstash-keystore add ES_API_KEY

# Relire la dead letter queue pour corriger les rejets
input { dead_letter_queue { path => "/var/lib/logstash/dead_letter_queue"  pipeline_id => "app"  commit_offsets => true } }

Erreur Courante : Une sortie bloquée (Elasticsearch indisponible) sans file persistante fait perdre les événements ou bloque les entrées. Activez queue.type: persisted et dimensionnez queue.max_bytes.

4. Performances et supervision #

Mesurer avant de régler.

API de monitoring et réglages #

curl -s localhost:9600/_node/stats/pipelines?pretty | jq '.pipelines | to_entries[] | {id: .key, in: .value.events.in, out: .value.events.out}'
curl -s 'localhost:9600/_node/stats/jvm?pretty' | jq '.jvm.mem.heap_used_percent'
curl -s 'localhost:9600/_node/hot_threads?human=true' | head -30          # où le CPU est consommé
bin/logstash --config.test_and_exit -f /etc/logstash/conf.d/             # validation

À savoir : Quand le débit augmente fortement, placez Kafka entre les collecteurs et Logstash pour absorber les pics et rejouer les données.

Schéma de référence et vérification rapide #

Visualiser le principe et le contrôler en ligne de commande.

Schéma de référence #