Traefik : Elasticsearch, Kibana, FileBeat et Logstash pour stocker vos logs

Dans un précédent article, j'ai eu l'occasion de vous présenter une stack qui avait comme objectif de stocker les access logs de Traefik dans Elasticsearch. Notamment via l'utilisation de Filebeat pour la génération des données structurées, et enfin, pouvoir les consulter via l'interface Kibana. Cette stack était donc composée des éléments suivants :

Filebeat,
ElasticSearch,
Kibana.

Mais ceci ne permettait pas de filtrer, voire de modifier certains éléments de nos logs. Un souci si l'on souhaite suivre les recommandations de la RGPD , par exemple, en anonymisant les adresses IP collectées. Mais il existe un outil que nous pouvons ajouter à cet ensemble afin de transformer nos logs en cas de besoin : Logstash.

Logstash, c'est quoi ?

Logstash est un outil qui fait partie de la suite Elastic, il permet de collecter, transformer et d'envoyer les données vers un système de stockage ( bien souvent elasticsearch 🙂 ).

Il possède une bibliothèque de filtres intégrée qui vous permette de répondre immédiatement à vos besoins en transformant rapidement vos logs dans le but de les enrichir ( par exemple l'utilisation de la GeoIP pour ajouter les informations géographiques liées à une adresse IP ).

Il est également possible d'utiliser des filtres Grok : ces filtres vont parser vos données non structurées.

Un exemple ( tiré de la documentation officielle ) permettra de comprendre très rapidement le fonctionnement.

Voici un extrait de log provenant d'un serveur web :

55.3.244.1 GET /index.html 15824 0.043

Ce log n'est pas structuré, comment récupérer uniquement certaines informations ? ( Oui, sed ou awk font le café aussi 🤣 ). Voici le pattern que je pourrais renseigner dans logstash afin de récupérer chaque élément facilement :

%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}

Ce qui me permettra de récupérer les champs suivants :

client: 55.3.244.1
method: GET
request: /index.html
bytes: 15824
duration: 0.043

Intégrons maintenant logstash !

Intégration

Afin de permettre à tout le monde de repartir avec les mêmes informations. Voici un récapitulatif visuel du cheminement de nos logs que nous souhaitons obtenir :

Je vais également en profiter pour expliquer de nouveau l'ensemble de ma configuration.

Tout d'abord il est nécessaire de demander à Traefik d'écrire son log d'entrée dans un fichier et non sur la sortie standard : le fonctionnement par défaut. Je vais utiliser mon fichier Traefik.yaml afin d'effectuer ce changement :

accessLog:
  filePath: "/var/log/traefik/access.log"
  fields:
    defaultMode: keep
    names:
      StartUTC: drop
    headers:
      defaultMode: keep

🚩 Les modifications du champ fields sont nécessaires pour passer l'heure de notre log sur le fuseau horaire Europe/Paris 🚩

Enfin je déclare un volume qui va stocker ce fichier. Ce volume sera partagé entre Traefik et Filebeat :

version: '3.7'
services:
  traefik:
    image: traefik:2.2
    container_name: traefik
    restart: always
    ports:
      - "80:80"
      - "443:443"
    environment:
      - "TZ=Europe/Paris" 
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
      - /usr/share/zoneinfo:/usr/share/zoneinfo:ro   
      - ./traefik.yaml:/etc/traefik/traefik.yaml:ro
      - ./custom/:/etc/traefik/custom/:ro
      - traefik_log:/var/log/traefik/
      - traefik_ssl:/letsencrypt
    networks:
      - traefik

  filebeat:
    image: docker.elastic.co/beats/filebeat:7.7.0
    container_name: filebeat
    volumes:
      - traefik_log:/var/log/traefik/
      - filebeat_data:/usr/share/filebeat/data
      - ./filebeat.yml:/usr/share/filebeat/filebeat.yml:ro
    networks:
      - elastic

🚩 Je profite de ce changement dans ma stack pour passer en version 7.7.0 🚩

Filebeat va ensuite envoyer les événements à Logstash. Cette configuration se trouve dans le fichier filebeat.yml qui est monté dans le répertoire /usr/share/filbeat/ :

