Cross Cluster Search: make remote clusters optional (elastic#27182)

Today Cross Cluster Search requires at least one node in each remote cluster to be up once the cross cluster search is run. Otherwise the whole search request fails despite some of the data (either local and/or remote) is available. This happens when performing the _search/shards calls to find out which remote shards the query has to be executed on. This scenario is different from shard failures that may happen later on when the query is actually executed, in case e.g. remote shards are missing, which is not going to fail the whole request but rather yield partial results, and the _shards section in the response will indicate that.

This commit introduces a boolean setting per cluster called search.remote.$cluster_alias.skip_if_disconnected, set to false by default, which allows to skip certain clusters if they are down when trying to reach them through a cross cluster search requests. By default all clusters are mandatory.

Scroll requests support such setting too when they are first initiated (first search request with scroll parameter), but subsequent scroll rounds (_search/scroll endpoint) will fail if some of the remote clusters went down meanwhile.

The search API response contains now a new _clusters section, similar to the _shards section, that gets returned whenever one or more clusters were disconnected and got skipped:

"_clusters" : {
    "total" : 3,
    "successful" : 2,
    "skipped" : 1
}
Such section won't be part of the response if no clusters have been skipped.

The per cluster skip_unavailable setting value has also been added to the output of the remote/info API.

client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/search/TransportNoopSearchAction.java

@@ -53,6 +53,7 @@ protected void doExecute(SearchRequest request, ActionListener<SearchResponse> l
                 new SearchHit[0], 0L, 0.0f),
             new InternalAggregations(Collections.emptyList()),
             new Suggest(Collections.emptyList()),
-            new SearchProfileShardResults(Collections.emptyMap()), false, false, 1), "", 1, 1, 0, 0, new ShardSearchFailure[0]));
+            new SearchProfileShardResults(Collections.emptyMap()), false, false, 1), "", 1, 1, 0, 0, ShardSearchFailure.EMPTY_ARRAY,
+                SearchResponse.Clusters.EMPTY));
     }
 }

client/rest-high-level/src/test/java/org/elasticsearch/client/RestHighLevelClientTests.java

@@ -166,7 +166,8 @@ public void testInfo() throws IOException {
     public void testSearchScroll() throws IOException {
         Header[] headers = randomHeaders(random(), "Header");
         SearchResponse mockSearchResponse = new SearchResponse(new SearchResponseSections(SearchHits.empty(), InternalAggregations.EMPTY,
-                null, false, false, null, 1), randomAlphaOfLengthBetween(5, 10), 5, 5, 0, 100, new ShardSearchFailure[0]);
+                null, false, false, null, 1), randomAlphaOfLengthBetween(5, 10), 5, 5, 0, 100, ShardSearchFailure.EMPTY_ARRAY,
+                SearchResponse.Clusters.EMPTY);
         mockResponse(mockSearchResponse);
         SearchResponse searchResponse = restHighLevelClient.searchScroll(new SearchScrollRequest(randomAlphaOfLengthBetween(5, 10)),
                 headers);

client/rest-high-level/src/test/java/org/elasticsearch/client/SearchIT.java

@@ -470,5 +470,6 @@ private static void assertSearchHeader(SearchResponse searchResponse) {
         assertThat(searchResponse.getTotalShards(), greaterThan(0));
         assertEquals(searchResponse.getTotalShards(), searchResponse.getSuccessfulShards());
         assertEquals(0, searchResponse.getShardFailures().length);
+        assertEquals(SearchResponse.Clusters.EMPTY, searchResponse.getClusters());
     }
 }

core/src/main/java/org/elasticsearch/action/admin/cluster/shards/ClusterSearchShardsResponse.java

