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
on_failure	N	-	See Handling Failures
on_success	N	-	See Handling Success
tag	N	-	Identifier

Details

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.

warning

Be careful with recursive pipeline references to avoid infinite loops. Currently, the processor does not detect circular dependencies.

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