Validate enrich index before completing policy execution (#100106) (#100158)

jbaiera · elasticmachine · web-flow · commit 88fef9a87241 · 2023-10-03T11:32:44.000-04:00
This PR adds a validation step to the end of an enrich policy run to ensure the integrity of the enrich index that is about to be promoted. (cherry picked from commit 225db31) Co-authored-by: Elastic Machine <elasticmachine@users.noreply.github.com>
diff --git a/docs/changelog/100106.yaml b/docs/changelog/100106.yaml
@@ -0,0 +1,5 @@
+pr: 100106
+summary: Validate enrich index before completing policy execution
+area: Ingest Node
+type: bug
+issues: []
diff --git a/x-pack/plugin/enrich/src/main/java/org/elasticsearch/xpack/enrich/EnrichPolicyRunner.java b/x-pack/plugin/enrich/src/main/java/org/elasticsearch/xpack/enrich/EnrichPolicyRunner.java
@@ -9,6 +9,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionResponse;
@@ -33,13 +34,15 @@
 import org.elasticsearch.client.internal.FilterClient;
 import org.elasticsearch.client.internal.OriginSettingClient;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.MappingMetadata;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.CheckedFunction;
+import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.index.reindex.BulkByScrollResponse;
@@ -123,13 +126,9 @@ public void run() {
             logger.debug("Policy [{}]: Checking source indices [{}]", policyName, sourceIndices);
             GetIndexRequest getIndexRequest = new GetIndexRequest().indices(sourceIndices);
             // This call does not set the origin to ensure that the user executing the policy has permission to access the source index
-            client.admin().indices().getIndex(getIndexRequest, listener.delegateFailure((l, getIndexResponse) -> {
-                try {
-                    validateMappings(getIndexResponse);
-                    prepareAndCreateEnrichIndex(toMappings(getIndexResponse));
-                } catch (Exception e) {
-                    l.onFailure(e);
-                }
+            client.admin().indices().getIndex(getIndexRequest, listener.delegateFailureAndWrap((l, getIndexResponse) -> {
+                validateMappings(getIndexResponse);
+                prepareAndCreateEnrichIndex(toMappings(getIndexResponse));
             }));
         } catch (Exception e) {
             listener.onFailure(e);
@@ -565,14 +564,66 @@ private void waitForIndexGreen(final String destinationIndexName) {
         ClusterHealthRequest request = new ClusterHealthRequest(destinationIndexName).waitForGreenStatus();
         enrichOriginClient().admin()
             .cluster()
-            .health(request, listener.delegateFailure((l, r) -> updateEnrichPolicyAlias(destinationIndexName)));
+            .health(request, listener.delegateFailureAndWrap((l, r) -> updateEnrichPolicyAlias(destinationIndexName)));
+    }
+
+    /**
+     * Ensures that the index we are about to promote at the end of a policy execution exists, is intact, and has not been damaged
+     * during the policy execution. In some cases, it is possible for the index being constructed to be deleted during the policy execution
+     * and recreated with invalid mappings/data. We validate that the mapping exists and that it contains the expected meta fields on it to
+     * guard against accidental removal and recreation during policy execution.
+     */
+    private void validateIndexBeforePromotion(String destinationIndexName, ClusterState clusterState) {
+        IndexMetadata destinationIndex = clusterState.metadata().index(destinationIndexName);
+        if (destinationIndex == null) {
+            throw new IndexNotFoundException(
+                "was not able to promote it as part of executing enrich policy [" + policyName + "]",
+                destinationIndexName
+            );
+        }
+        MappingMetadata mapping = destinationIndex.mapping();
+        if (mapping == null) {
+            throw new ResourceNotFoundException(
+                "Could not locate mapping for enrich index [{}] while completing [{}] policy run",
+                destinationIndexName,
+                policyName
+            );
+        }
+        Map<String, Object> mappingSource = mapping.sourceAsMap();
+        Object meta = mappingSource.get("_meta");
+        if (meta instanceof Map<?, ?> metaMap) {
+            Object policyNameMetaField = metaMap.get(ENRICH_POLICY_NAME_FIELD_NAME);
+            if (policyNameMetaField == null) {
+                throw new ElasticsearchException(
+                    "Could not verify enrich index [{}] metadata before completing [{}] policy run: policy name meta field missing",
+                    destinationIndexName,
+                    policyName
+                );
+            } else if (policyName.equals(policyNameMetaField) == false) {
+                throw new ElasticsearchException(
+                    "Could not verify enrich index [{}] metadata before completing [{}] policy run: policy name meta field does not "
+                        + "match expected value of [{}], was [{}]",
+                    destinationIndexName,
+                    policyName,
+                    policyName,
+                    policyNameMetaField.toString()
+                );
+            }
+        } else {
+            throw new ElasticsearchException(
+                "Could not verify enrich index [{}] metadata before completing [{}] policy run: mapping meta field missing",
+                destinationIndexName,
+                policyName
+            );
+        }
     }
 
     private void updateEnrichPolicyAlias(final String destinationIndexName) {
         String enrichIndexBase = EnrichPolicy.getBaseName(policyName);
         logger.debug("Policy [{}]: Promoting new enrich index [{}] to alias [{}]", policyName, destinationIndexName, enrichIndexBase);
         GetAliasesRequest aliasRequest = new GetAliasesRequest(enrichIndexBase);
         ClusterState clusterState = clusterService.state();
+        validateIndexBeforePromotion(destinationIndexName, clusterState);
         String[] concreteIndices = indexNameExpressionResolver.concreteIndexNamesWithSystemIndexAccess(clusterState, aliasRequest);
         String[] aliases = aliasRequest.aliases();
         IndicesAliasesRequest aliasToggleRequest = new IndicesAliasesRequest();
diff --git a/x-pack/plugin/enrich/src/test/java/org/elasticsearch/xpack/enrich/EnrichPolicyRunnerTests.java b/x-pack/plugin/enrich/src/test/java/org/elasticsearch/xpack/enrich/EnrichPolicyRunnerTests.java
@@ -16,6 +16,8 @@
 import org.elasticsearch.action.admin.indices.create.CreateIndexAction;
 import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
 import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexAction;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
 import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeAction;
 import org.elasticsearch.action.admin.indices.get.GetIndexAction;
 import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
@@ -1980,6 +1982,79 @@ public void testRunRangePolicyWithObjectFieldAsMatchField() throws Exception {
         assertThat(e.getMessage(), equalTo("Field 'field1' has type [object] which doesn't appear to be a range type"));
     }
 
+    public void testRunnerValidatesIndexIntegrity() throws Exception {
+        final String sourceIndex = "source-index";
+        IndexResponse indexRequest = client().index(new IndexRequest().index(sourceIndex).id("id").source("""
+            {
+              "field1": "value1",
+              "field2": 2,
+              "field3": "ignored",
+              "field4": "ignored",
+              "field5": "value5"
+            }""", XContentType.JSON).setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)).actionGet();
+        assertEquals(RestStatus.CREATED, indexRequest.status());
+
+        SearchResponse sourceSearchResponse = client().search(
+            new SearchRequest(sourceIndex).source(SearchSourceBuilder.searchSource().query(QueryBuilders.matchAllQuery()))
+        ).actionGet();
+        assertThat(sourceSearchResponse.getHits().getTotalHits().value, equalTo(1L));
+        Map<String, Object> sourceDocMap = sourceSearchResponse.getHits().getAt(0).getSourceAsMap();
+        assertNotNull(sourceDocMap);
+        assertThat(sourceDocMap.get("field1"), is(equalTo("value1")));
+        assertThat(sourceDocMap.get("field2"), is(equalTo(2)));
+        assertThat(sourceDocMap.get("field3"), is(equalTo("ignored")));
+        assertThat(sourceDocMap.get("field4"), is(equalTo("ignored")));
+        assertThat(sourceDocMap.get("field5"), is(equalTo("value5")));
+
+        List<String> enrichFields = List.of("field2", "field5");
+        EnrichPolicy policy = new EnrichPolicy(EnrichPolicy.MATCH_TYPE, null, List.of(sourceIndex), "field1", enrichFields);
+        String policyName = "test1";
+
+        final long createTime = randomNonNegativeLong();
+        String createdEnrichIndex = ".enrich-test1-" + createTime;
+        final AtomicReference<Exception> exception = new AtomicReference<>();
+        final CountDownLatch latch = new CountDownLatch(1);
+        ActionListener<ExecuteEnrichPolicyStatus> listener = createTestListener(latch, exception::set);
+
+        // Wrap the client so that when we receive the reindex action, we delete the index then resume operation. This mimics an invalid
+        // state for the resulting index.
+        Client client = new FilterClient(client()) {
+            @Override
+            protected <Request extends ActionRequest, Response extends ActionResponse> void doExecute(
+                ActionType<Response> action,
+                Request request,
+                ActionListener<Response> listener
+            ) {
+                if (action.equals(EnrichReindexAction.INSTANCE)) {
+                    super.doExecute(
+                        DeleteIndexAction.INSTANCE,
+                        new DeleteIndexRequest(createdEnrichIndex),
+                        listener.delegateFailureAndWrap((delegate, response) -> {
+                            if (response.isAcknowledged() == false) {
+                                fail("Enrich index should have been deleted but was not");
+                            }
+                            super.doExecute(action, request, delegate);
+                        })
+                    );
+                } else {
+                    super.doExecute(action, request, listener);
+                }
+            }
+        };
+        EnrichPolicyRunner enrichPolicyRunner = createPolicyRunner(client, policyName, policy, listener, createdEnrichIndex);
+
+        logger.info("Starting policy run");
+        enrichPolicyRunner.run();
+        latch.await();
+        Exception runnerException = exception.get();
+        if (runnerException == null) {
+            fail("Expected the runner to fail when the underlying index was deleted during policy execution!");
+        }
+        assertThat(runnerException, is(instanceOf(ElasticsearchException.class)));
+        assertThat(runnerException.getMessage(), containsString("Could not verify enrich index"));
+        assertThat(runnerException.getMessage(), containsString("mapping meta field missing"));
+    }
+
     private EnrichPolicyRunner createPolicyRunner(
         String policyName,
         EnrichPolicy policy,
