Make TransportLocalClusterStateAction wait for cluster to unblock

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: []

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}
 

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}
 

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]

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"
  }
  }
  }

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"
  }
  }
  }

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"
  }
  }
  }

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;

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

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/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;
+ }
+}

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();
+ }
+}

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()));