Skip to content
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.

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

Allow pipeline processor to ignore missing pipelines #87354

Merged
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
5 changes: 5 additions & 0 deletions docs/changelog/87354.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
pr: 87354
summary: Allow pipeline processor to ignore missing pipelines
area: Ingest
type: enhancement
issues: []
5 changes: 3 additions & 2 deletions docs/reference/ingest/processors/pipeline.asciidoc
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,9 @@ Executes another pipeline.
.Pipeline Options
[options="header"]
|======
| Name | Required | Default | Description
| `name` | yes | - | The name of the pipeline to execute. Supports <<template-snippets,template snippets>>.
| Name | Required | Default | Description
| `name` | yes | - | The name of the pipeline to execute. Supports <<template-snippets,template snippets>>.
| `ignore_missing_pipeline` | no | false | Whether to ignore missing pipelines instead of failing.
felixbarny marked this conversation as resolved.
Show resolved Hide resolved
include::common-options.asciidoc[]
|======

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -280,3 +280,94 @@ teardown:
- match: { _source.pipelines.0: "pipeline1" }
- match: { _source.pipelines.1: "another_pipeline" }
- match: { _source.pipelines.2: "another_pipeline2" }

---
"Test ignore_missing_pipeline parameter":
- do:
ingest.put_pipeline:
id: "my-pipeline"
body: >
{
"processors" : [
{
"pipeline" : {
"name": "non_existing_pipeline",
"ignore_missing_pipeline": false
}
}
]
}
- match: { acknowledged: true }

- do:
catch: /illegal_state_exception/
index:
index: test
id: "1"
pipeline: "my-pipeline"
body: {}
- match: { error.root_cause.0.type: "illegal_state_exception" }
- match: { error.root_cause.0.reason: "Pipeline processor configured for non-existent pipeline [non_existing_pipeline]" }

- do:
ingest.simulate:
id: "my-pipeline"
body: >
{
"docs": [
{
"_index": "index",
"_id": "id",
"_source": {
"foo": "bar"
}
}
]
}
- match: { docs.0.error.root_cause.0.type: "illegal_state_exception" }
- match: { docs.0.error.root_cause.0.reason: "Pipeline processor configured for non-existent pipeline [non_existing_pipeline]" }

- do:
ingest.put_pipeline:
id: "my-pipeline"
body: >
{
"processors" : [
{
"pipeline" : {
"name": "non_existing_pipeline",
"ignore_missing_pipeline": true
}
}
]
}
- match: { acknowledged: true }

- do:
index:
index: test
id: "1"
pipeline: "my-pipeline"
body: {}
- match: { _index: "test" }
- match: { _id: "1" }
- match: { _version: 1 }
- match: { result: "created" }

- do:
ingest.simulate:
id: "my-pipeline"
body: >
{
"docs": [
{
"_index": "index",
"_id": "id",
"_source": {
"foo": "bar"
}
}
]
}
- match: { docs.0.doc._index: "index" }
- match: { docs.0.doc._id: "id" }
Original file line number Diff line number Diff line change
Expand Up @@ -18,11 +18,19 @@ public class PipelineProcessor extends AbstractProcessor {
public static final String TYPE = "pipeline";

private final TemplateScript.Factory pipelineTemplate;
private final boolean ignoreMissingPipeline;
private final IngestService ingestService;

PipelineProcessor(String tag, String description, TemplateScript.Factory pipelineTemplate, IngestService ingestService) {
PipelineProcessor(
String tag,
String description,
TemplateScript.Factory pipelineTemplate,
boolean ignoreMissingPipeline,
IngestService ingestService
) {
super(tag, description);
this.pipelineTemplate = pipelineTemplate;
this.ignoreMissingPipeline = ignoreMissingPipeline;
this.ingestService = ingestService;
}

Expand All @@ -33,10 +41,14 @@ public void execute(IngestDocument ingestDocument, BiConsumer<IngestDocument, Ex
if (pipeline != null) {
Copy link
Member

Choose a reason for hiding this comment

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

From a performance aspect the getPipeline() operation is basically a regular Map lookup, right?

This path will get executed a lot if this feature is used in all Fleet data stream pipelines. I just wanted to confirm that lookup failures are not particularly expensive 😄 (like lazy loading on lookup failure).

Copy link
Member Author

Choose a reason for hiding this comment

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

From a performance aspect the getPipeline() operation is basically a regular Map lookup, right?

Yes, this method performs a map lookup. I don't think this map lookup will be noticeable during indexing (many other operations occur during indexing, so I don't think this will be visible).

ingestDocument.executePipeline(pipeline, handler);
} else {
handler.accept(
null,
new IllegalStateException("Pipeline processor configured for non-existent pipeline [" + pipelineName + ']')
);
if (ignoreMissingPipeline) {
handler.accept(ingestDocument, null);
} else {
handler.accept(
null,
new IllegalStateException("Pipeline processor configured for non-existent pipeline [" + pipelineName + ']')
);
}
}
}

Expand Down Expand Up @@ -85,7 +97,14 @@ public PipelineProcessor create(
"name",
ingestService.getScriptService()
);
return new PipelineProcessor(processorTag, description, pipelineTemplate, ingestService);
boolean ignoreMissingPipeline = ConfigurationUtils.readBooleanProperty(
TYPE,
processorTag,
config,
"ignore_missing_pipeline",
false
);
return new PipelineProcessor(processorTag, description, pipelineTemplate, ignoreMissingPipeline, ingestService);
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,25 @@ public void testThrowsOnMissingPipeline() throws Exception {
assertEquals("Pipeline processor configured for non-existent pipeline [missingPipelineId]", e[0].getMessage());
}

public void testIgnoreMissingPipeline() throws Exception {
var ingestService = createIngestService();
var testIngestDocument = RandomDocumentPicks.randomIngestDocument(random(), new HashMap<>());
var factory = new PipelineProcessor.Factory(ingestService);
var config = new HashMap<String, Object>();
config.put("name", "missingPipelineId");
config.put("ignore_missing_pipeline", true);

var r = new IngestDocument[1];
var e = new Exception[1];
var processor = factory.create(Collections.emptyMap(), null, null, config);
processor.execute(testIngestDocument, (result, e1) -> {
r[0] = result;
e[0] = e1;
});
assertNull(e[0]);
assertSame(testIngestDocument, r[0]);
}

public void testThrowsOnRecursivePipelineInvocations() throws Exception {
String innerPipelineId = "inner";
String outerPipelineId = "outer";
Expand Down Expand Up @@ -229,7 +248,7 @@ public String getType() {
});
if (i < (numPipelines - 1)) {
TemplateScript.Factory pipelineName = new TestTemplateService.MockTemplateScript.Factory(Integer.toString(i + 1));
processors.add(new PipelineProcessor(null, null, pipelineName, ingestService));
processors.add(new PipelineProcessor(null, null, pipelineName, false, ingestService));
}

Pipeline pipeline = new Pipeline(pipelineId, null, null, null, new CompoundProcessor(false, processors, List.of()));
Expand Down