[Remote Store] Fix sleep time bug during remote store sync (#14037)

--------- Signed-off-by: Gaurav Bafna <gbbafna@amazon.com>
opensearch-project · Jun 13, 2024 · afeddc2 · afeddc2
1 parent ccf5289
commit afeddc2
Show file tree

Hide file tree

Showing 3 changed files with 32 additions and 25 deletions.
diff --git a/...er/src/internalClusterTest/java/org/opensearch/remotemigration/MigrationBaseTestCase.java b/...er/src/internalClusterTest/java/org/opensearch/remotemigration/MigrationBaseTestCase.java
@@ -9,18 +9,24 @@
 package org.opensearch.remotemigration;
 
 import org.opensearch.action.DocWriteResponse;
+import org.opensearch.action.admin.cluster.health.ClusterHealthRequest;
+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.bulk.BulkRequest;
 import org.opensearch.action.bulk.BulkResponse;
 import org.opensearch.action.delete.DeleteResponse;
 import org.opensearch.action.index.IndexRequest;
 import org.opensearch.action.index.IndexResponse;
+import org.opensearch.client.Requests;
 import org.opensearch.cluster.ClusterState;
+import org.opensearch.cluster.health.ClusterHealthStatus;
 import org.opensearch.cluster.metadata.RepositoryMetadata;
 import org.opensearch.cluster.routing.RoutingNode;
+import org.opensearch.common.Priority;
 import org.opensearch.common.UUIDs;
 import org.opensearch.common.settings.Settings;
+import org.opensearch.common.unit.TimeValue;
 import org.opensearch.repositories.fs.ReloadableFsRepository;
 import org.opensearch.test.OpenSearchIntegTestCase;
 import org.junit.Before;
@@ -39,6 +45,7 @@
 import static org.opensearch.node.remotestore.RemoteStoreNodeService.REMOTE_STORE_COMPATIBILITY_MODE_SETTING;
 import static org.opensearch.repositories.fs.ReloadableFsRepository.REPOSITORIES_FAILRATE_SETTING;
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
+import static org.hamcrest.Matchers.equalTo;
 
 public class MigrationBaseTestCase extends OpenSearchIntegTestCase {
     protected static final String REPOSITORY_NAME = "test-remote-store-repo";
@@ -114,6 +121,10 @@ public void initDocRepToRemoteMigration() {
         );
     }
 
+    public ClusterHealthStatus ensureGreen(String... indices) {
+        return ensureGreen(TimeValue.timeValueSeconds(60), indices);
+    }
+
     public BulkResponse indexBulk(String indexName, int numDocs) {
         BulkRequest bulkRequest = new BulkRequest();
         for (int i = 0; i < numDocs; i++) {
@@ -181,14 +192,12 @@ private Thread getIndexingThread() {
                     long currentDocCount = indexedDocs.incrementAndGet();
                     if (currentDocCount > 0 && currentDocCount % refreshFrequency == 0) {
                         if (rarely()) {
-                            logger.info("--> [iteration {}] flushing index", currentDocCount);
                             client().admin().indices().prepareFlush(indexName).get();
+                            logger.info("Completed ingestion of {} docs. Flushing now", currentDocCount);
                         } else {
-                            logger.info("--> [iteration {}] refreshing index", currentDocCount);
                             client().admin().indices().prepareRefresh(indexName).get();
                         }
                     }
-                    logger.info("Completed ingestion of {} docs", currentDocCount);
                 }
             });
         }
@@ -218,4 +227,21 @@ public void stopShardRebalancing() {
                 .get()
         );
     }
+
+    public ClusterHealthStatus waitForRelocation() {
+        ClusterHealthRequest request = Requests.clusterHealthRequest()
+            .waitForNoRelocatingShards(true)
+            .timeout(TimeValue.timeValueSeconds(60))
+            .waitForEvents(Priority.LANGUID);
+        ClusterHealthResponse actionGet = client().admin().cluster().health(request).actionGet();
+        if (actionGet.isTimedOut()) {
+            logger.info(
+                "waitForRelocation timed out, cluster state:\n{}\n{}",
+                client().admin().cluster().prepareState().get().getState(),
+                client().admin().cluster().preparePendingClusterTasks().get()
+            );
+            assertThat("timed out waiting for relocation", actionGet.isTimedOut(), equalTo(false));
+        }
+        return actionGet.getStatus();
+    }
 }
diff --git a/...rc/internalClusterTest/java/org/opensearch/remotemigration/RemotePrimaryRelocationIT.java b/...rc/internalClusterTest/java/org/opensearch/remotemigration/RemotePrimaryRelocationIT.java
@@ -99,16 +99,7 @@ public void testRemotePrimaryRelocation() throws Exception {
             .add(new MoveAllocationCommand("test", 0, primaryNodeName("test"), remoteNode))
             .execute()
             .actionGet();
-        ClusterHealthResponse clusterHealthResponse = client().admin()
-            .cluster()
-            .prepareHealth()
-            .setTimeout(TimeValue.timeValueSeconds(60))
-            .setWaitForEvents(Priority.LANGUID)
-            .setWaitForNoRelocatingShards(true)
-            .execute()
-            .actionGet();
-
-        assertEquals(0, clusterHealthResponse.getRelocatingShards());
+        waitForRelocation();
         assertEquals(remoteNode, primaryNodeName("test"));
         logger.info("-->  relocation from docrep to remote  complete");
 
@@ -123,16 +114,7 @@ public void testRemotePrimaryRelocation() throws Exception {
             .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());
+        waitForRelocation();
         assertEquals(remoteNode2, primaryNodeName("test"));
 
         logger.info("-->  relocation from remote to remote  complete");
@@ -155,7 +137,6 @@ public void testRemotePrimaryRelocation() throws Exception {
 
     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());

diff --git a/server/src/main/java/org/opensearch/index/shard/IndexShard.java b/server/src/main/java/org/opensearch/index/shard/IndexShard.java
@@ -2146,7 +2146,7 @@ public void waitForRemoteStoreSync(Runnable onProgress) throws IOException {
                         segmentUploadeCount = directory.getSegmentsUploadedToRemoteStore().size();
                     }
                     try {
-                        Thread.sleep(TimeValue.timeValueSeconds(30).seconds());
+                        Thread.sleep(TimeValue.timeValueSeconds(30).millis());
                     } catch (InterruptedException ie) {
                         throw new OpenSearchException("Interrupted waiting for completion of [{}]", ie);
                     }