View on GitHub

Workshop Logstash/ElasticSearch/Kibana

Analyse en temps réel de vos logs applicatifs avec Logstash/Elasticsearch/Kibana

Dans ce workshop, vous aurez à votre disposition deux instances amazon :

Toutes les machines utilisent la même clef ssh, et ont le même login ubuntu. Dans le répertoire /home/ubuntu/tools, vous trouverez l'ensemble des outils nécéssaires durant le workshop, c'est à dire :

Pour vous connecter aux machines, récupérer le fichier kibana.pem ou le fichier kibana.ppk pour les utilisateurs Windows, dans votre répertoire courant (la clef sera révoquée après le workshop). N'oubliez pas de changer les permissions sur le fichier :

$ chmod 600 kibana.pem

Pour tester l'installation, et vous connecter à la première instance, taper la commande :

$ ssh -i kibana.pem ubuntu@io-0-kibana.aws.xebiatechevent.info

Logstash

Logstash est un pipe permettant de collecter, parser, et stocker des logs à l'aide d'entrées, de filtres et de sorties (input, filter, output). La phase de parsing permet d'ajouter de la sémantique à notre événement, en ajoutant, modifiant ou supprimant des champs, des tags, des types, etc...

Dans cette première partie de l'atelier, nous allons donc découvrir Logstash et le configurer pour structurer nos logs afin qu'ils soient facilement exploitables par la suite.

Découverte de Logstash

Commencer tout d'abord par créer le répertoire ~/workshop.

Copier le jar Logstash disponible dans le répertoire ~/tools dans le répertoire ~/workshop

Dans le répertoire ~/workshop, créer un fichier de configuration Logstash nommé logstash-logback.conf

input {
  stdin { } 
}

output {
  stdout { debug => true }
}

Vous pouvez maintenant exécuter Logstash grâce à la commande suivante:

$ java -jar logstash-1.2.1-flatjar.jar agent -f logstash-logback.conf

Logstash est prêt à interpréter ce qu'il recevera sur l'entrée standard. Pour un premier test, passez lui la ligne suivante en entrée:

02-10-2013 14:26:27.724 [pool-10-thread-1] INFO com.github.vspiewak.loggenerator.SearchRequest - id=9205,ip=217.109.49.180,cat=TSHIRT

Vous devez normalement voir un message de la forme suivante s'afficher:

{
    "message" => "02-10-2013 14:26:27.724 [pool-10-thread-1] INFO com.github.vspiewak.loggenerator.SearchRequest - id=9205,ip=217.109.49.180,cat=TSHIRT",
    "@timestamp" => "2013-10-25T07:50:26.232Z",
    "@version" => "1",
    "host" => "lucid64"
}

On constate que le timestamp enregistré n'est pas du tout lié à celui de la ligne de log. Logstash n'a pas pris en compte la date incluse dans notre log, il va falloir configurer entre autres un filtre pour cela.

Ajouter de la sémantique

Pour configurer Logstash afin qu'il puisse interpréter les données qu'il va recevoir en entrée, nous allons configurer plusieurs filtres.

Logstash execute le fichier de configuration en prenant en compte l'ordre de déclaration de vos filtres.

Filtre Grok

Le filtre Grok met à votre disposition plusieurs patterns pour parser les lignes de logs.

Ressources:

Dans un premier temps, nous voulons juste parser le niveau de log. Pour celà, le pattern LOGLEVEL va nous être utile.

filter {
   grok {
      match => ["message","%{LOGLEVEL:log_level}"]
   }
}

Vous pouvez relancer Logstash et repasser la ligne de log:

02-10-2013 14:26:27.724 [pool-10-thread-1] INFO com.github.vspiewak.loggenerator.SearchRequest - id=9205,ip=217.109.49.180,cat=TSHIRT

Cette fois, on remarque qu'un élément a été analysé, l'élément log_level

