[Remote Migration] Changes for Primary Relocation during migration

server/src/internalClusterTest/java/org/opensearch/index/shard/IndexShardIT.java

@@ -713,7 +713,8 @@ public static final IndexShard newIndexShard(
             null,
             () -> IndexSettings.DEFAULT_REMOTE_TRANSLOG_BUFFER_INTERVAL,
             nodeId,
-            null
+            null,
+            false
         );
     }
 

server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationClusterSettingIT.java

@@ -89,8 +89,8 @@ public void testIndexReplicationSettingOverridesSegRepClusterSetting() throws Ex
         Index index = resolveIndex(INDEX_NAME);
         Index anotherIndex = resolveIndex(ANOTHER_INDEX);
         IndicesService indicesService = internalCluster().getInstance(IndicesService.class, primaryNode);
-        assertEquals(indicesService.indexService(index).getIndexSettings().isSegRepEnabled(), false);
-        assertEquals(indicesService.indexService(anotherIndex).getIndexSettings().isSegRepEnabled(), true);
+        assertEquals(indicesService.indexService(index).getIndexSettings().isSegRepEnabledOrRemoteNode(), false);
+        assertEquals(indicesService.indexService(anotherIndex).getIndexSettings().isSegRepEnabledOrRemoteNode(), true);
     }
 
     public void testIndexReplicationSettingOverridesDocRepClusterSetting() throws Exception {
@@ -119,8 +119,8 @@ public void testIndexReplicationSettingOverridesDocRepClusterSetting() throws Ex
         Index index = resolveIndex(INDEX_NAME);
         Index anotherIndex = resolveIndex(ANOTHER_INDEX);
         IndicesService indicesService = internalCluster().getInstance(IndicesService.class, primaryNode);
-        assertEquals(indicesService.indexService(index).getIndexSettings().isSegRepEnabled(), true);
-        assertEquals(indicesService.indexService(anotherIndex).getIndexSettings().isSegRepEnabled(), false);
+        assertEquals(indicesService.indexService(index).getIndexSettings().isSegRepEnabledOrRemoteNode(), true);
+        assertEquals(indicesService.indexService(anotherIndex).getIndexSettings().isSegRepEnabledOrRemoteNode(), false);
     }
 
     public void testReplicationTypesOverrideNotAllowed_IndexAPI() {

server/src/internalClusterTest/java/org/opensearch/remotemigration/MigrationBaseTestCase.java

@@ -8,13 +8,19 @@
 
 package org.opensearch.remotemigration;
 
+import org.opensearch.action.admin.cluster.repositories.get.GetRepositoriesRequest;
+import org.opensearch.action.admin.cluster.repositories.get.GetRepositoriesResponse;
+import org.opensearch.cluster.metadata.RepositoryMetadata;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.util.FeatureFlags;
+import org.opensearch.repositories.fs.ReloadableFsRepository;
 import org.opensearch.test.OpenSearchIntegTestCase;
 
 import java.nio.file.Path;
+import java.util.concurrent.ExecutionException;
 
-import static org.opensearch.remotestore.RemoteStoreBaseIntegTestCase.remoteStoreClusterSettings;
+import static org.opensearch.repositories.fs.ReloadableFsRepository.REPOSITORIES_FAILRATE_SETTING;
+import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
 
 public class MigrationBaseTestCase extends OpenSearchIntegTestCase {
     protected static final String REPOSITORY_NAME = "test-remote-store-repo";
@@ -35,16 +41,27 @@ protected Settings nodeSettings(int nodeOrdinal) {
             return Settings.builder()
                 .put(super.nodeSettings(nodeOrdinal))
                 .put(remoteStoreClusterSettings(REPOSITORY_NAME, segmentRepoPath, REPOSITORY_2_NAME, translogRepoPath))
-                .put("discovery.initial_state_timeout", "500ms")
                 .build();
         } else {
             logger.info("Adding docrep node");
-            return Settings.builder().put(super.nodeSettings(nodeOrdinal)).put("discovery.initial_state_timeout", "500ms").build();
+            return Settings.builder().put(super.nodeSettings(nodeOrdinal)).build();
         }
     }
 
     @Override
     protected Settings featureFlagSettings() {
         return Settings.builder().put(super.featureFlagSettings()).put(FeatureFlags.REMOTE_STORE_MIGRATION_EXPERIMENTAL, "true").build();
     }
+
+    protected void setFailRate(String repoName, int value) throws ExecutionException, InterruptedException {
+        GetRepositoriesRequest gr = new GetRepositoriesRequest(new String[] { repoName });
+        GetRepositoriesResponse res = client().admin().cluster().getRepositories(gr).get();
+        RepositoryMetadata rmd = res.repositories().get(0);
+        Settings.Builder settings = Settings.builder()
+            .put("location", rmd.settings().get("location"))
+            .put(REPOSITORIES_FAILRATE_SETTING.getKey(), value);
+        assertAcked(
+            client().admin().cluster().preparePutRepository(repoName).setType(ReloadableFsRepository.TYPE).setSettings(settings).get()
+        );
+    }
 }

