[CCR] Add metadata to keep track of the index uuid of the leader index in the follow index (#33367)

The follow index api checks if the recorded uuid in the follow index matches
with uuid of the leader index and fails otherwise. This validation will
prevent a follow index from following an incompatible leader index.

The create_and_follow api will automatically add this custom index metadata
when it creates the follow index.

Closes #31505

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

@@ -87,6 +87,7 @@ public class Ccr extends Plugin implements ActionPlugin, PersistentTaskPlugin, E
     public static final String CCR_THREAD_POOL_NAME = "ccr";
     public static final String CCR_CUSTOM_METADATA_KEY = "ccr";
     public static final String CCR_CUSTOM_METADATA_LEADER_INDEX_SHARD_HISTORY_UUIDS = "leader_index_shard_history_uuids";
+    public static final String CCR_CUSTOM_METADATA_LEADER_INDEX_UUID_KEY = "leader_index_uuid";
 
     private final boolean enabled;
     private final Settings settings;

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

@@ -183,6 +183,7 @@ public final class TransportCreateAndFollowIndexAction
                 // Adding the leader index uuid for each shard as custom metadata:
                 Map<String, String> metadata = new HashMap<>();
                 metadata.put(Ccr.CCR_CUSTOM_METADATA_LEADER_INDEX_SHARD_HISTORY_UUIDS, String.join(",", historyUUIDs));
+                metadata.put(Ccr.CCR_CUSTOM_METADATA_LEADER_INDEX_UUID_KEY, leaderIndexMetaData.getIndexUUID());
                 imdBuilder.putCustom(Ccr.CCR_CUSTOM_METADATA_KEY, metadata);
 
                 // Copy all settings, but overwrite a few settings.

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

@@ -245,6 +245,14 @@ public class TransportFollowIndexAction extends HandledTransportAction<FollowInd
         if (followIndex == null) {
             throw new IllegalArgumentException("follow index [" + request.getFollowerIndex() + "] does not exist");
         }
+        String leaderIndexUUID = leaderIndex.getIndex().getUUID();
+        String recordedLeaderIndexUUID = followIndex
+                .getCustomData(Ccr.CCR_CUSTOM_METADATA_KEY)
+                .get(Ccr.CCR_CUSTOM_METADATA_LEADER_INDEX_UUID_KEY);
+        if (leaderIndexUUID.equals(recordedLeaderIndexUUID) == false) {
+            throw new IllegalArgumentException("follow index [" + request.getFollowerIndex() + "] should reference [" + leaderIndexUUID +
+                    "] as leader index but instead reference [" + recordedLeaderIndexUUID + "] as leader index");
+        }
 
         String[] recordedHistoryUUIDs = extractIndexShardHistoryUUIDs(followIndex);
         assert recordedHistoryUUIDs.length == leaderIndexHistoryUUID.length;

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/ShardChangesIT.java

@@ -39,14 +39,13 @@ import org.elasticsearch.tasks.TaskInfo;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.MockHttpTransport;
 import org.elasticsearch.test.discovery.TestZenDiscovery;
-import org.elasticsearch.test.junit.annotations.TestLogging;
-import org.elasticsearch.xpack.core.ccr.action.CreateAndFollowIndexAction;
-import org.elasticsearch.xpack.core.ccr.action.FollowIndexAction;
 import org.elasticsearch.xpack.ccr.action.ShardChangesAction;
 import org.elasticsearch.xpack.ccr.action.ShardFollowTask;
-import org.elasticsearch.xpack.core.ccr.action.UnfollowIndexAction;
 import org.elasticsearch.xpack.core.XPackSettings;
 import org.elasticsearch.xpack.core.ccr.ShardFollowNodeTaskStatus;
+import org.elasticsearch.xpack.core.ccr.action.CreateAndFollowIndexAction;
+import org.elasticsearch.xpack.core.ccr.action.FollowIndexAction;
+import org.elasticsearch.xpack.core.ccr.action.UnfollowIndexAction;
 
 import java.io.IOException;
 import java.util.Arrays;
@@ -314,10 +313,7 @@ public class ShardChangesIT extends ESIntegTestCase {
         internalCluster().ensureAtLeastNumDataNodes(3);
         String leaderIndexSettings = getIndexSettings(3, 1, singletonMap(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true"));
         assertAcked(client().admin().indices().prepareCreate("index1").setSource(leaderIndexSettings, XContentType.JSON));
-
+        ensureGreen("index1");
-        String followerIndexSettings = getIndexSettings(3, 1, singletonMap(CcrSettings.CCR_FOLLOWING_INDEX_SETTING.getKey(), "true"));
-        assertAcked(client().admin().indices().prepareCreate("index2").setSource(followerIndexSettings, XContentType.JSON));
-        ensureGreen("index1", "index2");
 
         AtomicBoolean run = new AtomicBoolean(true);
         Thread thread = new Thread(() -> {
@@ -339,7 +335,7 @@ public class ShardChangesIT extends ESIntegTestCase {
         final FollowIndexAction.Request followRequest = new FollowIndexAction.Request("index1", "index2", randomIntBetween(32, 2048),
             randomIntBetween(2, 10), Long.MAX_VALUE, randomIntBetween(2, 10),
             FollowIndexAction.DEFAULT_MAX_WRITE_BUFFER_SIZE, TimeValue.timeValueMillis(500), TimeValue.timeValueMillis(10));
-        client().execute(FollowIndexAction.INSTANCE, followRequest).get();
+        client().execute(CreateAndFollowIndexAction.INSTANCE, new CreateAndFollowIndexAction.Request(followRequest)).get();
 
         long maxNumDocsReplicated = Math.min(1000, randomLongBetween(followRequest.getMaxBatchOperationCount(),
             followRequest.getMaxBatchOperationCount() * 10));
@@ -416,34 +412,6 @@ public class ShardChangesIT extends ESIntegTestCase {
         expectThrows(IndexNotFoundException.class, () -> client().execute(FollowIndexAction.INSTANCE, followRequest3).actionGet());
     }
 
-    @TestLogging("_root:DEBUG")
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/33379")
-    public void testValidateFollowingIndexSettings() throws Exception {
-        assertAcked(client().admin().indices().prepareCreate("test-leader")
-            .setSettings(Settings.builder().put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true)));
-        // TODO: indexing should be optional but the current mapping logic requires for now.
-        client().prepareIndex("test-leader", "doc", "id").setSource("{\"f\": \"v\"}", XContentType.JSON).get();
-        assertAcked(client().admin().indices().prepareCreate("test-follower").get());
-        IllegalArgumentException followError = expectThrows(IllegalArgumentException.class, () -> client().execute(
-            FollowIndexAction.INSTANCE, createFollowRequest("test-leader", "test-follower")).actionGet());
-        assertThat(followError.getMessage(), equalTo("the following index [test-follower] is not ready to follow;" +
-            " the setting [index.xpack.ccr.following_index] must be enabled."));
-        // updating the `following_index` with an open index must not be allowed.
-        IllegalArgumentException updateError = expectThrows(IllegalArgumentException.class, () -> {
-            client().admin().indices().prepareUpdateSettings("test-follower")
-                .setSettings(Settings.builder().put(CcrSettings.CCR_FOLLOWING_INDEX_SETTING.getKey(), true)).get();
-        });
-        assertThat(updateError.getMessage(), containsString("Can't update non dynamic settings " +
-            "[[index.xpack.ccr.following_index]] for open indices [[test-follower/"));
-        assertAcked(client().admin().indices().prepareClose("test-follower"));
-        assertAcked(client().admin().indices().prepareUpdateSettings("test-follower")
-            .setSettings(Settings.builder().put(CcrSettings.CCR_FOLLOWING_INDEX_SETTING.getKey(), true)));
-        assertAcked(client().admin().indices().prepareOpen("test-follower"));
-        assertAcked(client().execute(FollowIndexAction.INSTANCE,
-            createFollowRequest("test-leader", "test-follower")).actionGet());
-        unfollowIndex("test-follower");
-    }
-
     public void testFollowIndex_lowMaxTranslogBytes() throws Exception {
         final String leaderIndexSettings = getIndexSettings(1, between(0, 1),
             singletonMap(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true"));
@@ -478,6 +446,37 @@ public class ShardChangesIT extends ESIntegTestCase {
         unfollowIndex("index2");
     }
 
+    public void testDontFollowTheWrongIndex() throws Exception {
+        String leaderIndexSettings = getIndexSettings(1, 0,
+            Collections.singletonMap(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true"));
+        assertAcked(client().admin().indices().prepareCreate("index1").setSource(leaderIndexSettings, XContentType.JSON));
+        ensureGreen("index1");
+        assertAcked(client().admin().indices().prepareCreate("index3").setSource(leaderIndexSettings, XContentType.JSON));
+        ensureGreen("index3");
+
+        FollowIndexAction.Request followRequest = new FollowIndexAction.Request("index1", "index2", 1024, 1, 1024L,
+            1, 10240, TimeValue.timeValueMillis(500), TimeValue.timeValueMillis(10));
+        CreateAndFollowIndexAction.Request createAndFollowRequest = new CreateAndFollowIndexAction.Request(followRequest);
+        client().execute(CreateAndFollowIndexAction.INSTANCE, createAndFollowRequest).get();
+
+        followRequest = new FollowIndexAction.Request("index3", "index4", 1024, 1, 1024L,
+            1, 10240, TimeValue.timeValueMillis(500), TimeValue.timeValueMillis(10));
+        createAndFollowRequest = new CreateAndFollowIndexAction.Request(followRequest);
+        client().execute(CreateAndFollowIndexAction.INSTANCE, createAndFollowRequest).get();
+        unfollowIndex("index2", "index4");
+
+        FollowIndexAction.Request wrongRequest1 = new FollowIndexAction.Request("index1", "index4", 1024, 1, 1024L,
+            1, 10240, TimeValue.timeValueMillis(500), TimeValue.timeValueMillis(10));
+        Exception e = expectThrows(IllegalArgumentException.class,
+            () -> client().execute(FollowIndexAction.INSTANCE, wrongRequest1).actionGet());
+        assertThat(e.getMessage(), containsString("follow index [index4] should reference"));
+
+        FollowIndexAction.Request wrongRequest2 = new FollowIndexAction.Request("index3", "index2", 1024, 1, 1024L,
+            1, 10240, TimeValue.timeValueMillis(500), TimeValue.timeValueMillis(10));
+        e = expectThrows(IllegalArgumentException.class, () -> client().execute(FollowIndexAction.INSTANCE, wrongRequest2).actionGet());
+        assertThat(e.getMessage(), containsString("follow index [index2] should reference"));
+    }
+
     private CheckedRunnable<Exception> assertTask(final int numberOfPrimaryShards, final Map<ShardId, Long> numDocsPerShard) {
         return () -> {
             final ClusterState clusterState = client().admin().cluster().prepareState().get().getState();
@@ -514,10 +513,12 @@ public class ShardChangesIT extends ESIntegTestCase {
         };
     }
 
-    private void unfollowIndex(String index) throws Exception {
+    private void unfollowIndex(String... indices) throws Exception {
-        final UnfollowIndexAction.Request unfollowRequest = new UnfollowIndexAction.Request();
+        for (String index : indices) {
-        unfollowRequest.setFollowIndex(index);
+            final UnfollowIndexAction.Request unfollowRequest = new UnfollowIndexAction.Request();
-        client().execute(UnfollowIndexAction.INSTANCE, unfollowRequest).get();
+            unfollowRequest.setFollowIndex(index);
+            client().execute(UnfollowIndexAction.INSTANCE, unfollowRequest).get();
+        }
         assertBusy(() -> {
             final ClusterState clusterState = client().admin().cluster().prepareState().get().getState();
             final PersistentTasksCustomMetaData tasks = clusterState.getMetaData().custom(PersistentTasksCustomMetaData.TYPE);

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

@@ -20,6 +20,7 @@ import org.elasticsearch.xpack.ccr.ShardChangesIT;
 import org.elasticsearch.xpack.core.ccr.action.FollowIndexAction;
 
 import java.io.IOException;
+import java.util.HashMap;
 import java.util.Map;
 
 import static java.util.Collections.emptyMap;
@@ -29,10 +30,11 @@ import static org.hamcrest.Matchers.equalTo;
 
 public class TransportFollowIndexActionTests extends ESTestCase {
 
-    private static final Map<String, String> CUSTOM_METADATA =
-        singletonMap(Ccr.CCR_CUSTOM_METADATA_LEADER_INDEX_SHARD_HISTORY_UUIDS, "uuid");
-
     public void testValidation() throws IOException {
+        final Map<String, String> customMetaData = new HashMap<>();
+        customMetaData.put(Ccr.CCR_CUSTOM_METADATA_LEADER_INDEX_SHARD_HISTORY_UUIDS, "uuid");
+        customMetaData.put(Ccr.CCR_CUSTOM_METADATA_LEADER_INDEX_UUID_KEY, "_na_");
+
         FollowIndexAction.Request request = ShardChangesIT.createFollowRequest("index1", "index2");
         String[] UUIDs = new String[]{"uuid"};
         {
@@ -47,12 +49,23 @@ public class TransportFollowIndexActionTests extends ESTestCase {
                 () -> validate(request, leaderIMD, null, null, null));
             assertThat(e.getMessage(), equalTo("follow index [index2] does not exist"));
         }
+        {
+            // should fail because the recorded leader index uuid is not equal to the leader actual index
+            IndexMetaData leaderIMD = createIMD("index1", 5, Settings.EMPTY, customMetaData);
+            IndexMetaData followIMD = createIMD("index2", 5, Settings.EMPTY,
+                singletonMap(Ccr.CCR_CUSTOM_METADATA_LEADER_INDEX_UUID_KEY, "another-value"));
+            Exception e = expectThrows(IllegalArgumentException.class,
+                () -> validate(request, leaderIMD, followIMD, UUIDs, null));
+            assertThat(e.getMessage(), equalTo("follow index [index2] should reference [_na_] as leader index but " +
+                "instead reference [another-value] as leader index"));
+        }
         {
             // should fail because the recorded leader index history uuid is not equal to the leader actual index history uuid:
             IndexMetaData leaderIMD = createIMD("index1", 5, Settings.EMPTY, emptyMap());
-            Map<String, String> customMetaData =
+            Map<String, String> anotherCustomMetaData = new HashMap<>();
-                singletonMap(Ccr.CCR_CUSTOM_METADATA_LEADER_INDEX_SHARD_HISTORY_UUIDS, "another-uuid");
+            anotherCustomMetaData.put(Ccr.CCR_CUSTOM_METADATA_LEADER_INDEX_UUID_KEY, "_na_");
-            IndexMetaData followIMD = createIMD("index2", 5, Settings.EMPTY, customMetaData);
+            anotherCustomMetaData.put(Ccr.CCR_CUSTOM_METADATA_LEADER_INDEX_SHARD_HISTORY_UUIDS, "another-uuid");
+            IndexMetaData followIMD = createIMD("index2", 5, Settings.EMPTY, anotherCustomMetaData);
             Exception e = expectThrows(IllegalArgumentException.class,
                 () -> validate(request, leaderIMD, followIMD, UUIDs, null));
             assertThat(e.getMessage(), equalTo("leader shard [index2][0] should reference [another-uuid] as history uuid but " +
@@ -61,7 +74,7 @@ public class TransportFollowIndexActionTests extends ESTestCase {
         {
             // should fail because leader index does not have soft deletes enabled
             IndexMetaData leaderIMD = createIMD("index1", 5, Settings.EMPTY, emptyMap());
-            IndexMetaData followIMD = createIMD("index2", 5, Settings.EMPTY, CUSTOM_METADATA);
+            IndexMetaData followIMD = createIMD("index2", 5, Settings.EMPTY, customMetaData);
             Exception e = expectThrows(IllegalArgumentException.class, () -> validate(request, leaderIMD, followIMD, UUIDs, null));
             assertThat(e.getMessage(), equalTo("leader index [index1] does not have soft deletes enabled"));
         }
@@ -69,7 +82,7 @@ public class TransportFollowIndexActionTests extends ESTestCase {
             // should fail because the number of primary shards between leader and follow index are not equal
             IndexMetaData leaderIMD = createIMD("index1", 5, Settings.builder()
                 .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true").build(), emptyMap());
-            IndexMetaData followIMD = createIMD("index2", 4, Settings.EMPTY, CUSTOM_METADATA);
+            IndexMetaData followIMD = createIMD("index2", 4, Settings.EMPTY, customMetaData);
             Exception e = expectThrows(IllegalArgumentException.class, () -> validate(request, leaderIMD, followIMD, UUIDs, null));
             assertThat(e.getMessage(),
                 equalTo("leader index primary shards [5] does not match with the number of shards of the follow index [4]"));
@@ -79,16 +92,28 @@ public class TransportFollowIndexActionTests extends ESTestCase {
             IndexMetaData leaderIMD = createIMD("index1", State.CLOSE, "{}", 5, Settings.builder()
                 .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true").build(), emptyMap());
             IndexMetaData followIMD = createIMD("index2", State.OPEN, "{}", 5, Settings.builder()
-                .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true").build(), CUSTOM_METADATA);
+                .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true").build(), customMetaData);
             Exception e = expectThrows(IllegalArgumentException.class, () -> validate(request, leaderIMD, followIMD, UUIDs, null));
             assertThat(e.getMessage(), equalTo("leader and follow index must be open"));
         }
+        {
+            // should fail, because index.xpack.ccr.following_index setting has not been enabled in leader index
+            IndexMetaData leaderIMD = createIMD("index1", 1,
+                Settings.builder().put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true").build(), customMetaData);
+            IndexMetaData followIMD = createIMD("index2", 1, Settings.EMPTY, customMetaData);
+            MapperService mapperService = MapperTestUtils.newMapperService(xContentRegistry(), createTempDir(), Settings.EMPTY, "index2");
+            mapperService.updateMapping(null, followIMD);
+            Exception e = expectThrows(IllegalArgumentException.class,
+                () -> validate(request, leaderIMD, followIMD, UUIDs, mapperService));
+            assertThat(e.getMessage(), equalTo("the following index [index2] is not ready to follow; " +
+                "the setting [index.xpack.ccr.following_index] must be enabled."));
+        }
         {
             // should fail, because leader has a field with the same name mapped as keyword and follower as text
             IndexMetaData leaderIMD = createIMD("index1", State.OPEN, "{\"properties\": {\"field\": {\"type\": \"keyword\"}}}", 5,
                 Settings.builder().put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true").build(), emptyMap());
             IndexMetaData followIMD = createIMD("index2", State.OPEN, "{\"properties\": {\"field\": {\"type\": \"text\"}}}", 5,
-                Settings.builder().put(CcrSettings.CCR_FOLLOWING_INDEX_SETTING.getKey(), true).build(), CUSTOM_METADATA);
+                Settings.builder().put(CcrSettings.CCR_FOLLOWING_INDEX_SETTING.getKey(), true).build(), customMetaData);
             MapperService mapperService = MapperTestUtils.newMapperService(xContentRegistry(), createTempDir(), Settings.EMPTY, "index2");
             mapperService.updateMapping(null, followIMD);
             Exception e = expectThrows(IllegalArgumentException.class, () -> validate(request, leaderIMD, followIMD, UUIDs, mapperService));
@@ -104,7 +129,7 @@ public class TransportFollowIndexActionTests extends ESTestCase {
             IndexMetaData followIMD = createIMD("index2", State.OPEN, mapping, 5, Settings.builder()
                 .put(CcrSettings.CCR_FOLLOWING_INDEX_SETTING.getKey(), true)
                 .put("index.analysis.analyzer.my_analyzer.type", "custom")
-                .put("index.analysis.analyzer.my_analyzer.tokenizer", "standard").build(), CUSTOM_METADATA);
+                .put("index.analysis.analyzer.my_analyzer.tokenizer", "standard").build(), customMetaData);
             Exception e = expectThrows(IllegalArgumentException.class, () -> validate(request, leaderIMD, followIMD, UUIDs, null));
             assertThat(e.getMessage(), equalTo("the leader and follower index settings must be identical"));
         }
@@ -114,7 +139,7 @@ public class TransportFollowIndexActionTests extends ESTestCase {
                 Settings.builder().put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true").build(), emptyMap());
             Settings followingIndexSettings = randomBoolean() ? Settings.EMPTY :
                 Settings.builder().put(CcrSettings.CCR_FOLLOWING_INDEX_SETTING.getKey(), false).build();
-            IndexMetaData followIMD = createIMD("index2", 5, followingIndexSettings, CUSTOM_METADATA);
+            IndexMetaData followIMD = createIMD("index2", 5, followingIndexSettings, customMetaData);
             MapperService mapperService = MapperTestUtils.newMapperService(xContentRegistry(), createTempDir(),
                 followingIndexSettings, "index2");
             mapperService.updateMapping(null, followIMD);
@@ -128,7 +153,7 @@ public class TransportFollowIndexActionTests extends ESTestCase {
             IndexMetaData leaderIMD = createIMD("index1", 5, Settings.builder()
                 .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true").build(), emptyMap());
             IndexMetaData followIMD = createIMD("index2", 5, Settings.builder()
-                .put(CcrSettings.CCR_FOLLOWING_INDEX_SETTING.getKey(), true).build(), CUSTOM_METADATA);
+                .put(CcrSettings.CCR_FOLLOWING_INDEX_SETTING.getKey(), true).build(), customMetaData);
             MapperService mapperService = MapperTestUtils.newMapperService(xContentRegistry(), createTempDir(), Settings.EMPTY, "index2");
             mapperService.updateMapping(null, followIMD);
             validate(request, leaderIMD, followIMD, UUIDs, mapperService);
@@ -143,7 +168,7 @@ public class TransportFollowIndexActionTests extends ESTestCase {
             IndexMetaData followIMD = createIMD("index2", State.OPEN, mapping, 5, Settings.builder()
                 .put(CcrSettings.CCR_FOLLOWING_INDEX_SETTING.getKey(), true)
                 .put("index.analysis.analyzer.my_analyzer.type", "custom")
-                .put("index.analysis.analyzer.my_analyzer.tokenizer", "standard").build(), CUSTOM_METADATA);
+                .put("index.analysis.analyzer.my_analyzer.tokenizer", "standard").build(), customMetaData);
             MapperService mapperService = MapperTestUtils.newMapperService(xContentRegistry(), createTempDir(),
                 followIMD.getSettings(), "index2");
             mapperService.updateMapping(null, followIMD);
@@ -161,7 +186,7 @@ public class TransportFollowIndexActionTests extends ESTestCase {
                 .put(CcrSettings.CCR_FOLLOWING_INDEX_SETTING.getKey(), true)
                 .put(IndexSettings.INDEX_REFRESH_INTERVAL_SETTING.getKey(), "10s")
                 .put("index.analysis.analyzer.my_analyzer.type", "custom")
-                .put("index.analysis.analyzer.my_analyzer.tokenizer", "standard").build(), CUSTOM_METADATA);
+                .put("index.analysis.analyzer.my_analyzer.tokenizer", "standard").build(), customMetaData);
             MapperService mapperService = MapperTestUtils.newMapperService(xContentRegistry(), createTempDir(),
                 followIMD.getSettings(), "index2");
             mapperService.updateMapping(null, followIMD);