[Backport 2.17] [Remote State] Upload incremental cluster state on master re-election (#15145) (#15792)

* [Remote State] Upload incremental cluster state on master re-election (#15145)

Signed-off-by: Shivansh Arora <[email protected]>
(cherry picked from commit cbdcbb7)

Author: shiv0408

release-notes/opensearch.release-notes-2.17.0.md

@@ -53,6 +53,7 @@
 - MultiTermQueries in keyword fields now default to `indexed` approach and gated behind cluster setting ([#15637](https://github.com/opensearch-project/OpenSearch/pull/15637))
 - Adding WithFieldName interface for QueryBuilders with fieldName ([#15705](https://github.com/opensearch-project/OpenSearch/pull/15705))
 - Static RemotePublication setting added, removed experimental feature flag ([#15478](https://github.com/opensearch-project/OpenSearch/pull/15478))
+-  [Remote Publication] Upload incremental cluster state on master re-election ([#15145](https://github.com/opensearch-project/OpenSearch/pull/15145))
 
 ### Dependencies
 - Bump `netty` from 4.1.111.Final to 4.1.112.Final ([#15081](https://github.com/opensearch-project/OpenSearch/pull/15081))

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

@@ -13,9 +13,12 @@
 import org.opensearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest;
 import org.opensearch.action.admin.cluster.state.ClusterStateResponse;
 import org.opensearch.client.Client;
+import org.opensearch.cluster.coordination.CoordinationState;
+import org.opensearch.cluster.coordination.PersistedStateRegistry;
 import org.opensearch.common.blobstore.BlobPath;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.discovery.DiscoveryStats;
+import org.opensearch.gateway.GatewayMetaState;
 import org.opensearch.gateway.remote.ClusterMetadataManifest.UploadedIndexMetadata;
 import org.opensearch.gateway.remote.model.RemoteClusterMetadataManifest;
 import org.opensearch.gateway.remote.model.RemoteRoutingTableBlobStore;
@@ -35,6 +38,7 @@
 import java.util.Base64;
 import java.util.Locale;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -62,15 +66,15 @@ public class RemoteStatePublicationIT extends RemoteStoreBaseIntegTestCase {
     private static final String REMOTE_STATE_PREFIX = "!";
     private static final String REMOTE_ROUTING_PREFIX = "_";
     private boolean isRemoteStateEnabled = true;
-    private String isRemotePublicationEnabled = "true";
+    private boolean isRemotePublicationEnabled = true;
     private boolean hasRemoteStateCharPrefix;
     private boolean hasRemoteRoutingCharPrefix;
 
     @Before
     public void setup() {
         asyncUploadMockFsRepo = false;
         isRemoteStateEnabled = true;
-        isRemotePublicationEnabled = "true";
+        isRemotePublicationEnabled = true;
         hasRemoteStateCharPrefix = randomBoolean();
         hasRemoteRoutingCharPrefix = randomBoolean();
     }
@@ -100,6 +104,7 @@ protected Settings nodeSettings(int nodeOrdinal) {
                 RemoteClusterStateService.REMOTE_CLUSTER_STATE_CHECKSUM_VALIDATION_MODE_SETTING.getKey(),
                 RemoteClusterStateService.RemoteClusterStateValidationMode.FAILURE
             )
+            .put(REMOTE_PUBLICATION_SETTING_KEY, isRemotePublicationEnabled)
             .put(
                 RemoteClusterStateService.CLUSTER_REMOTE_STORE_STATE_PATH_PREFIX.getKey(),
                 hasRemoteStateCharPrefix ? REMOTE_STATE_PREFIX : ""
@@ -220,6 +225,59 @@ public void testRemotePublicationDownloadStats() {
 
     }
 
+    public void testMasterReElectionUsesIncrementalUpload() throws IOException {
+        prepareCluster(3, 2, INDEX_NAME, 1, 1);
+        PersistedStateRegistry persistedStateRegistry = internalCluster().getClusterManagerNodeInstance(PersistedStateRegistry.class);
+        GatewayMetaState.RemotePersistedState remotePersistedState = (GatewayMetaState.RemotePersistedState) persistedStateRegistry
+            .getPersistedState(PersistedStateRegistry.PersistedStateType.REMOTE);
+        ClusterMetadataManifest manifest = remotePersistedState.getLastAcceptedManifest();
+        // force elected master to step down
+        internalCluster().stopCurrentClusterManagerNode();
+        ensureStableCluster(4);
+
+        persistedStateRegistry = internalCluster().getClusterManagerNodeInstance(PersistedStateRegistry.class);
+        CoordinationState.PersistedState persistedStateAfterElection = persistedStateRegistry.getPersistedState(
+            PersistedStateRegistry.PersistedStateType.REMOTE
+        );
+        ClusterMetadataManifest manifestAfterElection = persistedStateAfterElection.getLastAcceptedManifest();
+
+        // coordination metadata is updated, it will be unequal
+        assertNotEquals(manifest.getCoordinationMetadata(), manifestAfterElection.getCoordinationMetadata());
+        // all other attributes are not uploaded again and will be pointing to same files in manifest after new master is elected
+        assertEquals(manifest.getClusterUUID(), manifestAfterElection.getClusterUUID());
+        assertEquals(manifest.getIndices(), manifestAfterElection.getIndices());
+        assertEquals(manifest.getSettingsMetadata(), manifestAfterElection.getSettingsMetadata());
+        assertEquals(manifest.getTemplatesMetadata(), manifestAfterElection.getTemplatesMetadata());
+        assertEquals(manifest.getCustomMetadataMap(), manifestAfterElection.getCustomMetadataMap());
+        assertEquals(manifest.getRoutingTableVersion(), manifest.getRoutingTableVersion());
+        assertEquals(manifest.getIndicesRouting(), manifestAfterElection.getIndicesRouting());
+    }
+
+    public void testVotingConfigAreCommitted() throws ExecutionException, InterruptedException {
+        prepareCluster(3, 2, INDEX_NAME, 1, 2);
+        ensureStableCluster(5);
+        ensureGreen(INDEX_NAME);
+        // add two new nodes to the cluster, to update the voting config
+        internalCluster().startClusterManagerOnlyNodes(2, Settings.EMPTY);
+        ensureStableCluster(7);
+
+        internalCluster().getInstances(PersistedStateRegistry.class).forEach(persistedStateRegistry -> {
+            CoordinationState.PersistedState localState = persistedStateRegistry.getPersistedState(
+                PersistedStateRegistry.PersistedStateType.LOCAL
+            );
+            CoordinationState.PersistedState remoteState = persistedStateRegistry.getPersistedState(
+                PersistedStateRegistry.PersistedStateType.REMOTE
+            );
+            if (remoteState != null) {
+                assertEquals(
+                    localState.getLastAcceptedState().getLastCommittedConfiguration(),
+                    remoteState.getLastAcceptedState().getLastCommittedConfiguration()
+                );
+                assertEquals(5, remoteState.getLastAcceptedState().getLastCommittedConfiguration().getNodeIds().size());
+            }
+        });
+    }
+
     private void assertDataNodeDownloadStats(NodesStatsResponse nodesStatsResponse) {
         // assert cluster state stats for data node
         DiscoveryStats dataNodeDiscoveryStats = nodesStatsResponse.getNodes().get(0).getDiscoveryStats();

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

@@ -76,29 +76,4 @@ protected void verifyRestoredData(Map<String, Long> indexStats, String indexName
     protected void verifyRestoredData(Map<String, Long> indexStats, String indexName) throws Exception {
         verifyRestoredData(indexStats, indexName, true);
     }
-
-    public void prepareCluster(int numClusterManagerNodes, int numDataOnlyNodes, String indices, int replicaCount, int shardCount) {
-        prepareCluster(numClusterManagerNodes, numDataOnlyNodes, indices, replicaCount, shardCount, Settings.EMPTY);
-    }
-
-    public void prepareCluster(
-        int numClusterManagerNodes,
-        int numDataOnlyNodes,
-        String indices,
-        int replicaCount,
-        int shardCount,
-        Settings settings
-    ) {
-        prepareCluster(numClusterManagerNodes, numDataOnlyNodes, settings);
-        for (String index : indices.split(",")) {
-            createIndex(index, remoteStoreIndexSettings(replicaCount, shardCount));
-            ensureYellowAndNoInitializingShards(index);
-            ensureGreen(index);
-        }
-    }
-
-    public void prepareCluster(int numClusterManagerNodes, int numDataOnlyNodes, Settings settings) {
-        internalCluster().startClusterManagerOnlyNodes(numClusterManagerNodes, settings);
-        internalCluster().startDataOnlyNodes(numDataOnlyNodes, settings);
-    }
 }

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

@@ -351,13 +351,7 @@ protected void restore(boolean restoreAllShards, String... indices) {
     }
 
     protected void prepareCluster(int numClusterManagerNodes, int numDataOnlyNodes, String indices, int replicaCount, int shardCount) {
-        internalCluster().startClusterManagerOnlyNodes(numClusterManagerNodes);
-        internalCluster().startDataOnlyNodes(numDataOnlyNodes);
-        for (String index : indices.split(",")) {
-            createIndex(index, remoteStoreIndexSettings(replicaCount, shardCount));
-            ensureYellowAndNoInitializingShards(index);
-            ensureGreen(index);
-        }
+        prepareCluster(numClusterManagerNodes, numDataOnlyNodes, indices, replicaCount, shardCount, Settings.EMPTY);
     }
 
     protected void prepareCluster(
@@ -368,11 +362,16 @@ protected void prepareCluster(
         int shardCount,
         Settings settings
     ) {
-        internalCluster().startClusterManagerOnlyNodes(numClusterManagerNodes, settings);
-        internalCluster().startDataOnlyNodes(numDataOnlyNodes, settings);
+        prepareCluster(numClusterManagerNodes, numDataOnlyNodes, settings);
         for (String index : indices.split(",")) {
             createIndex(index, remoteStoreIndexSettings(replicaCount, shardCount));
+            ensureYellowAndNoInitializingShards(index);
             ensureGreen(index);
         }
     }
+
+    protected void prepareCluster(int numClusterManagerNodes, int numDataOnlyNodes, Settings settings) {
+        internalCluster().startClusterManagerOnlyNodes(numClusterManagerNodes, settings);
+        internalCluster().startDataOnlyNodes(numDataOnlyNodes, settings);
+    }
 }

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

@@ -40,6 +40,7 @@
 import org.opensearch.cluster.node.DiscoveryNode;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.util.io.IOUtils;
+import org.opensearch.gateway.remote.ClusterMetadataManifest;
 
 import java.io.Closeable;
 import java.io.IOException;
@@ -104,6 +105,7 @@ public CoordinationState(
             .getLastAcceptedConfiguration();
         this.publishVotes = new VoteCollection();
         this.isRemoteStateEnabled = isRemoteStoreClusterStateEnabled(settings);
+        // ToDo: revisit this check while making the setting dynamic
         this.isRemotePublicationEnabled = isRemoteStateEnabled
             && REMOTE_PUBLICATION_SETTING.get(settings)
             && localNode.isRemoteStatePublicationEnabled();
@@ -459,6 +461,9 @@ public PublishResponse handlePublishRequest(PublishRequest publishRequest) {
             clusterState.term()
         );
         persistedStateRegistry.getPersistedState(PersistedStateType.LOCAL).setLastAcceptedState(clusterState);
+        if (shouldUpdateRemotePersistedState(publishRequest)) {
+            updateRemotePersistedStateOnPublishRequest(publishRequest);
+        }
         assert getLastAcceptedState() == clusterState;
 
         return new PublishResponse(clusterState.term(), clusterState.version());
@@ -571,6 +576,9 @@ public void handleCommit(ApplyCommitRequest applyCommit) {
         );
 
         persistedStateRegistry.getPersistedState(PersistedStateType.LOCAL).markLastAcceptedStateAsCommitted();
+        if (shouldCommitRemotePersistedState()) {
+            persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE).markLastAcceptedStateAsCommitted();
+        }
         assert getLastCommittedConfiguration().equals(getLastAcceptedConfiguration());
     }
 
@@ -616,6 +624,33 @@ public void close() throws IOException {
         IOUtils.close(persistedStateRegistry);
     }
 
+    private boolean shouldUpdateRemotePersistedState(PublishRequest publishRequest) {
+        return persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE) != null
+            && publishRequest.getAcceptedState().getNodes().isLocalNodeElectedClusterManager() == false;
+    }
+
+    private void updateRemotePersistedStateOnPublishRequest(PublishRequest publishRequest) {
+        if (publishRequest instanceof RemoteStatePublishRequest) {
+            persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE).setLastAcceptedState(publishRequest.getAcceptedState());
+            persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE)
+                .setLastAcceptedManifest(((RemoteStatePublishRequest) publishRequest).getAcceptedManifest());
+        } else {
+            // We will end up here if PublishRequest was sent not using Remote Store even with remote persisted state on this node
+            persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE).setLastAcceptedState(null);
+            persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE).setLastAcceptedManifest(null);
+        }
+    }
+
+    private boolean shouldCommitRemotePersistedState() {
+        return persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE) != null
+            && persistedStateRegistry.getPersistedState(PersistedStateType.LOCAL)
+                .getLastAcceptedState()
+                .getNodes()
+                .isLocalNodeElectedClusterManager() == false
+            && persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE).getLastAcceptedState() != null
+            && persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE).getLastAcceptedManifest() != null;
+    }
+
     /**
      * Pluggable persistence layer for {@link CoordinationState}.
      *
@@ -653,6 +688,22 @@ public interface PersistedState extends Closeable {
          */
         PersistedStateStats getStats();
 
+        /**
+         * Returns the last accepted {@link ClusterMetadataManifest}.
+         *
+         * @return The last accepted {@link ClusterMetadataManifest}, or null if no manifest
+         *         has been accepted yet.
+         */
+        default ClusterMetadataManifest getLastAcceptedManifest() {
+            // return null by default, this method needs to be overridden wherever required
+            return null;
+        }
+
+        /**
+         * Sets the last accepted {@link ClusterMetadataManifest}.
+         */
+        default void setLastAcceptedManifest(ClusterMetadataManifest manifest) {}
+
         /**
          * Marks the last accepted cluster state as committed.
          * After a successful call to this method, {@link #getLastAcceptedState()} should return the last cluster state that was set,
@@ -661,14 +712,7 @@ public interface PersistedState extends Closeable {
          */
         default void markLastAcceptedStateAsCommitted() {
             final ClusterState lastAcceptedState = getLastAcceptedState();
-            Metadata.Builder metadataBuilder = null;
-            if (lastAcceptedState.getLastAcceptedConfiguration().equals(lastAcceptedState.getLastCommittedConfiguration()) == false) {
-                final CoordinationMetadata coordinationMetadata = CoordinationMetadata.builder(lastAcceptedState.coordinationMetadata())
-                    .lastCommittedConfiguration(lastAcceptedState.getLastAcceptedConfiguration())
-                    .build();
-                metadataBuilder = Metadata.builder(lastAcceptedState.metadata());
-                metadataBuilder.coordinationMetadata(coordinationMetadata);
-            }
+            Metadata.Builder metadataBuilder = commitVotingConfiguration(lastAcceptedState);
             // if we receive a commit from a Zen1 cluster-manager that has not recovered its state yet,
             // the cluster uuid might not been known yet.
             assert lastAcceptedState.metadata().clusterUUID().equals(Metadata.UNKNOWN_CLUSTER_UUID) == false
@@ -693,6 +737,18 @@ default void markLastAcceptedStateAsCommitted() {
             }
         }
 
+        default Metadata.Builder commitVotingConfiguration(ClusterState lastAcceptedState) {
+            Metadata.Builder metadataBuilder = null;
+            if (lastAcceptedState.getLastAcceptedConfiguration().equals(lastAcceptedState.getLastCommittedConfiguration()) == false) {
+                final CoordinationMetadata coordinationMetadata = CoordinationMetadata.builder(lastAcceptedState.coordinationMetadata())
+                    .lastCommittedConfiguration(lastAcceptedState.getLastAcceptedConfiguration())
+                    .build();
+                metadataBuilder = Metadata.builder(lastAcceptedState.metadata());
+                metadataBuilder.coordinationMetadata(coordinationMetadata);
+            }
+            return metadataBuilder;
+        }
+
         default void close() throws IOException {}
     }