[Backport 2.x] Restore ClusterState version during remote state restore (#10853)

CHANGELOG.md

@@ -17,6 +17,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Per request phase latency ([#10351](https://github.com/opensearch-project/OpenSearch/issues/10351))
 - Added cluster setting cluster.restrict.index.replication_type to restrict setting of index setting replication type ([#10866](https://github.com/opensearch-project/OpenSearch/pull/10866))
 - Add cluster state stats ([#10670](https://github.com/opensearch-project/OpenSearch/pull/10670))
+- [Remote cluster state] Restore cluster state version during remote state auto restore ([#10853](https://github.com/opensearch-project/OpenSearch/pull/10853))
 
 ### Dependencies
 - Bumps jetty version to 9.4.52.v20230823 to fix GMS-2023-1857 ([#9822](https://github.com/opensearch-project/OpenSearch/pull/9822))

server/src/internalClusterTest/java/org/opensearch/gateway/remote/RemoteClusterStateServiceIT.java

@@ -90,10 +90,10 @@ public void testFullClusterRestoreStaleDelete() throws Exception {
 
         assertEquals(10, repository.blobStore().blobContainer(baseMetadataPath.add("manifest")).listBlobsByPrefix("manifest").size());
 
-        Map<String, IndexMetadata> indexMetadataMap = remoteClusterStateService.getLatestMetadata(
+        Map<String, IndexMetadata> indexMetadataMap = remoteClusterStateService.getLatestClusterState(
             cluster().getClusterName(),
             getClusterState().metadata().clusterUUID()
-        ).getIndices();
+        ).getMetadata().getIndices();
         assertEquals(0, indexMetadataMap.values().stream().findFirst().get().getNumberOfReplicas());
         assertEquals(shardCount, indexMetadataMap.values().stream().findFirst().get().getNumberOfShards());
     }

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

@@ -30,6 +30,7 @@
 import java.nio.file.Path;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ExecutionException;
@@ -85,16 +86,25 @@ public void testFullClusterRestore() throws Exception {
         // Step - 1 index some data to generate files in remote directory
         Map<String, Long> indexStats = initialTestSetup(shardCount, replicaCount, dataNodeCount, 1);
         String prevClusterUUID = clusterService().state().metadata().clusterUUID();
+        long prevClusterStateVersion = clusterService().state().version();
 
         // Step - 2 Replace all nodes in the cluster with new nodes. This ensures new cluster state doesn't have previous index metadata
         resetCluster(dataNodeCount, clusterManagerNodeCount);
 
         String newClusterUUID = clusterService().state().metadata().clusterUUID();
         assert !Objects.equals(newClusterUUID, prevClusterUUID) : "cluster restart not successful. cluster uuid is same";
 
-        // Step - 3 Trigger full cluster restore and validate
+        // Step - 3 validate cluster state restored
+        long newClusterStateVersion = clusterService().state().version();
+        assert prevClusterStateVersion < newClusterStateVersion : String.format(
+            Locale.ROOT,
+            "ClusterState version is not restored. previousClusterVersion: [%s] is greater than current [%s]",
+            prevClusterStateVersion,
+            newClusterStateVersion
+        );
         validateMetadata(List.of(INDEX_NAME));
         verifyRedIndicesAndTriggerRestore(indexStats, INDEX_NAME, true);
+
     }
 
     /**
@@ -121,6 +131,7 @@ public void testFullClusterRestoreDoesntFailWithConflictingLocalState() throws E
         // index some data to generate files in remote directory
         Map<String, Long> indexStats = initialTestSetup(shardCount, replicaCount, dataNodeCount, 1);
         String prevClusterUUID = clusterService().state().metadata().clusterUUID();
+        long prevClusterStateVersion = clusterService().state().version();
 
         // stop all nodes
         internalCluster().stopAllNodes();
@@ -156,6 +167,14 @@ public Settings onNodeStopped(String nodeName) {
         newClusterUUID = clusterService().state().metadata().clusterUUID();
         assert !Objects.equals(newClusterUUID, ClusterState.UNKNOWN_UUID) : "cluster restart not successful. cluster uuid is still unknown";
         assert !Objects.equals(newClusterUUID, prevClusterUUID) : "cluster restart not successful. cluster uuid is same";
+
+        long newClusterStateVersion = clusterService().state().version();
+        assert prevClusterStateVersion < newClusterStateVersion : String.format(
+            Locale.ROOT,
+            "ClusterState version is not restored. previousClusterVersion: [%s] is greater than current [%s]",
+            prevClusterStateVersion,
+            newClusterStateVersion
+        );
         validateMetadata(List.of(INDEX_NAME));
 
         // start data nodes to trigger index data recovery
@@ -180,14 +199,22 @@ public void testFullClusterRestoreMultipleIndices() throws Exception {
         updateIndexBlock(true, secondIndexName);
 
         String prevClusterUUID = clusterService().state().metadata().clusterUUID();
+        long prevClusterStateVersion = clusterService().state().version();
 
         // Step - 2 Replace all nodes in the cluster with new nodes. This ensures new cluster state doesn't have previous index metadata
         resetCluster(dataNodeCount, clusterManagerNodeCount);
 
         String newClusterUUID = clusterService().state().metadata().clusterUUID();
         assert !Objects.equals(newClusterUUID, prevClusterUUID) : "cluster restart not successful. cluster uuid is same";
 
-        // Step - 3 Trigger full cluster restore
+        // Step - 3 validate cluster state restored
+        long newClusterStateVersion = clusterService().state().version();
+        assert prevClusterStateVersion < newClusterStateVersion : String.format(
+            Locale.ROOT,
+            "ClusterState version is not restored. previousClusterVersion: [%s] is greater than current [%s]",
+            prevClusterStateVersion,
+            newClusterStateVersion
+        );
         validateMetadata(List.of(INDEX_NAME, secondIndexName));
         verifyRedIndicesAndTriggerRestore(indexStats, INDEX_NAME, false);
         verifyRedIndicesAndTriggerRestore(indexStats2, secondIndexName, false);
@@ -239,6 +266,7 @@ public void testRemoteStateFullRestart() throws Exception {
 
         Map<String, Long> indexStats = initialTestSetup(shardCount, replicaCount, dataNodeCount, clusterManagerNodeCount);
         String prevClusterUUID = clusterService().state().metadata().clusterUUID();
+        long prevClusterStateVersion = clusterService().state().version();
         // Delete index metadata file in remote
         try {
             Files.move(
@@ -257,6 +285,14 @@ public void testRemoteStateFullRestart() throws Exception {
         ensureGreen(INDEX_NAME);
         String newClusterUUID = clusterService().state().metadata().clusterUUID();
         assert Objects.equals(newClusterUUID, prevClusterUUID) : "Full restart not successful. cluster uuid has changed";
+
+        long newClusterStateVersion = clusterService().state().version();
+        assert prevClusterStateVersion < newClusterStateVersion : String.format(
+            Locale.ROOT,
+            "ClusterState version is not restored. previousClusterVersion: [%s] is greater than current [%s]",
+            prevClusterStateVersion,
+            newClusterStateVersion
+        );
         validateCurrentMetadata();
         verifyRedIndicesAndTriggerRestore(indexStats, INDEX_NAME, true);
     }
@@ -309,6 +345,7 @@ public void testFullClusterRestoreGlobalMetadata() throws Exception {
         // Step - 1 index some data to generate files in remote directory
         Map<String, Long> indexStats = initialTestSetup(shardCount, replicaCount, dataNodeCount, 1);
         String prevClusterUUID = clusterService().state().metadata().clusterUUID();
+        long prevClusterStateVersion = clusterService().state().version();
 
         // Create global metadata - register a custom repo
         Path repoPath = registerCustomRepository();
@@ -328,8 +365,16 @@ public void testFullClusterRestoreGlobalMetadata() throws Exception {
         String newClusterUUID = clusterService().state().metadata().clusterUUID();
         assert !Objects.equals(newClusterUUID, prevClusterUUID) : "cluster restart not successful. cluster uuid is same";
 
-        // Step - 3 Trigger full cluster restore and validate
-        // validateCurrentMetadata();
+        // Step - 3 validate cluster state restored
+        long newClusterStateVersion = clusterService().state().version();
+        assert prevClusterStateVersion < newClusterStateVersion : String.format(
+            Locale.ROOT,
+            "ClusterState version is not restored. previousClusterVersion: [%s] is greater than current [%s]",
+            prevClusterStateVersion,
+            newClusterStateVersion
+        );
+
+        validateCurrentMetadata();
         assertEquals(Integer.valueOf(34), SETTING_CLUSTER_MAX_SHARDS_PER_NODE.get(clusterService().state().metadata().settings()));
         assertEquals(true, SETTING_READ_ONLY_SETTING.get(clusterService().state().metadata().settings()));
         assertTrue(clusterService().state().blocks().hasGlobalBlock(CLUSTER_READ_ONLY_BLOCK));

server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java

@@ -767,16 +767,16 @@ private IndexMetadata getIndexMetadata(String clusterName, String clusterUUID, U
     }
 
     /**
-     * Fetch latest metadata from remote cluster state including global metadata and index metadata
+     * Fetch latest ClusterState from remote, including global metadata, index metadata and cluster state version
      *
      * @param clusterUUID uuid of cluster state to refer to in remote
      * @param clusterName name of the cluster
      * @return {@link IndexMetadata}
      */
-    public Metadata getLatestMetadata(String clusterName, String clusterUUID) {
+    public ClusterState getLatestClusterState(String clusterName, String clusterUUID) {
         start();
         Optional<ClusterMetadataManifest> clusterMetadataManifest = getLatestClusterMetadataManifest(clusterName, clusterUUID);
-        if (!clusterMetadataManifest.isPresent()) {
+        if (clusterMetadataManifest.isEmpty()) {
             throw new IllegalStateException(
                 String.format(Locale.ROOT, "Latest cluster metadata manifest is not present for the provided clusterUUID: %s", clusterUUID)
             );
@@ -790,7 +790,10 @@ public Metadata getLatestMetadata(String clusterName, String clusterUUID) {
         Map<String, IndexMetadata> indexMetadataMap = new HashMap<>();
         indices.values().forEach(indexMetadata -> { indexMetadataMap.put(indexMetadata.getIndex().getName(), indexMetadata); });
 
-        return Metadata.builder(globalMetadata).indices(indexMetadataMap).build();
+        return ClusterState.builder(ClusterState.EMPTY_STATE)
+            .version(clusterMetadataManifest.get().getStateVersion())
+            .metadata(Metadata.builder(globalMetadata).indices(indexMetadataMap).build())
+            .build();
     }
 
     private Metadata getGlobalMetadata(String clusterName, String clusterUUID, ClusterMetadataManifest clusterMetadataManifest) {

server/src/main/java/org/opensearch/index/recovery/RemoteStoreRestoreService.java

@@ -138,7 +138,7 @@ public RemoteRestoreResult restore(
         String[] indexNames
     ) {
         Map<String, Tuple<Boolean, IndexMetadata>> indexMetadataMap = new HashMap<>();
-        Metadata remoteMetadata = null;
+        ClusterState remoteState = null;
         boolean metadataFromRemoteStore = (restoreClusterUUID == null
             || restoreClusterUUID.isEmpty()
             || restoreClusterUUID.isBlank()) == false;
@@ -150,8 +150,8 @@ public RemoteRestoreResult restore(
                     throw new IllegalArgumentException("clusterUUID to restore from should be different from current cluster UUID");
                 }
                 logger.info("Restoring cluster state from remote store from cluster UUID : [{}]", restoreClusterUUID);
-                remoteMetadata = remoteClusterStateService.getLatestMetadata(currentState.getClusterName().value(), restoreClusterUUID);
-                remoteMetadata.getIndices().values().forEach(indexMetadata -> {
+                remoteState = remoteClusterStateService.getLatestClusterState(currentState.getClusterName().value(), restoreClusterUUID);
+                remoteState.getMetadata().getIndices().values().forEach(indexMetadata -> {
                     indexMetadataMap.put(indexMetadata.getIndex().getName(), new Tuple<>(true, indexMetadata));
                 });
             } catch (Exception e) {
@@ -177,7 +177,7 @@ public RemoteRestoreResult restore(
                 }
             }
         }
-        return executeRestore(currentState, indexMetadataMap, restoreAllShards, remoteMetadata);
+        return executeRestore(currentState, indexMetadataMap, restoreAllShards, remoteState);
     }
 
     /**
@@ -191,7 +191,7 @@ private RemoteRestoreResult executeRestore(
         ClusterState currentState,
         Map<String, Tuple<Boolean, IndexMetadata>> indexMetadataMap,
         boolean restoreAllShards,
-        Metadata remoteMetadata
+        ClusterState remoteState
     ) {
         final String restoreUUID = UUIDs.randomBase64UUID();
         List<String> indicesToBeRestored = new ArrayList<>();
@@ -241,8 +241,11 @@ private RemoteRestoreResult executeRestore(
             totalShards += updatedIndexMetadata.getNumberOfShards();
         }
 
-        if (remoteMetadata != null) {
-            restoreGlobalMetadata(mdBuilder, remoteMetadata);
+        if (remoteState != null) {
+            restoreGlobalMetadata(mdBuilder, remoteState.getMetadata());
+            // Restore ClusterState version
+            logger.info("Restoring ClusterState with Remote State version [{}]", remoteState.version());
+            builder.version(remoteState.version());
         }
 
         RestoreInfo restoreInfo = new RestoreInfo("remote_store", indicesToBeRestored, totalShards, totalShards);

server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateServiceTests.java

@@ -665,7 +665,8 @@ public void testReadLatestMetadataManifestSuccessButNoIndexMetadata() throws IOE
 
         remoteClusterStateService.start();
         assertEquals(
-            remoteClusterStateService.getLatestMetadata(clusterState.getClusterName().value(), clusterState.metadata().clusterUUID())
+            remoteClusterStateService.getLatestClusterState(clusterState.getClusterName().value(), clusterState.metadata().clusterUUID())
+                .getMetadata()
                 .getIndices()
                 .size(),
             0
@@ -694,8 +695,10 @@ public void testReadLatestMetadataManifestSuccessButIndexMetadataFetchIOExceptio
         remoteClusterStateService.start();
         Exception e = assertThrows(
             IllegalStateException.class,
-            () -> remoteClusterStateService.getLatestMetadata(clusterState.getClusterName().value(), clusterState.metadata().clusterUUID())
-                .getIndices()
+            () -> remoteClusterStateService.getLatestClusterState(
+                clusterState.getClusterName().value(),
+                clusterState.metadata().clusterUUID()
+            ).getMetadata().getIndices()
         );
         assertEquals(e.getMessage(), "Error while downloading IndexMetadata - " + uploadedIndexMetadata.getUploadedFilename());
     }
@@ -740,10 +743,11 @@ public void testReadGlobalMetadata() throws IOException {
         final ClusterState clusterState = generateClusterStateWithGlobalMetadata().nodes(nodesWithLocalNodeClusterManager()).build();
         remoteClusterStateService.start();
 
+        long prevClusterStateVersion = 13L;
         final ClusterMetadataManifest expectedManifest = ClusterMetadataManifest.builder()
             .indices(List.of())
             .clusterTerm(1L)
-            .stateVersion(1L)
+            .stateVersion(prevClusterStateVersion)
             .stateUUID("state-uuid")
             .clusterUUID("cluster-uuid")
             .codecVersion(MANIFEST_CURRENT_CODEC_VERSION)
@@ -756,12 +760,20 @@ public void testReadGlobalMetadata() throws IOException {
         Metadata expactedMetadata = Metadata.builder().persistentSettings(Settings.builder().put("readonly", true).build()).build();
         mockBlobContainerForGlobalMetadata(mockBlobStoreObjects(), expectedManifest, expactedMetadata);
 
-        Metadata metadata = remoteClusterStateService.getLatestMetadata(
+        ClusterState newClusterState = remoteClusterStateService.getLatestClusterState(
             clusterState.getClusterName().value(),
             clusterState.metadata().clusterUUID()
         );
 
-        assertTrue(Metadata.isGlobalStateEquals(metadata, expactedMetadata));
+        assertTrue(Metadata.isGlobalStateEquals(newClusterState.getMetadata(), expactedMetadata));
+
+        long newClusterStateVersion = newClusterState.getVersion();
+        assert prevClusterStateVersion == newClusterStateVersion : String.format(
+            Locale.ROOT,
+            "ClusterState version is not restored. previousClusterVersion: [%s] is not equal to current [%s]",
+            prevClusterStateVersion,
+            newClusterStateVersion
+        );
     }
 
     public void testReadGlobalMetadataIOException() throws IOException {
@@ -793,7 +805,10 @@ public void testReadGlobalMetadataIOException() throws IOException {
         remoteClusterStateService.start();
         Exception e = assertThrows(
             IllegalStateException.class,
-            () -> remoteClusterStateService.getLatestMetadata(clusterState.getClusterName().value(), clusterState.metadata().clusterUUID())
+            () -> remoteClusterStateService.getLatestClusterState(
+                clusterState.getClusterName().value(),
+                clusterState.metadata().clusterUUID()
+            )
         );
         assertEquals(e.getMessage(), "Error while downloading Global Metadata - " + globalIndexMetadataName);
     }
@@ -824,16 +839,15 @@ public void testReadLatestIndexMetadataSuccess() throws IOException {
             .nodeId("nodeA")
             .opensearchVersion(VersionUtils.randomOpenSearchVersion(random()))
             .previousClusterUUID("prev-cluster-uuid")
-            .globalMetadataFileName("global-metadata-file")
             .codecVersion(ClusterMetadataManifest.CODEC_V0)
             .build();
 
         mockBlobContainer(mockBlobStoreObjects(), expectedManifest, Map.of(index.getUUID(), indexMetadata));
 
-        Map<String, IndexMetadata> indexMetadataMap = remoteClusterStateService.getLatestMetadata(
+        Map<String, IndexMetadata> indexMetadataMap = remoteClusterStateService.getLatestClusterState(
             clusterState.getClusterName().value(),
             clusterState.metadata().clusterUUID()
-        ).getIndices();
+        ).getMetadata().getIndices();
 
         assertEquals(indexMetadataMap.size(), 1);
         assertEquals(indexMetadataMap.get(index.getName()).getIndex().getName(), index.getName());