Skip to content

Commit

Permalink
Cross-cluster search: preserve cluster alias in shard failures (#32608)
Browse files Browse the repository at this point in the history
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.
  • Loading branch information
javanna authored Aug 6, 2018
1 parent 3cf0832 commit 826399f
Show file tree
Hide file tree
Showing 7 changed files with 335 additions and 34 deletions.
57 changes: 32 additions & 25 deletions server/src/main/java/org/elasticsearch/ExceptionsHelper.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand All @@ -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);
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -105,7 +106,7 @@ public RestStatus status() {
@Override
public String index() {
if (shardTarget != null) {
return shardTarget.getIndex();
return shardTarget.getFullyQualifiedIndexName();
}
return null;
}
Expand Down Expand Up @@ -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) {
Expand All @@ -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 {
Expand All @@ -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);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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());
}

/**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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.
*/
Expand Down Expand Up @@ -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());
Expand Down
115 changes: 115 additions & 0 deletions server/src/test/java/org/elasticsearch/ExceptionsHelperTests.java
Original file line number Diff line number Diff line change
Expand Up @@ -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 {

Expand Down Expand Up @@ -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));
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down Expand Up @@ -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());
}
}
Loading

0 comments on commit 826399f

Please sign in to comment.