server/src/internalClusterTest/java/org/opensearch/remotemigration/RemotePrimaryRelocationIT.java

@@ -0,0 +1,223 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.remotemigration;
+
+import com.carrotsearch.randomizedtesting.generators.RandomNumbers;
+
+import org.opensearch.action.DocWriteResponse;
+import org.opensearch.action.admin.cluster.health.ClusterHealthResponse;
+import org.opensearch.action.admin.cluster.repositories.get.GetRepositoriesRequest;
+import org.opensearch.action.admin.cluster.repositories.get.GetRepositoriesResponse;
+import org.opensearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest;
+import org.opensearch.action.delete.DeleteResponse;
+import org.opensearch.action.index.IndexResponse;
+import org.opensearch.client.Client;
+import org.opensearch.cluster.routing.allocation.command.MoveAllocationCommand;
+import org.opensearch.common.Priority;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.common.unit.TimeValue;
+import org.opensearch.index.query.QueryBuilders;
+import org.opensearch.plugins.Plugin;
+import org.opensearch.test.OpenSearchIntegTestCase;
+import org.opensearch.test.hamcrest.OpenSearchAssertions;
+import org.opensearch.test.transport.MockTransportService;
+
+import java.util.Collection;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static java.util.Arrays.asList;
+import static org.opensearch.node.remotestore.RemoteStoreNodeService.REMOTE_STORE_COMPATIBILITY_MODE_SETTING;
+import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
+
+@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0)
+public class RemotePrimaryRelocationIT extends MigrationBaseTestCase {
+    protected int maximumNumberOfShards() {
+        return 1;
+    }
+
+    // ToDo : Fix me when we support migration of replicas
+    protected int maximumNumberOfReplicas() {
+        return 0;
+    }
+
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return asList(MockTransportService.TestPlugin.class);
+    }
+
+    public void testMixedModeRelocation() throws Exception {
+        String docRepNode = internalCluster().startNode();
+        Client client = internalCluster().client(docRepNode);
+        ClusterUpdateSettingsRequest updateSettingsRequest = new ClusterUpdateSettingsRequest();
+        updateSettingsRequest.persistentSettings(Settings.builder().put(REMOTE_STORE_COMPATIBILITY_MODE_SETTING.getKey(), "mixed"));
+        assertAcked(client().admin().cluster().updateSettings(updateSettingsRequest).actionGet());
+
+        // create shard with 0 replica and 1 shard
+        client().admin().indices().prepareCreate("test").setSettings(indexSettings()).setMapping("field", "type=text").get();
+        ensureGreen("test");
+
+        AtomicInteger numAutoGenDocs = new AtomicInteger();
+        final AtomicBoolean finished = new AtomicBoolean(false);
+        Thread indexingThread = getIndexingThread(finished, numAutoGenDocs);
+
+        refresh("test");
+
+        // add remote node in mixed mode cluster
+        addRemote = true;
+        String remoteNode = internalCluster().startNode();
+        internalCluster().validateClusterFormed();
+
+        String remoteNode2 = internalCluster().startNode();
+        internalCluster().validateClusterFormed();
+
+        // assert repo gets registered
+        GetRepositoriesRequest gr = new GetRepositoriesRequest(new String[] { REPOSITORY_NAME });
+        GetRepositoriesResponse getRepositoriesResponse = client.admin().cluster().getRepositories(gr).actionGet();
+        assertEquals(1, getRepositoriesResponse.repositories().size());
+
+        // Index some more docs
+        int currentDoc = numAutoGenDocs.get();
+        int finalCurrentDoc1 = currentDoc;
+        waitUntil(() -> numAutoGenDocs.get() > finalCurrentDoc1 + 5);
+
+        logger.info("-->  relocating from {} to {} ", docRepNode, remoteNode);
+        client().admin().cluster().prepareReroute().add(new MoveAllocationCommand("test", 0, docRepNode, remoteNode)).execute().actionGet();
+        ClusterHealthResponse clusterHealthResponse = client().admin()
+            .cluster()
+            .prepareHealth()
+            .setTimeout(TimeValue.timeValueSeconds(60))
+            .setWaitForEvents(Priority.LANGUID)
+            .setWaitForNoRelocatingShards(true)
+            .execute()
+            .actionGet();
+
+        assertEquals(0, clusterHealthResponse.getRelocatingShards());
+        assertEquals(remoteNode, primaryNodeName("test"));
+        logger.info("-->  relocation from docrep to remote  complete");
+
+        // Index some more docs
+        currentDoc = numAutoGenDocs.get();
+        int finalCurrentDoc = currentDoc;
+        waitUntil(() -> numAutoGenDocs.get() > finalCurrentDoc + 5);
+
+        client().admin()
+            .cluster()
+            .prepareReroute()
+            .add(new MoveAllocationCommand("test", 0, remoteNode, remoteNode2))
+            .execute()
+            .actionGet();
+        clusterHealthResponse = client().admin()
+            .cluster()
+            .prepareHealth()
+            .setTimeout(TimeValue.timeValueSeconds(60))
+            .setWaitForEvents(Priority.LANGUID)
+            .setWaitForNoRelocatingShards(true)
+            .execute()
+            .actionGet();
+
+        assertEquals(0, clusterHealthResponse.getRelocatingShards());
+        assertEquals(remoteNode2, primaryNodeName("test"));
+
+        logger.info("-->  relocation from remote to remote  complete");
+
+        finished.set(true);
+        indexingThread.join();
+        refresh("test");
+        OpenSearchAssertions.assertHitCount(client().prepareSearch("test").setTrackTotalHits(true).get(), numAutoGenDocs.get());
+        OpenSearchAssertions.assertHitCount(
+            client().prepareSearch("test")
+                .setTrackTotalHits(true)// extra paranoia ;)
+                .setQuery(QueryBuilders.termQuery("auto", true))
+                .get(),
+            numAutoGenDocs.get()
+        );
+
+    }
+
+    public void testMixedModeRelocation_RemoteSeedingFail() throws Exception {
+        String docRepNode = internalCluster().startNode();
+        Client client = internalCluster().client(docRepNode);
+        ClusterUpdateSettingsRequest updateSettingsRequest = new ClusterUpdateSettingsRequest();
+        updateSettingsRequest.persistentSettings(Settings.builder().put(REMOTE_STORE_COMPATIBILITY_MODE_SETTING.getKey(), "mixed"));
+        assertAcked(client().admin().cluster().updateSettings(updateSettingsRequest).actionGet());
+
+        // create shard with 0 replica and 1 shard
+        client().admin().indices().prepareCreate("test").setSettings(indexSettings()).setMapping("field", "type=text").get();
+        ensureGreen("test");
+
+        AtomicInteger numAutoGenDocs = new AtomicInteger();
+        final AtomicBoolean finished = new AtomicBoolean(false);
+        Thread indexingThread = getIndexingThread(finished, numAutoGenDocs);
+
+        refresh("test");
+
+        // add remote node in mixed mode cluster
+        addRemote = true;
+        String remoteNode = internalCluster().startNode();
+        internalCluster().validateClusterFormed();
+
+        // assert repo gets registered
+        GetRepositoriesRequest gr = new GetRepositoriesRequest(new String[] { REPOSITORY_NAME });
+        GetRepositoriesResponse getRepositoriesResponse = client.admin().cluster().getRepositories(gr).actionGet();
+        assertEquals(1, getRepositoriesResponse.repositories().size());
+
+        setFailRate(REPOSITORY_NAME, 100);
+
+        logger.info("--> relocating from {} to {} ", docRepNode, remoteNode);
+        client().admin().cluster().prepareReroute().add(new MoveAllocationCommand("test", 0, docRepNode, remoteNode)).execute().actionGet();
+        ClusterHealthResponse clusterHealthResponse = client().admin()
+            .cluster()
+            .prepareHealth()
+            .setTimeout(TimeValue.timeValueSeconds(5))
+            .setWaitForEvents(Priority.LANGUID)
+            .setWaitForNoRelocatingShards(true)
+            .execute()
+            .actionGet();
+
+        assertTrue(clusterHealthResponse.getRelocatingShards() == 1);
+        setFailRate(REPOSITORY_NAME, 0);
+        Thread.sleep(RandomNumbers.randomIntBetween(random(), 0, 2000));
+        clusterHealthResponse = client().admin()
+            .cluster()
+            .prepareHealth()
+            .setTimeout(TimeValue.timeValueSeconds(45))
+            .setWaitForEvents(Priority.LANGUID)
+            .setWaitForNoRelocatingShards(true)
+            .execute()
+            .actionGet();
+        assertTrue(clusterHealthResponse.getRelocatingShards() == 0);
+        logger.info("--> remote to remote relocation complete");
+        finished.set(true);
+        indexingThread.join();
+        refresh("test");
+        OpenSearchAssertions.assertHitCount(client().prepareSearch("test").setTrackTotalHits(true).get(), numAutoGenDocs.get());
+        OpenSearchAssertions.assertHitCount(
+            client().prepareSearch("test")
+                .setTrackTotalHits(true)// extra paranoia ;)
+                .setQuery(QueryBuilders.termQuery("auto", true))
+                .get(),
+            numAutoGenDocs.get()
+        );
+    }
+
+    private static Thread getIndexingThread(AtomicBoolean finished, AtomicInteger numAutoGenDocs) {
+        Thread indexingThread = new Thread(() -> {
+            while (finished.get() == false && numAutoGenDocs.get() < 10_000) {
+                IndexResponse indexResponse = client().prepareIndex("test").setId("id").setSource("field", "value").get();
+                assertEquals(DocWriteResponse.Result.CREATED, indexResponse.getResult());
+                DeleteResponse deleteResponse = client().prepareDelete("test", "id").get();
+                assertEquals(DocWriteResponse.Result.DELETED, deleteResponse.getResult());
+                client().prepareIndex("test").setSource("auto", true).get();
+                numAutoGenDocs.incrementAndGet();
+            }
+        });
+        indexingThread.start();
+        return indexingThread;
+    }
+}

server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteIndexPrimaryRelocationIT.java

@@ -44,7 +44,6 @@ public Settings indexSettings() {
             .build();
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/opensearch-project/OpenSearch/issues/9191")
     public void testPrimaryRelocationWhileIndexing() throws Exception {
         internalCluster().startClusterManagerOnlyNode();
         super.testPrimaryRelocationWhileIndexing();