PR Fixes - Switch to local metadata action type and improve request handling

Author: lukewhiting

modules/streams/build.gradle

@@ -1,3 +1,12 @@
+/*
+ * 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", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
 apply plugin: 'elasticsearch.test-with-dependencies'
 apply plugin: 'elasticsearch.internal-cluster-test'
 apply plugin: 'elasticsearch.internal-yaml-rest-test'
@@ -11,8 +20,7 @@ esplugin {
 
 restResources {
   restApi {
-    // TODO: Limit this down to just required API's for faster build. See https://github.com/elastic/elasticsearch/blob/fb3149cc664eb7061d55741a87e7e8cf29db4989/TESTING.asciidoc#L443
-    include '*'
+    include '_common', 'streams'
   }
 }
 

modules/streams/src/main/java/org/elasticsearch/rest/streams/logs/RestStreamsStatusAction.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.rest.RestUtils;
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
+import org.elasticsearch.rest.action.RestCancellableNodeClient;
 import org.elasticsearch.rest.action.RestToXContentListener;
 
 import java.util.Collections;
@@ -27,7 +28,7 @@
 @ServerlessScope(Scope.PUBLIC)
 public class RestStreamsStatusAction extends BaseRestHandler {
 
-    public static final Set<String> SUPPORTED_PARAMS = Collections.singleton(RestUtils.REST_MASTER_TIMEOUT_PARAM);
+    public static final Set<String> SUPPORTED_PARAMS = Collections.singleton(RestUtils.REST_TIMEOUT_PARAM);
 
     @Override
     public String getName() {
@@ -41,8 +42,12 @@ public List<RestHandler.Route> routes() {
 
     @Override
     protected BaseRestHandler.RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) {
-        StreamsStatusAction.Request statusRequest = new StreamsStatusAction.Request(RestUtils.getMasterNodeTimeout(request));
-        return restChannel -> client.execute(StreamsStatusAction.INSTANCE, statusRequest, new RestToXContentListener<>(restChannel));
+        StreamsStatusAction.Request statusRequest = new StreamsStatusAction.Request(RestUtils.getAckTimeout(request));
+        return restChannel -> new RestCancellableNodeClient(client, request.getHttpChannel()).execute(
+            StreamsStatusAction.INSTANCE,
+            statusRequest,
+            new RestToXContentListener<>(restChannel)
+        );
     }
 
     @Override

modules/streams/src/main/java/org/elasticsearch/rest/streams/logs/StreamsStatusAction.java

@@ -9,40 +9,33 @@
 
 package org.elasticsearch.rest.streams.logs;
 
-import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.ActionType;
-import org.elasticsearch.action.support.master.MasterNodeReadRequest;
+import org.elasticsearch.action.support.local.LocalClusterStateRequest;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.tasks.CancellableTask;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.xcontent.ToXContentObject;
 import org.elasticsearch.xcontent.XContentBuilder;
 
 import java.io.IOException;
+import java.util.Map;
 
 public class StreamsStatusAction {
 
     public static ActionType<Response> INSTANCE = new ActionType<>("cluster:admin/streams/status");
 
-    public static class Request extends MasterNodeReadRequest<Request> {
-
-        public Request(TimeValue masterNodeTimeout) {
-            super(masterNodeTimeout);
-        }
-
-        public Request(StreamInput in) throws IOException {
-            super(in);
+    public static class Request extends LocalClusterStateRequest {
+        protected Request(TimeValue masterTimeout) {
+            super(masterTimeout);
         }
 
         @Override
-        public ActionRequestValidationException validate() {
-            return null;
-        }
-
-        @Override
-        public void writeTo(StreamOutput out) throws IOException {
-            super.writeTo(out);
+        public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
+            return new CancellableTask(id, type, action, "Streams status request", parentTaskId, headers);
         }
     }
 

modules/streams/src/main/java/org/elasticsearch/rest/streams/logs/TransportStreamsStatusAction.java

@@ -11,11 +11,10 @@
 
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.ActionFilters;
-import org.elasticsearch.action.support.master.TransportMasterNodeReadAction;
-import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.action.support.local.TransportLocalProjectMetadataAction;
+import org.elasticsearch.cluster.ProjectState;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
-import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.injection.guice.Inject;
@@ -28,9 +27,9 @@
  * Transport action to retrieve the status of logs streams in a project / cluster.
  * Results are broken down by stream type. Currently only logs streams are implemented.
  */
