Adding reindex data stream rest action (#118109)

* Adding a _migration/reindex endpoint

* Adding rest api spec and test

* Adding a feature flag for reindex data streams

* updating json spec

* fixing a typo

* Changing mode to an enum

* Moving ParseFields into public static finals

* Commenting out test that leaves task running, until we add a cancel API

* Removing persistent task id from output

* replacing a string with a variable

Author: masseyke

rest-api-spec/src/main/resources/rest-api-spec/api/migrate.reindex.json

@@ -0,0 +1,29 @@
+{
+  "migrate.reindex":{
+    "documentation":{
+      "url":"https://www.elastic.co/guide/en/elasticsearch/reference/master/data-stream-reindex.html",
+      "description":"This API reindexes all legacy backing indices for a data stream. It does this in a persistent task. The persistent task id is returned immediately, and the reindexing work is completed in that task"
+    },
+    "stability":"experimental",
+    "visibility":"private",
+    "headers":{
+      "accept": [ "application/json"],
+      "content_type": ["application/json"]
+    },
+    "url":{
+      "paths":[
+        {
+          "path":"/_migration/reindex",
+          "methods":[
+            "POST"
+          ]
+        }
+      ]
+    },
+    "body":{
+      "description":"The body contains the fields `mode` and `source.index, where the only mode currently supported is `upgrade`, and the `source.index` must be a data stream name",
+      "required":true
+    }
+  }
+}
+

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

@@ -51,7 +51,10 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
 
     public void testNonExistentDataStream() {
         String nonExistentDataStreamName = randomAlphaOfLength(50);
-        ReindexDataStreamRequest reindexDataStreamRequest = new ReindexDataStreamRequest(nonExistentDataStreamName);
+        ReindexDataStreamRequest reindexDataStreamRequest = new ReindexDataStreamRequest(
+            ReindexDataStreamAction.Mode.UPGRADE,
+            nonExistentDataStreamName
+        );
         assertThrows(
             ResourceNotFoundException.class,
             () -> client().execute(new ActionType<ReindexDataStreamResponse>(ReindexDataStreamAction.NAME), reindexDataStreamRequest)
@@ -61,7 +64,10 @@ public void testNonExistentDataStream() {
 
     public void testAlreadyUpToDateDataStream() throws Exception {
         String dataStreamName = randomAlphaOfLength(50).toLowerCase(Locale.ROOT);
-        ReindexDataStreamRequest reindexDataStreamRequest = new ReindexDataStreamRequest(dataStreamName);
+        ReindexDataStreamRequest reindexDataStreamRequest = new ReindexDataStreamRequest(
+            ReindexDataStreamAction.Mode.UPGRADE,
+            dataStreamName
+        );
         createDataStream(dataStreamName);
         ReindexDataStreamResponse response = client().execute(
             new ActionType<ReindexDataStreamResponse>(ReindexDataStreamAction.NAME),

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

@@ -11,21 +11,30 @@
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.common.settings.ClusterSettings;
+import org.elasticsearch.common.settings.IndexScopedSettings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.settings.SettingsFilter;
 import org.elasticsearch.common.settings.SettingsModule;
+import org.elasticsearch.features.NodeFeature;
 import org.elasticsearch.persistent.PersistentTaskParams;
 import org.elasticsearch.persistent.PersistentTaskState;
 import org.elasticsearch.persistent.PersistentTasksExecutor;
 import org.elasticsearch.plugins.ActionPlugin;
 import org.elasticsearch.plugins.PersistentTaskPlugin;
 import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.rest.RestController;
+import org.elasticsearch.rest.RestHandler;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xcontent.ParseField;
 import org.elasticsearch.xpack.migrate.action.ReindexDataStreamAction;
 import org.elasticsearch.xpack.migrate.action.ReindexDataStreamTransportAction;
+import org.elasticsearch.xpack.migrate.rest.RestMigrationReindexAction;
 import org.elasticsearch.xpack.migrate.task.ReindexDataStreamPersistentTaskExecutor;
 import org.elasticsearch.xpack.migrate.task.ReindexDataStreamPersistentTaskState;
 import org.elasticsearch.xpack.migrate.task.ReindexDataStreamStatus;
@@ -34,47 +43,80 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+
+import static org.elasticsearch.xpack.migrate.action.ReindexDataStreamAction.REINDEX_DATA_STREAM_FEATURE_FLAG;
 
 public class MigratePlugin extends Plugin implements ActionPlugin, PersistentTaskPlugin {
 
+    @Override
+    public List<RestHandler> getRestHandlers(
+        Settings unused,
+        NamedWriteableRegistry namedWriteableRegistry,
+        RestController restController,
+        ClusterSettings clusterSettings,
+        IndexScopedSettings indexScopedSettings,
+        SettingsFilter settingsFilter,
+        IndexNameExpressionResolver indexNameExpressionResolver,
+        Supplier<DiscoveryNodes> nodesInCluster,
+        Predicate<NodeFeature> clusterSupportsFeature
+    ) {
+        List<RestHandler> handlers = new ArrayList<>();
+        if (REINDEX_DATA_STREAM_FEATURE_FLAG.isEnabled()) {
+            handlers.add(new RestMigrationReindexAction());
+        }
+        return handlers;
+    }
+
     @Override
     public List<ActionHandler<? extends ActionRequest, ? extends ActionResponse>> getActions() {
         List<ActionHandler<? extends ActionRequest, ? extends ActionResponse>> actions = new ArrayList<>();
-        actions.add(new ActionHandler<>(ReindexDataStreamAction.INSTANCE, ReindexDataStreamTransportAction.class));
+        if (REINDEX_DATA_STREAM_FEATURE_FLAG.isEnabled()) {
+            actions.add(new ActionHandler<>(ReindexDataStreamAction.INSTANCE, ReindexDataStreamTransportAction.class));
+        }
         return actions;
     }
 
     @Override
     public List<NamedXContentRegistry.Entry> getNamedXContent() {
-        return List.of(
-            new NamedXContentRegistry.Entry(
-                PersistentTaskState.class,
-                new ParseField(ReindexDataStreamPersistentTaskState.NAME),
-                ReindexDataStreamPersistentTaskState::fromXContent
-            ),
-            new NamedXContentRegistry.Entry(
-                PersistentTaskParams.class,
-                new ParseField(ReindexDataStreamTaskParams.NAME),
-                ReindexDataStreamTaskParams::fromXContent
-            )
-        );
+        if (REINDEX_DATA_STREAM_FEATURE_FLAG.isEnabled()) {
+            return List.of(
+                new NamedXContentRegistry.Entry(
+                    PersistentTaskState.class,
+                    new ParseField(ReindexDataStreamPersistentTaskState.NAME),
+                    ReindexDataStreamPersistentTaskState::fromXContent
+                ),
+                new NamedXContentRegistry.Entry(
+                    PersistentTaskParams.class,
+                    new ParseField(ReindexDataStreamTaskParams.NAME),
+                    ReindexDataStreamTaskParams::fromXContent
+                )
+            );
+        } else {
+            return List.of();
+        }
     }
 
     @Override
     public List<NamedWriteableRegistry.Entry> getNamedWriteables() {
-        return List.of(
-            new NamedWriteableRegistry.Entry(
-                PersistentTaskState.class,
-                ReindexDataStreamPersistentTaskState.NAME,
-                ReindexDataStreamPersistentTaskState::new
-            ),
-            new NamedWriteableRegistry.Entry(
-                PersistentTaskParams.class,
-                ReindexDataStreamTaskParams.NAME,
-                ReindexDataStreamTaskParams::new
-            ),
-            new NamedWriteableRegistry.Entry(Task.Status.class, ReindexDataStreamStatus.NAME, ReindexDataStreamStatus::new)
-        );
+        if (REINDEX_DATA_STREAM_FEATURE_FLAG.isEnabled()) {
+            return List.of(
+                new NamedWriteableRegistry.Entry(
+                    PersistentTaskState.class,
+                    ReindexDataStreamPersistentTaskState.NAME,
+                    ReindexDataStreamPersistentTaskState::new
+                ),
+                new NamedWriteableRegistry.Entry(
+                    PersistentTaskParams.class,
+                    ReindexDataStreamTaskParams.NAME,
+                    ReindexDataStreamTaskParams::new
+                ),
+                new NamedWriteableRegistry.Entry(Task.Status.class, ReindexDataStreamStatus.NAME, ReindexDataStreamStatus::new)
+            );
+        } else {
+            return List.of();
+        }
     }
 
     @Override
@@ -85,6 +127,12 @@ public List<PersistentTasksExecutor<?>> getPersistentTasksExecutor(
         SettingsModule settingsModule,
         IndexNameExpressionResolver expressionResolver
     ) {
-        return List.of(new ReindexDataStreamPersistentTaskExecutor(client, clusterService, ReindexDataStreamTask.TASK_NAME, threadPool));
+        if (REINDEX_DATA_STREAM_FEATURE_FLAG.isEnabled()) {
+            return List.of(
+                new ReindexDataStreamPersistentTaskExecutor(client, clusterService, ReindexDataStreamTask.TASK_NAME, threadPool)
+            );
+        } else {
+            return List.of();
+        }
     }
 }

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

@@ -11,23 +11,41 @@
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.IndicesRequest;
+import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.util.FeatureFlag;
+import org.elasticsearch.features.NodeFeature;
+import org.elasticsearch.xcontent.ConstructingObjectParser;
+import org.elasticsearch.xcontent.ParseField;
+import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.ToXContentObject;
 import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.XContentParser;
 
 import java.io.IOException;
+import java.util.Locale;
 import java.util.Objects;
+import java.util.function.Predicate;
 
 public class ReindexDataStreamAction extends ActionType<ReindexDataStreamAction.ReindexDataStreamResponse> {
+    public static final FeatureFlag REINDEX_DATA_STREAM_FEATURE_FLAG = new FeatureFlag("reindex_data_stream");
 
     public static final ReindexDataStreamAction INSTANCE = new ReindexDataStreamAction();
     public static final String NAME = "indices:admin/data_stream/reindex";
+    public static final ParseField MODE_FIELD = new ParseField("mode");
+    public static final ParseField SOURCE_FIELD = new ParseField("source");
+    public static final ParseField INDEX_FIELD = new ParseField("index");
 
     public ReindexDataStreamAction() {
         super(NAME);
     }
 
+    public enum Mode {
+        UPGRADE
+    }
+
     public static class ReindexDataStreamResponse extends ActionResponse implements ToXContentObject {
         private final String taskId;
 
@@ -49,7 +67,7 @@ public void writeTo(StreamOutput out) throws IOException {
         @Override
         public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
             builder.startObject();
-            builder.field("task", getTaskId());
+            builder.field("acknowledged", true);
             builder.endObject();
             return builder;
         }
@@ -70,22 +88,52 @@ public boolean equals(Object other) {
 
     }
 
-    public static class ReindexDataStreamRequest extends ActionRequest {
+    public static class ReindexDataStreamRequest extends ActionRequest implements IndicesRequest, ToXContent {
+        private final Mode mode;
         private final String sourceDataStream;
 
-        public ReindexDataStreamRequest(String sourceDataStream) {
-            super();
+        public ReindexDataStreamRequest(Mode mode, String sourceDataStream) {
+            this.mode = mode;
             this.sourceDataStream = sourceDataStream;
         }
 
         public ReindexDataStreamRequest(StreamInput in) throws IOException {
             super(in);
+            this.mode = Mode.valueOf(in.readString());
             this.sourceDataStream = in.readString();
         }
 
+        private static final ConstructingObjectParser<ReindexDataStreamRequest, Predicate<NodeFeature>> PARSER =
+            new ConstructingObjectParser<>("migration_reindex", objects -> {
+                Mode mode = Mode.valueOf(((String) objects[0]).toUpperCase(Locale.ROOT));
+                String source = (String) objects[1];
+                return new ReindexDataStreamRequest(mode, source);
+            });
+
+        private static final ConstructingObjectParser<String, Void> SOURCE_PARSER = new ConstructingObjectParser<>(
+            SOURCE_FIELD.getPreferredName(),
+            false,
+            (a, id) -> (String) a[0]
+        );
+
+        static {
+            SOURCE_PARSER.declareString(ConstructingObjectParser.constructorArg(), INDEX_FIELD);
+            PARSER.declareString(ConstructingObjectParser.constructorArg(), MODE_FIELD);
+            PARSER.declareObject(
+                ConstructingObjectParser.constructorArg(),
+                (parser, id) -> SOURCE_PARSER.apply(parser, null),
+                SOURCE_FIELD
+            );
+        }
+
+        public static ReindexDataStreamRequest fromXContent(XContentParser parser) {
+            return PARSER.apply(parser, null);
+        }
+
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
+            out.writeString(mode.name());
             out.writeString(sourceDataStream);
         }
 
@@ -103,15 +151,42 @@ public String getSourceDataStream() {
             return sourceDataStream;
         }
 
+        public Mode getMode() {
+            return mode;
+        }
+
         @Override
         public int hashCode() {
-            return Objects.hashCode(sourceDataStream);
+            return Objects.hash(mode, sourceDataStream);
         }
 
         @Override
         public boolean equals(Object other) {
-            return other instanceof ReindexDataStreamRequest
-                && sourceDataStream.equals(((ReindexDataStreamRequest) other).sourceDataStream);
+            return other instanceof ReindexDataStreamRequest otherRequest
+                && mode.equals(otherRequest.mode)
+                && sourceDataStream.equals(otherRequest.sourceDataStream);
+        }
+
+        @Override
+        public String[] indices() {
+            return new String[] { sourceDataStream };
+        }
+
+        @Override
+        public IndicesOptions indicesOptions() {
+            return IndicesOptions.strictSingleIndexNoExpandForbidClosed();
+        }
+
+        /*
+         * This only exists for the sake of testing the xcontent parser
+         */
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            builder.field(MODE_FIELD.getPreferredName(), mode);
+            builder.startObject(SOURCE_FIELD.getPreferredName());
+            builder.field(INDEX_FIELD.getPreferredName(), sourceDataStream);
+            builder.endObject();
+            return builder;
         }
     }
 }