output.logstash:
  hosts: ["logstash:5044"]

filebeat.modules:
  - module: traefik
    access:
      enabled: true
      var.paths: ["/var/log/traefik/access.log*"]

Je vais maintenant pouvoir passer à la déclaration de mon service Logstash :

  logstash:
    image: docker.elastic.co/logstash/logstash:7.7.0
    container_name: logstash
    volumes: 
      - ./logstash.conf:/usr/share/logstash/pipeline/logstash.conf:ro
    networks:
      - elastic

Enfin voici mon fichier logstash.conf, on peut constater le stockage des données dans ma base Elasticsearch :

input {
  beats {
    port => 5044
  }
}

output {
    elasticsearch {
      hosts => "http://elasticsearch:9200"
      index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
  }
}

Voici enfin le fichier docker-compose.yaml complet, j'ajoute mon service Traefik dans le même fichier afin de simplifier le tout pour notre exemple :

version: '3.7'
services:
  traefik:
    image: traefik:2.2
    container_name: traefik
    restart: always
    ports:
      - "80:80"
      - "443:443"
    environment:
      - "TZ=Europe/Paris" 
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
      - /usr/share/zoneinfo:/usr/share/zoneinfo:ro   
      - ./traefik.yaml:/etc/traefik/traefik.yaml:ro
      - ./custom/:/etc/traefik/custom/:ro
      - traefik_log:/var/log/traefik/
      - traefik_ssl:/letsencrypt
    networks:
      - traefik

  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.7.0
    container_name: elasticsearch
    volumes:
      - es_data:/usr/share/elasticsearch/data
    environment:
      - ES_JAVA_OPTS=-Xms512m -Xmx512m
      - discovery.type=single-node
      - transport.host=localhost
      - transport.tcp.port=9300
      - http.port=9200
      - http.host=0.0.0.0
    ulimits:
      memlock:
        soft: -1
        hard: -1
    networks:
      - elastic

  kibana:
    image: docker.elastic.co/kibana/kibana:7.7.0
    container_name: kibana
    labels:
      - "traefik.enable=true"
      - "traefik.docker.network=traefik"
      - "traefik.http.routers.kibana.entrypoints=websecure"
      - "traefik.http.routers.kibana.rule=Host(`kibana.mydomain.com`)"
      - "traefik.http.services.kibana.loadbalancer.server.port=5601"
    environment:
      ELASTICSEARCH_HOSTS: 'http://ELASTICSEARCH:9200'
      ELASTICSEARCH_USERNAME: 'elastic'
      ELASTICSEARCH_PASSWORD: 'changeme'
      XPACK_MONITORING_ENABLED: 'false'
    networks:
      - elastic
      - traefik

  filebeat:
    image: docker.elastic.co/beats/filebeat:7.7.0
    container_name: filebeat
    volumes:
      - traefik_log:/var/log/traefik/
      - filebeat_data:/usr/share/filebeat/data
      - ./filebeat.yml:/usr/share/filebeat/filebeat.yml:ro
    networks:
      - elastic

  logstash:
    image: docker.elastic.co/logstash/logstash:7.7.0
    container_name: logstash
    volumes: 
      - ./logstash.conf:/usr/share/logstash/pipeline/logstash.conf:ro
    networks:
      - elastic

volumes:
  traefik_log:
  traefik_ssl:
  es_data:
  filebeat_data: 

networks:
  elastic:
  traefik:
    name: traefik

Mais si vous lancez cette stack, vous allez avoir un problème :

Le parsing réalisé par le module Traefik de Filebeat n'est pas conservé dans Logstash !

Exemple :

Les éléments contenus dans message ne sont pas extraits, notre donnée n'est donc pas structurée.

En réalité nous devons demander à Logstash de récupérer ce parsing :

When you use Filebeat modules with Logstash, you can use the ingest pipelines provided by Filebeat to parse the data. You need to load the pipelines into Elasticsearch and configure Logstash to use them.

