Skip to content

Commit 6d94636

Browse files
martijnvgmartijnvg
committed
[CCR] AutoFollowCoordinator and follower index already created (#36540)
The AutoFollowCoordinator should be resilient to the fact that the follower index has already been created and in that case it should only update the auto follow metadata with the fact that the follower index was created. Relates to #33007
1 parent 1d66c4c commit 6d94636

File tree

3 files changed

+117
-15
lines changed

3 files changed

+117
-15
lines changed

‎x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/CcrLicenseChecker.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -122,7 +122,7 @@ public void checkRemoteClusterLicenseAndFetchLeaderIndexMetadataAndHistoryUUIDs(
122122
client.getRemoteClusterClient(clusterAlias),
123123
request,
124124
onFailure,
125-
remoteClusterStateResponse -> {
125+
remoteClusterStateResponse -> {
126126
ClusterState remoteClusterState = remoteClusterStateResponse.getState();
127127
IndexMetaData leaderIndexMetaData = remoteClusterState.getMetaData().index(leaderIndex);
128128
if (leaderIndexMetaData == null) {

‎x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/AutoFollowCoordinator.java‎

Lines changed: 34 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
import org.elasticsearch.index.Index;
3333
import org.elasticsearch.index.IndexSettings;
3434
import org.elasticsearch.license.LicenseUtils;
35+
import org.elasticsearch.xpack.ccr.Ccr;
3536
import org.elasticsearch.xpack.ccr.CcrLicenseChecker;
3637
import org.elasticsearch.xpack.ccr.CcrSettings;
3738
import org.elasticsearch.xpack.core.ccr.AutoFollowMetadata;
@@ -364,8 +365,8 @@ private void autoFollowIndices(final AutoFollowMetadata autoFollowMetadata,
364365
Map<String, String> headers = autoFollowMetadata.getHeaders().get(autoFollowPatternName);
365366
List<String> followedIndices = autoFollowMetadata.getFollowedLeaderIndexUUIDs().get(autoFollowPatternName);
366367

367-
final List<Index> leaderIndicesToFollow = getLeaderIndicesToFollow(autoFollowPattern, remoteClusterState,
368-
clusterState, followedIndices);
368+
final List<Index> leaderIndicesToFollow =
369+
getLeaderIndicesToFollow(autoFollowPattern, remoteClusterState, followedIndices);
369370
if (leaderIndicesToFollow.isEmpty()) {
370371
finalise(slot, new AutoFollowResult(autoFollowPatternName));
371372
} else {
@@ -378,7 +379,7 @@ private void autoFollowIndices(final AutoFollowMetadata autoFollowMetadata,
378379

379380
Consumer<AutoFollowResult> resultHandler = result -> finalise(slot, result);
380381
checkAutoFollowPattern(autoFollowPatternName, remoteCluster, autoFollowPattern, leaderIndicesToFollow, headers,
381-
patternsForTheSameRemoteCluster, remoteClusterState.metaData(), resultHandler);
382+
patternsForTheSameRemoteCluster, remoteClusterState.metaData(), clusterState.metaData(), resultHandler);
382383
}
383384
i++;
384385
}
@@ -392,6 +393,7 @@ private void checkAutoFollowPattern(String autoFollowPattenName,
392393
Map<String, String> headers,
393394
List<Tuple<String, AutoFollowPattern>> patternsForTheSameRemoteCluster,
394395
MetaData remoteMetadata,
396+
MetaData localMetadata,
395397
Consumer<AutoFollowResult> resultHandler) {
396398

397399
final CountDown leaderIndicesCountDown = new CountDown(leaderIndicesToFollow.size());
@@ -427,7 +429,16 @@ private void checkAutoFollowPattern(String autoFollowPattenName,
427429
}
428430
});
429431
continue;
432+
} else if (leaderIndexAlreadyFollowed(autoFollowPattern, indexToFollow, localMetadata)) {
433+
updateAutoFollowMetadata(recordLeaderIndexAsFollowFunction(autoFollowPattenName, indexToFollow), error -> {
434+
results.set(slot, new Tuple<>(indexToFollow, error));
435+
if (leaderIndicesCountDown.countDown()) {
436+
resultHandler.accept(new AutoFollowResult(autoFollowPattenName, results.asList()));
437+
}
438+
});
439+
continue;
430440
}
441+
431442
followLeaderIndex(autoFollowPattenName, remoteCluster, indexToFollow, autoFollowPattern, headers, error -> {
432443
results.set(slot, new Tuple<>(indexToFollow, error));
433444
if (leaderIndicesCountDown.countDown()) {
@@ -438,6 +449,25 @@ private void checkAutoFollowPattern(String autoFollowPattenName,
438449
}
439450
}
440451

452+
private static boolean leaderIndexAlreadyFollowed(AutoFollowPattern autoFollowPattern,
453+
Index leaderIndex,
454+
MetaData localMetadata) {
455+
String followIndexName = getFollowerIndexName(autoFollowPattern, leaderIndex.getName());
456+
IndexMetaData indexMetaData = localMetadata.index(followIndexName);
457+
if (indexMetaData != null) {
458+
// If an index with the same name exists, but it is not a follow index for this leader index then
459+
// we should let the auto follower attempt to auto follow it, so it can fail later and
460+
// it is then visible in the auto follow stats. For example a cluster can just happen to have
461+
// an index with the same name as the new follower index.
462+
Map<String, String> customData = indexMetaData.getCustomData(Ccr.CCR_CUSTOM_METADATA_KEY);
463+
if (customData != null) {
464+
String recordedLeaderIndexUUID = customData.get(Ccr.CCR_CUSTOM_METADATA_LEADER_INDEX_UUID_KEY);
465+
return leaderIndex.getUUID().equals(recordedLeaderIndexUUID);
466+
}
467+
}
468+
return false;
469+
}
470+
441471
private void followLeaderIndex(String autoFollowPattenName,
442472
String remoteCluster,
443473
Index indexToFollow,
@@ -489,7 +519,6 @@ private void finalise(int slot, AutoFollowResult result) {
489519

490520
static List<Index> getLeaderIndicesToFollow(AutoFollowPattern autoFollowPattern,
491521
ClusterState remoteClusterState,
492-
ClusterState followerClusterState,
493522
List<String> followedIndexUUIDs) {
494523
List<Index> leaderIndicesToFollow = new ArrayList<>();
495524
for (IndexMetaData leaderIndexMetaData : remoteClusterState.getMetaData()) {
@@ -502,10 +531,7 @@ static List<Index> getLeaderIndicesToFollow(AutoFollowPattern autoFollowPattern,
502531
// this index will be auto followed.
503532
indexRoutingTable.allPrimaryShardsActive() &&
504533
followedIndexUUIDs.contains(leaderIndexMetaData.getIndex().getUUID()) == false) {
505-
// TODO: iterate over the indices in the followerClusterState and check whether a IndexMetaData
506-
// has a leader index uuid custom metadata entry that matches with uuid of leaderIndexMetaData variable
507-
// If so then handle it differently: not follow it, but just add an entry to
508-
// AutoFollowMetadata#followedLeaderIndexUUIDs
534+
509535
leaderIndicesToFollow.add(leaderIndexMetaData.getIndex());
510536
}
511537
}

‎x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/AutoFollowCoordinatorTests.java‎

Lines changed: 82 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.elasticsearch.index.Index;
2525
import org.elasticsearch.index.IndexSettings;
2626
import org.elasticsearch.test.ESTestCase;
27+
import org.elasticsearch.xpack.ccr.Ccr;
2728
import org.elasticsearch.xpack.ccr.CcrLicenseChecker;
2829
import org.elasticsearch.xpack.ccr.CcrSettings;
2930
import org.elasticsearch.xpack.ccr.action.AutoFollowCoordinator.AutoFollower;
@@ -345,8 +346,7 @@ public void testGetLeaderIndicesToFollow() {
345346
.routingTable(routingTableBuilder.build())
346347
.build();
347348

348-
List<Index> result = AutoFollower.getLeaderIndicesToFollow(autoFollowPattern, remoteState, clusterState,
349-
Collections.emptyList());
349+
List<Index> result = AutoFollower.getLeaderIndicesToFollow(autoFollowPattern, remoteState, Collections.emptyList());
350350
result.sort(Comparator.comparing(Index::getName));
351351
assertThat(result.size(), equalTo(5));
352352
assertThat(result.get(0).getName(), equalTo("metrics-0"));
@@ -356,7 +356,7 @@ public void testGetLeaderIndicesToFollow() {
356356
assertThat(result.get(4).getName(), equalTo("metrics-4"));
357357

358358
List<String> followedIndexUUIDs = Collections.singletonList(remoteState.metaData().index("metrics-2").getIndexUUID());
359-
result = AutoFollower.getLeaderIndicesToFollow(autoFollowPattern, remoteState, clusterState, followedIndexUUIDs);
359+
result = AutoFollower.getLeaderIndicesToFollow(autoFollowPattern, remoteState, followedIndexUUIDs);
360360
result.sort(Comparator.comparing(Index::getName));
361361
assertThat(result.size(), equalTo(4));
362362
assertThat(result.get(0).getName(), equalTo("metrics-0"));
@@ -390,8 +390,7 @@ public void testGetLeaderIndicesToFollow_shardsNotStarted() {
390390
.routingTable(RoutingTable.builder(remoteState.routingTable()).add(indexRoutingTable).build())
391391
.build();
392392

393-
List<Index> result = AutoFollower.getLeaderIndicesToFollow(autoFollowPattern, remoteState, clusterState,
394-
Collections.emptyList());
393+
List<Index> result = AutoFollower.getLeaderIndicesToFollow(autoFollowPattern, remoteState, Collections.emptyList());
395394
assertThat(result.size(), equalTo(1));
396395
assertThat(result.get(0).getName(), equalTo("index1"));
397396

@@ -404,7 +403,7 @@ public void testGetLeaderIndicesToFollow_shardsNotStarted() {
404403
.routingTable(RoutingTable.builder(remoteState.routingTable()).add(indexRoutingTable).build())
405404
.build();
406405

407-
result = AutoFollower.getLeaderIndicesToFollow(autoFollowPattern, remoteState, clusterState, Collections.emptyList());
406+
result = AutoFollower.getLeaderIndicesToFollow(autoFollowPattern, remoteState, Collections.emptyList());
408407
assertThat(result.size(), equalTo(2));
409408
result.sort(Comparator.comparing(Index::getName));
410409
assertThat(result.get(0).getName(), equalTo("index1"));
@@ -864,6 +863,83 @@ void cleanFollowedRemoteIndices(ClusterState remoteClusterState, List<String> pa
864863
"because soft deletes are not enabled"));
865864
}
866865

866+
public void testAutoFollowerFollowerIndexAlreadyExists() {
867+
Client client = mock(Client.class);
868+
when(client.getRemoteClusterClient(anyString())).thenReturn(client);
869+
870+
ClusterState remoteState = createRemoteClusterState("logs-20190101", true);
871+
872+
AutoFollowPattern autoFollowPattern = new AutoFollowPattern("remote", Collections.singletonList("logs-*"),
873+
null, null, null, null, null, null, null, null, null, null, null);
874+
Map<String, AutoFollowPattern> patterns = new HashMap<>();
875+
patterns.put("remote", autoFollowPattern);
876+
Map<String, List<String>> followedLeaderIndexUUIDS = new HashMap<>();
877+
followedLeaderIndexUUIDS.put("remote", new ArrayList<>());
878+
Map<String, Map<String, String>> autoFollowHeaders = new HashMap<>();
879+
autoFollowHeaders.put("remote", Collections.singletonMap("key", "val"));
880+
AutoFollowMetadata autoFollowMetadata = new AutoFollowMetadata(patterns, followedLeaderIndexUUIDS, autoFollowHeaders);
881+
882+
ClusterState currentState = ClusterState.builder(new ClusterName("name"))
883+
.metaData(MetaData.builder()
884+
.put(IndexMetaData.builder("logs-20190101")
885+
.settings(settings(Version.CURRENT).put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true))
886+
.putCustom(Ccr.CCR_CUSTOM_METADATA_KEY, Collections.singletonMap(Ccr.CCR_CUSTOM_METADATA_LEADER_INDEX_UUID_KEY,
887+
remoteState.metaData().index("logs-20190101").getIndexUUID()))
888+
.numberOfShards(1)
889+
.numberOfReplicas(0))
890+
.putCustom(AutoFollowMetadata.TYPE, autoFollowMetadata))
891+
.build();
892+
893+
894+
final Object[] resultHolder = new Object[1];
895+
Consumer<List<AutoFollowCoordinator.AutoFollowResult>> handler = results -> {
896+
resultHolder[0] = results;
897+
};
898+
AutoFollower autoFollower = new AutoFollower("remote", handler, localClusterStateSupplier(currentState), () -> 1L) {
899+
@Override
900+
void getRemoteClusterState(String remoteCluster,
901+
long metadataVersion,
902+
BiConsumer<ClusterStateResponse, Exception> handler) {
903+
assertThat(remoteCluster, equalTo("remote"));
904+
handler.accept(new ClusterStateResponse(new ClusterName("name"), remoteState, 1L, false), null);
905+
}
906+
907+
@Override
908+
void createAndFollow(Map<String, String> headers,
909+
PutFollowAction.Request followRequest,
910+
Runnable successHandler,
911+
Consumer<Exception> failureHandler) {
912+
fail("this should not be invoked");
913+
}
914+
915+
@Override
916+
void updateAutoFollowMetadata(Function<ClusterState, ClusterState> updateFunction,
917+
Consumer<Exception> handler) {
918+
ClusterState resultCs = updateFunction.apply(currentState);
919+
AutoFollowMetadata result = resultCs.metaData().custom(AutoFollowMetadata.TYPE);
920+
assertThat(result.getFollowedLeaderIndexUUIDs().size(), equalTo(1));
921+
assertThat(result.getFollowedLeaderIndexUUIDs().get("remote").size(), equalTo(1));
922+
handler.accept(null);
923+
}
924+
925+
@Override
926+
void cleanFollowedRemoteIndices(ClusterState remoteClusterState, List<String> patterns) {
927+
// Ignore, to avoid invoking updateAutoFollowMetadata(...) twice
928+
}
929+
};
930+
autoFollower.start();
931+
932+
@SuppressWarnings("unchecked")
933+
List<AutoFollowCoordinator.AutoFollowResult> results = (List<AutoFollowCoordinator.AutoFollowResult>) resultHolder[0];
934+
assertThat(results, notNullValue());
935+
assertThat(results.size(), equalTo(1));
936+
assertThat(results.get(0).clusterStateFetchException, nullValue());
937+
List<Map.Entry<Index, Exception>> entries = new ArrayList<>(results.get(0).autoFollowExecutionResults.entrySet());
938+
assertThat(entries.size(), equalTo(1));
939+
assertThat(entries.get(0).getKey().getName(), equalTo("logs-20190101"));
940+
assertThat(entries.get(0).getValue(), nullValue());
941+
}
942+
867943
private static ClusterState createRemoteClusterState(String indexName, Boolean enableSoftDeletes) {
868944
Settings.Builder indexSettings;
869945
if (enableSoftDeletes != null) {

0 commit comments

Comments
 (0)