12 - Filtres Date et GeoIP : temps et geolocalisation

Ce que tu vas apprendre

• Parser des dates dans n'importe quel format avec le filtre date
• Comprendre la différence entre @timestamp et un champ date custom
• Gerer les fuseaux horaires
• Ajouter la geolocalisation à partir d'une adresse IP avec GeoIP
• Monter la base GeoLite2 dans Docker

Prerequisites

• Savoir utiliser les filtres de base (voir articles 09 a 11)

Le filtre Date

Pourquoi @timestamp est si important

Le champ @timestamp est le champ temporel par défaut dans l'Elastic Stack. C'est celui que Kibana utilise pour la timeline, les filtres par date, les histogrammes. Sans un @timestamp correct, tes donnees sont inutilisables dans Kibana.

Par défaut, @timestamp contient le moment ou Logstash a reçu l'événement, pas le moment ou l'événement s'est produit. Si tu traites des logs avec 5 minutes de retard, le @timestamp a 5 minutes de decalage.

Le filtre date resout ca : il parse le timestamp de tes logs et le met dans @timestamp.

Configuration de base

filter {
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:log_date} %{LOGLEVEL:level} %{GREEDYDATA:msg}" }
  }

  date {
    match => ["log_date", "ISO8601"]
    target => "@timestamp"
    remove_field => ["log_date"]
  }
}

Le filtre date prend le champ log_date (extrait par Grok), le parse avec le format ISO8601, et ecrase @timestamp. Le champ log_date est supprime (plus besoin du string, on a le timestamp propre).

Les formats de date

Le paramètre match prend le nom du champ et un ou plusieurs formats. Logstash essaie les formats dans l'ordre.

Les tokens du format :

Plusieurs formats pour le meme champ

Si tes logs viennent de plusieurs sources avec des formats différents :

filter {
  date {
    match => [
      "log_date",
      "ISO8601",
      "yyyy-MM-dd HH:mm:ss",
      "dd/MMM/yyyy:HH:mm:ss Z",
      "UNIX"
    ]
  }
}

Logstash essaie chaque format dans l'ordre. Le premier qui matche gagne.

Gerer les fuseaux horaires

Si tes logs n'ont pas de timezone dans le timestamp, Logstash suppose UTC. C'est rarement ce que tu veux. Un serveur en France écrit 2026-03-31 14:23:01 en heure locale (CEST, UTC+2).

filter {
  date {
    match => ["log_date", "yyyy-MM-dd HH:mm:ss"]
    timezone => "Europe/Paris"
    target => "@timestamp"
  }
}

timezone dit a Logstash : "ce timestamp est en heure de Paris". Logstash convertit en UTC pour le stocker dans @timestamp. Dans Kibana, tu verras l'heure convertie selon le fuseau du navigateur.

Stocker dans un champ custom

Par défaut, target => "@timestamp". Mais si tu veux garder le timestamp original et en parser un deuxieme :

filter {
  date {
    match => ["created_at", "yyyy-MM-dd HH:mm:ss"]
    target => "created_timestamp"
  }
}

Le champ created_timestamp contiendra un objet date (pas un string), utilisable pour les calculs de duree et les aggregations par date dans Elasticsearch.

L'erreur classique : _dateparsefailure

Si le format ne matche pas, Logstash ajoute le tag _dateparsefailure. Les raisons les plus courantes :

• Le format ne correspond pas (tu as mis yyyy mais la date a 2 chiffres)
• Le champ contient des espaces en début ou fin (utilise mutate { strip => ["log_date"] } avant)
• Le champ est vide ou absent sur certains événements

filter {
  date {
    match => ["log_date", "yyyy-MM-dd HH:mm:ss"]
    tag_on_failure => ["_dateparsefailure"]
  }

  if "_dateparsefailure" in [tags] {
    mutate {
      add_field => { "date_parse_error" => "true" }
    }
  }
}

Le filtre GeoIP

A quoi ca sert

GeoIP prend une adresse IP et renvoie la localisation geographique : pays, ville, latitude, longitude, ASN (fournisseur d'acces). C'est ce qui te permet de faire des cartes dans Kibana.

Configuration de base

filter {
  geoip {
    source => "client_ip"
  }
}

Résultat ajoute a l'événement :

json{
  "geoip": {
    "country_code2": "FR",
    "country_code3": "FRA",
    "country_name": "France",
    "continent_code": "EU",
    "region_name": "Brittany",
    "city_name": "Vannes",
    "postal_code": "56000",
    "latitude": 47.6559,
    "longitude": -2.7603,
    "timezone": "Europe/Paris",
    "location": {
      "lon": -2.7603,
      "lat": 47.6559
    }
  }
}

Le champ geoip.location est au format geo_point d'Elasticsearch. Kibana sait le placer sur une carte automatiquement.

La base GeoLite2

GeoIP utilise la base de donnees GeoLite2 de MaxMind. Logstash 8.x inclut une version de la base dans l'image Docker. Mais cette base vieillit. Les IPs changent de proprietaire, les FAI changent.

Pour avoir une base à jour, créé un compte gratuit sur MaxMind et telecharge les bases mises à jour.

logstash-lab/
├── geoip/
│   ├── GeoLite2-City.mmdb
│   └── GeoLite2-ASN.mmdb

Monte le dossier dans Docker :

yamllogstash:
  volumes:
    - ./geoip/:/usr/share/logstash/geoip/:ro

Et référencé la base dans le pipeline :

filter {
  geoip {
    source => "client_ip"
    database => "/usr/share/logstash/geoip/GeoLite2-City.mmdb"
  }
}

Limiter les champs GeoIP

GeoIP ajoute beaucoup de champs. Si tu n'as besoin que du pays et des coordonnees :

filter {
  geoip {
    source => "client_ip"
    fields => ["country_name", "country_code2", "city_name", "location"]
  }
}

Ca economise du stockage dans Elasticsearch.

Ajouter l'ASN (fournisseur d'acces)

