Ingest ingest then create index #39607
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,119 @@ | ||
| --- | ||
| teardown: | ||
| - do: | ||
| ingest.delete_pipeline: | ||
| id: "retarget" | ||
| ignore: 404 | ||
|
|
||
| - do: | ||
| indices.delete: | ||
| index: foo | ||
|
|
||
| --- | ||
| "Test Change Target Index with Explicit Pipeline": | ||
|
|
||
| - do: | ||
| ingest.put_pipeline: | ||
| id: "retarget" | ||
| body: > | ||
| { | ||
| "processors": [ | ||
| { | ||
| "set" : { | ||
| "field" : "_index", | ||
| "value" : "foo" | ||
| } | ||
| } | ||
| ] | ||
| } | ||
| - match: { acknowledged: true } | ||
|
|
||
| # no indices | ||
| - do: | ||
| cat.indices: {} | ||
|
|
||
| - match: | ||
| $body: | | ||
| /^$/ | ||
|
|
||
| - do: | ||
| index: | ||
| index: test | ||
| id: 1 | ||
| pipeline: "retarget" | ||
| body: { | ||
| a: true | ||
| } | ||
|
|
||
| - do: | ||
| get: | ||
| index: foo | ||
| id: 1 | ||
| - match: { _source.a: true } | ||
|
|
||
| # only the foo index | ||
| - do: | ||
| cat.indices: | ||
| h: i | ||
|
|
||
| - match: | ||
| $body: | | ||
| /^foo\n$/ | ||
|
|
||
| --- | ||
| "Test Change Target Index with Default Pipeline": | ||
|
|
||
| - do: | ||
| indices.put_template: | ||
| name: index_template | ||
| body: | ||
| index_patterns: test | ||
| settings: | ||
| default_pipeline: "retarget" | ||
|
|
||
| - do: | ||
| ingest.put_pipeline: | ||
| id: "retarget" | ||
| body: > | ||
| { | ||
| "processors": [ | ||
| { | ||
| "set" : { | ||
| "field" : "_index", | ||
| "value" : "foo" | ||
| } | ||
| } | ||
| ] | ||
| } | ||
| - match: { acknowledged: true } | ||
|
|
||
| # no indices | ||
| - do: | ||
| cat.indices: {} | ||
|
|
||
| - match: | ||
| $body: | | ||
| /^$/ | ||
|
|
||
| - do: | ||
| index: | ||
| index: test | ||
| id: 1 | ||
| body: { | ||
| a: true | ||
| } | ||
|
|
||
| - do: | ||
| get: | ||
| index: foo | ||
| id: 1 | ||
| - match: { _source.a: true } | ||
|
|
||
| # only the foo index | ||
| - do: | ||
| cat.indices: | ||
| h: i | ||
|
|
||
| - match: | ||
| $body: | | ||
| /^foo\n$/ |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -47,11 +47,14 @@ | |
| import org.elasticsearch.cluster.metadata.AliasOrIndex; | ||
| import org.elasticsearch.cluster.metadata.IndexMetaData; | ||
| import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver; | ||
| import org.elasticsearch.cluster.metadata.IndexTemplateMetaData; | ||
| import org.elasticsearch.cluster.metadata.MappingMetaData; | ||
| import org.elasticsearch.cluster.metadata.MetaData; | ||
| import org.elasticsearch.cluster.metadata.MetaDataIndexTemplateService; | ||
| import org.elasticsearch.cluster.service.ClusterService; | ||
| import org.elasticsearch.common.collect.ImmutableOpenMap; | ||
| import org.elasticsearch.common.inject.Inject; | ||
| import org.elasticsearch.common.settings.Settings; | ||
| import org.elasticsearch.common.unit.TimeValue; | ||
| import org.elasticsearch.common.util.concurrent.AbstractRunnable; | ||
| import org.elasticsearch.common.util.concurrent.AtomicArray; | ||
|
|
@@ -151,6 +154,71 @@ protected void doExecute(Task task, BulkRequest bulkRequest, ActionListener<Bulk | |
| final long startTime = relativeTime(); | ||
| final AtomicArray<BulkItemResponse> responses = new AtomicArray<>(bulkRequest.requests.size()); | ||
|
|
||
| boolean hasIndexRequestsWithPipelines = false; | ||
| final MetaData metaData = clusterService.state().getMetaData(); | ||
| ImmutableOpenMap<String, IndexMetaData> indicesMetaData = metaData.indices(); | ||
| for (DocWriteRequest<?> actionRequest : bulkRequest.requests) { | ||
| IndexRequest indexRequest = getIndexWriteRequest(actionRequest); | ||
| if (indexRequest != null) { | ||
| // get pipeline from request | ||
| String pipeline = indexRequest.getPipeline(); | ||
| if (pipeline == null) { | ||
| // start to look for default pipeline via settings found in the index meta data | ||
| IndexMetaData indexMetaData = indicesMetaData.get(actionRequest.index()); | ||
| if (indexMetaData == null && indexRequest.index() != null) { | ||
| // if the write request if through an alias use the write index's meta data | ||
| AliasOrIndex indexOrAlias = metaData.getAliasAndIndexLookup().get(indexRequest.index()); | ||
| if (indexOrAlias != null && indexOrAlias.isAlias()) { | ||
| AliasOrIndex.Alias alias = (AliasOrIndex.Alias) indexOrAlias; | ||
| indexMetaData = alias.getWriteIndex(); | ||
| } | ||
| } | ||
| if (indexMetaData != null) { | ||
| // Find the the default pipeline if one is defined from and existing index. | ||
| String defaultPipeline = IndexSettings.DEFAULT_PIPELINE.get(indexMetaData.getSettings()); | ||
| indexRequest.setPipeline(defaultPipeline); | ||
| if (IngestService.NOOP_PIPELINE_NAME.equals(defaultPipeline) == false) { | ||
| hasIndexRequestsWithPipelines = true; | ||
| } | ||
| } else if (indexRequest.index() != null) { | ||
|
There was a problem hiding this comment. Moving the ingest node execution before the create index is the primary change of this PR. However, to allow the prevent re-introducing #32758, we need to look up the default pipeline from the matched templates which is what this code block does. |
||
| // No index exists yet (and is valid request), so matching index templates to look for a default pipeline | ||
| List<IndexTemplateMetaData> templates = MetaDataIndexTemplateService.findTemplates(metaData, indexRequest.index()); | ||
| assert (templates != null); | ||
| String defaultPipeline = IngestService.NOOP_PIPELINE_NAME; | ||
| // apply templates, here, in reverse order, since first ones are better matching | ||
| for (int i = templates.size() - 1; i >= 0; i--) { | ||
original-brownbear marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| Settings settings = templates.get(i).settings(); | ||
| if(IndexSettings.DEFAULT_PIPELINE.exists(settings)) { | ||
| defaultPipeline = IndexSettings.DEFAULT_PIPELINE.get(settings); | ||
| } | ||
| } | ||
| indexRequest.setPipeline(defaultPipeline); | ||
| if (IngestService.NOOP_PIPELINE_NAME.equals(defaultPipeline) == false) { | ||
| hasIndexRequestsWithPipelines = true; | ||
| } | ||
| } | ||
| } else if (IngestService.NOOP_PIPELINE_NAME.equals(pipeline) == false) { | ||
| hasIndexRequestsWithPipelines = true; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| if (hasIndexRequestsWithPipelines) { | ||
| // this method (doExecute) will be called again, but with the bulk requests updated from the ingest node processing but | ||
| // also with IngestService.NOOP_PIPELINE_NAME on each request. This ensures that this on the second time through this method, | ||
| // this path is never taken. | ||
| try { | ||
| if (clusterService.localNode().isIngestNode()) { | ||
| processBulkIndexIngestRequest(task, bulkRequest, listener); | ||
| } else { | ||
| ingestForwarder.forwardIngestRequest(BulkAction.INSTANCE, bulkRequest, listener); | ||
| } | ||
| } catch (Exception e) { | ||
| listener.onFailure(e); | ||
| } | ||
| return; | ||
| } | ||
|
|
||
| if (needToCheck()) { | ||
| // Attempt to create all the indices that we're going to need during the bulk before we start. | ||
| // Step 1: collect all the indices in the request | ||
|
|
@@ -181,15 +249,15 @@ protected void doExecute(Task task, BulkRequest bulkRequest, ActionListener<Bulk | |
| } | ||
| // Step 3: create all the indices that are missing, if there are any missing. start the bulk after all the creates come back. | ||
| if (autoCreateIndices.isEmpty()) { | ||
| executeBulk(task, bulkRequest, startTime, listener, responses, indicesThatCannotBeCreated); | ||
| } else { | ||
| final AtomicInteger counter = new AtomicInteger(autoCreateIndices.size()); | ||
| for (String index : autoCreateIndices) { | ||
| createIndex(index, bulkRequest.timeout(), new ActionListener<CreateIndexResponse>() { | ||
| @Override | ||
| public void onResponse(CreateIndexResponse result) { | ||
| if (counter.decrementAndGet() == 0) { | ||
| executeBulk(task, bulkRequest, startTime, listener, responses, indicesThatCannotBeCreated); | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -205,7 +273,7 @@ public void onFailure(Exception e) { | |
| } | ||
| } | ||
| if (counter.decrementAndGet() == 0) { | ||
| executeBulk(task, bulkRequest, startTime, ActionListener.wrap(listener::onResponse, inner -> { | ||
| inner.addSuppressed(e); | ||
| listener.onFailure(inner); | ||
| }), responses, indicesThatCannotBeCreated); | ||
|
|
@@ -215,56 +283,7 @@ public void onFailure(Exception e) { | |
| } | ||
| } | ||
| } else { | ||
| executeBulk(task, bulkRequest, startTime, listener, responses, emptyMap()); | ||
| } | ||
| } | ||
|
|
||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
these were removed because now that the ingest happens before index creation, if the index doesn't exist yet, and we only drop a document, the index never exists, and thus the response is different.