diff --git a/docs/reference/migration/migrate_8_0.asciidoc b/docs/reference/migration/migrate_8_0.asciidoc index 6a368ae903b44..efdc0d02e1adc 100644 --- a/docs/reference/migration/migrate_8_0.asciidoc +++ b/docs/reference/migration/migrate_8_0.asciidoc @@ -90,6 +90,7 @@ include::migrate_8_0/aggregations.asciidoc[] include::migrate_8_0/allocation.asciidoc[] include::migrate_8_0/analysis.asciidoc[] include::migrate_8_0/breaker.asciidoc[] +include::migrate_8_0/ccr.asciidoc[] include::migrate_8_0/cluster.asciidoc[] include::migrate_8_0/discovery.asciidoc[] include::migrate_8_0/eql.asciidoc[] diff --git a/docs/reference/migration/migrate_8_0/ccr.asciidoc b/docs/reference/migration/migrate_8_0/ccr.asciidoc new file mode 100644 index 0000000000000..e62b5f996a4e0 --- /dev/null +++ b/docs/reference/migration/migrate_8_0/ccr.asciidoc @@ -0,0 +1,22 @@ +[discrete] +[[breaking_80_ccr_changes]] +==== Cross Cluster Replication changes + +//NOTE: The notable-breaking-changes tagged regions are re-used in the +//Installation and Upgrade Guide + +//tag::notable-breaking-changes[] + +// end::notable-breaking-changes[] + +.Remote system indices are not followed automatically if they match an auto-follow pattern. +[%collapsible] +==== +*Details* + +Remote system indices matching an <> +won't be configured as a follower index automatically. + +*Impact* + +Explicitly <> to follow a remote system +index if that's the wanted behaviour. +==== diff --git a/x-pack/plugin/ccr/src/internalClusterTest/java/org/elasticsearch/xpack/ccr/AutoFollowIT.java b/x-pack/plugin/ccr/src/internalClusterTest/java/org/elasticsearch/xpack/ccr/AutoFollowIT.java index 7bc80530167ac..18409ae61795f 100644 --- a/x-pack/plugin/ccr/src/internalClusterTest/java/org/elasticsearch/xpack/ccr/AutoFollowIT.java +++ b/x-pack/plugin/ccr/src/internalClusterTest/java/org/elasticsearch/xpack/ccr/AutoFollowIT.java @@ -21,6 +21,9 @@ import org.elasticsearch.common.unit.ByteSizeValue; import org.elasticsearch.core.TimeValue; import org.elasticsearch.index.IndexNotFoundException; +import org.elasticsearch.indices.SystemIndexDescriptor; +import org.elasticsearch.plugins.Plugin; +import org.elasticsearch.plugins.SystemIndexPlugin; import org.elasticsearch.test.ESIntegTestCase; import org.elasticsearch.xpack.CcrIntegTestCase; import org.elasticsearch.xpack.core.ccr.AutoFollowMetadata; @@ -35,14 +38,17 @@ import org.elasticsearch.xpack.core.ccr.action.PutAutoFollowPatternAction; import java.util.Arrays; +import java.util.Collection; import java.util.Collections; import java.util.List; import java.util.Locale; +import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; +import java.util.stream.Stream; import static java.util.stream.Collectors.toUnmodifiableList; import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked; @@ -60,6 +66,30 @@ protected boolean reuseClusters() { return false; } + @Override + protected Collection> nodePlugins() { + return Stream.concat(super.nodePlugins().stream(), Stream.of(FakeSystemIndex.class)).collect(Collectors.toList()); + } + + public static class FakeSystemIndex extends Plugin implements SystemIndexPlugin { + public static final String SYSTEM_INDEX_NAME = ".fake-system-index"; + + @Override + public Collection getSystemIndexDescriptors(Settings settings) { + return Collections.singletonList(new SystemIndexDescriptor(SYSTEM_INDEX_NAME, "test index")); + } + + @Override + public String getFeatureName() { + return "fake system index"; + } + + @Override + public String getFeatureDescription() { + return "fake system index"; + } + } + public void testAutoFollow() throws Exception { Settings leaderIndexSettings = Settings.builder() .put(IndexMetadata.INDEX_NUMBER_OF_SHARDS_SETTING.getKey(), 1) @@ -93,6 +123,28 @@ public void testAutoFollow() throws Exception { assertFalse(ESIntegTestCase.indexExists("copy-logs-201812", followerClient())); } + public void testAutoFollowDoNotFollowSystemIndices() throws Exception { + putAutoFollowPatterns("my-pattern", new String[] {".*", "logs-*"}); + + // Trigger system index creation + leaderClient().prepareIndex(FakeSystemIndex.SYSTEM_INDEX_NAME) + .setSource(Map.of("a", "b")) + .execute() + .actionGet(); + + Settings leaderIndexSettings = Settings.builder() + .put(IndexMetadata.INDEX_NUMBER_OF_SHARDS_SETTING.getKey(), 1) + .put(IndexMetadata.INDEX_NUMBER_OF_REPLICAS_SETTING.getKey(), 0) + .build(); + createLeaderIndex("logs-201901", leaderIndexSettings); + assertLongBusy(() -> { + AutoFollowStats autoFollowStats = getAutoFollowStats(); + assertThat(autoFollowStats.getNumberOfSuccessfulFollowIndices(), equalTo(1L)); + assertTrue(ESIntegTestCase.indexExists("copy-logs-201901", followerClient())); + assertFalse(ESIntegTestCase.indexExists("copy-.fake-system-index", followerClient())); + }); + } + public void testCleanFollowedLeaderIndexUUIDs() throws Exception { Settings leaderIndexSettings = Settings.builder() .put(IndexMetadata.INDEX_NUMBER_OF_SHARDS_SETTING.getKey(), 1) diff --git a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/AutoFollowCoordinatorTests.java b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/AutoFollowCoordinatorTests.java index 53a030db5bf9e..6470cb04bbb5b 100644 --- a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/AutoFollowCoordinatorTests.java +++ b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/AutoFollowCoordinatorTests.java @@ -7,14 +7,15 @@ 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; -import org.elasticsearch.cluster.metadata.DataStreamTestHelper; import org.elasticsearch.cluster.metadata.DataStream; +import org.elasticsearch.cluster.metadata.DataStreamTestHelper; import org.elasticsearch.cluster.metadata.IndexMetadata; import org.elasticsearch.cluster.metadata.Metadata; import org.elasticsearch.cluster.routing.IndexRoutingTable; @@ -24,13 +25,13 @@ import org.elasticsearch.cluster.routing.TestShardRouting; import org.elasticsearch.cluster.service.ClusterService; import org.elasticsearch.common.UUIDs; -import org.elasticsearch.core.Tuple; import org.elasticsearch.common.settings.ClusterSettings; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.unit.ByteSizeValue; -import org.elasticsearch.core.TimeValue; import org.elasticsearch.common.util.concurrent.ConcurrentCollections; import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; +import org.elasticsearch.core.TimeValue; +import org.elasticsearch.core.Tuple; import org.elasticsearch.index.Index; import org.elasticsearch.index.IndexSettings; import org.elasticsearch.test.ESTestCase; @@ -71,6 +72,7 @@ import static java.util.Collections.singletonMap; import static org.elasticsearch.xpack.ccr.action.AutoFollowCoordinator.AutoFollower.cleanFollowedRemoteIndices; import static org.elasticsearch.xpack.ccr.action.AutoFollowCoordinator.AutoFollower.recordLeaderIndexAsFollowFunction; +import static org.hamcrest.Matchers.anEmptyMap; import static org.hamcrest.Matchers.containsInAnyOrder; import static org.hamcrest.Matchers.empty; import static org.hamcrest.Matchers.equalTo; @@ -2150,11 +2152,133 @@ void cleanFollowedRemoteIndices(ClusterState remoteClusterState, List pa } } + public void testSystemIndicesAreNotAutoFollowed() { + ClusterState clusterState = null; + final int nbLeaderSystemIndices = randomIntBetween(1, 15); + for (int i = 0; i < nbLeaderSystemIndices; i++) { + String indexName = "." + i; + if (clusterState == null) { + clusterState = createRemoteClusterState(indexName, true, 0, true); + } else { + clusterState = createRemoteClusterState(clusterState, true, indexName); + } + } + + Tuple, Set> autoFollowResults = executeAutoFollow(".*", clusterState); + assertThat(autoFollowResults.v1().size(), equalTo(1)); + assertThat(autoFollowResults.v1().get(0).autoFollowExecutionResults, is(anEmptyMap())); + assertThat(autoFollowResults.v2(), is(empty())); + } + + public void testSystemDataStreamsAreNotAutoFollowed() { + Tuple, Set> autoFollowResults = + executeAutoFollow("*.", createRemoteClusterStateWithDataStream(".test-data-stream")); + + assertThat(autoFollowResults.v1().size(), equalTo(1)); + assertThat(autoFollowResults.v1().get(0).autoFollowExecutionResults, is(anEmptyMap())); + assertThat(autoFollowResults.v2(), is(empty())); + } + + public void testFollowerIndexIsCreatedInExecuteAutoFollow() { + final String indexName = "idx-1"; + ClusterState clusterState = createRemoteClusterState(indexName, true, 0, false); + + Tuple, Set> autoFollowResults = executeAutoFollow("idx-*", clusterState); + assertThat(autoFollowResults.v1().size(), equalTo(1)); + assertThat(autoFollowResults.v1().get(0).autoFollowExecutionResults.size(), equalTo(1)); + for (Map.Entry autoFollowEntry : autoFollowResults.v1().get(0).autoFollowExecutionResults.entrySet()) { + assertThat(autoFollowEntry.getKey().getName(), equalTo(indexName)); + assertThat(autoFollowEntry.getValue(), nullValue()); + } + assertThat(autoFollowResults.v2().contains(indexName), equalTo(true)); + } + + private Tuple, Set> executeAutoFollow(String indexPattern, + ClusterState finalRemoteState) { + 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( + Map.of( + pattern, + new AutoFollowPattern( + "remote", + List.of(indexPattern), + Collections.emptyList(), + null, + Settings.EMPTY, + true, + null, + null, + null, + null, + null, + null, + null, + null, + null, + null + ) + ), + Map.of(pattern, List.of()), + Map.of(pattern, Map.of())))) + .build(); + + final AtomicReference lastModifiedClusterState = new AtomicReference<>(localState); + final List results = new ArrayList<>(); + final Set followedIndices = ConcurrentCollections.newConcurrentSet(); + final AutoFollower autoFollower = + new AutoFollower("remote", results::addAll, localClusterStateSupplier(localState), () -> 1L, Runnable::run) { + @Override + void getRemoteClusterState(String remoteCluster, + long metadataVersion, + BiConsumer handler) { + assertThat(remoteCluster, equalTo("remote")); + handler.accept(new ClusterStateResponse(new ClusterName("remote"), finalRemoteState, false), null); + } + + @Override + void createAndFollow(Map headers, + PutFollowAction.Request followRequest, + Runnable successHandler, + Consumer failureHandler) { + followedIndices.add(followRequest.getLeaderIndex()); + successHandler.run(); + } + + @Override + void updateAutoFollowMetadata(Function updateFunction, Consumer handler) { + lastModifiedClusterState.updateAndGet(updateFunction::apply); + handler.accept(null); + } + + @Override + void cleanFollowedRemoteIndices(ClusterState remoteClusterState, List patterns) { + // Ignore, to avoid invoking updateAutoFollowMetadata(...) twice + } + }; + autoFollower.start(); + + assertThat(results, notNullValue()); + return Tuple.tuple(results, followedIndices); + } + private static ClusterState createRemoteClusterState(String indexName, boolean enableSoftDeletes) { return createRemoteClusterState(indexName, enableSoftDeletes, 0L); } private static ClusterState createRemoteClusterState(String indexName, boolean enableSoftDeletes, long metadataVersion) { + return createRemoteClusterState(indexName, enableSoftDeletes, metadataVersion, false); + } + + private static ClusterState createRemoteClusterState(String indexName, + boolean enableSoftDeletes, + long metadataVersion, + boolean systemIndex) { Settings.Builder indexSettings; if (enableSoftDeletes == false) { indexSettings = settings(VersionUtils.randomPreviousCompatibleVersion(random(), Version.V_8_0_0)) @@ -2168,6 +2292,7 @@ private static ClusterState createRemoteClusterState(String indexName, boolean e .settings(indexSettings) .numberOfShards(1) .numberOfReplicas(0) + .system(systemIndex) .build(); ClusterState.Builder csBuilder = ClusterState.builder(new ClusterName("remote")) .metadata(Metadata.builder() @@ -2180,7 +2305,12 @@ private static ClusterState createRemoteClusterState(String indexName, boolean e return csBuilder.routingTable(RoutingTable.builder().add(indexRoutingTable).build()).build(); } + private static ClusterState createRemoteClusterState(final ClusterState previous, final String... indices) { + return createRemoteClusterState(previous, false, indices); + } + + private static ClusterState createRemoteClusterState(final ClusterState previous, boolean systemIndices, final String... indices) { if (indices == null) { return previous; } @@ -2192,6 +2322,7 @@ private static ClusterState createRemoteClusterState(final ClusterState previous .put(IndexMetadata.SETTING_INDEX_UUID, UUIDs.randomBase64UUID(random()))) .numberOfShards(1) .numberOfReplicas(0) + .system(systemIndices) .build(); metadataBuilder.put(indexMetadata, true); routingTableBuilder.add(IndexRoutingTable.builder(indexMetadata.getIndex()) @@ -2230,6 +2361,10 @@ private ClusterService mockClusterService() { } private static ClusterState createRemoteClusterStateWithDataStream(String dataStreamName) { + return createRemoteClusterStateWithDataStream(dataStreamName, false); + } + + private static ClusterState createRemoteClusterStateWithDataStream(String dataStreamName, boolean system) { Settings.Builder indexSettings = settings(Version.CURRENT); indexSettings.put(IndexMetadata.SETTING_INDEX_UUID, UUIDs.randomBase64UUID(random())); indexSettings.put("index.hidden", true); @@ -2238,9 +2373,10 @@ private static ClusterState createRemoteClusterStateWithDataStream(String dataSt .settings(indexSettings) .numberOfShards(1) .numberOfReplicas(0) + .system(system) .build(); DataStream dataStream = new DataStream(dataStreamName, new DataStream.TimestampField("@timestamp"), - List.of(indexMetadata.getIndex())); + List.of(indexMetadata.getIndex()), 1, null, false, false, system); ClusterState.Builder csBuilder = ClusterState.builder(new ClusterName("remote")) .metadata(Metadata.builder() .put(indexMetadata, true) diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ccr/AutoFollowMetadata.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ccr/AutoFollowMetadata.java index 05d935a237c48..912c145854748 100644 --- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ccr/AutoFollowMetadata.java +++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ccr/AutoFollowMetadata.java @@ -298,12 +298,16 @@ public boolean match(IndexAbstraction indexAbstraction) { public static boolean match(List leaderIndexPatterns, List leaderIndexExclusionPatterns, IndexAbstraction indexAbstraction) { - boolean matches = Regex.simpleMatch(leaderIndexExclusionPatterns, indexAbstraction.getName()) == false && - Regex.simpleMatch(leaderIndexPatterns, indexAbstraction.getName()); + boolean matches = indexAbstraction.isSystem() == false && + Regex.simpleMatch(leaderIndexExclusionPatterns, indexAbstraction.getName()) == false && + Regex.simpleMatch(leaderIndexPatterns, indexAbstraction.getName()); + if (matches) { return true; } else { - return indexAbstraction.getParentDataStream() != null && + final IndexAbstraction.DataStream parentDataStream = indexAbstraction.getParentDataStream(); + return parentDataStream != null && + parentDataStream.isSystem() == false && Regex.simpleMatch(leaderIndexExclusionPatterns, indexAbstraction.getParentDataStream().getName()) == false && Regex.simpleMatch(leaderIndexPatterns, indexAbstraction.getParentDataStream().getName()); }