Ingest Nodes/Pipelines, Processor Execution, on_failure, Conditional Processing, and Simulation

Learning outcomes

AtlasMart receives access logs from multiple services. The raw line is useful evidence, but search and alerting need typed fields such as timestamp, service, tenant, status code and latency. An ingest pipeline is a named, ordered sequence of processors executed before indexing. This lesson makes that execution path visible so a failed parser is not mistaken for a storage or search failure.

1. The write path has a transformation stage

A client sends an index or bulk operation, optionally naming a pipeline. The coordinating path sends the document through an ingest-capable node. Each processor sees the document resulting from the previous processor. Only after the pipeline finishes does the transformed document proceed to normal indexing, routing, primary-shard execution and replication. Therefore an acknowledgement is about the transformed document, not the original wire payload.

2. Build the disposable AtlasMart targets

PUT atlasmart-telemetry-v11
{
  "settings": {"number_of_shards":1,"number_of_replicas":0},
  "mappings": {
    "dynamic":"strict",
    "properties": {
      "@timestamp":{"type":"date"},
      "event_id":{"type":"keyword"},
      "service":{"type":"keyword"},
      "level":{"type":"keyword"},
      "tenant_id":{"type":"keyword"},
      "client_ip":{"type":"ip"},
      "http_method":{"type":"keyword"},
      "url_path":{"type":"keyword"},
      "status_code":{"type":"integer"},
      "duration_ms":{"type":"double"},
      "message":{"type":"text"},
      "ingest":{"properties":{
        "pipeline_version":{"type":"keyword"},
        "received_at":{"type":"date"},
        "failed":{"type":"boolean"},
        "error":{"type":"text","index":false}
      }}
    }
  }
}

PUT atlasmart-ingest-failures-v1
{
  "settings":{"number_of_shards":1,"number_of_replicas":0},
  "mappings":{"dynamic":true}
}

The success index is strict so the pipeline cannot accidentally invent fields forever. The failure index is intentionally permissive in this small lab because its job is preservation of diagnostic evidence; in production, quarantine schemas should also be governed and access controlled because failed records often contain malformed or sensitive input.

3. Define an ordered portable pipeline

PUT _ingest/pipeline/atlasmart-log-ingest-v1
{
  "description":"AtlasMart deterministic access-log parser v1",
  "processors":[
    {"set":{"tag":"stamp-pipeline-version","field":"ingest.pipeline_version","value":"v1"}},
    {"set":{"tag":"stamp-received-at","field":"ingest.received_at","value":"{{{_ingest.timestamp}}}"}},
    {"dissect":{
      "tag":"parse-access-line",
      "field":"message",
      "pattern":"%{event_time}|%{service}|%{level}|%{tenant_id}|%{client_ip}|%{http_method}|%{url_path}|%{status_code}|%{duration_ms}"
    }},
    {"date":{"tag":"parse-event-time","field":"event_time","target_field":"@timestamp","formats":["ISO8601"]}},
    {"convert":{"tag":"status-to-int","field":"status_code","type":"integer"}},
    {"convert":{"tag":"duration-to-double","field":"duration_ms","type":"double"}},
    {"remove":{"tag":"remove-temp-time","field":"event_time"}}
  ],
  "on_failure":[
    {"set":{"field":"_index","value":"atlasmart-ingest-failures-v1"}},
    {"set":{"field":"ingest.failed","value":true}},
    {"set":{"field":"ingest.error","value":"pipeline atlasmart-log-ingest-v1 failed"}}
  ]
}

Notice the tags. A processor tag is not decoration: it gives operators a stable stage name in diagnostics and per-processor statistics. The pipeline-level on_failure changes _index so a poison record is quarantined instead of being silently discarded. The lab stores a stable pipeline version and ingest receipt time so later replays can prove which transformation contract ran.

4. Simulate before writing

POST _ingest/pipeline/atlasmart-log-ingest-v1/_simulate?verbose=true
{
  "docs":[
    {"_index":"atlasmart-telemetry-v11","_id":"evt-good-1","_source":{
      "event_id":"evt-good-1",
      "message":"2026-09-11T07:30:00Z|catalog-api|INFO|tenant-a|192.0.2.10|GET|/products/P-1001|200|12.4"
    }},
    {"_index":"atlasmart-telemetry-v11","_id":"evt-bad-1","_source":{
      "event_id":"evt-bad-1",
      "message":"malformed|record"
    }}
  ]
}

