Moving over metadata mutate logic to IndexMetadataUpdater

Signed-off-by: Shourya Dutta Biswas <114977491+shourya035@users.noreply.github.com>
opensearch-project · Apr 20, 2024 · f14e231 · f14e231
1 parent d0fa3b0
commit f14e231
Show file tree

Hide file tree

Showing 15 changed files with 1,010 additions and 627 deletions.
diff --git a/...er/src/internalClusterTest/java/org/opensearch/remotemigration/MigrationBaseTestCase.java b/...er/src/internalClusterTest/java/org/opensearch/remotemigration/MigrationBaseTestCase.java
@@ -29,6 +29,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
+import static org.opensearch.cluster.routing.allocation.decider.EnableAllocationDecider.CLUSTER_ROUTING_REBALANCE_ENABLE_SETTING;
 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.repositories.fs.ReloadableFsRepository.REPOSITORIES_FAILRATE_SETTING;
@@ -157,4 +158,25 @@ private Thread getIndexingThread() {
             });
         }
     }
+
+    public void excludeNodeSet(String attr, String value) {
+        assertAcked(
+            internalCluster().client()
+                .admin()
+                .cluster()
+                .prepareUpdateSettings()
+                .setTransientSettings(Settings.builder().put("cluster.routing.allocation.exclude._" + attr, value))
+                .get()
+        );
+    }
+
+    public void stopShardRebalancing() {
+        assertAcked(
+            client().admin()
+                .cluster()
+                .prepareUpdateSettings()
+                .setPersistentSettings(Settings.builder().put(CLUSTER_ROUTING_REBALANCE_ENABLE_SETTING.getKey(), "none").build())
+                .get()
+        );
+    }
 }
