ES|QL CCS uses skip_unavailable setting for handling disconnected remote clusters

docs/changelog/115266.yaml

@@ -0,0 +1,6 @@
+pr: 115266
+summary: ES|QL CCS uses `skip_unavailable` setting for handling disconnected remote
+  clusters
+area: ES|QL
+type: enhancement
+issues: []

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

@@ -183,6 +183,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion INTRODUCE_ALL_APPLICABLE_SELECTOR = def(8_778_00_0);
     public static final TransportVersion INDEX_MODE_LOOKUP = def(8_779_00_0);
     public static final TransportVersion INDEX_REQUEST_REMOVE_METERING = def(8_780_00_0);
+    public static final TransportVersion ESQL_CCS_EXEC_INFO_WITH_FAILURES = def(8_781_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClustersQueryIT.java

@@ -25,6 +25,7 @@
 import org.elasticsearch.test.AbstractMultiClustersTestCase;
 import org.elasticsearch.test.InternalTestCluster;
 import org.elasticsearch.test.XContentTestUtils;
+import org.elasticsearch.transport.RemoteClusterAware;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.esql.plugin.EsqlPlugin;
 import org.elasticsearch.xpack.esql.plugin.QueryPragmas;
@@ -246,7 +247,8 @@ public void testSearchesWhereMissingIndicesAreSpecified() {
 
             EsqlExecutionInfo.Cluster localCluster = executionInfo.getCluster(LOCAL_CLUSTER);
             assertThat(localCluster.getIndexExpression(), equalTo("no_such_index"));
-            assertThat(localCluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.SKIPPED));
+            // TODO: a follow on PR will change this to throw an Exception when the local cluster requests a concrete index that is missing
+            assertThat(localCluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.SUCCESSFUL));
             assertThat(localCluster.getTook().millis(), greaterThanOrEqualTo(0L));
             assertThat(localCluster.getTook().millis(), lessThanOrEqualTo(overallTookMillis));
             assertThat(localCluster.getTotalShards(), equalTo(0));
@@ -499,7 +501,7 @@ public void testCCSExecutionOnSearchesWithLimit0() {
 
             EsqlExecutionInfo.Cluster localCluster = executionInfo.getCluster(LOCAL_CLUSTER);
             assertThat(localCluster.getIndexExpression(), equalTo("nomatch*"));
-            assertThat(localCluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.SKIPPED));
+            assertThat(localCluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.SUCCESSFUL));
             assertThat(localCluster.getTook().millis(), greaterThanOrEqualTo(0L));
             assertThat(localCluster.getTook().millis(), lessThanOrEqualTo(overallTookMillis));
             assertThat(remoteCluster.getTotalShards(), equalTo(0));
@@ -803,6 +805,14 @@ Map<String, Object> setupTwoClusters() {
         clusterInfo.put("local.index", localIndex);
         clusterInfo.put("remote.num_shards", numShardsRemote);
         clusterInfo.put("remote.index", remoteIndex);
+
+        String skipUnavailableKey = Strings.format("cluster.remote.%s.skip_unavailable", REMOTE_CLUSTER);
+        Setting<?> skipUnavailableSetting = cluster(REMOTE_CLUSTER).clusterService().getClusterSettings().get(skipUnavailableKey);
+        boolean skipUnavailable = (boolean) cluster(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY).clusterService()
+            .getClusterSettings()
+            .get(skipUnavailableSetting);
+        clusterInfo.put("remote.skip_unavailable", skipUnavailable);
+
         return clusterInfo;
     }
 

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlExecutionInfo.java

@@ -8,6 +8,7 @@
 package org.elasticsearch.xpack.esql.action;
 
 import org.elasticsearch.TransportVersions;
