Un monde d'octets

Ayant mis les logs de container docker en json, j'ai installé filebeat pour shipper directement ceux-ci à elasticsearch. Seul bémol, c'est que filebeat met comme timestamp source, l'heure à laquelle il à récolté (harvest) les logs.

Pour corriger ceci, il faut passer par un process d'ingest spécifique qui modifiera le document.

Nous créons notre ingest avec la commande suivante :

curl -XPUT pvalois:xxxxxxxxx@10.10.31.60:9200/_ingest/pipeline/json_timestamp \
-H "Content-type: application/json" \
-d '{
    "processors": [
      {
        "set": {
          "field": "_source.@timestamp",
          "value": "{{json.time}}",
	  "ignore_failure": true
        }
      }
    ]
}'

Comme ou le voir celle-ci crée un ingest préprocesseur nommé "json_timestamp" qui copie le content de json.time dans _source.@timestamp, corrigeant ainsi l'heure de réception du message.

Cependant la création de cet ingest ne suffit pas, il faut notifier a filebeat de l'utiliser lors de la communication avec elasticsearch. Pour cela, nous configurons le /etc/filebeat/filebeat.yml comme suit :

output.elasticsearch:
  hosts: [pvalois:xxxxxxxx@10.10.31.60:9200]
  pipeline: json_timestamp

Et voila, nos messages envoyés via filebeat seront maintenant livré à notre processor json.timestamp et tout sera propre à l'importation, même si celle-ci est faite à posteriori.

J'ai enfin écrit une configuration de docker qui me donne un résultat intéressant dans elasticsearch :

{
  "log-driver": "syslog",
  "log-opts":{
     "syslog-facility":"local6"
  },
  "live-restore": true,
  "metrics-addr" : "0.0.0.0:9323",
  "experimental" : true
}

Grace à cette configuration les logs des containers sont gratifié de la facility local6, et transittent par le syslog local. En paramétrant ce syslog avec la template json qu'il faut, et l'envoi a logstash, les logs seront envoyé a ES, et filtrable par la facilité. Mieux encore, il serait surement possible de passer par les mutate de logstash pour faire d'autres transformation, sur la simple reconnaissance de la facility.

Pour maintenir vos logs dans Elasticsearch, Elastic à développé un outil qui s'appelle curator

Installation

pip install curator

Configuration

Créer un fichier ~/.curator/curator.yml contenant :

client:
  hosts:
    - localhost
  port: 9200
  url_prefix:
  use_ssl: False
  certificate:
  client_cert:
  client_key:
  ssl_no_validate: False
  http_auth:
  timeout: 100
  master_only: False
logging:
  loglevel: INFO
  logfile:
  logformat: default
  blacklist: [‘elasticsearch’]

Paramêtrage d'une action

Voici une action delete.yml qui supprime tout les index logstash-* de plus de 7 jours :

actions:
 
  1:
    action: delete_indices
    description: >-
      Delete indices older than 7 days (based on index name).
    options:
      ignore_empty_list: True
      timeout_override:
      continue_if_exception: False
      disable_action: False
    filters:
    - filtertype: pattern
      kind: prefix
      value: logstash-
    - filtertype: age
      source: name
      direction: older
      timestring: '%Y.%m.%d'
      unit_count: 7
      unit: days

Cette action s'utilise comme suit :

curator delete.yml --dry-un
curator delete.yml

Pour installer Filebeat :

curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-5.5.2-amd64.deb
sudo dpkg -i filebeat-5.5.2-amd64.deb

Pour ajouter un input filebeat dans logstash, il suffit d'ajouter a la section input le segment suivant :

beats {
host => "mylogstashserver.mydomain"
port => 5044
type => "beat"
}

et avoir un bloc output qui accepte beats. J'ai simplifié mon output pour ne plus avoir de "if type";

output {
elasticsearch {
hosts => [ "127.0.0.1:9200" ]
}
}

Nous pouvons aussi segmenter la confiduration de logstash comme pour rsyslog. Dans mon cas, j'ai fait les fichiers suivants :

01-input.conf

, qui précise les inputs rsyslog et beat

10-ssh.conf

, qui contient un bloc filter pour groker les logs ssh et permettre leurs étiquetage et la géolicalisation du résultat

20-apache.conf

, qui contient un bloc filter pour groker les logs apache et permettre leurs étiquetage et la géolicalisation du résultat

30-outputs.conf

, qui définit que tous les logs sont envoyés au serveur ElasticSearch.

Voici la documentation d'installation d'un triplet Elk/Rsyslog/Logstash afin d'indexer des logs pour des requêtes plus efficaces.

Installation de Logstash et Elasticsearch

wget -qO - https://packages.elastic.co/GPG-KEY-elasticsearch | sudo apt-key add -
echo "deb http://artifacts.elastic.co/packages/5.x/apt stable main" > /etc/apt/sources.list.d/elk.list
apt update
apt install elasticsearch logstash

Ajoutez les lignes suivantes au fichier /etc/logstash/conf.d/logstash.conf

# This input block will listen on port 10514 for logs to come in.
# host should be an IP on the Logstash server.
# codec => "json" indicates that we expect the lines we're receiving to be in JSON format
# type => "rsyslog" is an optional identifier to help identify messaging streams in the pipeline.

input {
  udp {
    host => "127.0.0.1"
    port => 10514
    codec => "json"
    type => "rsyslog"
  }
}

# This is an empty filter block.  You can later add other filters here to further process
# your log lines

filter { }

# This output block will send all events of type "rsyslog" to Elasticsearch at the configured
# host and port into daily indices of the pattern, "rsyslog-YYYY.MM.DD"

output {
  if [type] == "rsyslog" {
    elasticsearch {
      hosts => [ "127.0.0.1:9200" ]
    }
  }
}

Maintenant nous allons configurer rsyslog pour envoyer a logstash :

Créer un fichier /etc/rsyslogd/01-json-template contenant :

template(name="json-template"
  type="list") {
    constant(value="{")
      constant(value="\"@timestamp\":\"")     property(name="timereported" dateFormat="rfc3339")
      constant(value="\",\"@version\":\"1")
      constant(value="\",\"message\":\"")     property(name="msg" format="json")
      constant(value="\",\"sysloghost\":\"")  property(name="hostname")
      constant(value="\",\"severity\":\"")    property(name="syslogseverity-text")
      constant(value="\",\"facility\":\"")    property(name="syslogfacility-text")
      constant(value="\",\"programname\":\"") property(name="programname")
      constant(value="\",\"procid\":\"")      property(name="procid")
    constant(value="\"}\n")
}

Puis indiquer a rsyslog qu'il faut envoyer en utilisant le template json a logstash :

Editer /etc/rsyslog.d/50-default et ajoutez la ligne :

*.*                         @127.0.0.1:10514;json-template

Lancer/Relancer les services :

service elasticsearch start
service logstash start
service rsyslog restart

Maintenant, Kibana !

apt install kibana
service kibana start

Consulter les liens :

Liste des indexes d'ElasticSearch

http://localhost:9200/_cat/indices?v

Kibana

http://localhost:5601/

Bien sur, si vous hébergez votre Elk + Kibana sur une autre machine, il faut changer l'ip dans /etc/rsyslog.d/50-default et localhost dans les urls de consultation et d'interrogation.

Nous avons aussi la possibilité d'avoir un rsyslog local à la machine qui héberge logstash, Elk, et kibana, ou que chaque machine envoie ses logs directement à logstash, ou encore que logstash soit local a chaque machine et envoie a un Elk distant. Multiples combinaisons selon vos besoins architecturaux.