-public class TransportStreamsStatusAction extends TransportMasterNodeReadAction<StreamsStatusAction.Request, StreamsStatusAction.Response> {
-
-    private final ProjectResolver projectResolver;
+public class TransportStreamsStatusAction extends TransportLocalProjectMetadataAction<
+    StreamsStatusAction.Request,
+    StreamsStatusAction.Response> {
 
     @Inject
     public TransportStreamsStatusAction(
@@ -42,33 +41,29 @@ public TransportStreamsStatusAction(
     ) {
         super(
             StreamsStatusAction.INSTANCE.name(),
-            transportService,
-            clusterService,
-            threadPool,
             actionFilters,
-            StreamsStatusAction.Request::new,
-            StreamsStatusAction.Response::new,
-            threadPool.executor(ThreadPool.Names.MANAGEMENT)
+            transportService.getTaskManager(),
+            clusterService,
+            threadPool.executor(ThreadPool.Names.MANAGEMENT),
+            projectResolver
         );
-        this.projectResolver = projectResolver;
     }
 
     @Override
-    protected void masterOperation(
+    protected ClusterBlockException checkBlock(StreamsStatusAction.Request request, ProjectState state) {
+        return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_READ);
+    }
+
+    @Override
+    protected void localClusterStateOperation(
         Task task,
         StreamsStatusAction.Request request,
-        ClusterState state,
+        ProjectState state,
         ActionListener<StreamsStatusAction.Response> listener
-    ) throws Exception {
-        ProjectId projectId = projectResolver.getProjectId();
-        StreamsMetadata streamsState = state.projectState(projectId).metadata().custom(StreamsMetadata.TYPE, StreamsMetadata.EMPTY);
+    ) {
+        StreamsMetadata streamsState = state.metadata().custom(StreamsMetadata.TYPE, StreamsMetadata.EMPTY);
         boolean logsEnabled = streamsState.isLogsEnabled();
         StreamsStatusAction.Response response = new StreamsStatusAction.Response(logsEnabled);
         listener.onResponse(response);
     }
-
-    @Override
-    protected ClusterBlockException checkBlock(StreamsStatusAction.Request request, ClusterState state) {
-        return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_READ);
-    }
 }

rest-api-spec/src/main/resources/rest-api-spec/api/streams.status.json

@@ -23,9 +23,9 @@
       ]
     },
     "params": {
-      "master_timeout": {
+      "timeout": {
         "type": "time",
-        "description": "Period to wait for a connection to the master node. If no response is received before the timeout expires, the request fails and returns an error."
+        "description": "Period to wait for a response. If no response is received before the timeout expires, the request fails and returns an error."
       }
     }
   }

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -299,6 +299,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion NONE_CHUNKING_STRATEGY = def(9_097_0_00);
     public static final TransportVersion PROJECT_DELETION_GLOBAL_BLOCK = def(9_098_0_00);
     public static final TransportVersion STREAMS_LOGS_SUPPORT = def(9_099_0_00);
+
     /*
      * STOP! READ THIS FIRST! No, really,
      *        ____ _____ ___  ____  _        ____  _____    _    ____    _____ _   _ ___ ____    _____ ___ ____  ____ _____ _

x-pack/qa/multi-project/core-rest-tests-with-multiple-projects/build.gradle

@@ -1,4 +1,13 @@
 apply plugin: 'elasticsearch.internal-yaml-rest-test'
+/*
+ * 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", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
 import org.elasticsearch.gradle.util.GradleUtils
 
 dependencies {
@@ -21,6 +30,7 @@ dependencies {
   restTestConfig project(path: ':modules:data-streams', configuration: "basicRestSpecs")
   restTestConfig project(path: ':modules:ingest-common', configuration: "basicRestSpecs")
   restTestConfig project(path: ':modules:reindex', configuration: "basicRestSpecs")
+  restTestConfig project(path: ':modules:streams', configuration: "basicRestSpecs")
 }
 
 // let the yamlRestTests see the classpath of test