+import org.elasticsearch.action.search.ShardSearchFailure;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
@@ -281,6 +282,7 @@ public static class Cluster implements ToXContentFragment, Writeable {
         private final Integer successfulShards;
         private final Integer skippedShards;
         private final Integer failedShards;
+        private final List<ShardSearchFailure> failures;
         private final TimeValue took;  // search latency for this cluster sub-search
 
         /**
@@ -300,7 +302,7 @@ public String toString() {
         }
 
         public Cluster(String clusterAlias, String indexExpression) {
-            this(clusterAlias, indexExpression, true, Cluster.Status.RUNNING, null, null, null, null, null);
+            this(clusterAlias, indexExpression, true, Cluster.Status.RUNNING, null, null, null, null, null, null);
         }
 
         /**
@@ -312,7 +314,7 @@ public Cluster(String clusterAlias, String indexExpression) {
          * @param skipUnavailable whether this Cluster is marked as skip_unavailable in remote cluster settings
          */
         public Cluster(String clusterAlias, String indexExpression, boolean skipUnavailable) {
-            this(clusterAlias, indexExpression, skipUnavailable, Cluster.Status.RUNNING, null, null, null, null, null);
+            this(clusterAlias, indexExpression, skipUnavailable, Cluster.Status.RUNNING, null, null, null, null, null, null);
         }
 
         /**
@@ -324,7 +326,7 @@ public Cluster(String clusterAlias, String indexExpression, boolean skipUnavaila
          * @param status current status of the search on this Cluster
          */
         public Cluster(String clusterAlias, String indexExpression, boolean skipUnavailable, Cluster.Status status) {
-            this(clusterAlias, indexExpression, skipUnavailable, status, null, null, null, null, null);
+            this(clusterAlias, indexExpression, skipUnavailable, status, null, null, null, null, null, null);
         }
 
         public Cluster(
@@ -336,6 +338,7 @@ public Cluster(
             Integer successfulShards,
             Integer skippedShards,
             Integer failedShards,
+            List<ShardSearchFailure> failures,
             TimeValue took
         ) {
             assert clusterAlias != null : "clusterAlias cannot be null";
@@ -349,6 +352,11 @@ public Cluster(
             this.successfulShards = successfulShards;
             this.skippedShards = skippedShards;
             this.failedShards = failedShards;
+            if (failures == null) {
+                this.failures = List.of();
+            } else {
+                this.failures = failures;
+            }
             this.took = took;
         }
 
@@ -362,6 +370,11 @@ public Cluster(StreamInput in) throws IOException {
             this.failedShards = in.readOptionalInt();
             this.took = in.readOptionalTimeValue();
             this.skipUnavailable = in.readBoolean();
+            if (in.getTransportVersion().onOrAfter(TransportVersions.ESQL_CCS_EXEC_INFO_WITH_FAILURES)) {
+                this.failures = Collections.unmodifiableList(in.readCollectionAsList(ShardSearchFailure::readShardSearchFailure));
+            } else {
+                this.failures = List.of();
+            }
         }
 
         @Override
@@ -375,6 +388,9 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeOptionalInt(failedShards);
             out.writeOptionalTimeValue(took);
             out.writeBoolean(skipUnavailable);
+            if (out.getTransportVersion().onOrAfter(TransportVersions.ESQL_CCS_EXEC_INFO_WITH_FAILURES)) {
+                out.writeCollection(failures);
+            }
         }
 
         /**
@@ -387,12 +403,12 @@ public void writeTo(StreamOutput out) throws IOException {
          * All other fields can be set and override the value in the "copyFrom" Cluster.
          */
         public static class Builder {
-            private String indexExpression;
             private Cluster.Status status;
             private Integer totalShards;
             private Integer successfulShards;
             private Integer skippedShards;
             private Integer failedShards;
+            private List<ShardSearchFailure> failures;
             private TimeValue took;
             private final Cluster original;
 
@@ -408,22 +424,18 @@ public Builder(Cluster copyFrom) {
             public Cluster build() {
                 return new Cluster(
                     original.getClusterAlias(),
-                    indexExpression == null ? original.getIndexExpression() : indexExpression,
+                    original.getIndexExpression(),
                     original.isSkipUnavailable(),
                     status != null ? status : original.getStatus(),
                     totalShards != null ? totalShards : original.getTotalShards(),
                     successfulShards != null ? successfulShards : original.getSuccessfulShards(),
                     skippedShards != null ? skippedShards : original.getSkippedShards(),
                     failedShards != null ? failedShards : original.getFailedShards(),
+                    failures != null ? failures : original.getFailures(),
                     took != null ? took : original.getTook()
                 );
             }
 
-            public Cluster.Builder setIndexExpression(String indexExpression) {
-                this.indexExpression = indexExpression;
-                return this;
-            }
-
             public Cluster.Builder setStatus(Cluster.Status status) {
                 this.status = status;
                 return this;
@@ -449,6 +461,11 @@ public Cluster.Builder setFailedShards(int failedShards) {
                 return this;
             }
 
+            public Cluster.Builder setFailures(List<ShardSearchFailure> failures) {
+                this.failures = failures;
+                return this;
+            }
+
             public Cluster.Builder setTook(TimeValue took) {
                 this.took = took;
                 return this;
@@ -466,7 +483,6 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
                 builder.field(STATUS_FIELD.getPreferredName(), getStatus().toString());
                 builder.field(INDICES_FIELD.getPreferredName(), indexExpression);
                 if (took != null) {
-                    // TODO: change this to took_nanos and call took.nanos?
                     builder.field(TOOK.getPreferredName(), took.millis());
                 }
                 if (totalShards != null) {
@@ -483,6 +499,13 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
                     }
                     builder.endObject();
                 }
+                if (failures != null && failures.size() > 0) {
+                    builder.startArray(RestActions.FAILURES_FIELD.getPreferredName());
+                    for (ShardSearchFailure failure : failures) {
+                        failure.toXContent(builder, params);
+                    }
+                    builder.endArray();
+                }
             }
             builder.endObject();
             return builder;
@@ -529,6 +552,10 @@ public Integer getFailedShards() {
             return failedShards;
         }
 
+        public List<ShardSearchFailure> getFailures() {
+            return failures;
+        }
+
         @Override
         public boolean equals(Object o) {
             if (this == o) return true;

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/EnrichResolution.java

@@ -23,6 +23,7 @@ public final class EnrichResolution {
 
     private final Map<Key, ResolvedEnrichPolicy> resolvedPolicies = ConcurrentCollections.newConcurrentMap();
     private final Map<Key, String> errors = ConcurrentCollections.newConcurrentMap();
+    private final Map<String, Exception> unavailableClusters = ConcurrentCollections.newConcurrentMap();
 
     public ResolvedEnrichPolicy getResolvedPolicy(String policyName, Enrich.Mode mode) {
         return resolvedPolicies.get(new Key(policyName, mode));
@@ -51,6 +52,14 @@ public void addError(String policyName, Enrich.Mode mode, String reason) {
         errors.putIfAbsent(new Key(policyName, mode), reason);
     }
 
+    public void addUnavailableCluster(String clusterAlias, Exception e) {
+        unavailableClusters.put(clusterAlias, e);
+    }
+
+    public Map<String, Exception> getUnavailableClusters() {
+        return unavailableClusters;
+    }
+
     private record Key(String policyName, Enrich.Mode mode) {
 
     }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/EnrichPolicyResolver.java

@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.esql.enrich;
 
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionListenerResponseHandler;
 import org.elasticsearch.action.search.SearchRequest;
@@ -50,6 +51,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -113,12 +115,27 @@ public void resolvePolicies(
         final boolean includeLocal = remoteClusters.remove(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY);
         lookupPolicies(remoteClusters, includeLocal, unresolvedPolicies, listener.map(lookupResponses -> {
             final EnrichResolution enrichResolution = new EnrichResolution();
+
+            Map<String, LookupResponse> lookupResponsesToProcess = new HashMap<>();
+
+            for (Map.Entry<String, LookupResponse> entry : lookupResponses.entrySet()) {
+                String clusterAlias = entry.getKey();
+                if (entry.getValue().connectionError != null) {
+                    enrichResolution.addUnavailableCluster(clusterAlias, entry.getValue().connectionError);
+                    // remove unavailable cluster from the list of clusters which is used below to create the ResolvedEnrichPolicy
+                    remoteClusters.remove(clusterAlias);
+                } else {
+                    lookupResponsesToProcess.put(clusterAlias, entry.getValue());
+                }
+            }
+
             for (UnresolvedPolicy unresolved : unresolvedPolicies) {
                 Tuple<ResolvedEnrichPolicy, String> resolved = mergeLookupResults(
                     unresolved,
                     calculateTargetClusters(unresolved.mode, includeLocal, remoteClusters),
-                    lookupResponses
+                    lookupResponsesToProcess
                 );
+
                 if (resolved.v1() != null) {
                     enrichResolution.addResolvedPolicy(unresolved.name, unresolved.mode, resolved.v1());
                 } else {
@@ -149,13 +166,16 @@ private Tuple<ResolvedEnrichPolicy, String> mergeLookupResults(
         Collection<String> targetClusters,
         Map<String, LookupResponse> lookupResults
     ) {
-        assert targetClusters.isEmpty() == false;
         String policyName = unresolved.name;
+        if (targetClusters.isEmpty()) {
+            return Tuple.tuple(null, "enrich policy [" + policyName + "] cannot be resolved since remote clusters are unavailable");
+        }
         final Map<String, ResolvedEnrichPolicy> policies = new HashMap<>();
         final List<String> failures = new ArrayList<>();
         for (String cluster : targetClusters) {
             LookupResponse lookupResult = lookupResults.get(cluster);
             if (lookupResult != null) {
+                assert lookupResult.connectionError == null : "Should never have a non-null connectionError here";
                 ResolvedEnrichPolicy policy = lookupResult.policies.get(policyName);
                 if (policy != null) {
                     policies.put(cluster, policy);
@@ -261,22 +281,34 @@ private void lookupPolicies(
             if (remotePolicies.isEmpty() == false) {
                 for (String cluster : remoteClusters) {
                     ActionListener<LookupResponse> lookupListener = refs.acquire(resp -> lookupResponses.put(cluster, resp));
-                    getRemoteConnection(
-                        cluster,
-                        lookupListener.delegateFailureAndWrap(
-                            (delegate, connection) -> transportService.sendRequest(
+                    getRemoteConnection(cluster, new ActionListener<Transport.Connection>() {
+                        @Override
+                        public void onResponse(Transport.Connection connection) {
+                            transportService.sendRequest(
                                 connection,
                                 RESOLVE_ACTION_NAME,
                                 new LookupRequest(cluster, remotePolicies),
                                 TransportRequestOptions.EMPTY,
-                                new ActionListenerResponseHandler<>(
-                                    delegate,
-                                    LookupResponse::new,
-                                    threadPool.executor(ThreadPool.Names.SEARCH)
-                                )
-                            )
-                        )
-                    );
+                                new ActionListenerResponseHandler<>(lookupListener.delegateResponse((l, e) -> {
+                                    if (ExceptionsHelper.isRemoteUnavailableException(e)
+                                        && remoteClusterService.isSkipUnavailable(cluster)) {
+                                        l.onResponse(new LookupResponse(e));
+                                    } else {
+                                        l.onFailure(e);
+                                    }
+                                }), LookupResponse::new, threadPool.executor(ThreadPool.Names.SEARCH))
+                            );
+                        }
+
+                        @Override
+                        public void onFailure(Exception e) {
+                            if (ExceptionsHelper.isRemoteUnavailableException(e) && remoteClusterService.isSkipUnavailable(cluster)) {
+                                lookupListener.onResponse(new LookupResponse(e));
+                            } else {
+                                lookupListener.onFailure(e);
+                            }
+                        }
+                    });
                 }
             }
             // local cluster
@@ -323,16 +355,30 @@ public void writeTo(StreamOutput out) throws IOException {
     private static class LookupResponse extends TransportResponse {
         final Map<String, ResolvedEnrichPolicy> policies;
         final Map<String, String> failures;
+        // does not need to be Writable since this indicates a failure to contact a remote cluster, so only set on querying cluster
+        final transient Exception connectionError;
 
         LookupResponse(Map<String, ResolvedEnrichPolicy> policies, Map<String, String> failures) {
             this.policies = policies;
             this.failures = failures;
+            this.connectionError = null;
+        }
+
+        /**
+         * Use this constructor when the remote cluster is unavailable to indicate inability to do the enrich policy lookup
+         * @param connectionError
+         */
+        LookupResponse(Exception connectionError) {
+            this.policies = Collections.emptyMap();
+            this.failures = Collections.emptyMap();
+            this.connectionError = connectionError;
         }
 
         LookupResponse(StreamInput in) throws IOException {
             PlanStreamInput planIn = new PlanStreamInput(in, in.namedWriteableRegistry(), null);
             this.policies = planIn.readMap(StreamInput::readString, ResolvedEnrichPolicy::new);
             this.failures = planIn.readMap(StreamInput::readString, StreamInput::readString);
+            this.connectionError = null;
         }
 
         @Override