La base ASN est séparée de la base City :

filter {
  geoip {
    source => "client_ip"
    database => "/usr/share/logstash/geoip/GeoLite2-ASN.mmdb"
    target => "asn"
  }
}

Résultat :

json{
  "asn": {
    "asn": 3215,
    "as_org": "Orange S.A."
  }
}

Le target séparé les champs ASN des champs City pour éviter les collisions.

Gerer les IPs privees

Les IPs privees (192.168.x.x, 10.x.x.x, 172.16-31.x.x) et localhost (127.0.0.1) ne sont pas dans la base GeoIP. Logstash ajoute le tag _geoip_lookup_failure.

filter {
  geoip {
    source => "client_ip"
    tag_on_failure => ["_geoip_lookup_failure"]
  }

  # Ne pas s'affoler sur les IPs privees
  if "_geoip_lookup_failure" in [tags] and [client_ip] =~ /^(10\.|172\.(1[6-9]|2|3[01])\.|192\.168\.)/ {
    mutate {
      remove_tag => ["_geoip_lookup_failure"]
      add_field => { "[geoip][country_name]" => "Private Network" }
    }
  }
}

Cas pratique : pipeline complet logs Nginx

Un pipeline qui parse des logs Nginx, corrige le timestamp et geolocalise les visiteurs :

input {
  file {
    path => "/data/nginx-access.log"
    start_position => "beginning"
    sincedb_path => "/dev/null"
  }
}

filter {
  # 1. Parser le log Nginx
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }

  # 2. Corriger le timestamp
  date {
    match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
    target => "@timestamp"
    remove_field => ["timestamp"]
  }

  # 3. Convertir les types
  mutate {
    convert => {
      "response" => "integer"
      "bytes" => "integer"
    }
  }

  # 4. Geolocaliser l'IP
  geoip {
    source => "clientip"
    fields => ["country_name", "city_name", "location"]
  }

  # 5. Ajouter l'ASN
  geoip {
    source => "clientip"
    database => "/usr/share/logstash/geoip/GeoLite2-ASN.mmdb"
    target => "asn"
  }

  # 6. Parser le user agent
  useragent {
    source => "agent"
    target => "user_agent"
    remove_field => ["agent"]
  }

  # 7. Nettoyer
  mutate {
    remove_field => ["message", "@version", "ident"]
  }
}

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

Ligne d'entree :

83.156.42.123 - - [31/Mar/2026:14:23:01 +0200] "GET /api/products HTTP/1.1" 200 4523 "https://paltemps.fr" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)"

Événement final dans Elasticsearch :

json{
  "@timestamp": "2026-03-31T12:23:01.000Z",
  "clientip": "83.156.42.123",
  "auth": "-",
  "verb": "GET",
  "request": "/api/products",
  "httpversion": "1.1",
  "response": 200,
  "bytes": 4523,
  "referrer": "\"https://paltemps.fr\"",
  "geoip": {
    "country_name": "France",
    "city_name": "Vannes",
    "location": { "lat": 47.6559, "lon": -2.7603 }
  },
  "asn": {
    "asn": 3215,
    "as_org": "Orange S.A."
  },
  "user_agent": {
    "name": "Safari",
    "os": "Mac OS X",
    "os_name": "Mac OS X",
    "device": "Mac"
  }
}

C'est ce genre de pipeline qu'on utilise sur paltemps.fr pour analyser le trafic. Dans Kibana, on peut faire une carte des visiteurs, un top 10 des FAI, et un histogramme des codes de réponse.

Résumé

• Le filtre date parse les timestamps et les met dans @timestamp (le champ que Kibana utilise)
• Plusieurs formats peuvent etre testes dans l'ordre avec un seul filtre
• timezone est obligatoire si tes logs n'incluent pas le fuseau horaire
• GeoIP ajoute pays, ville, coordonnees et ASN à partir d'une adresse IP
• La base GeoLite2 est incluse dans Logstash, mais il vaut mieux monter une version à jour
• Les IPs privees ne sont pas geolocalisables, gere le tag _geoip_lookup_failure

Precedent : 11 - Le filtre Mutate | Suivant : 13 - Filtres KV, JSON, XML

Sources

• Date Filter Plugin - elastic.co
• GeoIP Filter Plugin - elastic.co
• Joda-Time Format Tokens
• MaxMind GeoLite2 Databases