-
Notifications
You must be signed in to change notification settings - Fork 24.3k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Fix enrich cache corruption bug (#82540)
Backport #82441 to 7.17 branch. The enrich cache currently uses `SearchResponse` as cache value, which contains the hits used by the enrich processor for enrichment. What is actually used is by the enrich processor is `SearchHit.getSourceAsMap()`, which is a map of maps representation of a search hit. The problem is that this map is mutable and the map of maps is directly passed into `IngestDocument` and at the same time this is cached by the enrich cache via `SearchResponse` cache value. Any processor that modifies the content added by the enrich processor, also changes the map of maps representation of a search hit in the cache. This corrupts the cache, because if this the enrich cache serves a cache entry for the same key, a different snippet added to the document being enriched. The following changes have been made to fix this bug: * Use `List<Map<?, ?>>` as cache value for the enrich cache. * Upon caching lookup / search, make an immutable deep copy of `SearchHit.getSourceAsMap()`. * Upon serving an entry from the cache, make a normal deep copy, so the enrichent can be safely modified by subsequent processors. Closes #82340
- Loading branch information
Showing
9 changed files
with
321 additions
and
191 deletions.
There are no files selected for viewing
171 changes: 171 additions & 0 deletions
171
...enrich/src/internalClusterTest/java/org/elasticsearch/xpack/enrich/EnrichProcessorIT.java
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,171 @@ | ||
| /* | ||
| * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one | ||
| * or more contributor license agreements. Licensed under the Elastic License | ||
| * 2.0; you may not use this file except in compliance with the Elastic License | ||
| * 2.0. | ||
| */ | ||
|
|
||
| package org.elasticsearch.xpack.enrich; | ||
|
|
||
| import org.elasticsearch.action.index.IndexRequest; | ||
| import org.elasticsearch.action.ingest.SimulateDocumentBaseResult; | ||
| import org.elasticsearch.action.ingest.SimulatePipelineRequest; | ||
| import org.elasticsearch.action.ingest.SimulatePipelineResponse; | ||
| import org.elasticsearch.action.support.WriteRequest; | ||
| import org.elasticsearch.common.bytes.BytesArray; | ||
| import org.elasticsearch.common.settings.Settings; | ||
| import org.elasticsearch.ingest.common.IngestCommonPlugin; | ||
| import org.elasticsearch.plugins.Plugin; | ||
| import org.elasticsearch.reindex.ReindexPlugin; | ||
| import org.elasticsearch.test.ESSingleNodeTestCase; | ||
| import org.elasticsearch.xcontent.XContentType; | ||
| import org.elasticsearch.xpack.core.XPackSettings; | ||
| import org.elasticsearch.xpack.core.enrich.EnrichPolicy; | ||
| import org.elasticsearch.xpack.core.enrich.action.EnrichStatsAction; | ||
| import org.elasticsearch.xpack.core.enrich.action.ExecuteEnrichPolicyAction; | ||
| import org.elasticsearch.xpack.core.enrich.action.PutEnrichPolicyAction; | ||
|
|
||
| import java.util.Arrays; | ||
| import java.util.Collection; | ||
| import java.util.Collections; | ||
|
|
||
| import static org.elasticsearch.xpack.enrich.AbstractEnrichTestCase.createSourceIndices; | ||
| import static org.hamcrest.Matchers.equalTo; | ||
| import static org.hamcrest.Matchers.nullValue; | ||
|
|
||
| public class EnrichProcessorIT extends ESSingleNodeTestCase { | ||
|
|
||
| @Override | ||
| protected Collection<Class<? extends Plugin>> getPlugins() { | ||
| return Arrays.asList(LocalStateEnrich.class, ReindexPlugin.class, IngestCommonPlugin.class); | ||
| } | ||
|
|
||
| @Override | ||
| protected Settings nodeSettings() { | ||
| return Settings.builder() | ||
| // TODO Change this to run with security enabled | ||
| // https://github.com/elastic/elasticsearch/issues/75940 | ||
| .put(XPackSettings.SECURITY_ENABLED.getKey(), false) | ||
| .build(); | ||
| } | ||
|
|
||
| public void testEnrichCacheValuesCannotBeCorrupted() { | ||
| // Ensure enrich cache is empty | ||
| EnrichStatsAction.Request statsRequest = new EnrichStatsAction.Request(); | ||
| EnrichStatsAction.Response statsResponse = client().execute(EnrichStatsAction.INSTANCE, statsRequest).actionGet(); | ||
| assertThat(statsResponse.getCacheStats().size(), equalTo(1)); | ||
| assertThat(statsResponse.getCacheStats().get(0).getCount(), equalTo(0L)); | ||
| assertThat(statsResponse.getCacheStats().get(0).getMisses(), equalTo(0L)); | ||
| assertThat(statsResponse.getCacheStats().get(0).getHits(), equalTo(0L)); | ||
|
|
||
| String policyName = "device-enrich-policy"; | ||
| String sourceIndexName = "devices-idx"; | ||
|
|
||
| EnrichPolicy enrichPolicy = new EnrichPolicy( | ||
| EnrichPolicy.MATCH_TYPE, | ||
| null, | ||
| Collections.singletonList(sourceIndexName), | ||
| "host.ip", | ||
| Arrays.asList("device.name", "host.ip") | ||
| ); | ||
|
|
||
| // Create source index and add a single document: | ||
| createSourceIndices(client(), enrichPolicy); | ||
| IndexRequest indexRequest = new IndexRequest(sourceIndexName); | ||
| indexRequest.create(true); | ||
| indexRequest.source("{\"host\": {\"ip\": \"10.151.80.8\"},\"device\": {\"name\": \"bla\"}}", XContentType.JSON); | ||
| indexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); | ||
| client().index(indexRequest).actionGet(); | ||
|
|
||
| // Store policy and execute it: | ||
| PutEnrichPolicyAction.Request putPolicyRequest = new PutEnrichPolicyAction.Request(policyName, enrichPolicy); | ||
| client().execute(PutEnrichPolicyAction.INSTANCE, putPolicyRequest).actionGet(); | ||
| ExecuteEnrichPolicyAction.Request executePolicyRequest = new ExecuteEnrichPolicyAction.Request(policyName); | ||
| client().execute(ExecuteEnrichPolicyAction.INSTANCE, executePolicyRequest).actionGet(); | ||
|
|
||
| SimulatePipelineRequest simulatePipelineRequest = new SimulatePipelineRequest( | ||
| new BytesArray( | ||
| "{\n" | ||
| + " \"pipeline\": {\n" | ||
| + " \"processors\": [\n" | ||
| + " {\n" | ||
| + " \"enrich\": {\n" | ||
| + " \"policy_name\": \"device-enrich-policy\",\n" | ||
| + " \"field\": \"host.ip\",\n" | ||
| + " \"target_field\": \"_tmp.device\"\n" | ||
| + " }\n" | ||
| + " },\n" | ||
| + " {\n" | ||
| + " \"rename\" : {\n" | ||
| + " \"field\" : \"_tmp.device.device.name\",\n" | ||
| + " \"target_field\" : \"device.name\"\n" | ||
| + " }\n" | ||
| + " }\n" | ||
| + " ]\n" | ||
| + " },\n" | ||
| + " \"docs\": [\n" | ||
| + " {\n" | ||
| + " \"_source\": {\n" | ||
| + " \"host\": {\n" | ||
| + " \"ip\": \"10.151.80.8\"\n" | ||
| + " }\n" | ||
| + " }\n" | ||
| + " }\n" | ||
| + " ]\n" | ||
| + " }" | ||
| ), | ||
| XContentType.JSON | ||
| ); | ||
| SimulatePipelineResponse response = client().admin().cluster().simulatePipeline(simulatePipelineRequest).actionGet(); | ||
| SimulateDocumentBaseResult result = (SimulateDocumentBaseResult) response.getResults().get(0); | ||
| assertThat(result.getFailure(), nullValue()); | ||
| assertThat(result.getIngestDocument().getFieldValue("device.name", String.class), equalTo("bla")); | ||
|
|
||
| // Verify that there was a cache miss and a new entry was added to enrich cache. | ||
| statsResponse = client().execute(EnrichStatsAction.INSTANCE, statsRequest).actionGet(); | ||
| assertThat(statsResponse.getCacheStats().size(), equalTo(1)); | ||
| assertThat(statsResponse.getCacheStats().get(0).getCount(), equalTo(1L)); | ||
| assertThat(statsResponse.getCacheStats().get(0).getMisses(), equalTo(1L)); | ||
| assertThat(statsResponse.getCacheStats().get(0).getHits(), equalTo(0L)); | ||
|
|
||
| simulatePipelineRequest = new SimulatePipelineRequest( | ||
| new BytesArray( | ||
| "{\n" | ||
| + " \"pipeline\": {\n" | ||
| + " \"processors\": [\n" | ||
| + " {\n" | ||
| + " \"enrich\": {\n" | ||
| + " \"policy_name\": \"device-enrich-policy\",\n" | ||
| + " \"field\": \"host.ip\",\n" | ||
| + " \"target_field\": \"_tmp\"\n" | ||
| + " }\n" | ||
| + " }\n" | ||
| + " ]\n" | ||
| + " },\n" | ||
| + " \"docs\": [\n" | ||
| + " {\n" | ||
| + " \"_source\": {\n" | ||
| + " \"host\": {\n" | ||
| + " \"ip\": \"10.151.80.8\"\n" | ||
| + " }\n" | ||
| + " }\n" | ||
| + " }\n" | ||
| + " ]\n" | ||
| + " }" | ||
| ), | ||
| XContentType.JSON | ||
| ); | ||
| response = client().admin().cluster().simulatePipeline(simulatePipelineRequest).actionGet(); | ||
| result = (SimulateDocumentBaseResult) response.getResults().get(0); | ||
| assertThat(result.getFailure(), nullValue()); | ||
| assertThat(result.getIngestDocument().getFieldValue("_tmp.device.name", String.class), equalTo("bla")); | ||
|
|
||
| // Verify that enrich lookup was served from cache: | ||
| statsResponse = client().execute(EnrichStatsAction.INSTANCE, statsRequest).actionGet(); | ||
| assertThat(statsResponse.getCacheStats().size(), equalTo(1)); | ||
| assertThat(statsResponse.getCacheStats().get(0).getCount(), equalTo(1L)); | ||
| assertThat(statsResponse.getCacheStats().get(0).getMisses(), equalTo(1L)); | ||
| assertThat(statsResponse.getCacheStats().get(0).getHits(), equalTo(1L)); | ||
| } | ||
|
|
||
| } |
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
Oops, something went wrong.