Add pipeline to ensure unique Enrich index documents (#46348)

Adds a pipeline that removes ids and routing from documents before indexing
them into enrich indices. Enrich documents may come from multiple indices,
and thus have id collisions on them. This pipeline ensures that documents
with colliding id fields do not clobber one another during the reindex operation
while executing an enrich policy.

Author: jbaiera

x-pack/plugin/enrich/src/main/java/org/elasticsearch/xpack/enrich/EnrichPolicyReindexPipeline.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.enrich;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ingest.PutPipelineRequest;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.ingest.IngestMetadata;
+import org.elasticsearch.ingest.PipelineConfiguration;
+
+/**
+ * Manages the definitions and lifecycle of the ingest pipeline used by the reindex operation within the Enrich Policy execution.
+ */
+public class EnrichPolicyReindexPipeline {
+
+    /**
+     * The current version of the pipeline definition. Used in the pipeline's name to differentiate from breaking changes
+     * (separate from product version).
+     */
+    static final String CURRENT_PIPELINE_VERSION_NAME = "7";
+
+    /**
+     * The last version of the distribution that updated the pipelines definition.
+     * TODO: This should be the version of ES that Enrich first ships in, which likely doesn't exist yet.
+     */
+    static final int ENRICH_PIPELINE_LAST_UPDATED_VERSION = Version.V_7_4_0.id;
+
+    static String pipelineName() {
+        return "enrich-policy-reindex-" + CURRENT_PIPELINE_VERSION_NAME;
+    }
+
+    /**
+     * Checks if the current version of the pipeline definition is installed in the cluster
+     * @param clusterState The cluster state to check
+     * @return true if a pipeline exists that is compatible with this version of Enrich, false otherwise
+     */
+    static boolean exists(ClusterState clusterState) {
+        final IngestMetadata ingestMetadata = clusterState.getMetaData().custom(IngestMetadata.TYPE);
+        // we ensure that we both have the pipeline and its version represents the current (or later) version
+        if (ingestMetadata != null) {
+            final PipelineConfiguration pipeline = ingestMetadata.getPipelines().get(pipelineName());
+            if (pipeline != null) {
+                Object version = pipeline.getConfigAsMap().get("version");
+                return version instanceof Number && ((Number) version).intValue() >= ENRICH_PIPELINE_LAST_UPDATED_VERSION;
+            }
+        }
+        return false;
+    }
+
+    /**
+     * Creates a pipeline with the current version's pipeline definition
+     * @param client Client used to execute put pipeline
+     * @param listener Callback used after pipeline has been created
+     */
+    public static void create(Client client, ActionListener<AcknowledgedResponse> listener) {
+        final BytesReference pipeline = BytesReference.bytes(currentEnrichPipelineDefinition(XContentType.JSON));
+        final PutPipelineRequest request = new PutPipelineRequest(pipelineName(), pipeline, XContentType.JSON);
+        client.admin().cluster().putPipeline(request, listener);
+    }
+
+    private static XContentBuilder currentEnrichPipelineDefinition(XContentType xContentType) {
+        try {
+            return XContentBuilder.builder(xContentType.xContent())
+                .startObject()
+                    .field("description", "This pipeline sanitizes documents that will be stored in enrich indices for ingest lookup " +
+                        "purposes. It is an internal pipeline and should not be modified.")
+                    .field("version", ENRICH_PIPELINE_LAST_UPDATED_VERSION)
+                    .startArray("processors")
+                        .startObject()
+                            // remove the id from the document so that documents from multiple indices will always be unique.
+                            .startObject("remove")
+                                .field("field", "_id")
+                            .endObject()
+                        .endObject()
+                    .endArray()
+                .endObject();
+        } catch (final IOException e) {
+            throw new UncheckedIOException("Failed to create pipeline for enrich document sanitization", e);
+        }
+    }
+
+}

x-pack/plugin/enrich/src/main/java/org/elasticsearch/xpack/enrich/EnrichPolicyRunner.java

@@ -250,7 +250,7 @@ private void prepareAndCreateEnrichIndex() {
         client.admin().indices().create(createEnrichIndexRequest, new ActionListener<CreateIndexResponse>() {
             @Override
             public void onResponse(CreateIndexResponse createIndexResponse) {
-                transferDataToEnrichIndex(enrichIndexName);
+                prepareReindexOperation(enrichIndexName);
             }
 
             @Override
@@ -260,6 +260,25 @@ public void onFailure(Exception e) {
         });
     }
 
+    private void prepareReindexOperation(final String destinationIndexName) {
+        // Check to make sure that the enrich pipeline exists, and create it if it is missing.
+        if (EnrichPolicyReindexPipeline.exists(clusterService.state()) == false) {
+            EnrichPolicyReindexPipeline.create(client, new ActionListener<AcknowledgedResponse>() {
+                @Override
+                public void onResponse(AcknowledgedResponse acknowledgedResponse) {
+                    transferDataToEnrichIndex(destinationIndexName);
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    listener.onFailure(e);
+                }
+            });
+        } else {
+            transferDataToEnrichIndex(destinationIndexName);
+        }
+    }
+
     private void transferDataToEnrichIndex(final String destinationIndexName) {
         logger.debug("Policy [{}]: Transferring source data to new enrich index [{}]", policyName, destinationIndexName);
         // Filter down the source fields to just the ones required by the policy
@@ -277,6 +296,8 @@ private void transferDataToEnrichIndex(final String destinationIndexName) {
             .setSourceIndices(policy.getIndices().toArray(new String[0]));
         reindexRequest.getSearchRequest().source(searchSourceBuilder);
         reindexRequest.getDestination().source(new BytesArray(new byte[0]), XContentType.SMILE);
+        reindexRequest.getDestination().routing("discard");
+        reindexRequest.getDestination().setPipeline(EnrichPolicyReindexPipeline.pipelineName());
         client.execute(ReindexAction.INSTANCE, reindexRequest, new ActionListener<BulkByScrollResponse>() {
             @Override
             public void onResponse(BulkByScrollResponse bulkByScrollResponse) {

x-pack/plugin/enrich/src/test/java/org/elasticsearch/xpack/enrich/BasicEnrichTests.java

@@ -17,6 +17,7 @@
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.reindex.ReindexPlugin;
+import org.elasticsearch.ingest.common.IngestCommonPlugin;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.test.ESSingleNodeTestCase;
 import org.elasticsearch.xpack.core.enrich.EnrichPolicy;
@@ -46,7 +47,7 @@ public class BasicEnrichTests extends ESSingleNodeTestCase {
 
     @Override
     protected Collection<Class<? extends Plugin>> getPlugins() {
-        return Arrays.asList(LocalStateEnrich.class, ReindexPlugin.class);
+        return Arrays.asList(LocalStateEnrich.class, ReindexPlugin.class, IngestCommonPlugin.class);
     }
 
     @Override

x-pack/plugin/enrich/src/test/java/org/elasticsearch/xpack/enrich/EnrichMultiNodeIT.java

@@ -21,6 +21,7 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.reindex.ReindexPlugin;
+import org.elasticsearch.ingest.common.IngestCommonPlugin;
 import org.elasticsearch.node.Node;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.test.ESIntegTestCase;
@@ -59,7 +60,7 @@ public class EnrichMultiNodeIT extends ESIntegTestCase {
 
     @Override
     protected Collection<Class<? extends Plugin>> nodePlugins() {
-        return Arrays.asList(LocalStateEnrich.class, ReindexPlugin.class);
+        return Arrays.asList(LocalStateEnrich.class, ReindexPlugin.class, IngestCommonPlugin.class);
     }
 
     @Override