Run XPack usage actions on local node (#122933)

These actions solely need the cluster state, they can run on any node.

Relates #101805

Author: nielsbauman

docs/changelog/122933.yaml

@@ -0,0 +1,5 @@
+pr: 122933
+summary: Run XPack usage actions on local node
+area: Stats
+type: enhancement
+issues: []

server/src/main/java/org/elasticsearch/action/support/local/TransportLocalClusterStateAction.java

@@ -66,6 +66,9 @@ protected abstract void localClusterStateOperation(Task task, Request request, C
 
     @Override
     protected final void doExecute(Task task, Request request, ActionListener<Response> listener) {
+        if (task != null) {
+            request.setParentTask(clusterService.localNode().getId(), task.getId());
+        }
         final var state = clusterService.state();
         final var clusterBlockException = checkBlock(request, state);
         if (clusterBlockException != null) {

x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/action/AnalyticsUsageTransportAction.java

@@ -38,7 +38,7 @@ public AnalyticsUsageTransportAction(
     }
 
     @Override
-    protected void masterOperation(
+    protected void localClusterStateOperation(
         Task task,
         XPackUsageRequest request,
         ClusterState state,

x-pack/plugin/analytics/src/test/java/org/elasticsearch/xpack/analytics/action/AnalyticsInfoTransportActionTests.java

@@ -73,7 +73,7 @@ public void testAvailable() throws Exception {
             client
         );
         PlainActionFuture<XPackUsageFeatureResponse> future = new PlainActionFuture<>();
-        usageAction.masterOperation(task, null, clusterState, future);
+        usageAction.localClusterStateOperation(task, null, clusterState, future);
         XPackFeatureUsage usage = future.get().getUsage();
         assertThat(usage.available(), is(true));
 
@@ -100,7 +100,7 @@ public void testEnabled() throws Exception {
             client
         );
         PlainActionFuture<XPackUsageFeatureResponse> future = new PlainActionFuture<>();
-        usageAction.masterOperation(task, null, clusterState, future);
+        usageAction.localClusterStateOperation(task, null, clusterState, future);
         XPackFeatureUsage usage = future.get().getUsage();
         assertTrue(usage.enabled());
 

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/CCRUsageTransportAction.java

@@ -48,7 +48,7 @@ public CCRUsageTransportAction(
     }
 
     @Override
-    protected void masterOperation(
+    protected void localClusterStateOperation(
         Task task,
         XPackUsageRequest request,
         ClusterState state,

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/CCRInfoTransportActionTests.java

@@ -152,7 +152,7 @@ public void testUsageStats() throws Exception {
             licenseState
         );
         PlainActionFuture<XPackUsageFeatureResponse> future = new PlainActionFuture<>();
-        usageAction.masterOperation(null, null, clusterState, future);
+        usageAction.localClusterStateOperation(null, null, clusterState, future);
         CCRInfoTransportAction.Usage ccrUsage = (CCRInfoTransportAction.Usage) future.get().getUsage();
         assertThat(ccrUsage.enabled(), equalTo(true));
         assertThat(ccrUsage.available(), equalTo(false));

x-pack/plugin/core/src/internalClusterTest/java/org/elasticsearch/xpack/core/rest/action/XPackUsageRestCancellationIT.java

@@ -150,7 +150,7 @@ public BlockingXPackUsageAction(
         }
 
         @Override
-        protected void masterOperation(
+        protected void localClusterStateOperation(
             Task task,
             XPackUsageRequest request,
             ClusterState state,
@@ -185,7 +185,7 @@ public NonBlockingXPackUsageAction(
         }
 
         @Override
-        protected void masterOperation(
+        protected void localClusterStateOperation(
             Task task,
             XPackUsageRequest request,
             ClusterState state,

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

@@ -7,7 +7,7 @@
 package org.elasticsearch.protocol.xpack;
 
 import org.elasticsearch.action.ActionRequestValidationException;
-import org.elasticsearch.action.support.master.MasterNodeRequest;
+import org.elasticsearch.action.support.local.LocalClusterStateRequest;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.tasks.CancellableTask;
@@ -17,14 +17,14 @@
 import java.io.IOException;
 import java.util.Map;
 
-public class XPackUsageRequest extends MasterNodeRequest<XPackUsageRequest> {
+public class XPackUsageRequest extends LocalClusterStateRequest {
 
     public XPackUsageRequest(TimeValue masterNodeTimeout) {
         super(masterNodeTimeout);
     }
 
     public XPackUsageRequest(StreamInput in) throws IOException {
-        super(in);
+        super(in, false);
     }
 
     @Override

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

@@ -42,7 +42,7 @@ public HealthApiUsageTransportAction(
     }
 
     @Override
-    protected void masterOperation(
+    protected void localClusterStateOperation(
         Task task,
         XPackUsageRequest request,
         ClusterState state,

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

@@ -23,6 +23,8 @@
 
 public class RemoteClusterUsageTransportAction extends XPackUsageFeatureTransportAction {
 
+    private TransportService transportService;
+
     @Inject
     public RemoteClusterUsageTransportAction(
         TransportService transportService,
@@ -31,10 +33,11 @@ public RemoteClusterUsageTransportAction(
         ActionFilters actionFilters
     ) {
         super(XPackUsageFeatureAction.REMOTE_CLUSTERS.name(), transportService, clusterService, threadPool, actionFilters);
+        this.transportService = transportService;
     }
 
     @Override
-    protected void masterOperation(
+    protected void localClusterStateOperation(
         Task task,
         XPackUsageRequest request,
         ClusterState state,

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

@@ -47,7 +47,7 @@ public DataStreamLifecycleUsageTransportAction(
     }
 
     @Override
-    protected void masterOperation(
+    protected void localClusterStateOperation(
         Task task,
         XPackUsageRequest request,
         ClusterState state,

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

@@ -39,7 +39,7 @@ public DataStreamUsageTransportAction(
     }
 
     @Override
-    protected void masterOperation(
+    protected void localClusterStateOperation(
         Task task,
         XPackUsageRequest request,
         ClusterState state,

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

@@ -10,11 +10,13 @@
 import org.elasticsearch.action.ActionRunnable;
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.support.ActionFilters;
-import org.elasticsearch.action.support.master.TransportMasterNodeAction;
+import org.elasticsearch.action.support.ChannelActionListener;
+import org.elasticsearch.action.support.local.TransportLocalClusterStateAction;
 import org.elasticsearch.client.internal.node.NodeClient;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.core.UpdateForV10;
 import org.elasticsearch.injection.guice.Inject;
 import org.elasticsearch.protocol.xpack.XPackUsageRequest;
 import org.elasticsearch.tasks.Task;
@@ -25,11 +27,16 @@
 import java.util.ArrayList;
 import java.util.List;
 
-public class TransportXPackUsageAction extends TransportMasterNodeAction<XPackUsageRequest, XPackUsageResponse> {
+public class TransportXPackUsageAction extends TransportLocalClusterStateAction<XPackUsageRequest, XPackUsageResponse> {
 
     private final NodeClient client;
     private final List<ActionType<XPackUsageFeatureResponse>> usageActions;
 
+    /**
+     * 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 TransportXPackUsageAction(
@@ -41,16 +48,22 @@ public TransportXPackUsageAction(
     ) {
         super(
             XPackUsageAction.NAME,
-            transportService,
-            clusterService,
-            threadPool,
             actionFilters,
-            XPackUsageRequest::new,
-            XPackUsageResponse::new,
+            transportService.getTaskManager(),
+            clusterService,
             threadPool.executor(ThreadPool.Names.MANAGEMENT)
         );
         this.client = client;
         this.usageActions = usageActions();
+
+        transportService.registerRequestHandler(
+            actionName,
+            executor,
+            false,
+            true,
+            XPackUsageRequest::new,
+            (request, channel, task) -> executeDirect(task, request, new ChannelActionListener<>(channel))
+        );
     }
 
     // overrideable for tests
@@ -59,14 +72,19 @@ protected List<ActionType<XPackUsageFeatureResponse>> usageActions() {
     }
 
     @Override
-    protected void masterOperation(Task task, XPackUsageRequest request, ClusterState state, ActionListener<XPackUsageResponse> listener) {
+    protected void localClusterStateOperation(
+        Task task,
+        XPackUsageRequest request,
+        ClusterState state,
+        ActionListener<XPackUsageResponse> listener
+    ) {
         new ActionRunnable<>(listener) {
             final List<XPackFeatureUsage> responses = new ArrayList<>(usageActions.size());
 
             @Override
             protected void doRun() {
                 if (responses.size() < usageActions().size()) {
-                    final var childRequest = new XPackUsageRequest(request.masterNodeTimeout());
+                    final var childRequest = new XPackUsageRequest(request.masterTimeout());
                     childRequest.setParentTask(request.getParentTask());
                     client.executeLocally(
                         usageActions.get(responses.size()),

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

@@ -7,32 +7,42 @@
 package org.elasticsearch.xpack.core.action;
 
 import org.elasticsearch.action.support.ActionFilters;
-import org.elasticsearch.action.support.master.TransportMasterNodeAction;
+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.service.ClusterService;
+import org.elasticsearch.core.UpdateForV10;
 import org.elasticsearch.protocol.xpack.XPackUsageRequest;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
-public abstract class XPackUsageFeatureTransportAction extends TransportMasterNodeAction<XPackUsageRequest, XPackUsageFeatureResponse> {
+public abstract class XPackUsageFeatureTransportAction extends TransportLocalClusterStateAction<
+    XPackUsageRequest,
+    XPackUsageFeatureResponse> {
 
+    /**
+     * 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")
     public XPackUsageFeatureTransportAction(
         String name,
         TransportService transportService,
         ClusterService clusterService,
         ThreadPool threadPool,
         ActionFilters actionFilters
     ) {
-        super(
-            name,
-            transportService,
-            clusterService,
-            threadPool,
-            actionFilters,
+        super(name, actionFilters, transportService.getTaskManager(), clusterService, threadPool.executor(ThreadPool.Names.MANAGEMENT));
+
+        transportService.registerRequestHandler(
+            actionName,
+            executor,
+            false,
+            true,
             XPackUsageRequest::new,
-            XPackUsageFeatureResponse::new,
-            threadPool.executor(ThreadPool.Names.MANAGEMENT)
+            (request, channel, task) -> executeDirect(task, request, new ChannelActionListener<>(channel))
         );
     }
 

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

@@ -57,7 +57,7 @@ public DataTiersUsageTransportAction(
     }
 
     @Override
-    protected void masterOperation(
+    protected void localClusterStateOperation(
         Task task,
         XPackUsageRequest request,
         ClusterState state,

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

@@ -33,7 +33,7 @@ public EnrichUsageTransportAction(
     }
 
     @Override
-    protected void masterOperation(
+    protected void localClusterStateOperation(
         Task task,
         XPackUsageRequest request,
         ClusterState state,

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/EnterpriseSearchUsageTransportAction.java

@@ -78,7 +78,7 @@ public EnterpriseSearchUsageTransportAction(
     }
 
     @Override
-    protected void masterOperation(
+    protected void localClusterStateOperation(
         Task task,
         XPackUsageRequest request,
         ClusterState state,
@@ -153,7 +153,7 @@ protected void masterOperation(
 
         // Step 1: Fetch analytics collections count
         GetAnalyticsCollectionAction.Request analyticsCollectionsCountRequest = new GetAnalyticsCollectionAction.Request(
-            request.masterNodeTimeout(),
+            request.masterTimeout(),
             new String[] { "*" }
         );
 

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/EqlUsageTransportAction.java

@@ -46,7 +46,7 @@ public EqlUsageTransportAction(
     }
 
     @Override
-    protected void masterOperation(
+    protected void localClusterStateOperation(
         Task task,
         XPackUsageRequest request,
         ClusterState state,

x-pack/plugin/eql/src/test/java/org/elasticsearch/xpack/eql/EqlInfoTransportActionTests.java

@@ -100,7 +100,7 @@ public void testUsageStats() throws Exception {
         TransportService transportService = MockUtils.setupTransportServiceWithThreadpoolExecutor(threadPool);
         var usageAction = new EqlUsageTransportAction(transportService, clusterService, threadPool, mock(ActionFilters.class), client);
         PlainActionFuture<XPackUsageFeatureResponse> future = new PlainActionFuture<>();
-        usageAction.masterOperation(mock(Task.class), null, null, future);
+        usageAction.localClusterStateOperation(mock(Task.class), null, null, future);
         EqlFeatureSetUsage eqlUsage = (EqlFeatureSetUsage) future.get().getUsage();
 
         long fooBarBaz = ObjectPath.eval("foo.bar.baz", eqlUsage.stats());

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/EsqlUsageTransportAction.java

@@ -46,7 +46,7 @@ public EsqlUsageTransportAction(
     }
 
     @Override
-    protected void masterOperation(
+    protected void localClusterStateOperation(
         Task task,
         XPackUsageRequest request,
         ClusterState state,

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/EsqlInfoTransportActionTests.java

@@ -102,7 +102,7 @@ public void testUsageStats() throws Exception {
 
         var usageAction = new EsqlUsageTransportAction(transportService, clusterService, threadPool, mock(ActionFilters.class), client);
         PlainActionFuture<XPackUsageFeatureResponse> future = new PlainActionFuture<>();
-        usageAction.masterOperation(mock(Task.class), null, null, future);
+        usageAction.localClusterStateOperation(mock(Task.class), null, null, future);
         EsqlFeatureSetUsage esqlUsage = (EsqlFeatureSetUsage) future.get().getUsage();
 
         long fooBarBaz = ObjectPath.eval("foo.bar.baz", esqlUsage.stats());

x-pack/plugin/frozen-indices/src/main/java/org/elasticsearch/xpack/frozen/FrozenIndicesUsageTransportAction.java

@@ -35,7 +35,7 @@ public FrozenIndicesUsageTransportAction(
     }
 
     @Override
-    protected void masterOperation(
+    protected void localClusterStateOperation(
         Task task,
         XPackUsageRequest request,
         ClusterState state,

x-pack/plugin/graph/src/main/java/org/elasticsearch/xpack/graph/GraphUsageTransportAction.java

@@ -43,7 +43,7 @@ public GraphUsageTransportAction(
     }
 
     @Override
-    protected void masterOperation(
+    protected void localClusterStateOperation(
         Task task,
         XPackUsageRequest request,
         ClusterState state,

x-pack/plugin/graph/src/test/java/org/elasticsearch/xpack/graph/GraphInfoTransportActionTests.java

@@ -54,7 +54,7 @@ public void testAvailable() throws Exception {
             licenseState
         );
         PlainActionFuture<XPackUsageFeatureResponse> future = new PlainActionFuture<>();
-        usageAction.masterOperation(null, null, null, future);
+        usageAction.localClusterStateOperation(null, null, null, future);
         XPackFeatureUsage usage = future.get().getUsage();
         assertThat(usage.available(), is(available));
 
@@ -93,7 +93,7 @@ public void testEnabled() throws Exception {
             licenseState
         );
         PlainActionFuture<XPackUsageFeatureResponse> future = new PlainActionFuture<>();
-        usageAction.masterOperation(null, null, null, future);
+        usageAction.localClusterStateOperation(null, null, null, future);
         XPackFeatureUsage usage = future.get().getUsage();
         assertThat(usage.enabled(), is(enabled));