Cross-cluster search: preserve cluster alias in shard failures (#32608)

When some remote clusters return shard failures as part of a cross-cluster search request, the cluster alias currently gets lost. As a result, if the shard failures are all caused by the same error, and against indices belonging to different clusters, but with the same index name, only one failure gets returned as part of the search response, meaning that failures are grouped by index name, ignoring the cluster alias.

With this commit we make sure that `ShardSearchFailure` returns the cluster alias as part of the index name. Also, we set the fully qualfied index name when creating a `QueryShardException`. That way shard failures are grouped by cluster:index. Such fixes should cover at least most of the cases where either 1) the shard target is set but we don't have the index in the cause (we were previously reading it only from the cause that did not have the cluster alias) 2) the shard target is missing but if the cause is a `QueryShardException` the cluster alias does not get lost.

We also prevent NPE in case the failure cause is not set and test such scenario.

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

@@ -39,6 +39,7 @@
 import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Queue;
 import java.util.Set;
@@ -278,7 +279,7 @@ public static ShardOperationFailedException[] groupBy(ShardOperationFailedExcept
         List<ShardOperationFailedException> uniqueFailures = new ArrayList<>();
         Set<GroupBy> reasons = new HashSet<>();
         for (ShardOperationFailedException failure : failures) {
-            GroupBy reason = new GroupBy(failure.getCause());
+            GroupBy reason = new GroupBy(failure);
             if (reasons.contains(reason) == false) {
                 reasons.add(reason);
                 uniqueFailures.add(failure);
@@ -287,46 +288,52 @@ public static ShardOperationFailedException[] groupBy(ShardOperationFailedExcept
         return uniqueFailures.toArray(new ShardOperationFailedException[0]);
     }
 
-    static class GroupBy {
+    private static class GroupBy {
         final String reason;
         final String index;
         final Class<? extends Throwable> causeType;
 
-        GroupBy(Throwable t) {
-            if (t instanceof ElasticsearchException) {
-                final Index index = ((ElasticsearchException) t).getIndex();
-                if (index != null) {
-                    this.index = index.getName();
-                } else {
-                    this.index = null;
+        GroupBy(ShardOperationFailedException failure) {
+            Throwable cause = failure.getCause();
+            //the index name from the failure contains the cluster alias when using CCS. Ideally failures should be grouped by
+            //index name and cluster alias. That's why the failure index name has the precedence over the one coming from the cause,
+            //which does not include the cluster alias.
+            String indexName = failure.index();
+            if (indexName == null) {
+                if (cause instanceof ElasticsearchException) {
+                    final Index index = ((ElasticsearchException) cause).getIndex();
+                    if (index != null) {
+                        indexName = index.getName();
+                    }
                 }
+            }
+            this.index = indexName;
+            if (cause == null) {
+                this.reason = failure.reason();
+                this.causeType = null;
             } else {
-                index = null;
+                this.reason = cause.getMessage();
+                this.causeType = cause.getClass();
             }
-            reason = t.getMessage();
-            causeType = t.getClass();
         }
 
         @Override
         public boolean equals(Object o) {
-            if (this == o) return true;
-            if (o == null || getClass() != o.getClass()) return false;
-
+            if (this == o) {
+                return true;
+            }
+            if (o == null || getClass() != o.getClass()) {
+                return false;
+            }
             GroupBy groupBy = (GroupBy) o;
-
-            if (!causeType.equals(groupBy.causeType)) return false;
-            if (index != null ? !index.equals(groupBy.index) : groupBy.index != null) return false;
-            if (reason != null ? !reason.equals(groupBy.reason) : groupBy.reason != null) return false;
-
-            return true;
+            return Objects.equals(reason, groupBy.reason) &&
+                Objects.equals(index, groupBy.index) &&
+                Objects.equals(causeType, groupBy.causeType);
         }
 
         @Override
         public int hashCode() {
-            int result = reason != null ? reason.hashCode() : 0;
-            result = 31 * result + (index != null ? index.hashCode() : 0);
-            result = 31 * result + causeType.hashCode();
-            return result;
+            return Objects.hash(reason, index, causeType);
         }
     }
 }

server/src/main/java/org/elasticsearch/action/search/ShardSearchFailure.java

@@ -34,6 +34,7 @@
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.search.SearchException;
 import org.elasticsearch.search.SearchShardTarget;
+import org.elasticsearch.transport.RemoteClusterAware;
 
 import java.io.IOException;
 
@@ -66,7 +67,7 @@ public ShardSearchFailure(Exception e) {
 
     public ShardSearchFailure(Exception e, @Nullable SearchShardTarget shardTarget) {
         final Throwable actual = ExceptionsHelper.unwrapCause(e);
-        if (actual != null && actual instanceof SearchException) {
+        if (actual instanceof SearchException) {
             this.shardTarget = ((SearchException) actual).shard();
         } else if (shardTarget != null) {
             this.shardTarget = shardTarget;
@@ -105,7 +106,7 @@ public RestStatus status() {
     @Override
     public String index() {
         if (shardTarget != null) {
-            return shardTarget.getIndex();
+            return shardTarget.getFullyQualifiedIndexName();
         }
         return null;
     }
@@ -186,6 +187,7 @@ public static ShardSearchFailure fromXContent(XContentParser parser) throws IOEx
         String currentFieldName = null;
         int shardId = -1;
         String indexName = null;
+        String clusterAlias = null;
         String nodeId = null;
         ElasticsearchException exception = null;
         while((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
@@ -196,6 +198,11 @@ public static ShardSearchFailure fromXContent(XContentParser parser) throws IOEx
                     shardId  = parser.intValue();
                 } else if (INDEX_FIELD.equals(currentFieldName)) {
                     indexName  = parser.text();
+                    int indexOf = indexName.indexOf(RemoteClusterAware.REMOTE_CLUSTER_INDEX_SEPARATOR);
+                    if (indexOf > 0) {
+                        clusterAlias = indexName.substring(0, indexOf);
+                        indexName = indexName.substring(indexOf + 1);
+                    }
                 } else if (NODE_FIELD.equals(currentFieldName)) {
                     nodeId  = parser.text();
                 } else {
@@ -214,7 +221,7 @@ public static ShardSearchFailure fromXContent(XContentParser parser) throws IOEx
         SearchShardTarget searchShardTarget = null;
         if (nodeId != null) {
             searchShardTarget = new SearchShardTarget(nodeId,
-                    new ShardId(new Index(indexName, IndexMetaData.INDEX_UUID_NA_VALUE), shardId), null, OriginalIndices.NONE);
+                    new ShardId(new Index(indexName, IndexMetaData.INDEX_UUID_NA_VALUE), shardId), clusterAlias, OriginalIndices.NONE);
         }
         return new ShardSearchFailure(exception, searchShardTarget);
     }

server/src/main/java/org/elasticsearch/index/query/QueryShardException.java

@@ -38,7 +38,7 @@ public QueryShardException(QueryShardContext context, String msg, Object... args
 
     public QueryShardException(QueryShardContext context, String msg, Throwable cause, Object... args) {
         super(msg, cause, args);
-        setIndex(context.index());
+        setIndex(context.getFullyQualifiedIndexName());
     }
 
     /**

server/src/main/java/org/elasticsearch/search/SearchShardTarget.java

@@ -19,8 +19,6 @@
 
 package org.elasticsearch.search;
 
-import java.io.IOException;
-
 import org.elasticsearch.Version;
 import org.elasticsearch.action.OriginalIndices;
 import org.elasticsearch.common.Nullable;
@@ -32,6 +30,8 @@
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.transport.RemoteClusterAware;
 
+import java.io.IOException;
+
 /**
  * The target that the search request was executed on.
  */
@@ -96,6 +96,13 @@ public String getClusterAlias() {
         return clusterAlias;
     }
 
+    /**
+     * Returns the fully qualified index name, including the cluster alias.
+     */
+    public String getFullyQualifiedIndexName() {
+        return RemoteClusterAware.buildRemoteIndexName(getClusterAlias(), getIndex());
+    }
+
     @Override
     public int compareTo(SearchShardTarget o) {
         int i = shardId.getIndexName().compareTo(o.getIndex());

server/src/test/java/org/elasticsearch/ExceptionsHelperTests.java

@@ -20,15 +20,27 @@
 package org.elasticsearch;
 
 import org.apache.commons.codec.DecoderException;
+import org.elasticsearch.action.OriginalIndices;
+import org.elasticsearch.action.ShardOperationFailedException;
+import org.elasticsearch.action.search.ShardSearchFailure;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
+import org.elasticsearch.common.ParsingException;
 import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
+import org.elasticsearch.index.Index;
+import org.elasticsearch.index.query.QueryShardException;
+import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.rest.RestStatus;
+import org.elasticsearch.search.SearchShardTarget;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.transport.RemoteClusterAware;
 
 import java.util.Optional;
 
 import static org.elasticsearch.ExceptionsHelper.MAX_ITERATIONS;
 import static org.elasticsearch.ExceptionsHelper.maybeError;
 import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.nullValue;
 
 public class ExceptionsHelperTests extends ESTestCase {
 
@@ -91,4 +103,107 @@ public void testStatus() {
         assertThat(ExceptionsHelper.status(new EsRejectedExecutionException("rejected")), equalTo(RestStatus.TOO_MANY_REQUESTS));
     }
 
+    public void testGroupBy() {
+        ShardOperationFailedException[] failures = new ShardOperationFailedException[]{
+            createShardFailureParsingException("error", "node0", "index", 0, null),
+            createShardFailureParsingException("error", "node1", "index", 1, null),
+            createShardFailureParsingException("error", "node2", "index2", 2, null),
+            createShardFailureParsingException("error", "node0", "index", 0, "cluster1"),
+            createShardFailureParsingException("error", "node1", "index", 1, "cluster1"),
+            createShardFailureParsingException("error", "node2", "index", 2, "cluster1"),
+            createShardFailureParsingException("error", "node0", "index", 0, "cluster2"),
+            createShardFailureParsingException("error", "node1", "index", 1, "cluster2"),
+            createShardFailureParsingException("error", "node2", "index", 2, "cluster2"),
+            createShardFailureParsingException("another error", "node2", "index", 2, "cluster2")
+        };
+
+        ShardOperationFailedException[] groupBy = ExceptionsHelper.groupBy(failures);
+        assertThat(groupBy.length, equalTo(5));
+        String[] expectedIndices = new String[]{"index", "index2", "cluster1:index", "cluster2:index", "cluster2:index"};
+        String[] expectedErrors = new String[]{"error", "error", "error", "error", "another error"};
+        int i = 0;
+        for (ShardOperationFailedException shardOperationFailedException : groupBy) {
+            assertThat(shardOperationFailedException.getCause().getMessage(), equalTo(expectedErrors[i]));
+            assertThat(shardOperationFailedException.index(), equalTo(expectedIndices[i++]));
+        }
+    }
+
+    private static ShardSearchFailure createShardFailureParsingException(String error, String nodeId,
+                                                                         String index, int shardId, String clusterAlias) {
+        ParsingException ex = new ParsingException(0, 0, error, new IllegalArgumentException("some bad argument"));
+        ex.setIndex(index);
+        return new ShardSearchFailure(ex, createSearchShardTarget(nodeId, shardId, index, clusterAlias));
+    }
+
+    private static SearchShardTarget createSearchShardTarget(String nodeId, int shardId, String index, String clusterAlias) {
+        return new SearchShardTarget(nodeId,
+            new ShardId(new Index(index, IndexMetaData.INDEX_UUID_NA_VALUE), shardId), clusterAlias, OriginalIndices.NONE);
+    }
+
+    public void testGroupByNullTarget() {
+        ShardOperationFailedException[] failures = new ShardOperationFailedException[] {
+            createShardFailureQueryShardException("error", "index", null),
+            createShardFailureQueryShardException("error", "index", null),
+            createShardFailureQueryShardException("error", "index", null),
+            createShardFailureQueryShardException("error", "index", "cluster1"),
+            createShardFailureQueryShardException("error", "index", "cluster1"),
+            createShardFailureQueryShardException("error", "index", "cluster1"),
+            createShardFailureQueryShardException("error", "index", "cluster2"),
+            createShardFailureQueryShardException("error", "index", "cluster2"),
+            createShardFailureQueryShardException("error", "index2", null),
+            createShardFailureQueryShardException("another error", "index2", null),
+        };
+
+        ShardOperationFailedException[] groupBy = ExceptionsHelper.groupBy(failures);
+        assertThat(groupBy.length, equalTo(5));
+        String[] expectedIndices = new String[]{"index", "cluster1:index", "cluster2:index", "index2", "index2"};
+        String[] expectedErrors = new String[]{"error", "error", "error", "error", "another error"};
+        int i = 0;
+        for (ShardOperationFailedException shardOperationFailedException : groupBy) {
+            assertThat(shardOperationFailedException.index(), nullValue());
+            assertThat(shardOperationFailedException.getCause(), instanceOf(ElasticsearchException.class));
+            ElasticsearchException elasticsearchException = (ElasticsearchException) shardOperationFailedException.getCause();
+            assertThat(elasticsearchException.getMessage(), equalTo(expectedErrors[i]));
+            assertThat(elasticsearchException.getIndex().getName(), equalTo(expectedIndices[i++]));
+        }
+    }
+
+    private static ShardSearchFailure createShardFailureQueryShardException(String error, String indexName, String clusterAlias) {
+        Index index = new Index(RemoteClusterAware.buildRemoteIndexName(clusterAlias, indexName), "uuid");
+        QueryShardException queryShardException = new QueryShardException(index, error, new IllegalArgumentException("parse error"));
+        return new ShardSearchFailure(queryShardException, null);
+    }
+
+    public void testGroupByNullCause() {
+        ShardOperationFailedException[] failures = new ShardOperationFailedException[] {
+            new ShardSearchFailure("error", createSearchShardTarget("node0", 0, "index", null)),
+            new ShardSearchFailure("error", createSearchShardTarget("node1", 1, "index", null)),
+            new ShardSearchFailure("error", createSearchShardTarget("node1", 1, "index2", null)),
+            new ShardSearchFailure("error", createSearchShardTarget("node2", 2, "index", "cluster1")),
+            new ShardSearchFailure("error", createSearchShardTarget("node1", 1, "index", "cluster1")),
+            new ShardSearchFailure("a different error", createSearchShardTarget("node3", 3, "index", "cluster1"))
+        };
+
+        ShardOperationFailedException[] groupBy = ExceptionsHelper.groupBy(failures);
+        assertThat(groupBy.length, equalTo(4));
+        String[] expectedIndices = new String[]{"index", "index2", "cluster1:index", "cluster1:index"};
+        String[] expectedErrors = new String[]{"error", "error", "error", "a different error"};
+
+        int i = 0;
+        for (ShardOperationFailedException shardOperationFailedException : groupBy) {
+            assertThat(shardOperationFailedException.reason(), equalTo(expectedErrors[i]));
+            assertThat(shardOperationFailedException.index(), equalTo(expectedIndices[i++]));
+        }
+    }
+
+    public void testGroupByNullIndex() {
+        ShardOperationFailedException[] failures = new ShardOperationFailedException[] {
+            new ShardSearchFailure("error", null),
+            new ShardSearchFailure(new IllegalArgumentException("error")),
+            new ShardSearchFailure(new ParsingException(0, 0, "error", null)),
+        };
+
+        ShardOperationFailedException[] groupBy = ExceptionsHelper.groupBy(failures);
+        assertThat(groupBy.length, equalTo(3));
+    }
 }

server/src/test/java/org/elasticsearch/action/search/ShardSearchFailureTests.java

@@ -45,8 +45,9 @@ public static ShardSearchFailure createTestItem() {
         if (randomBoolean()) {
             String nodeId = randomAlphaOfLengthBetween(5, 10);
             String indexName = randomAlphaOfLengthBetween(5, 10);
+            String clusterAlias = randomBoolean() ? randomAlphaOfLengthBetween(5, 10) : null;
             searchShardTarget = new SearchShardTarget(nodeId,
-                    new ShardId(new Index(indexName, IndexMetaData.INDEX_UUID_NA_VALUE), randomInt()), null, null);
+                    new ShardId(new Index(indexName, IndexMetaData.INDEX_UUID_NA_VALUE), randomInt()), clusterAlias, OriginalIndices.NONE);
         }
         return new ShardSearchFailure(ex, searchShardTarget);
     }
@@ -115,4 +116,22 @@ public void testToXContent() throws IOException {
                 + "}",
                 xContent.utf8ToString());
     }
+
+    public void testToXContentWithClusterAlias() throws IOException {
+        ShardSearchFailure failure = new ShardSearchFailure(new ParsingException(0, 0, "some message", null),
+            new SearchShardTarget("nodeId", new ShardId(new Index("indexName", "indexUuid"), 123), "cluster1", OriginalIndices.NONE));
+        BytesReference xContent = toXContent(failure, XContentType.JSON, randomBoolean());
+        assertEquals(
+            "{\"shard\":123,"
+                + "\"index\":\"cluster1:indexName\","
+                + "\"node\":\"nodeId\","
+                + "\"reason\":{"
+                + "\"type\":\"parsing_exception\","
+                + "\"reason\":\"some message\","
+                + "\"line\":0,"
+                + "\"col\":0"
+                + "}"
+                + "}",
+            xContent.utf8ToString());
+    }
 }