Replace adhoc pipeline registration with IndexTemplateRegistry

Author: parkertimmins

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ClientHelper.java

@@ -196,6 +196,7 @@ private static String maybeRewriteSingleAuthenticationHeaderForVersion(
     public static final String APM_ORIGIN = "apm";
     public static final String OTEL_ORIGIN = "otel";
     public static final String REINDEX_DATA_STREAM_ORIGIN = "reindex_data_stream";
+    public static final String MIGRATE_ORIGIN = "migrate";
 
     private ClientHelper() {}
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/user/InternalUsers.java

@@ -25,7 +25,6 @@
 import org.elasticsearch.action.datastreams.ModifyDataStreamsAction;
 import org.elasticsearch.action.downsample.DownsampleAction;
 import org.elasticsearch.action.index.TransportIndexAction;
-import org.elasticsearch.action.ingest.PutPipelineTransportAction;
 import org.elasticsearch.action.search.TransportSearchAction;
 import org.elasticsearch.action.search.TransportSearchScrollAction;
 import org.elasticsearch.index.reindex.ReindexAction;
@@ -197,9 +196,7 @@ public class InternalUsers {
         UsernamesField.REINDEX_DATA_STREAM_NAME,
         new RoleDescriptor(
             UsernamesField.REINDEX_DATA_STREAM_ROLE,
-            new String[] {
-                PutPipelineTransportAction.TYPE.name()
-            },
+            new String[] {},
             new RoleDescriptor.IndicesPrivileges[] {
                 RoleDescriptor.IndicesPrivileges.builder()
                     .indices("*")

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/user/InternalUsersTests.java

@@ -303,8 +303,7 @@ public void testReindexDataStreamUser() {
 
         final SimpleRole role = getLocalClusterRole(InternalUsers.REINDEX_DATA_STREAM_USER);
 
-        assertThat(role.cluster().privileges(), hasSize(1));
-        checkClusterAccess(InternalUsers.REINDEX_DATA_STREAM_USER, role, PutPipelineTransportAction.TYPE.name(), true);
+        assertThat(role.cluster(), is(ClusterPermission.NONE));
         assertThat(role.runAs(), is(RunAsPermission.NONE));
         assertThat(role.application(), is(ApplicationPermission.NONE));
         assertThat(role.remoteIndices(), is(RemoteIndicesPermission.NONE));

x-pack/plugin/core/template-resources/src/main/resources/reindex-data-stream-pipeline.json

@@ -0,0 +1,16 @@
+{
+  "description": "This pipeline sanitizes documents that are being reindexed into a data stream using the reindex data stream API. It is an internal pipeline and should not be modified.",
+  "processors": [
+    {
+      "set": {
+        "field": "@timestamp",
+        "value": 0,
+        "override": false
+      }
+    }
+  ],
+  "_meta": {
+    "managed": true
+  },
+  "version": ${xpack.migrate.reindex.pipeline.version}
+}

x-pack/plugin/migrate/src/internalClusterTest/java/org/elasticsearch/xpack/migrate/action/ReindexDatastreamIndexTransportActionIT.java

@@ -50,7 +50,7 @@
 import org.elasticsearch.test.transport.MockTransportService;
 import org.elasticsearch.xcontent.XContentType;
 import org.elasticsearch.xpack.migrate.MigratePlugin;
-import org.elasticsearch.xpack.migrate.ReindexDataStreamPipeline;
+import org.elasticsearch.xpack.migrate.MigrateTemplateRegistry;
 import org.junit.After;
 
 import java.io.IOException;
@@ -73,7 +73,7 @@ public class ReindexDatastreamIndexTransportActionIT extends ESIntegTestCase {
     private void cleanupCluster() throws Exception {
         clusterAdmin().execute(
             DeletePipelineTransportAction.TYPE,
-            new DeletePipelineRequest(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT, ReindexDataStreamPipeline.PIPELINE_NAME)
+            new DeletePipelineRequest(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT, MigrateTemplateRegistry.REINDEX_DATA_STREAM_PIPELINE_NAME)
         );
         super.cleanUpCluster();
     }
@@ -167,14 +167,15 @@ public void testCustomReindexPipeline() {
                         "value": "gorgonzola"
                       }
                     }
-                  ]
+                  ],
+                  "version": 1000
                 }
             """;
 
         PutPipelineRequest putRequest = new PutPipelineRequest(
             TEST_REQUEST_TIMEOUT,
             TEST_REQUEST_TIMEOUT,
-            ReindexDataStreamPipeline.PIPELINE_NAME,
+            MigrateTemplateRegistry.REINDEX_DATA_STREAM_PIPELINE_NAME,
             new BytesArray(customPipeline),
             XContentType.JSON
         );

x-pack/plugin/migrate/src/main/java/org/elasticsearch/xpack/migrate/MigratePlugin.java

@@ -55,6 +55,7 @@
 import org.elasticsearch.xpack.migrate.task.ReindexDataStreamTaskParams;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
@@ -64,6 +65,18 @@
 import static org.elasticsearch.xpack.migrate.task.ReindexDataStreamPersistentTaskExecutor.MAX_CONCURRENT_INDICES_REINDEXED_PER_DATA_STREAM_SETTING;
 
 public class MigratePlugin extends Plugin implements ActionPlugin, PersistentTaskPlugin {
+    @Override
+    public Collection<?> createComponents(PluginServices services) {
+        var registry = new MigrateTemplateRegistry(
+            services.environment().settings(),
+            services.clusterService(),
+            services.threadPool(),
+            services.client(),
+            services.xContentRegistry()
+        );
+        registry.initialize();
+        return List.of(registry);
+    }
 
     @Override
     public List<RestHandler> getRestHandlers(

x-pack/plugin/migrate/src/main/java/org/elasticsearch/xpack/migrate/MigrateTemplateRegistry.java

@@ -0,0 +1,55 @@
+/*
+ * 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.migrate;
+
+import org.elasticsearch.client.internal.Client;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.xcontent.NamedXContentRegistry;
+import org.elasticsearch.xpack.core.ClientHelper;
+import org.elasticsearch.xpack.core.template.IndexTemplateRegistry;
+import org.elasticsearch.xpack.core.template.IngestPipelineConfig;
+import org.elasticsearch.xpack.core.template.JsonIngestPipelineConfig;
+
+import java.util.List;
+
+public class MigrateTemplateRegistry extends IndexTemplateRegistry {
+
+    // This number must be incremented when we make changes to built-in pipeline.
+    // If a specific user pipeline is needed instead, its version should be set to a value higher than the REGISTRY_VERSION.
+    static final int REGISTRY_VERSION = 1;
+    public static final String REINDEX_DATA_STREAM_PIPELINE_NAME = "reindex-data-stream-pipeline";
+    private static final String TEMPLATE_VERSION_VARIABLE = "xpack.migrate.reindex.pipeline.version";
+
+    public MigrateTemplateRegistry(
+        Settings nodeSettings,
+        ClusterService clusterService,
+        ThreadPool threadPool,
+        Client client,
+        NamedXContentRegistry xContentRegistry
+    ) {
+        super(nodeSettings, clusterService, threadPool, client, xContentRegistry);
+    }
+
+    @Override
+    protected List<IngestPipelineConfig> getIngestPipelines() {
+        return List.of(
+            new JsonIngestPipelineConfig(
+                REINDEX_DATA_STREAM_PIPELINE_NAME,
+                "/" + REINDEX_DATA_STREAM_PIPELINE_NAME + ".json",
+                REGISTRY_VERSION,
+                TEMPLATE_VERSION_VARIABLE
+            )
+        );
+    }
+
+    @Override
+    protected String getOrigin() {
+        return ClientHelper.MIGRATE_ORIGIN;
+    }
+}

x-pack/plugin/migrate/src/main/java/org/elasticsearch/xpack/migrate/ReindexDataStreamPipeline.java

@@ -53,7 +53,7 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.deprecation.DeprecatedIndexPredicate;
-import org.elasticsearch.xpack.migrate.ReindexDataStreamPipeline;
+import org.elasticsearch.xpack.migrate.MigrateTemplateRegistry;
 
 import java.util.Locale;
 import java.util.Map;
@@ -156,8 +156,7 @@ protected void doExecute(
             return;
         }
         final boolean wasClosed = isClosed(sourceIndex);
-        SubscribableListener.<AcknowledgedResponse>newForked(l -> prepareReindexOperation(l, taskId))
-            .<AcknowledgedResponse>andThen(l -> setBlockWrites(sourceIndexName, l, taskId))
+        SubscribableListener.<AcknowledgedResponse>newForked(l -> setBlockWrites(sourceIndexName, l, taskId))
             .<OpenIndexResponse>andThen(l -> openIndexIfClosed(sourceIndexName, wasClosed, l, taskId))
             .<BroadcastResponse>andThen(l -> refresh(sourceIndexName, l, taskId))
             .<AcknowledgedResponse>andThen(l -> deleteDestIfExists(destIndexName, l, taskId))
@@ -201,14 +200,6 @@ private static boolean isClosed(IndexMetadata indexMetadata) {
         return indexMetadata.getState().equals(IndexMetadata.State.CLOSE);
     }
 
-    private void prepareReindexOperation(ActionListener<AcknowledgedResponse> listener, TaskId parentTaskId) {
-        if (ReindexDataStreamPipeline.exists(clusterService.state())) {
-            listener.onResponse(null);
-        } else {
-            ReindexDataStreamPipeline.create(client, listener, parentTaskId);
-        }
-    }
-
     private void setBlockWrites(String sourceIndexName, ActionListener<AcknowledgedResponse> listener, TaskId parentTaskId) {
         logger.debug("Setting write block on source index [{}]", sourceIndexName);
         addBlockToIndex(WRITE, sourceIndexName, new ActionListener<>() {
@@ -281,7 +272,7 @@ void reindex(String sourceIndexName, String destIndexName, ActionListener<BulkBy
         logger.debug("Reindex to destination index [{}] from source index [{}]", destIndexName, sourceIndexName);
         var reindexRequest = new ReindexRequest();
         reindexRequest.setSourceIndices(sourceIndexName);
-        reindexRequest.setDestPipeline(ReindexDataStreamPipeline.PIPELINE_NAME);
+        reindexRequest.setDestPipeline(MigrateTemplateRegistry.REINDEX_DATA_STREAM_PIPELINE_NAME);
         reindexRequest.getSearchRequest().allowPartialSearchResults(false);
         reindexRequest.getSearchRequest().source().fetchSource(true);
         reindexRequest.setDestIndex(destIndexName);

x-pack/plugin/migrate/src/main/java/org/elasticsearch/xpack/migrate/action/ReindexDataStreamIndexTransportAction.java

@@ -37,6 +37,7 @@
 import static org.elasticsearch.xpack.core.ClientHelper.INDEX_LIFECYCLE_ORIGIN;
 import static org.elasticsearch.xpack.core.ClientHelper.INFERENCE_ORIGIN;
 import static org.elasticsearch.xpack.core.ClientHelper.LOGSTASH_MANAGEMENT_ORIGIN;
+import static org.elasticsearch.xpack.core.ClientHelper.MIGRATE_ORIGIN;
 import static org.elasticsearch.xpack.core.ClientHelper.ML_ORIGIN;
 import static org.elasticsearch.xpack.core.ClientHelper.MONITORING_ORIGIN;
 import static org.elasticsearch.xpack.core.ClientHelper.OTEL_ORIGIN;
@@ -164,6 +165,7 @@ public static void switchUserBasedOnActionOriginAndExecute(
             case ENT_SEARCH_ORIGIN:
             case CONNECTORS_ORIGIN:
             case INFERENCE_ORIGIN:
+            case MIGRATE_ORIGIN:
             case TASKS_ORIGIN:   // TODO use a more limited user for tasks
                 securityContext.executeAsInternalUser(InternalUsers.XPACK_USER, version, consumer);
                 break;