Run TransportExplainDataStreamLifecycleAction on local node

This action solely needs the cluster state, it can run on any node.
Additionally, it needs to be cancellable to avoid doing unnecessary work
after a client failure or timeout.

Relates #101805

Author: nielsbauman

modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/action/TransportExplainDataStreamLifecycleAction.java

@@ -14,7 +14,8 @@
 import org.elasticsearch.action.datastreams.lifecycle.ExplainDataStreamLifecycleAction;
 import org.elasticsearch.action.datastreams.lifecycle.ExplainIndexDataStreamLifecycle;
 import org.elasticsearch.action.support.ActionFilters;
-import org.elasticsearch.action.support.master.TransportMasterNodeReadAction;
+import org.elasticsearch.action.support.ChannelActionListener;
+import org.elasticsearch.action.support.local.TransportLocalClusterStateAction;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
@@ -28,8 +29,10 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.core.UpdateForV10;
 import org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleErrorStore;
 import org.elasticsearch.injection.guice.Inject;
+import org.elasticsearch.tasks.CancellableTask;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
@@ -40,13 +43,20 @@
 /**
  * Transport action handling the explain the data stream lifecycle requests for one or more data stream lifecycle managed indices.
  */
-public class TransportExplainDataStreamLifecycleAction extends TransportMasterNodeReadAction<
+public class TransportExplainDataStreamLifecycleAction extends TransportLocalClusterStateAction<
     ExplainDataStreamLifecycleAction.Request,
     ExplainDataStreamLifecycleAction.Response> {
 
+    private final IndexNameExpressionResolver indexNameExpressionResolver;
     private final DataStreamLifecycleErrorStore errorStore;
     private final DataStreamGlobalRetentionSettings globalRetentionSettings;
 
+    /**
+     * NB prior to 9.0 this was a TransportMasterNodeReadAction so for BwC it must be registered with the TransportService until
+     * we no longer need to support calling this action remotely.
+     */
+    @UpdateForV10(owner = UpdateForV10.Owner.DATA_MANAGEMENT)
+    @SuppressWarnings("this-escape")
     @Inject
     public TransportExplainDataStreamLifecycleAction(
         TransportService transportService,
@@ -58,28 +68,33 @@ public TransportExplainDataStreamLifecycleAction(
         DataStreamGlobalRetentionSettings globalRetentionSettings
     ) {
         super(
-            ExplainDataStreamLifecycleAction.INSTANCE.name(),
-            transportService,
-            clusterService,
-            threadPool,
+            ExplainDataStreamLifecycleAction.NAME,
             actionFilters,
-            ExplainDataStreamLifecycleAction.Request::new,
-            indexNameExpressionResolver,
-            ExplainDataStreamLifecycleAction.Response::new,
+            transportService.getTaskManager(),
+            clusterService,
             threadPool.executor(ThreadPool.Names.MANAGEMENT)
         );
+        this.indexNameExpressionResolver = indexNameExpressionResolver;
         this.errorStore = dataLifecycleServiceErrorStore;
         this.globalRetentionSettings = globalRetentionSettings;
+
+        transportService.registerRequestHandler(
+            actionName,
+            executor,
+            false,
+            true,
+            ExplainDataStreamLifecycleAction.Request::new,
+            (request, channel, task) -> executeDirect(task, request, new ChannelActionListener<>(channel))
+        );
     }
 
     @Override
-    protected void masterOperation(
+    protected void localClusterStateOperation(
         Task task,
         ExplainDataStreamLifecycleAction.Request request,
         ClusterState state,
         ActionListener<ExplainDataStreamLifecycleAction.Response> listener
     ) throws Exception {
-
         String[] concreteIndices = indexNameExpressionResolver.concreteIndexNames(state, request);
         List<ExplainIndexDataStreamLifecycle> explainIndices = new ArrayList<>(concreteIndices.length);
         Metadata metadata = state.metadata();
@@ -114,6 +129,7 @@ protected void masterOperation(
             explainIndices.add(explainIndexDataStreamLifecycle);
         }
 
+        ((CancellableTask) task).ensureNotCancelled();
         ClusterSettings clusterSettings = clusterService.getClusterSettings();
         listener.onResponse(
             new ExplainDataStreamLifecycleAction.Response(

modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/rest/RestExplainDataStreamLifecycleAction.java

@@ -18,6 +18,7 @@
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
+import org.elasticsearch.rest.action.RestCancellableNodeClient;
 import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 
 import java.util.List;
@@ -48,7 +49,7 @@ protected RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient
         );
         explainRequest.includeDefaults(restRequest.paramAsBoolean("include_defaults", false));
         explainRequest.indicesOptions(IndicesOptions.fromRequest(restRequest, IndicesOptions.strictExpandOpen()));
-        return channel -> client.execute(
+        return channel -> new RestCancellableNodeClient(client, restRequest.getHttpChannel()).execute(
             ExplainDataStreamLifecycleAction.INSTANCE,
             explainRequest,
             new RestRefCountedChunkedToXContentListener<>(channel)

server/src/main/java/org/elasticsearch/action/datastreams/lifecycle/ExplainDataStreamLifecycleAction.java

@@ -16,7 +16,7 @@
 import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.admin.indices.rollover.RolloverConfiguration;
 import org.elasticsearch.action.support.IndicesOptions;
-import org.elasticsearch.action.support.master.MasterNodeReadRequest;
+import org.elasticsearch.action.support.local.LocalClusterStateRequest;
 import org.elasticsearch.cluster.metadata.DataStreamGlobalRetention;
 import org.elasticsearch.cluster.metadata.DataStreamLifecycle;
 import org.elasticsearch.common.collect.Iterators;
@@ -25,13 +25,18 @@
 import org.elasticsearch.common.xcontent.ChunkedToXContentObject;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.core.UpdateForV10;
+import org.elasticsearch.tasks.CancellableTask;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.xcontent.ParseField;
 import org.elasticsearch.xcontent.ToXContent;
 
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 
 /**
@@ -46,7 +51,7 @@ private ExplainDataStreamLifecycleAction() {/* no instances */}
     /**
      * Request explaining the data stream lifecycle for one or more indices.
      */
-    public static class Request extends MasterNodeReadRequest<Request> implements IndicesRequest.Replaceable {
+    public static class Request extends LocalClusterStateRequest implements IndicesRequest.Replaceable {
         private String[] names;
         private boolean includeDefaults;
         private IndicesOptions indicesOptions = IndicesOptions.strictExpandOpen();
@@ -70,26 +75,28 @@ public ActionRequestValidationException validate() {
             return null;
         }
 
+        @Override
+        public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
+            return new CancellableTask(id, type, action, "", parentTaskId, headers);
+        }
+
         @Override
         public boolean includeDataStreams() {
             return true;
         }
 
+        /**
+         * NB prior to 9.0 this was a TransportMasterNodeReadAction so for BwC we must remain able to read these requests until
+         * we no longer need to support calling this action remotely.
+         */
+        @UpdateForV10(owner = UpdateForV10.Owner.DATA_MANAGEMENT)
         public Request(StreamInput in) throws IOException {
             super(in);
             this.names = in.readOptionalStringArray();
             this.indicesOptions = IndicesOptions.readIndicesOptions(in);
             this.includeDefaults = in.readBoolean();
         }
 
-        @Override
-        public void writeTo(StreamOutput out) throws IOException {
-            super.writeTo(out);
-            out.writeOptionalStringArray(names);
-            indicesOptions.writeIndicesOptions(out);
-            out.writeBoolean(includeDefaults);
-        }
-
         @Override
         public boolean equals(Object o) {
             if (this == o) {
@@ -159,15 +166,6 @@ public Response(
             this.globalRetention = globalRetention;
         }
 
-        public Response(StreamInput in) throws IOException {
-            super(in);
-            this.indices = in.readCollectionAsList(ExplainIndexDataStreamLifecycle::new);
-            this.rolloverConfiguration = in.readOptionalWriteable(RolloverConfiguration::new);
-            this.globalRetention = in.getTransportVersion().onOrAfter(TransportVersions.V_8_14_0)
-                ? in.readOptionalWriteable(DataStreamGlobalRetention::read)
-                : null;
-        }
-
         public List<ExplainIndexDataStreamLifecycle> getIndices() {
             return indices;
         }
@@ -180,6 +178,11 @@ public DataStreamGlobalRetention getGlobalRetention() {
             return globalRetention;
         }
 
+        /**
+         * NB prior to 9.0 this was a TransportMasterNodeReadAction so for BwC we must remain able to write these responses until
+         * we no longer need to support calling this action remotely.
+         */
+        @UpdateForV10(owner = UpdateForV10.Owner.DATA_MANAGEMENT)
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             out.writeCollection(indices);

server/src/test/java/org/elasticsearch/action/datastreams/lifecycle/ExplainDataStreamLifecycleResponseTests.java

@@ -13,22 +13,16 @@
 import org.elasticsearch.action.admin.indices.rollover.MinPrimaryShardDocsCondition;
 import org.elasticsearch.action.admin.indices.rollover.RolloverConditions;
 import org.elasticsearch.action.admin.indices.rollover.RolloverConfiguration;
-import org.elasticsearch.cluster.metadata.DataStreamGlobalRetention;
 import org.elasticsearch.cluster.metadata.DataStreamLifecycle;
 import org.elasticsearch.cluster.metadata.DataStreamTestHelper;
 import org.elasticsearch.common.bytes.BytesReference;
-import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
-import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
-import org.elasticsearch.indices.IndicesModule;
 import org.elasticsearch.test.AbstractChunkedSerializingTestCase;
-import org.elasticsearch.test.AbstractWireSerializingTestCase;
-import org.elasticsearch.xcontent.NamedXContentRegistry;
+import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentFactory;
-import org.junit.Before;
 
 import java.io.IOException;
 import java.util.List;
@@ -39,26 +33,7 @@
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.nullValue;
 
-public class ExplainDataStreamLifecycleResponseTests extends AbstractWireSerializingTestCase<Response> {
-
-    private NamedWriteableRegistry namedWriteableRegistry;
-    private NamedXContentRegistry xContentRegistry;
-
-    @Before
-    public void setupNamedWriteableRegistry() {
-        namedWriteableRegistry = new NamedWriteableRegistry(IndicesModule.getNamedWriteables());
-        xContentRegistry = new NamedXContentRegistry(IndicesModule.getNamedXContents());
-    }
-
-    @Override
-    protected NamedXContentRegistry xContentRegistry() {
-        return xContentRegistry;
-    }
-
-    @Override
-    protected NamedWriteableRegistry getNamedWriteableRegistry() {
-        return namedWriteableRegistry;
-    }
+public class ExplainDataStreamLifecycleResponseTests extends ESTestCase {
 
     @SuppressWarnings("unchecked")
     public void testToXContent() throws IOException {
@@ -280,38 +255,4 @@ private static ExplainIndexDataStreamLifecycle createRandomIndexDataStreamLifecy
                 : null
         );
     }
-
-    @Override
-    protected Writeable.Reader<Response> instanceReader() {
-        return Response::new;
-    }
-
-    @Override
-    protected Response createTestInstance() {
-        return randomResponse();
-    }
-
-    @Override
-    protected Response mutateInstance(Response instance) {
-        return randomResponse();
-    }
-
-    private Response randomResponse() {
-        return new Response(
-            List.of(createRandomIndexDataStreamLifecycleExplanation(System.nanoTime(), randomBoolean() ? new DataStreamLifecycle() : null)),
-            randomBoolean()
-                ? new RolloverConfiguration(
-                    new RolloverConditions(
-                        Map.of(MaxPrimaryShardDocsCondition.NAME, new MaxPrimaryShardDocsCondition(randomLongBetween(1000, 199_999_000)))
-                    )
-                )
-                : null,
-            randomBoolean()
-                ? new DataStreamGlobalRetention(
-                    TimeValue.timeValueDays(randomIntBetween(1, 10)),
-                    TimeValue.timeValueDays(randomIntBetween(10, 20))
-                )
-                : null
-        );
-    }
 }