Ingestion Elasticsearch : Bulk API, ingest pipelines, enrich et reindex

1. La Bulk API #

Indexer en masse.

Format NDJSON #

POST _bulk
{ "create": { "_index": "logs-nginx-default" } }
{ "@timestamp": "2026-10-20T10:00:00Z", "message": "203.0.113.5 - - [20/Oct/2026:10:00:00 +0000] \"GET /api/cart HTTP/1.1\" 200 512 \"-\" \"Mozilla/5.0\"" }
{ "create": { "_index": "logs-nginx-default" } }
{ "@timestamp": "2026-10-20T10:00:01Z", "message": "198.51.100.7 - - [20/Oct/2026:10:00:01 +0000] \"POST /api/order HTTP/1.1\" 500 87 \"-\" \"curl/8.5\"" }

Bonne Pratique : Visez des requêtes bulk de 5 à 15 Mo et vérifiez toujours le champ errors de la réponse : un bulk « réussi » peut contenir des documents rejetés.

2. Les ingest pipelines #

Transformer à l'ingestion.

Pipeline Nginx #

PUT _ingest/pipeline/nginx-access
{
  "processors": [
    { "grok": { "field": "message", "patterns": [ "%{IPORHOST:source.ip} - %{DATA:user.name} \\[%{HTTPDATE:nginx.time}\\] \"%{WORD:http.request.method} %{DATA:url.path} HTTP/%{NUMBER:http.version}\" %{NUMBER:http.response.status_code:int} %{NUMBER:http.response.body.bytes:long} \"%{DATA:http.request.referrer}\" \"%{DATA:user_agent.original}\"" ] } },
    { "date":       { "field": "nginx.time", "formats": [ "dd/MMM/yyyy:HH:mm:ss Z" ] } },
    { "geoip":      { "field": "source.ip", "target_field": "source.geo", "ignore_missing": true } },
    { "user_agent": { "field": "user_agent.original", "ignore_missing": true } },
    { "remove":     { "field": [ "nginx.time" ] } }
  ],
  "on_failure": [
    { "set": { "field": "event.kind", "value": "pipeline_error" } },
    { "set": { "field": "error.message", "value": "{{ _ingest.on_failure_message }}" } }
  ]
}

POST _ingest/pipeline/nginx-access/_simulate
{ "docs": [ { "_source": { "message": "203.0.113.5 - - [20/Oct/2026:10:00:00 +0000] \"GET /api/cart HTTP/1.1\" 200 512 \"-\" \"Mozilla/5.0\"" } } ] }

À savoir : dissect est plus rapide que grok quand le format est fixe (séparateurs connus) ; gardez grok pour les formats variables.

3. Enrich et reindex #

Compléter et retraiter.

Enrich #

PUT _enrich/policy/customers
{ "match": { "indices": "customers", "match_field": "customer.id", "enrich_fields": [ "customer.tier", "customer.country" ] } }
POST _enrich/policy/customers/_execute

# dans un pipeline
{ "enrich": { "policy_name": "customers", "field": "customer.id", "target_field": "customer_info" } }

# retraiter un index existant avec un pipeline
POST _reindex
{ "source": { "index": "old-logs" }, "dest": { "index": "logs-nginx-default", "op_type": "create", "pipeline": "nginx-access" } }

Logstash et Elastic Agent restent utiles quand il faut des sources variées, des tampons ou des sorties multiples ; les ingest pipelines suffisent pour transformer ce qui arrive déjà dans Elasticsearch.