Pipeline

Control Flow Elastic Compatible

Synopsis

Executes another pipeline by name, allowing for pipeline reuse and modular configurations.

Schema

- pipeline:
    name: <string>
    description: <text>
    if: <script>
    ignore_failure: <boolean>
    ignore_missing_pipeline: <boolean>
    on_failure: <processor[]>
    on_success: <processor[]>
    tag: <string>

Configuration

The following fields are used to define the processor:

Field	Required	Default	Description
name	Y	-	Name or reference of the pipeline to execute
description	N	-	Explanatory note
if	N	-	Condition to run
ignore_failure	N	false	See Handling Failures
ignore_missing_pipeline	N	false	If true, silently continue when referenced pipeline is not found
return_on_missing_pipeline	N	false	With ignore_missing_pipeline, stop the CURRENT pipeline when the named one is absent instead of continuing to the next processor
clone	N	false	Run the named pipeline against a deep copy of the event. See below — it also changes error handling
merge_on_success	N	false	clone only. Replace the event with the clone when the sub-pipeline succeeds. Without it the clone is discarded
on_failure	N	-	See Handling Failures
on_success	N	-	See Handling Success
tag	N	-	Identifier
disabled	N	false	When true, the processor is skipped and the event continues to the next one. Lets you take a processor out of the path without removing its configuration

Details

clone also suppresses failures

clone: true does two things, and only the first is obvious.

It runs the named pipeline against a deep copy of the event, so the sub-pipeline's changes do not reach the original unless merge_on_success is set. What a clone leaves behind either way is the vendor name, which is copied back to the source event.

It also makes every failure in that sub-pipeline non-fatal. pp.IgnoreFailure || pp.Clone gates the error path, so clone implies ignore_failure whether or not you set it — and a stopper error, which normally ends the parent pipeline, is not propagated either (IsStopperError(errBase) && !pp.Clone).

That is defensible for a side-branch whose result is discarded, but it means a cloned sub-pipeline cannot fail the parent, and its errors reach only the debug log. Do not use clone for work whose failure should stop processing.

Missing pipelines

ignore_missing_pipeline continues past a pipeline that does not exist. Adding return_on_missing_pipeline changes what "continue" means: instead of moving to the next processor it ends the current pipeline cleanly.

The pipeline processor can reference other pipelines using the syntax {{ IngestPipeline "pipeline-name" }}. The names can be specified with or without the .yml/.yaml extension.

warning

As pipeline references are resolved at runtime, make sure all referenced pipelines exist in your configuration, or set ignore_missing_pipeline to true if they are optional.

note

The processor detects circular pipeline references at runtime. If a pipeline cycle is detected, the processor returns an error identifying the pipeline that caused the cycle.

Examples

Basic

Another pipeline to be executed...	pipelines: - name: main processors: - pipeline: name: '{{ IngestPipeline "parse-logs" }}'
must be defined in the configuration:	pipelines: - name: parse-logs processors: - grok: field: message pattern: '%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level}'

Conditionals

Executing the pipeline based on criteria...	- pipeline: name: '{{ IngestPipeline "enrich-data" }}' if: 'ctx.level == "ERROR"' ignore_failure: true
helps control the flow:	pipelines: - name: enrich-data processors: - append: field: tags value: needs-review

Error Handling

Handle missing pipelines...	- pipeline: name: '{{ IngestPipeline "optional-pipeline" }}' ignore_missing_pipeline: true on_failure: - append: field: tags value: pipeline-failed
and specify fallback actions:	pipelines: - name: main processors: - append: field: status value: processed