Cross-cluster search: preserve cluster alias in shard failures

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