{
    "message" => "02-10-2013 14:26:27.724 [pool-10-thread-1] INFO com.github.vspiewak.loggenerator.SearchRequest - id=9205,ip=217.109.49.180,cat=TSHIRT",
    "@timestamp" => "2013-10-27T08:46:51.132Z",
    "@version" => "1",
    "host" => "lucid64",
    "log_level" => "INFO"
}

Nous allons enrichir notre filtre pour parser le reste des lignes:

filter {
   grok {
      match => ["message","(?<log_date>%{MONTHDAY}-%{MONTHNUM}-%{YEAR} %{HOUR}:%{MINUTE}:%{SECOND}.[0-9]{3}) \[%{NOTSPACE:thread}\] %{LOGLEVEL:log_level} %{NOTSPACE:classname} - %{GREEDYDATA:msg}"]
   }
}

De nouveau, nous pouvons relancer Logstash et lui passer notre ligne de log, cette fois, nous allons avoir un résultat de la forme suivante:

{
    "message" => "02-10-2013 14:26:27.724 [pool-10-thread-1] INFO com.github.vspiewak.loggenerator.SearchRequest - id=9205,ip=217.109.49.180,cat=TSHIRT",
    "@timestamp" => "2013-10-27T10:30:32.242Z",
    "@version" => "1",
    "host" => "lucid64",
    "log_date" => "02-10-2013 14:26:27.724",
    "thread" => "pool-10-thread-1",
    "log_level" => "INFO",
    "classname" => "com.github.vspiewak.loggenerator.SearchRequest",
    "msg" => "id=9205,ip=217.109.49.180,cat=TSHIRT"
}

Pour alléger la configuration, nous allons extraire l'expression régulière de la date et la définir en tant que pattern:

Indiquez ensuite à Grok le dossier contenant vos fichiers de patterns via l'attribut "patterns_dir".

La configuration Losgstash devient:

filter { 
   grok {
      patterns_dir => "./patterns"
      match => [ "message", "%{LOG_DATE:log_date} \[%{NOTSPACE:thread}\] %{LOGLEVEL:log_level} %{NOTSPACE:classname} - %{GREEDYDATA:msg}"]
   }
}

À ce niveau, nous constatons que l'intégralité du message a été parsé et est maintenant interprété. Par contre, même si nous avons correctement interprété la date du log, le champ @timestamp contient toujours la date de lecture par Logstash. Il serait plus intéressant de mettre dans ce champ la date de log. Pour cela, il va falloir utiliser un autre type de filtre, le filtre date.

Filtre date

Ressources:

Le filtre date est l'un des filtres les plus important. Il permet de préciser quelle date utiliser pour l'événement généré et alimentant le champ @timestamp. Rajoutez le filtre dans le fichier de configuration:

filter {
    [...]
    date {
        match => ["log_date","dd-MM-YYYY HH:mm:ss.SSS"]
    }
}

En redonnant notre ligne de log en entrée, nous récupérons un retour de la forme suivante:

{
    "message" => "02-10-2013 14:26:27.724 [pool-10-thread-1] INFO com.github.vspiewak.loggenerator.SearchRequest - id=9205,ip=217.109.49.180,cat=TSHIRT",
    "@timestamp" => "2013-10-02T12:26:27.724Z",
    "@version" => "1",
    "host" => "lucid64",
    "log_date" => "02-10-2013 14:26:27.724",
    "thread" => "pool-10-thread-1",
    "log_level" => "INFO",
    "classname" => "com.github.vspiewak.loggenerator.SearchRequest",
    "msg" => "id=9205,ip=217.109.49.180,cat=TSHIRT"
}

Logstash normalise les dates au format UTC automatiquement.

Filtre kv

Le filtre kv s'avère très utile lorsque vous voulez parser un champ de type foo=bar comme par exemple une requête HTTP. Ajoutez le filtre kv pour notre example:

filter {
    [...]
    kv {
      field_split => ","
      source => "msg"
    }
}

