Refactoring IT and naming changes

Signed-off-by: Gaurav Bafna <gbbafna@amazon.com>

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

@@ -0,0 +1,38 @@
+/*
+ * 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 org.opensearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest;
+import org.opensearch.client.Client;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.test.OpenSearchIntegTestCase;
+
+import java.util.List;
+
+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, autoManageMasterNodes = false)
+public class DocRepMigrationTestCase extends MigrationBaseTestCase {
+
+    public void testMixedModeAddDocRep() throws Exception {
+        internalCluster().setBootstrapClusterManagerNodeIndex(0);
+        List<String> cmNodes = internalCluster().startNodes(1);
+
+        Client client = internalCluster().client(cmNodes.get(0));
+        ClusterUpdateSettingsRequest updateSettingsRequest = new ClusterUpdateSettingsRequest();
+        updateSettingsRequest.persistentSettings(Settings.builder().put(REMOTE_STORE_COMPATIBILITY_MODE_SETTING.getKey(), "mixed"));
+        assertAcked(client().admin().cluster().updateSettings(updateSettingsRequest).actionGet());
+        addRemote = false;
+        internalCluster().startNode();
+        String[] allNodes = internalCluster().getNodeNames();
+        assertBusy(() -> { assertEquals(client.admin().cluster().prepareClusterStats().get().getNodes().size(), allNodes.length); });
+    }
+
+}

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

@@ -0,0 +1,50 @@
+/*
+ * 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 org.opensearch.common.settings.Settings;
+import org.opensearch.common.util.FeatureFlags;
+import org.opensearch.test.OpenSearchIntegTestCase;
+
+import java.nio.file.Path;
+
+import static org.opensearch.remotestore.RemoteStoreBaseIntegTestCase.remoteStoreClusterSettings;
+
+public class MigrationBaseTestCase extends OpenSearchIntegTestCase {
+    protected static final String REPOSITORY_NAME = "test-remote-store-repo";
+    protected static final String REPOSITORY_2_NAME = "test-remote-store-repo-2";
+
+    protected Path segmentRepoPath;
+    protected Path translogRepoPath;
+
+    boolean addRemote = false;
+
+    protected Settings nodeSettings(int nodeOrdinal) {
+        if (segmentRepoPath == null || translogRepoPath == null) {
+            segmentRepoPath = randomRepoPath().toAbsolutePath();
+            translogRepoPath = randomRepoPath().toAbsolutePath();
+        }
+        if (addRemote) {
+            logger.info("Adding remote store node");
+            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();
+        }
+    }
+
+    @Override
+    protected Settings featureFlagSettings() {
+        return Settings.builder().put(super.featureFlagSettings()).put(FeatureFlags.REMOTE_STORE_MIGRATION_EXPERIMENTAL, "true").build();
+    }
+}

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

@@ -0,0 +1,74 @@
+/*
+ * 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 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.client.Client;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.test.OpenSearchIntegTestCase;
+
+import java.util.List;
+
+import static org.opensearch.node.remotestore.RemoteStoreNodeService.MIGRATION_DIRECTION_SETTING;
+import static org.opensearch.node.remotestore.RemoteStoreNodeService.REMOTE_STORE_COMPATIBILITY_MODE_SETTING;
+import static org.opensearch.remotestore.RemoteStoreBaseIntegTestCase.remoteStoreClusterSettings;
+import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
+
+@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0, autoManageMasterNodes = false)
+public class RemoteStoreMigrationTestCase extends MigrationBaseTestCase {
+    public void testMixedModeAddRemoteNodes() throws Exception {
+        internalCluster().setBootstrapClusterManagerNodeIndex(0);
+        List<String> cmNodes = internalCluster().startNodes(1);
+        Client client = internalCluster().client(cmNodes.get(0));
+        ClusterUpdateSettingsRequest updateSettingsRequest = new ClusterUpdateSettingsRequest();
+        updateSettingsRequest.persistentSettings(Settings.builder().put(REMOTE_STORE_COMPATIBILITY_MODE_SETTING.getKey(), "mixed"));
+        assertAcked(client().admin().cluster().updateSettings(updateSettingsRequest).actionGet());
+
+        // add remote node in mixed mode cluster
+        addRemote = true;
+        internalCluster().startNode();
+        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());
+
+        // add docrep mode in mixed mode cluster
+        addRemote = true;
+        internalCluster().startNode();
+        assertBusy(() -> {
+            assertEquals(client.admin().cluster().prepareClusterStats().get().getNodes().size(), internalCluster().getNodeNames().length);
+        });
+
+        // add incompatible remote node in remote mixed cluster
+        Settings.Builder badSettings = Settings.builder()
+            .put(remoteStoreClusterSettings(REPOSITORY_NAME, segmentRepoPath, "REPOSITORY_2_NAME", translogRepoPath))
+            .put("discovery.initial_state_timeout", "500ms");
+        String badNode = internalCluster().startNode(badSettings);
+        assertTrue(client.admin().cluster().prepareClusterStats().get().getNodes().size() < internalCluster().getNodeNames().length);
+        internalCluster().stopRandomNode(settings -> settings.get("node.name").equals(badNode));
+    }
+
+    public void testMigrationDirections() {
+        ClusterUpdateSettingsRequest updateSettingsRequest = new ClusterUpdateSettingsRequest();
+        // add remote node in docrep cluster
+        updateSettingsRequest.persistentSettings(Settings.builder().put(MIGRATION_DIRECTION_SETTING.getKey(), "docrep"));
+        assertAcked(client().admin().cluster().updateSettings(updateSettingsRequest).actionGet());
+
+        updateSettingsRequest.persistentSettings(Settings.builder().put(MIGRATION_DIRECTION_SETTING.getKey(), "remote_store"));
+        assertAcked(client().admin().cluster().updateSettings(updateSettingsRequest).actionGet());
+
+        updateSettingsRequest.persistentSettings(Settings.builder().put(MIGRATION_DIRECTION_SETTING.getKey(), "random"));
+        assertThrows(IllegalArgumentException.class, () -> client().admin().cluster().updateSettings(updateSettingsRequest).actionGet());
+    }
+}

server/src/main/java/org/opensearch/cluster/coordination/JoinTaskExecutor.java

@@ -180,12 +180,7 @@ public ClusterTasksResult<Task> execute(ClusterState currentState, List<Task> jo
         // An optimization can be done as this will get invoked
         // for every set of node join task which we can optimize to not compute if cluster state already has
         // repository information.
-        Optional<DiscoveryNode> remoteDN = newState.nodes()
-            .getNodes()
-            .values()
-            .stream()
-            .filter(DiscoveryNode::isRemoteStoreNode)
-            .findFirst();
+        Optional<DiscoveryNode> remoteDN = currentNodes.getNodes().values().stream().filter(DiscoveryNode::isRemoteStoreNode).findFirst();
         DiscoveryNode dn = remoteDN.orElseGet(() -> (currentNodes.getNodes().values()).stream().findFirst().get());
         RepositoriesMetadata repositoriesMetadata = remoteStoreNodeService.updateRepositoriesMetadata(
             dn,
@@ -221,8 +216,8 @@ public ClusterTasksResult<Task> execute(ClusterState currentState, List<Task> jo
                     nodesBuilder.add(node);
 
                     if (remoteDN.isEmpty()) {
-                        // This is hit only on cases where we encounter first remote node in remote store migration
-                        logger.info("Updating system repository now for remote store migration");
+                        // This is hit only on cases where we encounter first remote node
+                        logger.info("Updating system repository now for remote store");
                         repositoriesMetadata = remoteStoreNodeService.updateRepositoriesMetadata(
                             node,
                             currentState.getMetadata().custom(RepositoriesMetadata.TYPE)

server/src/main/java/org/opensearch/common/settings/ClusterSettings.java

@@ -698,7 +698,7 @@ public void apply(Settings value, Settings current, Settings previous) {
                 RemoteClusterStateService.GLOBAL_METADATA_UPLOAD_TIMEOUT_SETTING,
                 RemoteClusterStateService.METADATA_MANIFEST_UPLOAD_TIMEOUT_SETTING,
                 RemoteStoreNodeService.REMOTE_STORE_COMPATIBILITY_MODE_SETTING,
-                RemoteStoreNodeService.DIRECTION_SETTING,
+                RemoteStoreNodeService.MIGRATION_DIRECTION_SETTING,
                 IndicesService.CLUSTER_REMOTE_TRANSLOG_BUFFER_INTERVAL_SETTING,
                 IndicesService.CLUSTER_REMOTE_INDEX_RESTRICT_ASYNC_DURABILITY_SETTING,
                 AdmissionControlSettings.ADMISSION_CONTROL_TRANSPORT_LAYER_MODE,

server/src/main/java/org/opensearch/node/remotestore/RemoteStoreNodeService.java

@@ -22,6 +22,7 @@
 import org.opensearch.threadpool.ThreadPool;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
@@ -56,14 +57,14 @@ public class RemoteStoreNodeService {
         Setting.Property.NodeScope
     );
 
-    public static final Setting<Direction> DIRECTION_SETTING = new Setting<>(
-        "direction",
+    public static final Setting<Direction> MIGRATION_DIRECTION_SETTING = new Setting<>(
+        "migration.direction",
         Direction.NONE.name(),
         Direction::parseString,
         value -> {
             if (value != Direction.NONE && FeatureFlags.isEnabled(FeatureFlags.REMOTE_STORE_MIGRATION_EXPERIMENTAL_SETTING) == false) {
                 throw new IllegalArgumentException(
-                    " direction is under an experimental feature and can be activated only by enabling "
+                    " migration.direction is under an experimental feature and can be activated only by enabling "
                         + REMOTE_STORE_MIGRATION_EXPERIMENTAL
                         + " feature flag in the JVM options "
                 );
@@ -97,7 +98,7 @@ public static CompatibilityMode parseString(String compatibilityMode) {
                         + compatibilityMode
                         + "] compatibility mode is not supported. "
                         + "supported modes are ["
-                        + CompatibilityMode.values().toString()
+                        + Arrays.toString(CompatibilityMode.values())
                         + "]"
                 );
             }
@@ -119,14 +120,7 @@ public static Direction parseString(String direction) {
             try {
                 return Direction.valueOf(direction.toUpperCase(Locale.ROOT));
             } catch (IllegalArgumentException e) {
-                throw new IllegalArgumentException(
-                    "["
-                        + direction
-                        + "] direction is not supported. "
-                        + "supported modes are ["
-                        + CompatibilityMode.values().toString()
-                        + "]"
-                );
+                throw new IllegalArgumentException("[" + direction + "] migration.direction is not supported.");
             }
         }
     }