Comme le décrit la documentation officielle, je dois également charger la pipeline dans Elasticsearch afin que Logstash puisse l'utiliser. Si vous ne le faites pas, vous allez obtenir le message suivant :

[2020-05-22T07:53:43,940][WARN ][logstash.outputs.elasticsearch][main][cc883128956b6c09f3d9268a85acffbc068fc7461cbb54b2dffd4aefc0b09930] Could not index event to Elasticsearch. {:status=>400, :action=>["index", {:_id=>nil, :_index=>"filebeat-7.7.0-2020.05.22", :routing=>nil, :_type=>"_doc", :pipeline=>"filebeat-7.7.0-traefik-access-pipeline"}, #LogStash::Event:0x741a17da], :response=>{"index"=>{"_index"=>"filebeat-7.7.0-2020.05.22", "_type"=>"_doc", "_id"=>nil, "status"=>400, "error"=>{"type"=>"illegal_argument_exception", "reason"=>"pipeline with id [filebeat-7.7.0-traefik-access-pipeline] does not exist"}}}}

🚩 La pipeline n'existe pas, il est nécessaire de la charger ! 🚩

Il va être nécessaire de modifier notre fichier filebeat.yml pour y renseigner de nouveaux éléments :

setup.kibana:
  host: "kibana:5601"

#output.logstash:
#  hosts: ["logstash:5044"]

output.elasticsearch:
  hosts: ["elasticsearch:9200"]
  indices:
    - index: "filebeat-%{[agent.version]}-%{+yyyy.MM.dd}"

filebeat.modules:
  - module: traefik
    access:
      enabled: true
      var.paths: ["/var/log/traefik/access.log*"]

Je vais effectuer un premier démarrage avec :

Ma sortie Logstash commentée,
Une sortie vers elasticsearch qui va charger la pipeline Filebeat.

Je vais arrêter mon conteneur logstash :

$ docker-compose stop filebeat

Et le relancer pour prendre en compte le nouveau fichier :

$ docker-compose start filebeat

Ensuite je vais me connecter sur mon instance afin de charger la pipeline dans Elasticsearch :

$ docker-compose exec -it filebeat sh
sh-4.2$ filebeat setup -e
....
2020-05-22T08:22:53.894Z        INFO    fileset/pipelines.go:134        Elasticsearch pipeline with ID 'filebeat-7.7.0-traefik-access-pipeline' loaded
Loaded Ingest pipelines

Une fois terminée, je coupe mon instance :

$ docker-compose stop filebeat

Je modifie de nouveau le fichier filebeat.yml afin d'envoyer mes informations à Logstash ( je commente l'output Elasticsearch ) :

setup.kibana:
  host: "kibana:5601"

output.logstash:
  hosts: ["logstash:5044"]

#output.elasticsearch:
#  hosts: ["elasticsearch:9200"]
#  indices:
#    - index: "filebeat-%{[agent.version]}-%{+yyyy.MM.dd}"

filebeat.modules:
  - module: traefik
    access:
      enabled: true
      var.paths: ["/var/log/traefik/access.log*"]

Et je relance mon instance :

$ docker-compose start filebeat

Et cette fois-ci, je peux valider sur Kibana que me logs sont bien structurés :

Nous venons de voir comment intégrer Logstash à notre traitement tout en conservant la pipeline de Filbeat ! Cet article n'est bien sûr pas techniquement parfait :

Pas de mode cluster sur notre base Elasticsearch,
Pas de protection pour la connexion à Kibana,
Les accès par défaut d'Elasticsearch ...
Pas de rotation de mon fichier de log !
Grafana ?

De plus, je n'ai toujours pas anonymisé mes journaux access logs de Traefik ! Mais cet article étant déjà conséquent, je reviendrai une autre fois sur ces quelques points.

Ce prochain article sera également l'occasion de se poser une question : Est-ce intéressant de lancer plusieurs instances de Traefik sur un même hôte ?!

En tout cas n'hésitez pas à m'apporter des remarques ou des commentaires sur Twitter ou via les commentaires ! C'est toujours un plaisir d'avoir des retours ! 😇

Ressources externes concernant le sujet :