Relancez Logstash et passez lui en entrée la ligne suivante:

08-10-2013 16:33:49.629 [pool-1-thread-1] INFO com.github.vspiewak.loggenerator.SearchRequest - id=41,ip=157.55.34.94,brand=Apple,name=iPhone 5C,model=iPhone 5C - Blanc - Disque 16Go,category=Mobile,color=Blanc,options=Disque 16Go,price=599.0

Vous devez obtenir le résultat suivant:

{
    "message" => "08-10-2013 16:33:49.629 [pool-1-thread-1] INFO com.github.vspiewak.loggenerator.SearchRequest - id=41,ip=157.55.34.94,brand=Apple,name=iPhone 5C,model=iPhone 5C - Blanc - Disque 16Go,category=Mobile,color=Blanc,options=Disque 16Go,price=599.0",
    "@timestamp" => "2013-10-08T14:33:49.629Z",
    "@version" => "1",
    "host" => "lucid64",
    "log_date" => "08-10-2013 16:33:49.629",
    "thread" => "pool-1-thread-1",
    "log_level" => "INFO",
    "classname" => "com.github.vspiewak.loggenerator.SearchRequest",
    "msg" => "id=41,ip=157.55.34.94,brand=Apple,name=iPhone 5C,model=iPhone 5C - Blanc - Disque 16Go,category=Mobile,color=Blanc,options=Disque 16Go,price=599.0",
    "id" => "41",
    "ip" => "157.55.34.94",
    "brand" => "Apple",
    "name" => "iPhone 5C",
    "model" => "iPhone 5C - Blanc - Disque 16Go",
    "category" => "Mobile",
    "color" => "Blanc",
    "options" => "Disque 16Go",
    "price" => "599.0"
}
Logstash parse maintenant notre ligne de vente et ajoute automatiquement les champs category, brand, name, model, color, options et price.

Filtre GeoIP

Le filtre geoip permet d'ajouter des informations de géolocalisation via une adresse ip (ou hostname). Logstash utilise la base de donnée GeoCityLite de Maxmind sous license CCA-ShareAlike 3.0.

Nous allons utiliser une version de GeoCity téléchargée au préalable sur le site Maxmind plutôt que la version embarquée dans Logstash. Copiez le fichier ~/tools/GeoLiteCity.dat dans ~/workshop et rajoutez le filtre dans la configuration Logstash:
filter {
    [...]

    geoip {
        source => "ip"
        database => "./GeoLiteCity.dat"
    }
}

Filtre mutate

Le filtre mutate est un filtre "couteaux suisses" permettant une multitude de modifications.

Ajout de tag

Nous allons ajouter un tag à nos logs afin de différencier les recherches des ventes:

filter {
    [...]
    if [classname] =~ /SellRequest/ {
        mutate { add_tag => "sell" }
    } else if [classname] =~ /SearchRequest$/ {
        mutate { add_tag => "search" }
    }
}

Relancez Logstash et passez lui en entrée la ligne suivante:

08-10-2013 16:33:49.629 [pool-1-thread-1] INFO com.github.vspiewak.loggenerator.SearchRequest - id=41,ip=157.55.34.94,brand=Apple,name=iPhone 5C,model=iPhone 5C - Blanc - Disque 16Go,category=Mobile,color=Blanc,options=Disque 16Go,price=599.0
08-10-2013 16:33:49.629 [pool-1-thread-1] INFO com.github.vspiewak.loggenerator.SellRequest - id=41,ip=157.55.34.94,brand=Apple,name=iPhone 5C,model=iPhone 5C - Blanc - Disque 16Go,category=Mobile,color=Blanc,options=Disque 16Go,price=599.0

Notez au passage que Logstash permet l'utilisation de conditions, pour en savoir plus:

Conversion de type

Le filtre mutate permet de convertir certains champs en entier, flottant ou string. Nous ajoutons à notre configuration la conversion des champs id et price:

filter {
    [...]
    mutate {
        convert => [ "id", "integer" ]
        convert => [ "price", "float" ]
    }   
}

Suppression d'un champ

Toujours avec le filtre mutate, nous allons supprimer le champ "msg". Nous avons en effet parsé ce champ avec le filtre kv et n'avons plus besoin de ce doublon d'information.

filter {
    [...] 
    mutate {
        [...]
        remove_field => [ "msg" ]  
    }  
}

Split d'un champ

Pour finir avec le filtre mutate, nous allons splité notre champ "options" afin d'avoir un tableau d'options.

filter {
    [...]
    mutate {
        [...]
       split => [ "options", "|" ]
    }
}

GeoIP et Bettermap

Le panel Bettermap de Kibana requiert un champ contenant les coordonnées GPS au format Geo_JSON (i.e: un tableau de deux float au format: [ longitude, latitude ]).

Ajoutez un champ "geoip.lnglat" contenant le tableau de coordonnées via le "hack" suivant:

filter {
    [...]
    # geoip.lnglat contiendra le point geo_json, 'tmplat' contient la latitude (temporaire)
    # les deux champs sont de type string.
    mutate {
        add_field => [ "[geoip][lnglat]", "%{[geoip][longitude]}", "tmplat", "%{[geoip][latitude]}" ]
    }

    # merge du champ tmplat dans geoip.lnglat. 
    # le champ geoip.lnglat devient un tableau de string
    mutate {
        merge => [ "[geoip][lnglat]", "tmplat" ]
    }

    # conversion du tableau de string en float
    # suppression du champ tmplat
    mutate {
        convert => [ "[geoip][lnglat]", "float" ]
        remove_field => [ "tmplat" ]
    }
}

Résultat final

input {
  stdin { } 
}

filter {
    grok {
        patterns_dir => "./patterns"
        match => ["message","%{LOG_DATE:log_date} \[%{NOTSPACE:thread}\] %{LOGLEVEL:log_level} %{NOTSPACE:classname} - %{GREEDYDATA:msg}"]
    }

    date {
        #timezone => "UTC"
        match => ["log_date","dd-MM-YYYY HH:mm:ss.SSS"]
    }

    kv {
        field_split => ","
        source => "msg"
    }

    geoip {
        source => "ip"
        database => "./GeoLiteCity.dat"
    }

    if [classname] =~ /SellRequest/ {
        mutate { add_tag => "sell" }
    } else if [classname] =~ /SearchRequest$/ {
        mutate { add_tag => "search" }
    }

    mutate {
        convert => [ "id", "integer" ]
        convert => [ "price", "float" ]
        remove_field => [ "msg" ]
        split => [ "options", "|" ]
    }

    # hack pour Bettermap panel de Kibana
    mutate {
        add_field => [ "[geoip][lnglat]", "%{[geoip][longitude]}", "tmplat", "%{[geoip][latitude]}" ]
    }

    mutate {
        merge => [ "[geoip][lnglat]", "tmplat" ]
    }

    mutate {
        convert => [ "[geoip][lnglat]", "float" ]
        remove_field => [ "tmplat" ]
    }

}

output {
    stdout { debug => true }
}

Lancer la génération de log

Maintenant que Logstash est capable d'analyser nos logs, nous allons lancer notre générateur.

Création du dossier contenant les futurs logs

$ mkdir /tmp/logstash

Modifier l'entrée de Logstash

Afin que Logstash analyse tous les fichiers de log contenu dans le dossier précédemment créé.

input {
    file {
        path => "/tmp/logstash/*.log"
    }
}

Lancez le générateur de log

$ java -jar log-generator.jar -n 10 -r 1000 > /tmp/logstash/workshop.log &

ElasticSearch

Maintenant que Logstash est configuré pour parser nos logs et les transformer dans un format convenable, nous allons stocker ces logs dans ElasticSearch.

