Make GetIndexAction cancellable (#87681)

DaveCTurner · web-flow · commit eb55b48fe61a · 2022-06-16T10:28:51.000+01:00
The get-indices API does some nontrivial work on the master and at high index counts the response may be very large and could take a long time to compute. Some clients will time out and retry if it takes too long. Today this API is not properly cancellable which leads to a good deal of wasted work in this situation, and the potentially-enormous response is serialized on a transport worker thread. With this commit we make the API cancellable and move the serialization to a `MANAGEMENT` thread. Relates #77466
diff --git a/docs/changelog/87681.yaml b/docs/changelog/87681.yaml
@@ -0,0 +1,5 @@
+pr: 87681
+summary: Make `GetIndexAction` cancellable
+area: Indices APIs
+type: bug
+issues: []
diff --git a/qa/smoke-test-http/src/javaRestTest/java/org/elasticsearch/http/RestClusterInfoActionCancellationIT.java b/qa/smoke-test-http/src/javaRestTest/java/org/elasticsearch/http/RestClusterInfoActionCancellationIT.java
@@ -9,6 +9,7 @@
 package org.elasticsearch.http;
 
 import org.apache.http.client.methods.HttpGet;
+import org.elasticsearch.action.admin.indices.get.GetIndexAction;
 import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsAction;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
@@ -37,21 +38,28 @@
 import static org.hamcrest.core.IsEqual.equalTo;
 
 @ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, numDataNodes = 0, numClientNodes = 0)
