Skip to content

Do not auto-follow closed indices (#47721) #47801

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Oct 9, 2019
Merged

Do not auto-follow closed indices (#47721) #47801

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4fbf36b
Do not auto-follow closed indices (#47721)
tlrx Oct 9, 2019
43962ec
Fix AutoFollowCoordinatorTests
tlrx Oct 9, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions ...ck/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/AutoFollowCoordinator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -596,6 +596,9 @@ static List<Index> getLeaderIndicesToFollow(AutoFollowPattern autoFollowPattern,
List<String> followedIndexUUIDs) {
List<Index> leaderIndicesToFollow = new ArrayList<>();
for (IndexMetaData leaderIndexMetaData : remoteClusterState.getMetaData()) {
if (leaderIndexMetaData.getState() != IndexMetaData.State.OPEN) {
continue;
}
if (autoFollowPattern.match(leaderIndexMetaData.getIndex().getName())) {
IndexRoutingTable indexRoutingTable = remoteClusterState.routingTable().index(leaderIndexMetaData.getIndex());
if (indexRoutingTable != null &&
Expand Down
120 changes: 116 additions & 4 deletions ...ugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/AutoFollowCoordinatorTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,10 @@
*/
package org.elasticsearch.xpack.ccr.action;

import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
import org.elasticsearch.Version;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.action.support.replication.ClusterStateCreationUtils;
import org.elasticsearch.client.Client;
import org.elasticsearch.cluster.ClusterName;
import org.elasticsearch.cluster.ClusterState;
Expand All @@ -18,11 +20,13 @@
import org.elasticsearch.cluster.routing.ShardRoutingState;
import org.elasticsearch.cluster.routing.TestShardRouting;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.UUIDs;
import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.common.settings.ClusterSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
import org.elasticsearch.index.Index;
import org.elasticsearch.index.IndexSettings;
Expand All @@ -44,10 +48,12 @@
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
Expand All @@ -57,6 +63,7 @@
import static org.elasticsearch.xpack.ccr.action.AutoFollowCoordinator.AutoFollower.recordLeaderIndexAsFollowFunction;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.hasItem;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.nullValue;
Expand Down Expand Up @@ -416,6 +423,26 @@ public void testGetLeaderIndicesToFollow_shardsNotStarted() {
assertThat(result.get(1).getName(), equalTo("index2"));
}

public void testGetLeaderIndicesToFollowWithClosedIndices() {
final AutoFollowPattern autoFollowPattern = new AutoFollowPattern("remote", Collections.singletonList("*"),
null, null, null, null, null, null, null, null, null, null, null);

// index is opened
ClusterState remoteState = ClusterStateCreationUtils.stateWithActivePrimary("test-index", true, randomIntBetween(1, 3), 0);
List<Index> result = AutoFollower.getLeaderIndicesToFollow(autoFollowPattern, remoteState, Collections.emptyList());
assertThat(result.size(), equalTo(1));
assertThat(result, hasItem(remoteState.metaData().index("test-index").getIndex()));

// index is closed
remoteState = ClusterState.builder(remoteState)
.metaData(MetaData.builder(remoteState.metaData())
.put(IndexMetaData.builder(remoteState.metaData().index("test-index")).state(IndexMetaData.State.CLOSE).build(), true)
.build())
.build();
result = AutoFollower.getLeaderIndicesToFollow(autoFollowPattern, remoteState, Collections.emptyList());
assertThat(result.size(), equalTo(0));
}

public void testRecordLeaderIndexAsFollowFunction() {
AutoFollowMetadata autoFollowMetadata = new AutoFollowMetadata(Collections.emptyMap(),
Collections.singletonMap("pattern1", Collections.emptyList()), Collections.emptyMap());
Expand Down Expand Up @@ -763,7 +790,9 @@ void updateAutoFollowMetadata(Function<ClusterState, ClusterState> updateFunctio
autoFollower.start();
assertThat(allResults.size(), equalTo(states.length));
for (int i = 0; i < states.length; i++) {
assertThat(allResults.get(i).autoFollowExecutionResults.containsKey(new Index("logs-" + i, "_na_")), is(true));
final String indexName = "logs-" + i;
assertThat(allResults.get(i).autoFollowExecutionResults.keySet().stream()
.anyMatch(index -> index.getName().equals(indexName)), is(true));
}
}

Expand Down Expand Up @@ -1049,6 +1078,87 @@ void updateAutoFollowMetadata(
}
}

public void testClosedIndicesAreNotAutoFollowed() {
final Client client = mock(Client.class);
when(client.getRemoteClusterClient(anyString())).thenReturn(client);

final String pattern = "pattern1";
final ClusterState localState = ClusterState.builder(new ClusterName("local"))
.metaData(MetaData.builder()
.putCustom(AutoFollowMetadata.TYPE,
new AutoFollowMetadata(Collections.singletonMap(pattern,
new AutoFollowPattern("remote", Collections.singletonList("docs-*"), null, null, null, null, null, null, null, null,
null, null, null)),
Collections.singletonMap(pattern, Collections.emptyList()),
Collections.singletonMap(pattern, Collections.emptyMap()))))
.build();

ClusterState remoteState = null;
final int nbLeaderIndices = randomIntBetween(1, 15);
for (int i = 0; i < nbLeaderIndices; i++) {
String indexName = "docs-" + i;
if (remoteState == null) {
remoteState = createRemoteClusterState(indexName, true);
} else {
remoteState = createRemoteClusterState(remoteState, indexName);
}
if (randomBoolean()) {
// randomly close the index
remoteState = ClusterState.builder(remoteState.getClusterName())
.routingTable(remoteState.routingTable())
.metaData(MetaData.builder(remoteState.metaData())
.put(IndexMetaData.builder(remoteState.metaData().index(indexName)).state(IndexMetaData.State.CLOSE).build(), true)
.build())
.build();
}
}

final ClusterState finalRemoteState = remoteState;
final AtomicReference<ClusterState> lastModifiedClusterState = new AtomicReference<>(localState);
final List<AutoFollowCoordinator.AutoFollowResult> results = new ArrayList<>();
final Set<Object> followedIndices = ConcurrentCollections.newConcurrentSet();
final AutoFollower autoFollower =
new AutoFollower("remote", results::addAll, localClusterStateSupplier(localState), () -> 1L, Runnable::run) {
@Override
void getRemoteClusterState(String remoteCluster,
long metadataVersion,
BiConsumer<ClusterStateResponse, Exception> handler) {
assertThat(remoteCluster, equalTo("remote"));
handler.accept(new ClusterStateResponse(new ClusterName("remote"), finalRemoteState, false), null);
}

@Override
void createAndFollow(Map<String, String> headers,
PutFollowAction.Request followRequest,
Runnable successHandler,
Consumer<Exception> failureHandler) {
followedIndices.add(followRequest.getLeaderIndex());
successHandler.run();
}

@Override
void updateAutoFollowMetadata(Function<ClusterState, ClusterState> updateFunction, Consumer<Exception> handler) {
lastModifiedClusterState.updateAndGet(updateFunction::apply);
handler.accept(null);
}

@Override
void cleanFollowedRemoteIndices(ClusterState remoteClusterState, List<String> patterns) {
// Ignore, to avoid invoking updateAutoFollowMetadata(...) twice
}
};
autoFollower.start();

assertThat(results, notNullValue());
assertThat(results.size(), equalTo(1));

for (ObjectObjectCursor<String, IndexMetaData> index : remoteState.metaData().indices()) {
boolean expect = index.value.getState() == IndexMetaData.State.OPEN;
assertThat(results.get(0).autoFollowExecutionResults.containsKey(index.value.getIndex()), is(expect));
assertThat(followedIndices.contains(index.key), is(expect));
}
}

private static ClusterState createRemoteClusterState(String indexName, Boolean enableSoftDeletes) {
Settings.Builder indexSettings;
if (enableSoftDeletes != null) {
Expand All @@ -1075,19 +1185,21 @@ private static ClusterState createRemoteClusterState(String indexName, Boolean e

private static ClusterState createRemoteClusterState(ClusterState previous, String indexName) {
IndexMetaData indexMetaData = IndexMetaData.builder(indexName)
.settings(settings(Version.CURRENT).put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true))
.settings(settings(Version.CURRENT)
.put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true)
.put(IndexMetaData.SETTING_INDEX_UUID, UUIDs.randomBase64UUID(random())))
.numberOfShards(1)
.numberOfReplicas(0)
.build();
ClusterState.Builder csBuilder = ClusterState.builder(new ClusterName("remote"))
ClusterState.Builder csBuilder = ClusterState.builder(previous.getClusterName())
.metaData(MetaData.builder(previous.metaData())
.version(previous.metaData().version() + 1)
.put(indexMetaData, true));

ShardRouting shardRouting =
TestShardRouting.newShardRouting(indexName, 0, "1", true, ShardRoutingState.INITIALIZING).moveToStarted();
IndexRoutingTable indexRoutingTable = IndexRoutingTable.builder(indexMetaData.getIndex()).addShard(shardRouting).build();
csBuilder.routingTable(RoutingTable.builder().add(indexRoutingTable).build()).build();
csBuilder.routingTable(RoutingTable.builder(previous.routingTable()).add(indexRoutingTable).build()).build();

return csBuilder.build();
}
Expand Down