Skip to content

Support terminate processor. #345

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Jul 16, 2025
Merged

Support terminate processor. #345

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ba40399
Support termiate processor.
mashhurs Jul 14, 2025
f817630
Update the integration test for terminate processor to it check the e…
mashhurs Jul 15, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 29 additions & 0 deletions spec/integration/elastic_integration_spec.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1134,6 +1134,35 @@ def path; @path; end
# end
end

describe 'with terminate processor' do
let(:pipeline_processor) {
'{
"terminate": {
"if": "ctx.error != null",
"tag": "terminated_ingest_pipeline"
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

}
},
{
"append": {
"field": "append_field",
"value": ["integration", "test"]
}
}'
}

it 'terminates the ingest pipeline' do
events = [LogStash::Event.new(
"message" => "Send message to pipeline which gets terminated.",
"error" => "This is intentionally placed error.",
"data_stream" => data_stream)]

subject.multi_filter(events).each do |event|
expect(event.get("[@metadata][target_ingest_pipeline]")).to include("_none")
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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"
        }
      }
    }
  ]
}

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks! That is very helpful. Love that idea!

# intentionally placed append processor to check if it is not executed after terminate processor
expect(event.get("append_field")).to be_nil
end
end
end
end

context '#multi-pipeline execution' do
Expand Down
1 change: 1 addition & 0 deletions src/main/java/co/elastic/logstash/filters/elasticintegration/EventProcessorBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,7 @@ public EventProcessorBuilder() {
org.elasticsearch.ingest.common.SetProcessor.TYPE,
org.elasticsearch.ingest.common.SortProcessor.TYPE,
org.elasticsearch.ingest.common.SplitProcessor.TYPE,
"terminate", // note: upstream constant is package-private
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

org.elasticsearch.ingest.common.TrimProcessor.TYPE,
org.elasticsearch.ingest.common.URLDecodeProcessor.TYPE,
org.elasticsearch.ingest.common.UppercaseProcessor.TYPE,
Expand Down