-public class RestGetMappingsCancellationIT extends HttpSmokeTestCase {
+public class RestClusterInfoActionCancellationIT extends HttpSmokeTestCase {
 
     public void testGetMappingsCancellation() throws Exception {
+        runTest(GetMappingsAction.NAME, "/test/_mappings");
+    }
+
+    public void testGetIndicesCancellation() throws Exception {
+        runTest(GetIndexAction.NAME, "/test");
+    }
+
+    private void runTest(String actionName, String endpoint) throws Exception {
         internalCluster().startMasterOnlyNode();
         internalCluster().startDataOnlyNode();
         ensureStableCluster(2);
 
         createIndex("test");
         ensureGreen("test");
-        final String actionName = GetMappingsAction.NAME;
         // Add a retryable cluster block that would block the request execution
         updateClusterState(currentState -> {
             ClusterBlock clusterBlock = new ClusterBlock(
                 1000,
-                "Get mappings cancellation test cluster block",
+                actionName + " cancellation test cluster block",
                 true,
                 false,
                 false,
@@ -62,7 +70,7 @@ public void testGetMappingsCancellation() throws Exception {
             return ClusterState.builder(currentState).blocks(ClusterBlocks.builder().addGlobalBlock(clusterBlock).build()).build();
         });
 
-        final Request request = new Request(HttpGet.METHOD_NAME, "/test/_mappings");
+        final Request request = new Request(HttpGet.METHOD_NAME, endpoint);
         final PlainActionFuture<Response> future = new PlainActionFuture<>();
         final Cancellable cancellable = getRestClient().performRequestAsync(request, wrapAsRestResponseListener(future));
 
diff --git a/server/src/main/java/org/elasticsearch/action/ActionModule.java b/server/src/main/java/org/elasticsearch/action/ActionModule.java
@@ -734,7 +734,7 @@ public void initRestHandlers(Supplier<DiscoveryNodes> nodesInCluster) {
         registerHandler.accept(new RestResetFeatureStateAction());
         registerHandler.accept(new RestGetFeatureUpgradeStatusAction());
         registerHandler.accept(new RestPostFeatureUpgradeAction());
-        registerHandler.accept(new RestGetIndicesAction());
+        registerHandler.accept(new RestGetIndicesAction(threadPool));
         registerHandler.accept(new RestIndicesStatsAction());
         registerHandler.accept(new RestIndicesSegmentsAction(threadPool));
         registerHandler.accept(new RestIndicesShardStoresAction());
diff --git a/server/src/main/java/org/elasticsearch/action/admin/indices/get/GetIndexRequest.java b/server/src/main/java/org/elasticsearch/action/admin/indices/get/GetIndexRequest.java
@@ -14,12 +14,16 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.util.ArrayUtils;
 import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.tasks.CancellableTask;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.tasks.TaskId;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Locale;
+import java.util.Map;
 import java.util.Set;
 
 /**
@@ -161,4 +165,9 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeBoolean(humanReadable);
         out.writeBoolean(includeDefaults);
     }
+
+    @Override
+    public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
+        return new CancellableTask(id, type, action, getDescription(), parentTaskId, headers);
+    }
 }
diff --git a/server/src/main/java/org/elasticsearch/action/admin/indices/get/TransportGetIndexAction.java b/server/src/main/java/org/elasticsearch/action/admin/indices/get/TransportGetIndexAction.java
@@ -17,14 +17,14 @@
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.MappingMetadata;
-import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.collect.ImmutableOpenMap;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.IndexScopedSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsFilter;
 import org.elasticsearch.indices.IndicesService;
+import org.elasticsearch.tasks.CancellableTask;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
@@ -94,11 +94,12 @@ protected void doMasterOperation(
         boolean doneMappings = false;
         boolean doneSettings = false;
         for (Feature feature : features) {
+            checkCancellation(task);
             switch (feature) {
                 case MAPPINGS:
                     if (doneMappings == false) {
                         mappingsResult = state.metadata()
-                            .findMappings(concreteIndices, indicesService.getFieldFilter(), Metadata.ON_NEXT_INDEX_FIND_MAPPINGS_NOOP);
+                            .findMappings(concreteIndices, indicesService.getFieldFilter(), () -> checkCancellation(task));
                         doneMappings = true;
                     }
                     break;
@@ -113,6 +114,7 @@ protected void doMasterOperation(
                         ImmutableOpenMap.Builder<String, Settings> settingsMapBuilder = ImmutableOpenMap.builder();
                         ImmutableOpenMap.Builder<String, Settings> defaultSettingsMapBuilder = ImmutableOpenMap.builder();
                         for (String index : concreteIndices) {
+                            checkCancellation(task);
                             Settings indexSettings = state.metadata().index(index).getSettings();
                             if (request.humanReadable()) {
                                 indexSettings = IndexMetadata.addHumanReadableSettings(indexSettings);
@@ -137,4 +139,10 @@ protected void doMasterOperation(
         }
         listener.onResponse(new GetIndexResponse(concreteIndices, mappingsResult, aliasesResult, settings, defaultSettings, dataStreams));
     }
+
+    private static void checkCancellation(Task task) {
+        if (task instanceof CancellableTask cancellableTask) {
+            cancellableTask.ensureNotCancelled();
+        }
+    }
 }
diff --git a/server/src/main/java/org/elasticsearch/action/admin/indices/mapping/get/TransportGetMappingsAction.java b/server/src/main/java/org/elasticsearch/action/admin/indices/mapping/get/TransportGetMappingsAction.java
@@ -26,7 +26,6 @@
 import org.elasticsearch.transport.TransportService;
 
 import java.util.Map;
-import java.util.concurrent.CancellationException;
 
 public class TransportGetMappingsAction extends TransportClusterInfoAction<GetMappingsRequest, GetMappingsResponse> {
 
@@ -75,8 +74,8 @@ protected void doMasterOperation(
     }
 
     private static void checkCancellation(Task task) {
-        if (task instanceof CancellableTask && ((CancellableTask) task).isCancelled()) {
-            throw new CancellationException("Task cancelled");
+        if (task instanceof CancellableTask cancellableTask) {
+            cancellableTask.ensureNotCancelled();
         }
     }
 }
diff --git a/server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetIndicesAction.java b/server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetIndicesAction.java
@@ -17,7 +17,9 @@
 import org.elasticsearch.core.RestApiVersion;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestRequest;
-import org.elasticsearch.rest.action.RestToXContentListener;
+import org.elasticsearch.rest.action.DispatchingRestToXContentListener;
+import org.elasticsearch.rest.action.RestCancellableNodeClient;
+import org.elasticsearch.threadpool.ThreadPool;
 
 import java.io.IOException;
 import java.util.Collections;
@@ -42,6 +44,12 @@ public class RestGetIndicesAction extends BaseRestHandler {
             .collect(Collectors.toSet())
     );
 
+    private final ThreadPool threadPool;
+
+    public RestGetIndicesAction(ThreadPool threadPool) {
+        this.threadPool = threadPool;
+    }
+
     @Override
     public List<Route> routes() {
         return List.of(new Route(GET, "/{index}"), new Route(HEAD, "/{index}"));
@@ -70,7 +78,13 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         getIndexRequest.humanReadable(request.paramAsBoolean("human", false));
         getIndexRequest.includeDefaults(request.paramAsBoolean("include_defaults", false));
         getIndexRequest.features(GetIndexRequest.Feature.fromRequest(request));
-        return channel -> client.admin().indices().getIndex(getIndexRequest, new RestToXContentListener<>(channel));
+        final var httpChannel = request.getHttpChannel();
+        return channel -> new RestCancellableNodeClient(client, httpChannel).admin()
+            .indices()
+            .getIndex(
+                getIndexRequest,
+                new DispatchingRestToXContentListener<>(threadPool.executor(ThreadPool.Names.MANAGEMENT), channel, request)
+            );
     }
 
     /**
diff --git a/server/src/test/java/org/elasticsearch/rest/action/admin/indices/RestGetIndicesActionTests.java b/server/src/test/java/org/elasticsearch/rest/action/admin/indices/RestGetIndicesActionTests.java
@@ -9,6 +9,7 @@
 package org.elasticsearch.rest.action.admin.indices;
 
 import org.elasticsearch.client.internal.node.NodeClient;
+import org.elasticsearch.common.util.concurrent.DeterministicTaskQueue;
 import org.elasticsearch.core.RestApiVersion;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.test.ESTestCase;
@@ -36,7 +37,7 @@ public void testIncludeTypeNamesWarning() throws IOException {
             Map.of("Content-Type", contentTypeHeader, "Accept", contentTypeHeader)
         ).withMethod(RestRequest.Method.GET).withPath("/some_index").withParams(params).build();
 
-        RestGetIndicesAction handler = new RestGetIndicesAction();
+        RestGetIndicesAction handler = new RestGetIndicesAction(new DeterministicTaskQueue().getThreadPool());
         handler.prepareRequest(request, mock(NodeClient.class));
         assertCriticalWarnings(RestGetIndicesAction.TYPES_DEPRECATION_MESSAGE);
 
@@ -57,7 +58,7 @@ public void testIncludeTypeNamesWarningExists() throws IOException {
             Map.of("Content-Type", contentTypeHeader, "Accept", contentTypeHeader)
         ).withMethod(RestRequest.Method.HEAD).withPath("/some_index").withParams(params).build();
 
-        RestGetIndicesAction handler = new RestGetIndicesAction();
+        RestGetIndicesAction handler = new RestGetIndicesAction(new DeterministicTaskQueue().getThreadPool());
         handler.prepareRequest(request, mock(NodeClient.class));
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -26,7 +26,6 @@`
`26`	`26`	`import org.elasticsearch.transport.TransportService;`
`27`	`27`
`28`	`28`	`import java.util.Map;`
`29`		`-import java.util.concurrent.CancellationException;`
`30`	`29`
`31`	`30`	`public class TransportGetMappingsAction extends TransportClusterInfoAction<GetMappingsRequest, GetMappingsResponse> {`
`32`	`31`
`@@ -75,8 +74,8 @@ protected void doMasterOperation(`
`75`	`74`	`}`
`76`	`75`
`77`	`76`	`private static void checkCancellation(Task task) {`
`78`		`- if (task instanceof CancellableTask && ((CancellableTask) task).isCancelled()) {`
`79`		`- throw new CancellationException("Task cancelled");`
	`77`	`+ if (task instanceof CancellableTask cancellableTask) {`
	`78`	`+ cancellableTask.ensureNotCancelled();`
`80`	`79`	`}`
`81`	`80`	`}`
`82`	`81`	`}`