Support terminate processor.

[mashhurs]

Registers terminate processor to the support list.

[mergify]

This pull request does not have a backport label. Could you fix it @mashhurs? 🙏
To fixup this pull request, you need to add the backport labels for the needed
branches, such as:

• backport-8./d is the label to automatically backport to the 8./d branch. /d is the digit.
• If no backport is necessary, please add the backport-skip label

[mashhurs]

I wished this tag be available in _ingest_document so that I can expect(event.get("[tag]")).to eql? "terminated_ingest_pipeline" but no tag exists.

[mashhurs]

TerminateProcessor is also package private like dot_expander https://github.com/elastic/logstash-filter-elastic_integration/blob/main/src/main/java/co/elastic/logstash/filters/elasticintegration/EventProcessorBuilder.java#L103

[donoghuc]

Can you help me understand what exactly this assertion proves WRT the termination? Looks like with "if": "ctx.error != null" the terminate will trigger.

[mashhurs]

After plugin successfully executes the pipeline, it let's know (set the [@metadata][target_ingest_pipeline] to _none) es-output to avoid ingest pipeline execution in Elasticsearch side.

what exactly this assertion proves WRT the termination?

Yeah, this one I tried to highlight here that terminate processor doesn't attach any specific fields we can validate (it seems tag is for debug purpose). I have updated the integration test which adds append processor after terminate and validates to make sure append processor didn't execute.
Correspond ES query as an integration test would like this:

POST /_ingest/pipeline/_simulate
{
  "pipeline" :
  {
    "processors": [
      {
        "terminate": {
          "if": "ctx.error != null",
          "tag": "terminated_pipeline"
        }
      },
      {
        "append": {
          "field": "message",
          "value": "my message value"
        }
      }
    ]
  },
  "docs": [
    {
      "_index": "index",
      "_id": "id1",
      "_source": {
        "foo": "bar",
        "error": "some exception"
      }
    },
    {
      "_index": "index",
      "_id": "id2",
      "_source": {
        "foo": "rab"
      }
    }
  ]
}

In the result, we can see id1 will not have message since pipeline was terminated by terminate processor:

{
  "docs": [
    {
      "doc": {
        "_index": "index",
        "_version": "-3",
        "_id": "id1",
        "_source": {
          "error": "some exception",
          "foo": "bar"
        },
        "_ingest": {
          "timestamp": "2025-07-15T23:56:43.427866851Z"
        }
      }
    },
    {
      "doc": {
        "_index": "index",
        "_version": "-3",
        "_id": "id2",
        "_source": {
          "message": [
            "my message value"
          ],
          "foo": "rab"
        },
        "_ingest": {
          "timestamp": "2025-07-15T23:56:43.427882279Z"
        }
      }
    }
  ]
}

[donoghuc]

Thanks! That is very helpful. Love that idea!

[elasticmachine]

💚 Build Succeeded

• Buildkite Build
• Commit: f817630

History

• 💚 Build #595 succeeded ba40399

[donoghuc]

Thanks for the detailed breakdown. That is very helpful. Appreciate the extra context.

[mashhurs]

Just FYI: terminate was introduced in ES 8.16 versions, so worth to backport to all active branches.

[github-actions]

@Mergifyio backport 8.17 8.18 8.19 9.0 9.1

[mergify]

backport 8.17 8.18 8.19 9.0 9.1

✅ Backports have been created

• #346 [8.17] (backport #345) Support terminate processor. has been created for branch 8.17
• #347 [8.18] (backport #345) Support terminate processor. has been created for branch 8.18
• #348 [8.19] (backport #345) Support terminate processor. has been created for branch 8.19
• #349 [9.0] (backport #345) Support terminate processor. has been created for branch 9.0
• #350 [9.1] (backport #345) Support terminate processor. has been created for branch 9.1