Make TransportLocalClusterStateAction wait for cluster to unblock (#117230)

nielsbauman · web-flow · commit 032b42fcf7c3 · 2024-12-04T12:17:13.000+01:00
This will make `TransportLocalClusterStateAction` wait for a new state
that is not blocked. This means we need a timeout (again). For
consistency's sake, we're reusing the REST param `master_timeout` for
this timeout as well.

The only class that was using `TransportLocalClusterStateAction` was
`TransportGetAliasesAction`, so its request needed to accept a timeout
again as well.
diff --git a/docs/changelog/117230.yaml b/docs/changelog/117230.yaml
@@ -0,0 +1,5 @@
+pr: 117230
+summary: Make various alias retrieval APIs wait for cluster to unblock
+area: Distributed
+type: enhancement
+issues: []
diff --git a/docs/reference/cat/alias.asciidoc b/docs/reference/cat/alias.asciidoc
@@ -51,6 +51,8 @@ include::{es-ref-dir}/rest-api/common-parms.asciidoc[tag=cat-v]
 
 include::{es-ref-dir}/rest-api/common-parms.asciidoc[tag=expand-wildcards]
 
+include::{es-ref-dir}/rest-api/common-parms.asciidoc[tag=master-timeout]
+
 [[cat-alias-api-example]]
 ==== {api-examples-title}
 
diff --git a/docs/reference/indices/alias-exists.asciidoc b/docs/reference/indices/alias-exists.asciidoc
@@ -52,6 +52,8 @@ Defaults to `all`.
 (Optional, Boolean) If `false`, requests that include a missing data stream or
 index in the `<target>` return an error. Defaults to `false`.
 
+include::{es-ref-dir}/rest-api/common-parms.asciidoc[tag=master-timeout]
+
 [[alias-exists-api-response-codes]]
 ==== {api-response-codes-title}
 
diff --git a/docs/reference/indices/get-alias.asciidoc b/docs/reference/indices/get-alias.asciidoc
@@ -58,3 +58,5 @@ Defaults to `all`.
 `ignore_unavailable`::
 (Optional, Boolean) If `false`, requests that include a missing data stream or
 index in the `<target>` return an error. Defaults to `false`.
+
+include::{es-ref-dir}/rest-api/common-parms.asciidoc[tag=master-timeout]
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/cat.aliases.json b/rest-api-spec/src/main/resources/rest-api-spec/api/cat.aliases.json
@@ -65,6 +65,11 @@
         ],
         "default": "all",
         "description":"Whether to expand wildcard expression to concrete indices that are open, closed or both."
+      },
+      "master_timeout":{
+        "type":"time",
+        "description":"Timeout for waiting for new cluster state in case it is blocked",
+        "default":"30s"
       }
     }
   }
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/indices.exists_alias.json b/rest-api-spec/src/main/resources/rest-api-spec/api/indices.exists_alias.json
@@ -61,6 +61,11 @@
         ],
         "default":"all",
         "description":"Whether to expand wildcard expression to concrete indices that are open, closed or both."
+      },
+      "master_timeout":{
+        "type":"time",
+        "description":"Timeout for waiting for new cluster state in case it is blocked",
+        "default":"30s"
       }
     }
   }
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/indices.get_alias.json b/rest-api-spec/src/main/resources/rest-api-spec/api/indices.get_alias.json
@@ -79,6 +79,11 @@
         ],
         "default": "all",
         "description":"Whether to expand wildcard expression to concrete indices that are open, closed or both."
+      },
+      "master_timeout":{
+        "type":"time",
+        "description":"Timeout for waiting for new cluster state in case it is blocked",
+        "default":"30s"
       }
     }
   }
diff --git a/server/src/main/java/module-info.java b/server/src/main/java/module-info.java
@@ -147,6 +147,7 @@
     exports org.elasticsearch.action.support.master;
     exports org.elasticsearch.action.support.master.info;
     exports org.elasticsearch.action.support.nodes;