Configuration d'ElasticSearch

Connectez vous à la vm io-1-kibana.aws.xebiatechevent.info et effectuez les opérations suivantes :

$ ssh -i kibana.pem ubuntu@io-1-kibana.aws.xebiatechevent.info

Template de mapping ElasticSearch pour les index Logstash

Lancez ElasticSearch après avoir installé le plugin head:

$ ~/workshop/elasticsearch-0.90.5/bin/plugin -install mobz/elasticsearch-head
$ ~/workshop/elasticsearch-0.90.5/bin/elasticsearch

Une fois elasticsearch lancé, vous pouvez visualiser facilement votre cluster Elasticsearch via l'url: http://io-1-kibana.aws.xebiatechevent.info:9200/_plugin/head

Ajoutez le template suivante afin d'utiliser l'analyser "keyword" pour les champs "ip", name", "model", "options" et "email":

curl -XPUT http://localhost:9200/_template/logstash_per_index -d '{
    "template" : "logstash*",
    "mappings" : {
        "_default_" : {
           "_all" : {"enabled" : false},
           "properties" : {
              "@timestamp": { "type": "date", "index": "not_analyzed" },
              "tags": { "type": "string", "index": "not_analyzed" },
              "ip": { "type" : "ip", "analyzer": "keyword", "index": "analyzed" },
              "name": { "type" : "string", "analyzer": "keyword", "index": "analyzed" },
              "model": { "type" : "string", "analyzer": "keyword", "index": "analyzed" },
              "options": { "type" : "string", "analyzer": "keyword", "index": "analyzed" },
              "email": { "type" : "string", "analyzer": "keyword", "index": "analyzed" }
 
            }
        }
   }
}
'

Ce mapping vous permettra de ne pas avoir décueils lors de la construction de votre dashboard Kibana.

Branchement de Logstash avec ElasticSearch

ElasticSearch est maintenant configuré. Nous allons donc configurer Logstash pour qu'il envoit les logs analysés dans le moteur de recherche.

Modification de la sortie de Logstash:

output {
    elasticsearch {
        host => "io-1-kibana.aws.xebiatechevent.info"
    }
}

C'est tout ce qu'il y a à faire. Vous pouvez maintenant redémarrer Logstash et pour vérifier qu'ElasticSearch est bien alimenté, retourner sur la vm d'ElasticSearch et lancez les deux commandes suivantes.

Pour lister les index:

$ curl -s http://127.0.0.1:9200/_status?pretty=true | grep logstash

Pour voir le nombre de documents indexés:

$ curl -gs -XGET "http://localhost:9200/logstash-*/_count"

Kibana

Installation

Installation de Kibana dans Apache:

cd /var/www
sudo unzip /home/ubuntu/tools/kibana-latest.zip 
sudo mv kibana-latest kibana
sudo /etc/init.d/apache2 restart

Vous devriez pouvoir acceder au dashboard via l'url: io-1-kibana.aws.xebiatechevent.info/kibana

Dashboard Logstash

Kibana vous propose un dashboard pré-configuré si vos données viennent de Logstash/ES. Vous pouvez y acceder via l'url: io-1-kibana.aws.xebiatechevent.info/kibana/index.html#/dashboard/file/logstash.json.

Dashboard GeekShop

Nous allons créer un dashboard personnalisé pour notre boutique en ligne. Commencez avec un dashboard vide accessible à l'adresse: io-1-kibana.aws.xebiatechevent.info/kibana/index.html#/dashboard/file/blank.json. N'oubliez pas de sauvegarder régulièrement votre dashboard dans Elasticsearch grâce au menu situé en haut à droite.

Configuration générale

Cliquez sur la roue crantée tout en haut à droite afin de faire apparaître le menu "Dashboard Settings":

Configuration des queries

Ajoutez des barres de recherches en cliquant sur l'icône "+".

Entrez les queries suivantes: