Pipeline processor

The pipeline processor allows a pipeline to reference and include another predefined pipeline. This can be useful when you have a set of common processors that need to be shared across multiple pipelines. Instead of redefining those common processors in each pipeline, you can create a separate base pipeline containing the shared processors and then reference that base pipeline from other pipelines using the pipeline processor.

The following is the syntax for the pipeline processor:

{
  "pipeline": {
    "name": "general-pipeline"
  }
}

Configuration parameters

The following table lists the required and optional parameters for the pipeline processor.

Using the processor

Follow these steps to use the processor in a pipeline.

Step 1: Create a pipeline

The following query creates a general pipeline named general-pipeline and then creates a new pipeline named outer-pipeline, which references the general-pipeline:

PUT _ingest/pipeline/general-pipeline  
{  
  "description": "a general pipeline",  
  "processors": [  
    {  
      "uppercase": {  
        "field": "protocol"  
      }  
    },  
    {  
      "remove": {  
        "field": "name"  
      }  
    }  
  ]  
}

PUT _ingest/pipeline/outer-pipeline  
{  
  "description": "an outer pipeline referencing the general pipeline",  
  "processors": [  
    {  
      "pipeline": {  
        "name": "general-pipeline"  
      }  
    }  
  ]  
}

Step 2 (Optional): Test the pipeline

It is recommended that you test your pipeline before you ingest documents.

To test the pipeline, run the following query:

POST _ingest/pipeline/outer-pipeline/_simulate
{  
  "docs": [  
    {  
      "_source": {  
        "protocol": "https",  
        "name":"test"  
      }  
    }  
  ]  
}  

Response

The following example response confirms that the pipeline is working as expected:

{  
  "docs": [  
    {  
      "doc": {  
        "_index": "_index",  
        "_id": "_id",  
        "_source": {  
          "protocol": "HTTPS"  
        },  
        "_ingest": {  
          "timestamp": "2024-05-24T02:43:43.700735801Z"  
        }  
      }  
    }  
  ]  
}

Step 3: Ingest a document

The following query ingests a document into an index named testindex1:

POST testindex1/_doc/1?pipeline=outer-pipeline  
{  
  "protocol": "https",  
  "name": "test"  
}  

Response

The request indexes the document into the testindex1 index with the protocol field converted to uppercase and the name field removed, as shown in the following response:

{  
  "_index": "testindex1",  
  "_id": "1",  
  "_version": 2,  
  "result": "created",  
  "_shards": {  
    "total": 2,  
    "successful": 2,  
    "failed": 0  
  },  
  "_seq_no": 1,  
  "_primary_term": 1  
}  

Step 4 (Optional): Retrieve the document

To retrieve the document, run the following query:

GET testindex1/_doc/1

Response

The response shows the document with the protocol field converted to uppercase and the name field removed:

{  
  "_index": "testindex1",  
  "_id": "1",  
  "_version": 2,  
  "_seq_no": 1,  
  "_primary_term": 1,  
  "found": true,  
  "_source": {  
    "protocol": "HTTPS"  
  }  
}  

Selecting a pipeline dynamically

The name parameter supports template snippets, so you can select the pipeline to run based on a field value in the document. The following request creates a pipeline that parses NGINX logs:

PUT _ingest/pipeline/logs-nginx
{
  "processors": [
    {
      "set": {
        "field": "parsed_by",
        "value": "{{_ingest.pipeline}}"
      }
    }
  ]
}

The following request creates a pipeline that runs the pipeline named logs- followed by the value of the document’s service field:

PUT _ingest/pipeline/logs-router
{
  "processors": [
    {
      "set": {
        "field": "routed_by",
        "value": "{{_ingest.pipeline}}"
      }
    },
    {
      "pipeline": {
        "name": "logs-{{service}}"
      }
    }
  ]
}

Index a document whose service field is nginx:

PUT logs/_doc/1?pipeline=logs-router
{
  "service": "nginx",
  "message": "GET /index.html 200"
}

Retrieve the document:

GET logs/_doc/1

The response shows that the logs-router pipeline ran the logs-nginx pipeline:

{
  "_index": "logs",
  "_id": "1",
  "_version": 1,
  "_seq_no": 0,
  "_primary_term": 1,
  "found": true,
  "_source": {
    "parsed_by": "logs-nginx",
    "service": "nginx",
    "message": "GET /index.html 200",
    "routed_by": "logs-router"
  }
}

If the resolved pipeline doesn’t exist, or if running it would create a cycle in which a pipeline runs itself, the pipeline processor fails.

Accessing the current pipeline name

The _ingest.pipeline ingest metadata field contains the ID of the pipeline that is currently running. In the preceding example, the routed_by field contains logs-router and the parsed_by field contains logs-nginx because each set processor ran in a different pipeline. After the inner pipeline completes, _ingest.pipeline again contains the ID of the outer pipeline.