+    exports org.elasticsearch.action.support.local;
     exports org.elasticsearch.action.support.replication;
     exports org.elasticsearch.action.support.single.instance;
     exports org.elasticsearch.action.support.single.shard;
diff --git a/server/src/main/java/org/elasticsearch/action/admin/indices/alias/get/GetAliasesRequest.java b/server/src/main/java/org/elasticsearch/action/admin/indices/alias/get/GetAliasesRequest.java
@@ -8,21 +8,20 @@
  */
 package org.elasticsearch.action.admin.indices.alias.get;
 
-import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.AliasesRequest;
 import org.elasticsearch.action.support.IndicesOptions;
-import org.elasticsearch.action.support.TransportAction;
+import org.elasticsearch.action.support.local.LocalClusterStateRequest;
+import org.elasticsearch.action.support.master.MasterNodeRequest;
 import org.elasticsearch.common.Strings;
-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 java.io.IOException;
 import java.util.Map;
 
-public class GetAliasesRequest extends ActionRequest implements AliasesRequest {
+public class GetAliasesRequest extends LocalClusterStateRequest implements AliasesRequest {
 
     public static final IndicesOptions DEFAULT_INDICES_OPTIONS = IndicesOptions.strictExpandHidden();
 
@@ -31,18 +30,20 @@ public class GetAliasesRequest extends ActionRequest implements AliasesRequest {
     private String[] indices = Strings.EMPTY_ARRAY;
     private IndicesOptions indicesOptions = DEFAULT_INDICES_OPTIONS;
 
+    @Deprecated
     public GetAliasesRequest(String... aliases) {
-        this.aliases = aliases;
-        this.originalAliases = aliases;
+        this(MasterNodeRequest.TRAPPY_IMPLICIT_DEFAULT_MASTER_NODE_TIMEOUT, aliases);
     }
 
+    @Deprecated
     public GetAliasesRequest() {
         this(Strings.EMPTY_ARRAY);
     }
 
-    @Override
-    public void writeTo(StreamOutput out) throws IOException {
-        TransportAction.localOnly();
+    public GetAliasesRequest(TimeValue masterTimeout, String... aliases) {
+        super(masterTimeout);
+        this.aliases = aliases;
+        this.originalAliases = aliases;
     }
 
     @Override
diff --git a/server/src/main/java/org/elasticsearch/action/admin/indices/alias/get/TransportGetAliasesAction.java b/server/src/main/java/org/elasticsearch/action/admin/indices/alias/get/TransportGetAliasesAction.java
@@ -10,7 +10,7 @@
 
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.ActionFilters;
-import org.elasticsearch.action.support.TransportLocalClusterStateAction;
+import org.elasticsearch.action.support.local.TransportLocalClusterStateAction;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
diff --git a/server/src/main/java/org/elasticsearch/action/support/TransportLocalClusterStateAction.java b/server/src/main/java/org/elasticsearch/action/support/TransportLocalClusterStateAction.java
diff --git a/server/src/main/java/org/elasticsearch/action/support/local/LocalClusterStateRequest.java b/server/src/main/java/org/elasticsearch/action/support/local/LocalClusterStateRequest.java
@@ -0,0 +1,43 @@
+/*
+ * 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".
+ */
+
+package org.elasticsearch.action.support.local;
+
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.support.TransportAction;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.core.TimeValue;
+
+import java.io.IOException;
+import java.util.Objects;
+
+/**
+ * A base request for actions that are executed locally on the node that receives the request.
+ */
+public abstract class LocalClusterStateRequest extends ActionRequest {
+
+    /**
+     * The timeout for waiting until the cluster is unblocked.
+     * We use the name <code>masterTimeout</code> to be consistent with the master node actions.
+     */
+    private final TimeValue masterTimeout;
+
+    protected LocalClusterStateRequest(TimeValue masterTimeout) {
+        this.masterTimeout = Objects.requireNonNull(masterTimeout);
+    }
+
+    @Override
+    public final void writeTo(StreamOutput out) throws IOException {
+        TransportAction.localOnly();
+    }
+
+    public TimeValue masterTimeout() {
+        return masterTimeout;
+    }
+}
diff --git a/server/src/main/java/org/elasticsearch/action/support/local/TransportLocalClusterStateAction.java b/server/src/main/java/org/elasticsearch/action/support/local/TransportLocalClusterStateAction.java
@@ -0,0 +1,128 @@
+/*
+ * 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".
+ */
+
+package org.elasticsearch.action.support.local;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.elasticsearch.ElasticsearchTimeoutException;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.action.ActionRunnable;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.TransportAction;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateObserver;
+import org.elasticsearch.cluster.block.ClusterBlockException;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
+import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.node.NodeClosedException;
+import org.elasticsearch.tasks.CancellableTask;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.tasks.TaskManager;
+
+import java.util.concurrent.Executor;
+
+import static org.elasticsearch.common.Strings.format;
+
+/**
+ * Analogue of {@link org.elasticsearch.action.support.master.TransportMasterNodeReadAction} except that it runs on the local node rather
+ * than delegating to the master.
+ */
+public abstract class TransportLocalClusterStateAction<Request extends LocalClusterStateRequest, Response extends ActionResponse> extends
+    TransportAction<Request, Response> {
+
+    private static final Logger logger = LogManager.getLogger(TransportLocalClusterStateAction.class);
+
+    protected final ClusterService clusterService;
+    protected final Executor executor;
+
+    protected TransportLocalClusterStateAction(
+        String actionName,
+        ActionFilters actionFilters,
+        TaskManager taskManager,
+        ClusterService clusterService,
+        Executor executor
+    ) {
+        // TODO replace DIRECT_EXECUTOR_SERVICE when removing workaround for https://github.com/elastic/elasticsearch/issues/97916
+        super(actionName, actionFilters, taskManager, EsExecutors.DIRECT_EXECUTOR_SERVICE);
+        this.clusterService = clusterService;
+        this.executor = executor;
+    }
+
+    protected abstract ClusterBlockException checkBlock(Request request, ClusterState state);
+
+    protected abstract void localClusterStateOperation(Task task, Request request, ClusterState state, ActionListener<Response> listener)
+        throws Exception;
+
+    @Override
+    protected final void doExecute(Task task, Request request, ActionListener<Response> listener) {
+        final var state = clusterService.state();
+        final var clusterBlockException = checkBlock(request, state);
+        if (clusterBlockException != null) {
+            if (clusterBlockException.retryable() == false) {
+                listener.onFailure(clusterBlockException);
+            } else {
+                waitForClusterUnblock(task, request, listener, state, clusterBlockException);
+            }
+        } else {
+            innerDoExecute(task, request, listener, state);
+        }
+    }
+
+    private void innerDoExecute(Task task, Request request, ActionListener<Response> listener, ClusterState state) {
+        if (task instanceof CancellableTask cancellableTask && cancellableTask.notifyIfCancelled(listener)) {
+            return;
+        }
+        // Workaround for https://github.com/elastic/elasticsearch/issues/97916 - TODO remove this when we can
+        executor.execute(ActionRunnable.wrap(listener, l -> localClusterStateOperation(task, request, state, l)));
+    }
+
+    private void waitForClusterUnblock(
+        Task task,
+        Request request,
+        ActionListener<Response> listener,
+        ClusterState initialState,
+        ClusterBlockException exception
+    ) {
+        var observer = new ClusterStateObserver(
+            initialState,
+            clusterService,
+            request.masterTimeout(),
+            logger,
+            clusterService.threadPool().getThreadContext()
+        );
+        observer.waitForNextChange(new ClusterStateObserver.Listener() {
+            @Override
+            public void onNewClusterState(ClusterState state) {
+                logger.trace("retrying with cluster state version [{}]", state.version());
+                innerDoExecute(task, request, listener, state);
+            }
+
+            @Override
+            public void onClusterServiceClose() {
+                listener.onFailure(new NodeClosedException(clusterService.localNode()));
+            }
+
+            @Override
+            public void onTimeout(TimeValue timeout) {
+                logger.debug(
+                    () -> format("timed out while waiting for cluster to unblock in [%s] (timeout [%s])", actionName, timeout),
+                    exception
+                );
+                listener.onFailure(new ElasticsearchTimeoutException("timed out while waiting for cluster to unblock", exception));
+            }
+        }, clusterState -> isTaskCancelled(task) || checkBlock(request, clusterState) == null);
+    }
+
+    private boolean isTaskCancelled(Task task) {
+        return task instanceof CancellableTask cancellableTask && cancellableTask.isCancelled();
+    }
+}
diff --git a/server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetAliasesAction.java b/server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetAliasesAction.java
@@ -27,6 +27,7 @@
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.RestStatus;
+import org.elasticsearch.rest.RestUtils;
 import org.elasticsearch.rest.Scope;
 import org.elasticsearch.rest.ServerlessScope;
 import org.elasticsearch.rest.action.RestBuilderListener;
@@ -207,7 +208,8 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
 
         final boolean namesProvided = request.hasParam("name");
         final String[] aliases = request.paramAsStringArrayOrEmptyIfAll("name");
-        final GetAliasesRequest getAliasesRequest = new GetAliasesRequest(aliases);
+        final var masterNodeTimeout = RestUtils.getMasterNodeTimeout(request);
+        final GetAliasesRequest getAliasesRequest = new GetAliasesRequest(masterNodeTimeout, aliases);
         final String[] indices = Strings.splitStringByCommaToArray(request.param("index"));
         getAliasesRequest.indices(indices);
         getAliasesRequest.indicesOptions(IndicesOptions.fromRequest(request, getAliasesRequest.indicesOptions()));
diff --git a/server/src/main/java/org/elasticsearch/rest/action/cat/RestAliasAction.java b/server/src/main/java/org/elasticsearch/rest/action/cat/RestAliasAction.java
diff --git a/server/src/test/java/org/elasticsearch/action/support/local/TransportLocalClusterStateActionTests.java b/server/src/test/java/org/elasticsearch/action/support/local/TransportLocalClusterStateActionTests.java

Original file line number	Diff line number	Diff line change
`@@ -65,6 +65,11 @@`
`65`	`65`	`],`
`66`	`66`	`"default": "all",`
`67`	`67`	`"description":"Whether to expand wildcard expression to concrete indices that are open, closed or both."`
	`68`	`+ },`
	`69`	`+ "master_timeout":{`
	`70`	`+ "type":"time",`
	`71`	`+ "description":"Timeout for waiting for new cluster state in case it is blocked",`
	`72`	`+ "default":"30s"`
`68`	`73`	`}`
`69`	`74`	`}`
`70`	`75`	`}`
Original file line number	Diff line number	Diff line change
`@@ -61,6 +61,11 @@`
`61`	`61`	`],`
`62`	`62`	`"default":"all",`
`63`	`63`	`"description":"Whether to expand wildcard expression to concrete indices that are open, closed or both."`
	`64`	`+ },`
	`65`	`+ "master_timeout":{`
	`66`	`+ "type":"time",`
	`67`	`+ "description":"Timeout for waiting for new cluster state in case it is blocked",`
	`68`	`+ "default":"30s"`
`64`	`69`	`}`
`65`	`70`	`}`
`66`	`71`	`}`
Original file line number	Diff line number	Diff line change
`@@ -79,6 +79,11 @@`
`79`	`79`	`],`
`80`	`80`	`"default": "all",`
`81`	`81`	`"description":"Whether to expand wildcard expression to concrete indices that are open, closed or both."`
	`82`	`+ },`
	`83`	`+ "master_timeout":{`
	`84`	`+ "type":"time",`
	`85`	`+ "description":"Timeout for waiting for new cluster state in case it is blocked",`
	`86`	`+ "default":"30s"`
`82`	`87`	`}`
`83`	`88`	`}`
`84`	`89`	`}`