@@ -30,11 +30,15 @@
 
 import java.io.IOException;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 
 public class ClusterSearchShardsResponse extends ActionResponse implements ToXContentObject {
 
+    public static final ClusterSearchShardsResponse EMPTY = new ClusterSearchShardsResponse(new ClusterSearchShardsGroup[0],
+            new DiscoveryNode[0], Collections.emptyMap());
+
     private ClusterSearchShardsGroup[] groups;
     private DiscoveryNode[] nodes;
     private Map<String, AliasFilter> indicesAndFilters;

core/src/main/java/org/elasticsearch/action/search/AbstractSearchAsyncAction.java

@@ -68,15 +68,16 @@ abstract class AbstractSearchAsyncAction<Result extends SearchPhaseResult> exten
     private final AtomicInteger successfulOps = new AtomicInteger();
     private final AtomicInteger skippedOps = new AtomicInteger();
     private final TransportSearchAction.SearchTimeProvider timeProvider;
-
+    private final SearchResponse.Clusters clusters;
 
     protected AbstractSearchAsyncAction(String name, Logger logger, SearchTransportService searchTransportService,
                                         BiFunction<String, String, Transport.Connection> nodeIdToConnection,
                                         Map<String, AliasFilter> aliasFilter, Map<String, Float> concreteIndexBoosts,
                                         Executor executor, SearchRequest request,
                                         ActionListener<SearchResponse> listener, GroupShardsIterator<SearchShardIterator> shardsIts,
                                         TransportSearchAction.SearchTimeProvider timeProvider, long clusterStateVersion,
-                                        SearchTask task, SearchPhaseResults<Result> resultConsumer, int maxConcurrentShardRequests) {
+                                        SearchTask task, SearchPhaseResults<Result> resultConsumer, int maxConcurrentShardRequests,
+                                        SearchResponse.Clusters clusters) {
         super(name, request, shardsIts, logger, maxConcurrentShardRequests, executor);
         this.timeProvider = timeProvider;
         this.logger = logger;
@@ -90,6 +91,7 @@ protected AbstractSearchAsyncAction(String name, Logger logger, SearchTransportS
         this.concreteIndexBoosts = concreteIndexBoosts;
         this.aliasFilter = aliasFilter;
         this.results = resultConsumer;
+        this.clusters = clusters;
     }
 
     /**
@@ -108,7 +110,7 @@ public final void start() {
             //no search shards to search on, bail with empty response
             //(it happens with search across _all with no indices around and consistent with broadcast operations)
             listener.onResponse(new SearchResponse(InternalSearchResponse.empty(), null, 0, 0, 0, buildTookInMillis(),
-                ShardSearchFailure.EMPTY_ARRAY));
+                ShardSearchFailure.EMPTY_ARRAY, clusters));
             return;
         }
         executePhase(this);
@@ -264,7 +266,7 @@ public final SearchRequest getRequest() {
     @Override
     public final SearchResponse buildSearchResponse(InternalSearchResponse internalSearchResponse, String scrollId) {
         return new SearchResponse(internalSearchResponse, scrollId, getNumShards(), successfulOps.get(),
-            skippedOps.get(), buildTookInMillis(), buildShardFailures());
+            skippedOps.get(), buildTookInMillis(), buildShardFailures(), clusters);
     }
 
     @Override

core/src/main/java/org/elasticsearch/action/search/CanMatchPreFilterSearchPhase.java

@@ -33,7 +33,7 @@
 import java.util.stream.Stream;
 
 /**
- * This search phrase can be used as an initial search phase to pre-filter search shards based on query rewriting.
+ * This search phase can be used as an initial search phase to pre-filter search shards based on query rewriting.
  * The queries are rewritten against the shards and based on the rewrite result shards might be able to be excluded
  * from the search. The extra round trip to the search shards is very cheap and is not subject to rejections
  * which allows to fan out to more shards at the same time without running into rejections even if we are hitting a
@@ -50,13 +50,15 @@ final class CanMatchPreFilterSearchPhase extends AbstractSearchAsyncAction<Searc
                                         Executor executor, SearchRequest request,
                                         ActionListener<SearchResponse> listener, GroupShardsIterator<SearchShardIterator> shardsIts,
                                         TransportSearchAction.SearchTimeProvider timeProvider, long clusterStateVersion,
-                                        SearchTask task, Function<GroupShardsIterator<SearchShardIterator>, SearchPhase> phaseFactory) {
+                                        SearchTask task, Function<GroupShardsIterator<SearchShardIterator>, SearchPhase> phaseFactory,
+                                        SearchResponse.Clusters clusters) {
         /*
          * We set max concurrent shard requests to the number of shards to otherwise avoid deep recursing that would occur if the local node
          * is the coordinating node for the query, holds all the shards for the request, and there are a lot of shards.
          */
         super("can_match", logger, searchTransportService, nodeIdToConnection, aliasFilter, concreteIndexBoosts, executor, request,
-            listener, shardsIts, timeProvider, clusterStateVersion, task, new BitSetSearchPhaseResults(shardsIts.size()), shardsIts.size());
+            listener, shardsIts, timeProvider, clusterStateVersion, task, new BitSetSearchPhaseResults(shardsIts.size()), shardsIts.size(),
+                clusters);
         this.phaseFactory = phaseFactory;
         this.shardsIts = shardsIts;
     }

