Reindexing 220M legacy documents through a failing ingest pipeline

An ingest pipeline is per-document. When a processor throws and there is no on_failure (and no ignore_failure), that document is rejected - but the transport-level response is still successful. In _reindex the failures surface only in the task's response.failures array; in _bulk the request returns 200 with "errors": true and per-item error entries.

So: check the item-level result, never the HTTP status.

1. Inspect the running/finished task: GET _tasks/<task_id> → response.created, response.failures[] (each has cause.reason, e.g. date_time_parse_exception or failed to parse field [bytes] of type [long]).
2. Inspect processor counters: GET _nodes/stats/ingest?filter_path=**.processors - stats.failed per processor tells you which processor, not which document.
3. Add a dead-letter path so failures become observable data rather than silence:

   {"parse-legacy": {
     "processors": [ ... ],
     "on_failure": [
       {"set": {"field": "_index", "value": "failed-events"}},
       {"set": {"field": "error.type", "value": "{{{_ingest.on_failure_processor_type}}}"}},
       {"set": {"field": "error.message", "value": "{{{_ingest.on_failure_message}}}"}}
     ]
   }}
   

   Every rejected document is then written to failed-events with its original _source and the failure reason, and can be replayed after the pipeline is fixed.

Because the source is gone, this dead-letter index is now part of your recovery plan, not debug output.

Key realisation: the date processor was targeting @timestamp while the document already had an @timestamp - so successes also overwrote the original value, and failures left a misleading value behind. Parse into a distinct field, then rename deliberately.

{"parse-legacy": {
  "processors": [
    {"grok": {
      "field": "message",
      "patterns": ["%{TIMESTAMP_ISO8601:event_time}", "%{HTTPDATE:event_time}"],
      "ignore_failure": true}}
    ,
    {"date": {
      "field": "event_time",
      "target_field": "event_time_parsed",
      "formats": ["dd/MMM/yyyy:HH:mm:ss Z", "ISO8601"],
      "timezone": "Europe/Berlin",
      "locale": "en",
      "ignore_failure": true}}
    ,
    {"rename": {
      "field": "event_time_parsed",
      "target_field": "@timestamp",
      "ignore_missing": true}}
    ,
    {"convert": {"field": "bytes", "type": "long", "ignore_missing": true}}
    ,
    {"gsub": {"field": "bytes", "pattern": "[,\\s]", "replacement": "", "ignore_failure": true}}
  ]
}}

Reasoning points to state explicitly:

• formats is an ordered list - put the most common format first; matching stops at the first success.
• timezone only applies when the pattern has no zone. dd/MMM/yyyy:HH:mm:ss Z carries Z, so the parsed offset wins; ISO8601 also wins. Do not assume timezone overrides them.
• locale: en guards MMM month names - a node with a non-English default locale will otherwise fail on every Jan/Feb.
• gsub must run before convert, and the convert must tolerate missing/empty values; ignore_failure on convert alone would drop the whole field.
• Prefer on_failure → dead-letter index over scattered ignore_failure for anything you cannot afford to lose; ignore_failure hides the failure, the dead-letter index preserves it.

POST _reindex?wait_for_completion=false&slices=auto&requests_per_second=2000
{
  "source": {"index": "legacy-events", "size": 2000, "query": {"range": {"event_time": {"lt": "now-1d/d"}}}},
  "dest":   {"index": "events-v2", "op_type": "create", "pipeline": "parse-legacy"},
  "conflicts": "proceed"
}

Reasoning:

• op_type: create with the original _id is what makes the job retriable. index would silently overwrite; create turns a replay into a cheap per-item 409 (version_conflict_engine_exception) that conflicts: proceed records in response.version_conflicts. You can now kill and restart the job safely.
• wait_for_completion=false + GET _tasks/<task_id> - a 220M-document job runs for hours; a synchronous client will time out and leave you blind about progress. Poll response.created, response.version_conflicts, response.failures.
• slices: auto parallelises across source shards; more slices than shards adds overhead without throughput. Watch for search rejections.
• requests_per_second (throttle) - start conservative and raise while watching _nodes/stats/thread_pool for search/write queue growth and _cat/indices merge pressure.
• Pause/resume via the task API: POST _tasks/<task_id>/_cancel then re-run with the same settings; the create semantics make that safe. An alternative is manual search_after slicing, which gives you a stable resume cursor but more code.
• Run the migration into an index with zero replicas, then set number_of_replicas: 1 and refresh_interval back afterwards; merge and refresh churn during ingest is otherwise the dominant cost.