Make TransportLocalClusterStateAction wait for cluster to unblock

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.

Author: nielsbauman

rest-api-spec/src/main/resources/rest-api-spec/api/cat.aliases.json

@@ -65,6 +65,10 @@
         ],
         "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"
       }
     }
   }

rest-api-spec/src/main/resources/rest-api-spec/api/indices.exists_alias.json

@@ -61,6 +61,10 @@
         ],
         "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"
       }
     }
   }

rest-api-spec/src/main/resources/rest-api-spec/api/indices.get_alias.json

@@ -79,6 +79,10 @@
         ],
         "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"
       }
     }
   }

server/src/main/java/module-info.java

@@ -146,6 +146,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;

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,24 @@ 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 clusterUpdateTimeout, String... aliases) {
+        super(clusterUpdateTimeout);
+        this.aliases = aliases;
+        this.originalAliases = aliases;
+    }
+
+    public GetAliasesRequest(TimeValue clusterUpdateTimeout) {
+        this(clusterUpdateTimeout, Strings.EMPTY_ARRAY);
     }
 
     @Override

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;

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

@@ -0,0 +1,39 @@
+/*
+ * 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 {
+
+    private final TimeValue clusterUpdateTimeout;
+
+    protected LocalClusterStateRequest(TimeValue clusterUpdateTimeout) {
+        this.clusterUpdateTimeout = Objects.requireNonNull(clusterUpdateTimeout);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        TransportAction.localOnly();
+    }
+
+    public TimeValue clusterUpdateTimeout() {
+        return clusterUpdateTimeout;
+    }
+}

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

@@ -0,0 +1,129 @@
+/*
+ * 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);
+
+    @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);
+            }
+            return;
+        }
+
+        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.clusterUpdateTimeout(),
+            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();
+    }
+
+    protected abstract void localClusterStateOperation(Task task, Request request, ClusterState state, ActionListener<Response> listener)
+        throws Exception;
+}

server/src/main/java/org/elasticsearch/action/support/local/TransportLocalClusterStateAction.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()));