core/src/main/java/org/elasticsearch/action/search/SearchDfsQueryThenFetchAsyncAction.java

@@ -40,10 +40,10 @@ final class SearchDfsQueryThenFetchAsyncAction extends AbstractSearchAsyncAction
             final Map<String, Float> concreteIndexBoosts, final SearchPhaseController searchPhaseController, final Executor executor,
             final SearchRequest request, final ActionListener<SearchResponse> listener,
             final GroupShardsIterator<SearchShardIterator> shardsIts, final TransportSearchAction.SearchTimeProvider timeProvider,
-            final long clusterStateVersion, final SearchTask task) {
+            final long clusterStateVersion, final SearchTask task, SearchResponse.Clusters clusters) {
         super("dfs", logger, searchTransportService, nodeIdToConnection, aliasFilter, concreteIndexBoosts, executor, request, listener,
                 shardsIts, timeProvider, clusterStateVersion, task, new ArraySearchPhaseResults<>(shardsIts.size()),
-                request.getMaxConcurrentShardRequests());
+                request.getMaxConcurrentShardRequests(), clusters);
         this.searchPhaseController = searchPhaseController;
     }
 

core/src/main/java/org/elasticsearch/action/search/SearchQueryThenFetchAsyncAction.java

@@ -40,10 +40,10 @@ final class SearchQueryThenFetchAsyncAction extends AbstractSearchAsyncAction<Se
             final Map<String, Float> concreteIndexBoosts, final SearchPhaseController searchPhaseController, final Executor executor,
             final SearchRequest request, final ActionListener<SearchResponse> listener,
             final GroupShardsIterator<SearchShardIterator> shardsIts, final TransportSearchAction.SearchTimeProvider timeProvider,
-            long clusterStateVersion, SearchTask task) {
+            long clusterStateVersion, SearchTask task, SearchResponse.Clusters clusters) {
         super("query", logger, searchTransportService, nodeIdToConnection, aliasFilter, concreteIndexBoosts, executor, request, listener,
             shardsIts, timeProvider, clusterStateVersion, task, searchPhaseController.newSearchPhaseResults(request, shardsIts.size()),
-                request.getMaxConcurrentShardRequests());
+                request.getMaxConcurrentShardRequests(), clusters);
         this.searchPhaseController = searchPhaseController;
     }
 

core/src/main/java/org/elasticsearch/action/search/SearchResponse.java

@@ -26,8 +26,10 @@
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.xcontent.StatusToXContentObject;
+import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.rest.RestStatus;
@@ -43,6 +45,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 import static org.elasticsearch.action.search.ShardSearchFailure.readShardSearchFailure;
 import static org.elasticsearch.common.xcontent.XContentParserUtils.ensureExpectedToken;
@@ -71,15 +74,18 @@ public class SearchResponse extends ActionResponse implements StatusToXContentOb
 
     private ShardSearchFailure[] shardFailures;
 
