Make Remote Publication a dynamic setting

Signed-off-by: Shivansh Arora <[email protected]>

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

@@ -38,6 +38,7 @@
 import org.opensearch.cluster.coordination.PersistedStateRegistry.PersistedStateType;
 import org.opensearch.cluster.metadata.Metadata;
 import org.opensearch.cluster.node.DiscoveryNode;
+import org.opensearch.common.settings.ClusterSettings;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.util.io.IOUtils;
 import org.opensearch.gateway.remote.ClusterMetadataManifest;
@@ -87,7 +88,8 @@ public CoordinationState(
         DiscoveryNode localNode,
         PersistedStateRegistry persistedStateRegistry,
         ElectionStrategy electionStrategy,
-        Settings settings
+        Settings settings,
+        ClusterSettings clusterSettings
     ) {
         this.localNode = localNode;
 
@@ -105,10 +107,10 @@ 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();
+            && localNode.isRemoteStatePublicationConfigured();
+        clusterSettings.addSettingsUpdateConsumer(REMOTE_PUBLICATION_SETTING, this::setRemotePublicationSetting);
     }
 
     public boolean isRemotePublicationEnabled() {
@@ -651,6 +653,15 @@ private boolean shouldCommitRemotePersistedState() {
             && persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE).getLastAcceptedManifest() != null;
     }
 
+    private void setRemotePublicationSetting(boolean remotePublicationSetting) {
+        if (remotePublicationSetting == false) {
+            this.isRemotePublicationEnabled = false;
+        } else {
+            this.isRemotePublicationEnabled = isRemoteStateEnabled && localNode.isRemoteStatePublicationConfigured();
+        }
+
+    }
+
     /**
      * Pluggable persistence layer for {@link CoordinationState}.
      *

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

@@ -187,8 +187,10 @@ public class Coordinator extends AbstractLifecycleComponent implements Discovery
     private Optional<CoordinatorPublication> currentPublication = Optional.empty();
     private final NodeHealthService nodeHealthService;
     private final PersistedStateRegistry persistedStateRegistry;
+    private final RemoteClusterStateService remoteClusterStateService;
     private final RemoteStoreNodeService remoteStoreNodeService;
     private NodeConnectionsService nodeConnectionsService;
+    private final ClusterSettings clusterSettings;
 
     /**
      * @param nodeName The name of the node, used to name the {@link java.util.concurrent.ExecutorService} of the {@link SeedHostsResolver}.
@@ -312,6 +314,8 @@ public Coordinator(
         this.persistedStateRegistry = persistedStateRegistry;
         this.localNodeCommissioned = true;
         this.remoteStoreNodeService = remoteStoreNodeService;
+        this.remoteClusterStateService = remoteClusterStateService;
+        this.clusterSettings = clusterSettings;
     }
 
     private ClusterFormationState getClusterFormationState() {
@@ -867,7 +871,9 @@ boolean publicationInProgress() {
     @Override
     protected void doStart() {
         synchronized (mutex) {
-            coordinationState.set(new CoordinationState(getLocalNode(), persistedStateRegistry, electionStrategy, settings));
+            coordinationState.set(
+                new CoordinationState(getLocalNode(), persistedStateRegistry, electionStrategy, settings, clusterSettings)
+            );
             peerFinder.setCurrentTerm(getCurrentTerm());
             configuredHostsResolver.start();
             final ClusterState lastAcceptedState = coordinationState.get().getLastAcceptedState();
@@ -912,9 +918,9 @@ public DiscoveryStats stats() {
                 stats.add(persistedStateRegistry.getPersistedState(stateType).getStats());
             }
         });
-        if (coordinationState.get().isRemotePublicationEnabled()) {
-            stats.add(publicationHandler.getFullDownloadStats());
-            stats.add(publicationHandler.getDiffDownloadStats());
+        if (remoteClusterStateService != null) {
+            stats.add(remoteClusterStateService.getFullDownloadStats());
+            stats.add(remoteClusterStateService.getDiffDownloadStats());
         }
         clusterStateStats.setPersistenceStats(stats);
         return new DiscoveryStats(new PendingClusterStateStats(0, 0, 0), publicationHandler.stats(), clusterStateStats);

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

@@ -513,10 +513,10 @@ private static void ensureRemoteClusterStateNodesCompatibility(DiscoveryNode joi
 
         assert existingNodes.isEmpty() == false;
         Optional<DiscoveryNode> remotePublicationNode = existingNodes.stream()
-            .filter(DiscoveryNode::isRemoteStatePublicationEnabled)
+            .filter(DiscoveryNode::isRemoteStatePublicationConfigured)
             .findFirst();
 
-        if (remotePublicationNode.isPresent() && joiningNode.isRemoteStatePublicationEnabled()) {
+        if (remotePublicationNode.isPresent() && joiningNode.isRemoteStatePublicationConfigured()) {
             ensureRepositoryCompatibility(joiningNode, remotePublicationNode.get(), REMOTE_CLUSTER_PUBLICATION_REPO_NAME_ATTRIBUTES);
         }
     }

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

@@ -178,14 +178,6 @@ public PublishClusterStateStats stats() {
         );
     }
 
-    public PersistedStateStats getFullDownloadStats() {
-        return remoteClusterStateService.getFullDownloadStats();
-    }
-
-    public PersistedStateStats getDiffDownloadStats() {
-        return remoteClusterStateService.getDiffDownloadStats();
-    }
-
     private PublishWithJoinResponse handleIncomingPublishRequest(BytesTransportRequest request) throws IOException {
         try (StreamInput in = CompressedStreamUtils.decompressBytes(request, namedWriteableRegistry)) {
             ClusterState incomingState;
@@ -356,7 +348,7 @@ public PublicationContext newPublicationContext(
     ) {
         if (isRemotePublicationEnabled == true) {
             if (allNodesRemotePublicationEnabled.get() == false) {
-                if (validateRemotePublicationOnAllNodes(clusterChangedEvent.state().nodes()) == true) {
+                if (validateRemotePublicationConfiguredOnAllNodes(clusterChangedEvent.state().nodes()) == true) {
                     allNodesRemotePublicationEnabled.set(true);
                 }
             }
@@ -374,11 +366,11 @@ public PublicationContext newPublicationContext(
         return publicationContext;
     }
 
-    private boolean validateRemotePublicationOnAllNodes(DiscoveryNodes discoveryNodes) {
+    private boolean validateRemotePublicationConfiguredOnAllNodes(DiscoveryNodes discoveryNodes) {
         assert ClusterMetadataManifest.getCodecForVersion(discoveryNodes.getMinNodeVersion()) >= ClusterMetadataManifest.CODEC_V0;
         for (DiscoveryNode node : discoveryNodes.getNodes().values()) {
             // if a node is non-remote then created local publication context
-            if (node.isRemoteStatePublicationEnabled() == false) {
+            if (node.isRemoteStatePublicationConfigured() == false) {
                 return false;
             }
         }

server/src/main/java/org/opensearch/cluster/node/DiscoveryNode.java

@@ -518,7 +518,7 @@ public boolean isRemoteStoreNode() {
      * Returns whether remote cluster state publication is enabled on this node
      * @return true if the node contains remote cluster state node attribute and remote routing table node attribute
      */