Expected invariant for the good sample: the final source contains typed status_code=200, duration_ms=12.4, an ISO timestamp, stable service/tenant fields and ingest.pipeline_version=v1. Expected invariant for the bad sample: the normal parse path fails and the pipeline-level handler changes its target to atlasmart-ingest-failures-v1 while retaining the original message. Exact verbose response shape is version dependent; assert fields and destination, not incidental ordering.

5. Conditions should express intentional optionality

PUT _ingest/pipeline/atlasmart-log-ingest-v1-cond
{
  "processors":[
    {"uppercase":{
      "tag":"normalize-level",
      "field":"level",
      "if":"ctx.level != null"
    }},
    {"set":{
      "tag":"mark-server-error",
      "field":"is_server_error",
      "value":true,
      "if":"ctx.status_code != null && ctx.status_code >= 500"
    }}
  ]
}

A condition says “this field is optional and this branch is defined.” ignore_failure says “this processor failed and we decided to continue.” Those are different operational facts and should not be conflated.

6. Index through the pipeline, then inspect evidence

POST atlasmart-telemetry-v11/_doc/evt-good-1?pipeline=atlasmart-log-ingest-v1&refresh=true
{
  "event_id":"evt-good-1",
  "message":"2026-09-11T07:30:00Z|catalog-api|INFO|tenant-a|192.0.2.10|GET|/products/P-1001|200|12.4"
}

POST atlasmart-telemetry-v11/_doc/evt-bad-1?pipeline=atlasmart-log-ingest-v1&refresh=true
{
  "event_id":"evt-bad-1",
  "message":"malformed|record"
}

GET atlasmart-telemetry-v11/_doc/evt-good-1
GET atlasmart-ingest-failures-v1/_doc/evt-bad-1
GET _nodes/stats/ingest?filter_path=nodes.*.ingest

Do not infer throughput from one request. The node statistics are cumulative counters: capture a before snapshot, run a controlled batch, capture an after snapshot, then compute deltas. Pair that with client-side elapsed time and system CPU so you can distinguish processor time from networking, indexing, refresh and storage effects.

7. Deliberately wrong approach: hide every error

Setting ignore_failure:true on the parser can make writes appear successful while required fields remain absent. AtlasMart then gets a worse failure: documents are searchable but semantically unusable for dashboards, alerting or tenant filters. Repair by making required parsing fail closed, quarantine the raw event with a stable error classification, and reserve optional conditions for genuinely optional fields.

Production judgment

Keep pipeline logic bounded, versioned and testable. Measure p95/p99 end-to-end indexing latency and sustained throughput with and without the pipeline, plus processor time, CPU, failure rate, quarantine growth and indexing lag. Protect pipeline-management privileges, because a malicious or accidental pipeline change can rewrite tenant fields, routing, destination indices or sensitive data. Snapshot/recovery protects indexed state; it does not replace retaining replayable raw events when your business requires deterministic reprocessing.

Summary and next step

You can now trace a document through an ingest pipeline and distinguish transformation failure from indexing failure. Next we choose specific processors from the structure of the source data and compare their correctness and CPU tradeoffs.

Authoritative references

• Elastic ingest pipelines — Pipeline execution, conditionals, failure handling, node statistics, default/final pipeline concepts and operational limitations.
• Elastic simulate pipeline API — Test an existing or inline pipeline; verbose mode exposes per-processor intermediate results.
• Elastic ingest error handling — Processor-level and pipeline-level on_failure semantics and ingest failure metadata.
• Elastic enrich processor setup — Source data, enrich policy execution, generated enrich index and ingest processor workflow.
• Elastic enrich processor reference — policy_name, match field, target field, max_matches and lookup behavior.
• Elastic script processor — Painless ingest context and script processor behavior.
• OpenSearch ingest pipelines — Pipeline model, ingest node prerequisite and guidance on using Data Prepper for larger or more complex preprocessing.
• OpenSearch ingest processors — Current core processor inventory and Nodes Info inspection endpoint.
• OpenSearch pipeline failures — Failure handling and ingest-pipeline node statistics.
• OpenSearch access data in a pipeline — Source, metadata such as _index/_routing, and _ingest.timestamp access.
• OpenSearch grok processor — Pattern-based parsing and debug controls.
• OpenSearch dissect processor — Delimiter-based parsing for stable log formats.
• OpenSearch user-agent processor — User-agent parsing and target field configuration.
• OpenSearch Nodes Stats API — Per-node, pipeline and processor ingest counts, time, current work and failures.