+    private Clusters clusters;
+
     private long tookInMillis;
 
     public SearchResponse() {
     }
 
     public SearchResponse(SearchResponseSections internalResponse, String scrollId, int totalShards, int successfulShards,
-                          int skippedShards, long tookInMillis, ShardSearchFailure[] shardFailures) {
+                          int skippedShards, long tookInMillis, ShardSearchFailure[] shardFailures, Clusters clusters) {
         this.internalResponse = internalResponse;
         this.scrollId = scrollId;
+        this.clusters = clusters;
         this.totalShards = totalShards;
         this.successfulShards = successfulShards;
         this.skippedShards = skippedShards;
@@ -199,6 +205,15 @@ public Map<String, ProfileShardResult> getProfileResults() {
         return internalResponse.profile();
     }
 
+    /**
+     * Returns info about what clusters the search was executed against. Available only in responses obtained
+     * from a Cross Cluster Search request, otherwise <code>null</code>
+     * @see Clusters
+     */
+    public Clusters getClusters() {
+        return clusters;
+    }
+
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
@@ -221,6 +236,7 @@ public XContentBuilder innerToXContent(XContentBuilder builder, Params params) t
         }
         RestActions.buildBroadcastShardsHeader(builder, params, getTotalShards(), getSuccessfulShards(), getSkippedShards(),
             getFailedShards(), getShardFailures());
+        clusters.toXContent(builder, params);
         internalResponse.toXContent(builder, params);
         return builder;
     }
@@ -242,6 +258,7 @@ public static SearchResponse fromXContent(XContentParser parser) throws IOExcept
         int skippedShards = 0; // 0 for BWC
         String scrollId = null;
         List<ShardSearchFailure> failures = new ArrayList<>();
+        Clusters clusters = Clusters.EMPTY;
         while((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
             if (token == XContentParser.Token.FIELD_NAME) {
                 currentFieldName = parser.currentName();
@@ -296,6 +313,28 @@ public static SearchResponse fromXContent(XContentParser parser) throws IOExcept
                             parser.skipChildren();
                         }
                     }
+                } else if (Clusters._CLUSTERS_FIELD.match(currentFieldName)) {
+                    int successful = -1;
+                    int total = -1;
+                    int skipped = -1;
+                    while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
+                        if (token == XContentParser.Token.FIELD_NAME) {
+                            currentFieldName = parser.currentName();
+                        } else if (token.isValue()) {
+                            if (Clusters.SUCCESSFUL_FIELD.match(currentFieldName)) {
+                                successful = parser.intValue();
+                            } else if (Clusters.TOTAL_FIELD.match(currentFieldName)) {
+                                total = parser.intValue();
+                            } else if (Clusters.SKIPPED_FIELD.match(currentFieldName)) {
+                                skipped = parser.intValue();
+                            } else {
+                                parser.skipChildren();
+                            }
+                        } else {
+                            parser.skipChildren();
+                        }
+                    }
+                    clusters = new Clusters(total, successful, skipped);
                 } else {
                     parser.skipChildren();
                 }
@@ -304,7 +343,7 @@ public static SearchResponse fromXContent(XContentParser parser) throws IOExcept
         SearchResponseSections searchResponseSections = new SearchResponseSections(hits, aggs, suggest, timedOut, terminatedEarly,
                 profile, numReducePhases);
         return new SearchResponse(searchResponseSections, scrollId, totalShards, successfulShards, skippedShards, tookInMillis,
-                failures.toArray(new ShardSearchFailure[failures.size()]));
+                failures.toArray(new ShardSearchFailure[failures.size()]), clusters);
     }
 
     @Override
@@ -322,6 +361,12 @@ public void readFrom(StreamInput in) throws IOException {
                 shardFailures[i] = readShardSearchFailure(in);
             }
         }