-    public boolean isRemoteStatePublicationEnabled() {
+    public boolean isRemoteStatePublicationConfigured() {
         return this.getAttributes()
             .keySet()
             .stream()

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

@@ -111,6 +111,7 @@
 import static org.opensearch.gateway.remote.model.RemoteTemplatesMetadata.TEMPLATES_METADATA;
 import static org.opensearch.gateway.remote.model.RemoteTransientSettingsMetadata.TRANSIENT_SETTING_METADATA;
 import static org.opensearch.gateway.remote.routingtable.RemoteIndexRoutingTable.INDEX_ROUTING_METADATA_PREFIX;
+import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.isRemoteRoutingTableEnabled;
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.isRemoteStoreClusterStateEnabled;
 
 /**
@@ -132,7 +133,7 @@ public class RemoteClusterStateService implements Closeable {
         REMOTE_PUBLICATION_SETTING_KEY,
         false,
         Property.NodeScope,
-        Property.Final
+        Property.Dynamic
     );
 
     /**
@@ -232,7 +233,7 @@ public static RemoteClusterStateValidationMode parseString(String mode) {
     private final String METADATA_UPDATE_LOG_STRING = "wrote metadata for [{}] indices and skipped [{}] unchanged "
         + "indices, coordination metadata updated : [{}], settings metadata updated : [{}], templates metadata "
         + "updated : [{}], custom metadata updated : [{}], indices routing updated : [{}]";
-    private final boolean isPublicationEnabled;
+    private boolean isPublicationEnabled;
     private final String remotePathPrefix;
 
     private final RemoteClusterStateCache remoteClusterStateCache;
@@ -273,9 +274,10 @@ public RemoteClusterStateService(
         this.remoteStateStats = new RemotePersistenceStats();
         this.namedWriteableRegistry = namedWriteableRegistry;
         this.indexMetadataUploadListeners = indexMetadataUploadListeners;
-        this.isPublicationEnabled = REMOTE_PUBLICATION_SETTING.get(settings)
+        this.isPublicationEnabled = clusterSettings.get(REMOTE_PUBLICATION_SETTING)
             && RemoteStoreNodeAttribute.isRemoteStoreClusterStateEnabled(settings)
             && RemoteStoreNodeAttribute.isRemoteRoutingTableEnabled(settings);
+        clusterSettings.addSettingsUpdateConsumer(REMOTE_PUBLICATION_SETTING, this::setRemotePublicationSetting);
         this.remotePathPrefix = CLUSTER_REMOTE_STORE_STATE_PATH_PREFIX.get(settings);
         this.remoteRoutingTableService = RemoteRoutingTableServiceFactory.getService(
             repositoriesService,
@@ -1109,6 +1111,14 @@ private void setChecksumValidationMode(RemoteClusterStateValidationMode remoteCl
         this.remoteClusterStateValidationMode = remoteClusterStateValidationMode;
     }
 
+    private void setRemotePublicationSetting(boolean remotePublicationSetting) {
+        if (remotePublicationSetting == false) {
+            this.isPublicationEnabled = false;
+        } else {
+            this.isPublicationEnabled = isRemoteStoreClusterStateEnabled(settings) && isRemoteRoutingTableEnabled(settings);
+        }
+    }
+
     // Package private for unit test
     RemoteRoutingTableService getRemoteRoutingTableService() {
         return this.remoteRoutingTableService;

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

@@ -21,8 +21,8 @@ public class RemotePersistenceStats {
     RemoteDownloadStats remoteDiffDownloadStats;
     RemoteDownloadStats remoteFullDownloadStats;
 
-    final String FULL_DOWNLOAD_STATS = "remote_full_download";
-    final String DIFF_DOWNLOAD_STATS = "remote_diff_download";
+    public static final String FULL_DOWNLOAD_STATS = "remote_full_download";
+    public static final String DIFF_DOWNLOAD_STATS = "remote_diff_download";
 
     public RemotePersistenceStats() {
         remoteUploadStats = new RemoteUploadStats();

server/src/test/java/org/opensearch/cluster/coordination/CoordinationStateTests.java

@@ -1283,7 +1283,7 @@ public static CoordinationState createCoordinationState(
         DiscoveryNode localNode,
         Settings settings
     ) {
-        return new CoordinationState(localNode, persistedStateRegistry, ElectionStrategy.DEFAULT_INSTANCE, settings);
+        return new CoordinationState(localNode, persistedStateRegistry, ElectionStrategy.DEFAULT_INSTANCE, settings, null);
     }
 
     public static ClusterState clusterState(

server/src/test/java/org/opensearch/cluster/coordination/PreVoteCollectorTests.java

@@ -302,7 +302,8 @@ public void testPrevotingIndicatesElectionSuccess() {
             localNode,
             persistedStateRegistry,
             ElectionStrategy.DEFAULT_INSTANCE,
-            Settings.EMPTY
+            Settings.EMPTY,
+            null
         );
 
         final long newTerm = randomLongBetween(currentTerm + 1, Long.MAX_VALUE);

server/src/test/java/org/opensearch/cluster/coordination/PublicationTests.java

@@ -94,7 +94,13 @@ class MockNode {
             );
             PersistedStateRegistry persistedStateRegistry = persistedStateRegistry();
             persistedStateRegistry.addPersistedState(PersistedStateType.LOCAL, new InMemoryPersistedState(0L, initialState));
-            coordinationState = new CoordinationState(localNode, persistedStateRegistry, ElectionStrategy.DEFAULT_INSTANCE, Settings.EMPTY);
+            coordinationState = new CoordinationState(
+                localNode,
+                persistedStateRegistry,
+                ElectionStrategy.DEFAULT_INSTANCE,
+                Settings.EMPTY,
+                null
+            );
         }
 
         final DiscoveryNode localNode;

test/framework/src/main/java/org/opensearch/cluster/coordination/CoordinationStateTestCluster.java

@@ -150,7 +150,7 @@ static class ClusterNode {
             persistedStateRegistry.addPersistedState(PersistedStateType.LOCAL, persistedState);
 
             this.electionStrategy = electionStrategy;
-            state = new CoordinationState(localNode, persistedStateRegistry, electionStrategy, Settings.EMPTY);
+            state = new CoordinationState(localNode, persistedStateRegistry, electionStrategy, Settings.EMPTY, null);
         }
 
         void reboot() {
@@ -189,7 +189,7 @@ void reboot() {
                 localNode.getVersion()
             );
 
-            state = new CoordinationState(localNode, persistedStateRegistry, electionStrategy, Settings.EMPTY);
+            state = new CoordinationState(localNode, persistedStateRegistry, electionStrategy, Settings.EMPTY, null);
         }
 
         void setInitialState(CoordinationMetadata.VotingConfiguration initialConfig, long initialValue) {