Ce guide illustre la mise en place d’une chaîne de collecet et de traitement de logs applicatifs. Filebeat assure la capture des fichiers, Redis joue le rôle de tampon distribué, Logstash parse et enrichit les événements, et Elasticsearch indexe les données pour la recherche.

Configuration de Filebeat

# Entrées de collecte
filebeat.inputs:
- type: log
  paths:
    - /var/log/application/service-a/*.log
  encoding: utf-8
  document_type: app_service_a
  multiline.pattern: '^\d{4}-\d{2}-\d{2}'
  multiline.negate: true
  multiline.match: after

filebeat.registry_file: ${path.data}/registry-app

# Sortie vers Redis avec équilibrage de charge
output.redis:
  hosts: ["redis1.exemple.com:6379", "redis2.exemple.com:6379", "redis3.exemple.com:6379"]
  key: canal-logs
  datatype: list
  loadbalance: true

Configuration d'Elasticsearch

cluster.name: cluster-production
node.name: es-node-b
bootstrap.memory_lock: true
network.host: 10.0.1.12
http.port: 9200
transport.tcp.port: 9300
discovery.zen.ping.unicast.hosts: ["10.0.1.11", "10.0.1.12", "10.0.1.13"]
discovery.zen.minimum_master_nodes: 2
http.cors.enabled: true
http.cors.allow-origin: "*"

Configuraton de Logstash

input {
  redis {
    data_type => "list"
    key => "canal-logs"
    host => "redis1.exemple.com"
    port => 6379
  }
  redis {
    data_type => "list"
    key => "canal-logs"
    host => "redis2.exemple.com"
    port => 6379
  }
  redis {
    data_type => "list"
    key => "canal-logs"
    host => "redis3.exemple.com"
    port => 6379
  }
}

filter {
  if [type] == "app_service_a" {
    grok {
      match => {
        "message" => "%{TIMESTAMP_ISO8601:horodatage}\s+\[%{DATA:thread}\]\s+%{DATA:niveau}\s+%{DATA:classe}\s+-\s+\[%{DATA:id_metier}\]%{DATA:source},%{DATA:cible},%{DATA:api},%{DATA:succes},%{DATA:duree},%{GREEDYDATA:erreur}"
      }
    }
    mutate {
      convert => { "duree" => "integer" }
    }
  } else if [type] == "app_service_b" {
    grok {
      match => {
        "message" => "%{TIMESTAMP_ISO8601:horodatage}\s+\[%{DATA:thread}\]\s+%{DATA:niveau}\s+%{DATA:classe}\s+"
      }
    }
  } else {
    grok {
      match => {
        "message" => "%{TIMESTAMP_ISO8601:horodatage}\s+%{DATA:thread}\s+%{DATA:niveau}\s+%{DATA:classe}\s+"
      }
    }
  }
  date {
    match => ["horodatage", "ISO8601"]
    target => "@timestamp"
  }
}

output {
  elasticsearch {
    hosts => ["es1.exemple.com:9200", "es2.exemple.com:9200", "es3.exemple.com:9200"]
    index => "%{type}-%{+YYYY.MM.dd}"
    template_name => "modele_logstash"
    pool_max_per_route => 300
    flush_size => 2000
    idle_flush_time => 5
  }
}