diff --git a/.../src/internalClusterTest/java/org/opensearch/remotemigration/RemoteDualReplicationIT.java b/.../src/internalClusterTest/java/org/opensearch/remotemigration/RemoteDualReplicationIT.java
@@ -132,8 +132,8 @@ public void testRemotePrimaryDocRepReplica() throws Exception {
 
     /*
     Scenario:
-    - Starts 1 docrep backed data node
-    - Creates an index with 0 replica
+    - Starts 2 docrep backed data node
+    - Creates an index with 1 replica
     - Starts 1 remote backed data node
     - Index some docs
     - Move primary copy from docrep to remote through _cluster/reroute
@@ -145,14 +145,14 @@ public void testRemotePrimaryDocRepReplica() throws Exception {
     public void testRemotePrimaryDocRepAndRemoteReplica() throws Exception {
         internalCluster().startClusterManagerOnlyNode();
 
-        logger.info("---> Starting 1 docrep data nodes");
-        String docrepNodeName = internalCluster().startDataOnlyNode();
+        logger.info("---> Starting 2 docrep data nodes");
+        internalCluster().startDataOnlyNodes(2);
         internalCluster().validateClusterFormed();
         assertEquals(internalCluster().client().admin().cluster().prepareGetRepositories().get().repositories().size(), 0);
 
-        logger.info("---> Creating index with 0 replica");
+        logger.info("---> Creating index with 1 replica");
         Settings zeroReplicas = Settings.builder()
-            .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+            .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 1)
             .put(IndexService.RETENTION_LEASE_SYNC_INTERVAL_SETTING.getKey(), "1s")
             .put(IndexService.GLOBAL_CHECKPOINT_SYNC_INTERVAL_SETTING.getKey(), "1s")
             .build();
@@ -245,14 +245,26 @@ RLs on remote enabled copies are brought up to (GlobalCkp + 1) upon a flush requ
         pollAndCheckRetentionLeases(REMOTE_PRI_DOCREP_REMOTE_REP);
     }
 
+    /*
+    Scenario:
+    - Starts 2 docrep backed data node
+    - Creates an index with 1 replica
+    - Starts 1 remote backed data node
+    - Index some docs
+    - Move primary copy from docrep to remote through _cluster/reroute
+    - Starts another remote backed data node
+    - Expands index to 2 replicas. One replica copy lies in remote backed node and other in docrep backed node
+    - Index some more docs
+    - Assert retention lease consistency
+     */
     public void testMissingRetentionLeaseCreatedOnFailedOverRemoteReplica() throws Exception {
         internalCluster().startClusterManagerOnlyNode();
 
-        logger.info("---> Starting docrep data node");
-        internalCluster().startDataOnlyNode();
+        logger.info("---> Starting 2 docrep data nodes");
+        internalCluster().startDataOnlyNodes(2);
 
         Settings zeroReplicasAndOverridenSyncIntervals = Settings.builder()
-            .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+            .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 1)
             .put(IndexService.GLOBAL_CHECKPOINT_SYNC_INTERVAL_SETTING.getKey(), "100ms")
             .put(IndexService.RETENTION_LEASE_SYNC_INTERVAL_SETTING.getKey(), "100ms")
             .build();
@@ -323,25 +335,24 @@ private void pollAndCheckRetentionLeases(String indexName) throws Exception {
 
     /*
     Scenario:
-    - Starts 1 docrep backed data node
-    - Creates an index with 0 replica
+    - Starts 2 docrep backed data node
+    - Creates an index with 1 replica
     - Starts 1 remote backed data node
     - Move primary copy from docrep to remote through _cluster/reroute
-    - Expands index to 1 replica
     - Stops remote enabled node
     - Ensure doc count is same after failover
     - Index some more docs to ensure working of failed-over primary
      */
     public void testFailoverRemotePrimaryToDocrepReplica() throws Exception {
         internalCluster().startClusterManagerOnlyNode();
 
-        logger.info("---> Starting 1 docrep data nodes");
-        String docrepNodeName = internalCluster().startDataOnlyNode();
+        logger.info("---> Starting 2 docrep data nodes");
+        internalCluster().startDataOnlyNodes(2);
         internalCluster().validateClusterFormed();
         assertEquals(internalCluster().client().admin().cluster().prepareGetRepositories().get().repositories().size(), 0);
 
         logger.info("---> Creating index with 0 replica");
-        Settings excludeRemoteNode = Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0).build();
+        Settings excludeRemoteNode = Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 1).build();
         createIndex(FAILOVER_REMOTE_TO_DOCREP, excludeRemoteNode);
         ensureGreen(FAILOVER_REMOTE_TO_DOCREP);
         initDocRepToRemoteMigration();
@@ -376,8 +387,8 @@ public void testFailoverRemotePrimaryToDocrepReplica() throws Exception {
         );
         ensureGreen(FAILOVER_REMOTE_TO_DOCREP);
 
-        logger.info("---> Expanding index to 1 replica copy");
-        Settings twoReplicas = Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 1).build();
+        logger.info("---> Expanding index to 2 replica copies");
+        Settings twoReplicas = Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 2).build();
         assertAcked(
             internalCluster().client()
                 .admin()
@@ -412,7 +423,7 @@ public void testFailoverRemotePrimaryToDocrepReplica() throws Exception {
 
         logger.info("---> Stop remote store enabled node");
         internalCluster().stopRandomNode(InternalTestCluster.nameFilter(remoteNodeName));
-        ensureStableCluster(2);
+        ensureStableCluster(3);
         ensureYellow(FAILOVER_REMOTE_TO_DOCREP);
 
         shardStatsMap = internalCluster().client().admin().indices().prepareStats(FAILOVER_REMOTE_TO_DOCREP).setDocs(true).get().asMap();
@@ -433,19 +444,18 @@ public void testFailoverRemotePrimaryToDocrepReplica() throws Exception {
         refreshAndWaitForReplication(FAILOVER_REMOTE_TO_DOCREP);
 
         shardStatsMap = internalCluster().client().admin().indices().prepareStats(FAILOVER_REMOTE_TO_DOCREP).setDocs(true).get().asMap();
-        assertEquals(1, shardStatsMap.size());
+        assertEquals(2, shardStatsMap.size());
         shardStatsMap.forEach(
             (shardRouting, shardStats) -> { assertEquals(firstBatch + secondBatch, shardStats.getStats().getDocs().getCount()); }
         );
     }
 
     /*
     Scenario:
-    - Starts 1 docrep backed data node
-    - Creates an index with 0 replica
+    - Starts 2 docrep backed data node
+    - Creates an index with 1 replica
     - Starts 1 remote backed data node
     - Move primary copy from docrep to remote through _cluster/reroute
-    - Expands index to 1 replica
     - Stops remote enabled node
     - Ensure doc count is same after failover
     - Index some more docs to ensure working of failed-over primary
@@ -529,7 +539,8 @@ private void assertReplicaAndPrimaryConsistency(String indexName, int firstBatch
                     RemoteSegmentStats remoteSegmentStats = shardStats.getSegments().getRemoteSegmentStats();
                     assertTrue(remoteSegmentStats.getUploadBytesSucceeded() > 0);
                     assertTrue(remoteSegmentStats.getTotalUploadTime() > 0);
-                } else {
+                }
+                if (shardRouting.unassigned() == false && shardRouting.primary() == false) {
                     boolean remoteNode = nodes.get(shardRouting.currentNodeId()).isRemoteStoreNode();
                     assertEquals(
                         "Mismatched doc count. Is this on remote node ? " + remoteNode,