+        //TODO update version once backported
+        if (in.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
+            clusters = new Clusters(in);
+        } else {
+            clusters = Clusters.EMPTY;
+        }
         scrollId = in.readOptionalString();
         tookInMillis = in.readVLong();
         if (in.getVersion().onOrAfter(Version.V_5_6_0)) {
@@ -340,7 +385,10 @@ public void writeTo(StreamOutput out) throws IOException {
         for (ShardSearchFailure shardSearchFailure : shardFailures) {
             shardSearchFailure.writeTo(out);
         }
-
+        //TODO update version once backported
+        if (out.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
+            clusters.writeTo(out);
+        }
         out.writeOptionalString(scrollId);
         out.writeVLong(tookInMillis);
         if(out.getVersion().onOrAfter(Version.V_5_6_0)) {
@@ -353,4 +401,101 @@ public String toString() {
         return Strings.toString(this);
     }
 
+    /**
+     * Holds info about the clusters that the search was executed on: how many in total, how many of them were successful
+     * and how many of them were skipped.
+     */
+    public static class Clusters implements ToXContent, Writeable {
+
+        public static final Clusters EMPTY = new Clusters(0, 0, 0);
+
+        static final ParseField _CLUSTERS_FIELD = new ParseField("_clusters");
+        static final ParseField SUCCESSFUL_FIELD = new ParseField("successful");
+        static final ParseField SKIPPED_FIELD = new ParseField("skipped");
+        static final ParseField TOTAL_FIELD = new ParseField("total");
+
+        private final int total;
+        private final int successful;
+        private final int skipped;
+
+        Clusters(int total, int successful, int skipped) {
+            assert total >= 0 && successful >= 0 && skipped >= 0
+                    : "total: " + total + " successful: " + successful + " skipped: " + skipped;
+            assert successful <= total && skipped == total - successful
+                    : "total: " + total + " successful: " + successful + " skipped: " + skipped;
+            this.total = total;
+            this.successful = successful;
+            this.skipped = skipped;
+        }
+
+        private Clusters(StreamInput in) throws IOException {
+            this.total = in.readVInt();
+            this.successful = in.readVInt();
+            this.skipped = in.readVInt();
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            out.writeVInt(total);
+            out.writeVInt(successful);
+            out.writeVInt(skipped);
+        }
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            if (this != EMPTY) {
+                builder.startObject(_CLUSTERS_FIELD.getPreferredName());
+                builder.field(TOTAL_FIELD.getPreferredName(), total);
+                builder.field(SUCCESSFUL_FIELD.getPreferredName(), successful);
+                builder.field(SKIPPED_FIELD.getPreferredName(), skipped);
+                builder.endObject();
+            }
+            return builder;
+        }
+
+        /**
+         * Returns how many total clusters the search was requested to be executed on
+         */
+        public int getTotal() {
+            return total;
+        }
+
+        /**
+         * Returns how many total clusters the search was executed successfully on
+         */
+        public int getSuccessful() {
+            return successful;
+        }
+
+        /**
+         * Returns how many total clusters were during the execution of the search request
+         */
+        public int getSkipped() {
+            return skipped;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) {
+                return true;
+            }
+            if (o == null || getClass() != o.getClass()) {
+                return false;
+            }
+            Clusters clusters = (Clusters) o;
+            return total == clusters.total &&
+                    successful == clusters.successful &&
+                    skipped == clusters.skipped;
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(total, successful, skipped);
+        }
+
+        @Override
+        public String toString() {
+            return "Clusters{total=" + total + ", successful=" + successful + ", skipped=" + skipped + '}';
+        }
+    }
 }

core/src/main/java/org/elasticsearch/action/search/SearchScrollAsyncAction.java

@@ -249,7 +249,7 @@ protected final void sendResponse(SearchPhaseController.ReducedQueryPhase queryP
                 scrollId = request.scrollId();
             }
             listener.onResponse(new SearchResponse(internalResponse, scrollId, this.scrollId.getContext().length, successfulOps.get(),
-                0, buildTookInMillis(), buildShardFailures()));
+                0, buildTookInMillis(), buildShardFailures(), SearchResponse.Clusters.EMPTY));
         } catch (Exception e) {
             listener.onFailure(new ReduceSearchPhaseException("fetch", "inner finish failed", e, buildShardFailures()));
         }