Make remote publication static setting

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

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

@@ -34,12 +34,12 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 
-import static org.opensearch.common.util.FeatureFlags.REMOTE_PUBLICATION_EXPERIMENTAL;
 import static org.opensearch.gateway.remote.RemoteClusterStateCleanupManager.CLUSTER_STATE_CLEANUP_INTERVAL_DEFAULT;
 import static org.opensearch.gateway.remote.RemoteClusterStateCleanupManager.REMOTE_CLUSTER_STATE_CLEANUP_INTERVAL_SETTING;
 import static org.opensearch.gateway.remote.RemoteClusterStateCleanupManager.RETAINED_MANIFESTS;
 import static org.opensearch.gateway.remote.RemoteClusterStateCleanupManager.SKIP_CLEANUP_STATE_CHANGES;
 import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_PUBLICATION_SETTING_KEY;
 import static org.opensearch.gateway.remote.routingtable.RemoteIndexRoutingTable.INDEX_ROUTING_TABLE;
 import static org.opensearch.indices.IndicesService.CLUSTER_DEFAULT_INDEX_REFRESH_INTERVAL_SETTING;
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY;
@@ -187,7 +187,7 @@ public void testRemoteCleanupDeleteStaleIndexRoutingFiles() throws Exception {
             RemoteStoreEnums.PathType.HASHED_PREFIX.toString()
         )
             .put("node.attr." + REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY, REMOTE_ROUTING_TABLE_REPO)
-            .put(REMOTE_PUBLICATION_EXPERIMENTAL, true);
+            .put(REMOTE_PUBLICATION_SETTING_KEY, true);
 
         int shardCount = randomIntBetween(1, 2);
         int replicaCount = 1;

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

@@ -36,8 +36,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import static org.opensearch.common.util.FeatureFlags.REMOTE_PUBLICATION_EXPERIMENTAL;
 import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_PUBLICATION_SETTING_KEY;
 import static org.opensearch.gateway.remote.routingtable.RemoteIndexRoutingTable.INDEX_ROUTING_TABLE;
 import static org.opensearch.indices.IndicesService.CLUSTER_DEFAULT_INDEX_REFRESH_INTERVAL_SETTING;
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY;
@@ -65,7 +65,7 @@ protected Settings nodeSettings(int nodeOrdinal) {
                 RemoteStoreEnums.PathType.HASHED_PREFIX.toString()
             )
             .put("node.attr." + REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY, REMOTE_ROUTING_TABLE_REPO)
-            .put(REMOTE_PUBLICATION_EXPERIMENTAL, true)
+            .put(REMOTE_PUBLICATION_SETTING_KEY, true)
             .build();
     }
 

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

@@ -13,7 +13,6 @@
 import org.opensearch.client.Client;
 import org.opensearch.common.blobstore.BlobPath;
 import org.opensearch.common.settings.Settings;
-import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.gateway.remote.model.RemoteClusterMetadataManifest;
 import org.opensearch.indices.recovery.RecoverySettings;
 import org.opensearch.remotestore.RemoteStoreBaseIntegTestCase;
@@ -34,6 +33,7 @@
 
 import static org.opensearch.gateway.remote.RemoteClusterStateAttributesManager.DISCOVERY_NODES;
 import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_PUBLICATION_SETTING_KEY;
 import static org.opensearch.gateway.remote.RemoteClusterStateUtils.DELIMITER;
 import static org.opensearch.gateway.remote.model.RemoteClusterBlocks.CLUSTER_BLOCKS;
 import static org.opensearch.gateway.remote.model.RemoteCoordinationMetadata.COORDINATION_METADATA;
@@ -60,14 +60,6 @@ public void setup() {
         isRemotePublicationEnabled = "true";
     }
 
-    @Override
-    protected Settings featureFlagSettings() {
-        return Settings.builder()
-            .put(super.featureFlagSettings())
-            .put(FeatureFlags.REMOTE_PUBLICATION_EXPERIMENTAL, isRemotePublicationEnabled)
-            .build();
-    }
-
     @Override
     protected Settings nodeSettings(int nodeOrdinal) {
         String routingTableRepoName = "remote-routing-repo";
@@ -84,6 +76,7 @@ protected Settings nodeSettings(int nodeOrdinal) {
         return Settings.builder()
             .put(super.nodeSettings(nodeOrdinal))
             .put(REMOTE_CLUSTER_STATE_ENABLED_SETTING.getKey(), isRemoteStateEnabled)
+            .put(REMOTE_PUBLICATION_SETTING_KEY, isRemotePublicationEnabled)
             .put("node.attr." + REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY, routingTableRepoName)
             .put(routingTableRepoTypeAttributeKey, ReloadableFsRepository.TYPE)
             .put(routingTableRepoSettingsAttributeKeyPrefix + "location", segmentRepoPath)

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

@@ -39,7 +39,6 @@
 import org.opensearch.cluster.metadata.Metadata;
 import org.opensearch.cluster.node.DiscoveryNode;
 import org.opensearch.common.settings.Settings;
-import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.common.util.io.IOUtils;
 
 import java.io.Closeable;
@@ -53,7 +52,7 @@
 import java.util.Set;
 
 import static org.opensearch.cluster.coordination.Coordinator.ZEN1_BWC_TERM;
-import static org.opensearch.common.util.FeatureFlags.REMOTE_PUBLICATION_EXPERIMENTAL;
+import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_PUBLICATION_SETTING;
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.isRemoteStoreClusterStateEnabled;
 
 /**
@@ -81,7 +80,8 @@ public class CoordinationState {
     private VotingConfiguration lastPublishedConfiguration;
     private VoteCollection publishVotes;
     private final boolean isRemoteStateEnabled;
-    private final boolean isRemotePublicationEnabled;
+    private boolean isRemotePublicationEnabled;
+    private boolean sendFullStateAfterRemotePublicationDisable;
 
     public CoordinationState(
         DiscoveryNode localNode,
@@ -106,14 +106,22 @@ public CoordinationState(
         this.publishVotes = new VoteCollection();
         this.isRemoteStateEnabled = isRemoteStoreClusterStateEnabled(settings);
         this.isRemotePublicationEnabled = isRemoteStateEnabled
-            && FeatureFlags.isEnabled(REMOTE_PUBLICATION_EXPERIMENTAL)
+            && REMOTE_PUBLICATION_SETTING.get(settings)
             && localNode.isRemoteStatePublicationEnabled();
+        this.sendFullStateAfterRemotePublicationDisable = false;
     }
 
     public boolean isRemotePublicationEnabled() {
         return isRemotePublicationEnabled;
     }
 
+    // after the remote publication is disabled, we will send full state to all the nodes once
+    public boolean shouldPublishFullStateAndToggle() {
+        boolean shouldPublishFullState = this.sendFullStateAfterRemotePublicationDisable;
+        this.sendFullStateAfterRemotePublicationDisable = false;
+        return shouldPublishFullState;
+    }
+
     public long getCurrentTerm() {
         return persistedStateRegistry.getPersistedState(PersistedStateType.LOCAL).getCurrentTerm();
     }

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

@@ -187,6 +187,7 @@ public class Coordinator extends AbstractLifecycleComponent implements Discovery
     private final NodeHealthService nodeHealthService;
     private final PersistedStateRegistry persistedStateRegistry;
     private final RemoteStoreNodeService remoteStoreNodeService;
+    private final ClusterSettings clusterSettings;
 
     /**
      * @param nodeName The name of the node, used to name the {@link java.util.concurrent.ExecutorService} of the {@link SeedHostsResolver}.
@@ -310,6 +311,7 @@ public Coordinator(
         this.persistedStateRegistry = persistedStateRegistry;
         this.localNodeCommissioned = true;
         this.remoteStoreNodeService = remoteStoreNodeService;
+        this.clusterSettings = clusterSettings;
     }
 
     private ClusterFormationState getClusterFormationState() {
@@ -1335,7 +1337,8 @@ assert getLocalNode().equals(clusterState.getNodes().get(getLocalNode().getId())
                 final PublicationTransportHandler.PublicationContext publicationContext = publicationHandler.newPublicationContext(
                     clusterChangedEvent,
                     coordinationState.get().isRemotePublicationEnabled(),
-                    persistedStateRegistry
+                    persistedStateRegistry,
+                    coordinationState.get().shouldPublishFullStateAndToggle()
                 );
 
                 final PublishRequest publishRequest = coordinationState.get().handleClientValue(clusterState);

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

@@ -330,12 +330,14 @@ private PublishWithJoinResponse acceptRemoteStateOnLocalNode(RemotePublishReques
     public PublicationContext newPublicationContext(
         ClusterChangedEvent clusterChangedEvent,
         boolean isRemotePublicationEnabled,
-        PersistedStateRegistry persistedStateRegistry
+        PersistedStateRegistry persistedStateRegistry,
+        boolean shouldSendFullState
     ) {
         final PublicationContext publicationContext = new PublicationContext(
             clusterChangedEvent,
             isRemotePublicationEnabled,
-            persistedStateRegistry
+            persistedStateRegistry,
+            shouldSendFullState
         );
 
         // Build the serializations we expect to need now, early in the process, so that an error during serialization fails the publication
@@ -397,12 +399,13 @@ public class PublicationContext {
         PublicationContext(
             ClusterChangedEvent clusterChangedEvent,
             boolean isRemotePublicationEnabled,
-            PersistedStateRegistry persistedStateRegistry
+            PersistedStateRegistry persistedStateRegistry,
+            boolean shouldSendFullState
         ) {
             discoveryNodes = clusterChangedEvent.state().nodes();
             newState = clusterChangedEvent.state();
             previousState = clusterChangedEvent.previousState();
-            sendFullVersion = previousState.getBlocks().disableStatePersistence();
+            sendFullVersion = previousState.getBlocks().disableStatePersistence() || shouldSendFullState;
             sendRemoteState = isRemotePublicationEnabled;
             this.persistedStateRegistry = persistedStateRegistry;
         }

server/src/main/java/org/opensearch/common/settings/ClusterSettings.java

@@ -726,6 +726,7 @@ public void apply(Settings value, Settings current, Settings previous) {
                 // Remote cluster state settings
                 RemoteClusterStateCleanupManager.REMOTE_CLUSTER_STATE_CLEANUP_INTERVAL_SETTING,
                 RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING,
+                RemoteClusterStateService.REMOTE_PUBLICATION_SETTING,
                 INDEX_METADATA_UPLOAD_TIMEOUT_SETTING,
                 GLOBAL_METADATA_UPLOAD_TIMEOUT_SETTING,
                 METADATA_MANIFEST_UPLOAD_TIMEOUT_SETTING,

server/src/main/java/org/opensearch/common/settings/FeatureFlagSettings.java

@@ -37,7 +37,6 @@ protected FeatureFlagSettings(
         FeatureFlags.TIERED_REMOTE_INDEX_SETTING,
         FeatureFlags.REMOTE_STORE_MIGRATION_EXPERIMENTAL_SETTING,
         FeatureFlags.PLUGGABLE_CACHE_SETTING,
-        FeatureFlags.REMOTE_PUBLICATION_EXPERIMENTAL_SETTING,
         FeatureFlags.STAR_TREE_INDEX_SETTING,
         FeatureFlags.APPLICATION_BASED_CONFIGURATION_TEMPLATES_SETTING
     );

server/src/main/java/org/opensearch/common/util/FeatureFlags.java

@@ -67,11 +67,6 @@ public class FeatureFlags {
      */
     public static final String PLUGGABLE_CACHE = "opensearch.experimental.feature.pluggable.caching.enabled";
 
-    /**
-     * Gates the functionality of remote routing table.
-     */
-    public static final String REMOTE_PUBLICATION_EXPERIMENTAL = "opensearch.experimental.feature.remote_store.publication.enabled";
-
     /**
      * Gates the functionality of background task execution.
      */
@@ -99,12 +94,6 @@ public class FeatureFlags {
 
     public static final Setting<Boolean> PLUGGABLE_CACHE_SETTING = Setting.boolSetting(PLUGGABLE_CACHE, false, Property.NodeScope);
 
-    public static final Setting<Boolean> REMOTE_PUBLICATION_EXPERIMENTAL_SETTING = Setting.boolSetting(
-        REMOTE_PUBLICATION_EXPERIMENTAL,
-        false,
-        Property.NodeScope
-    );
-
     /**
      * Gates the functionality of star tree index, which improves the performance of search
      * aggregations.
@@ -130,7 +119,6 @@ public class FeatureFlags {
         DATETIME_FORMATTER_CACHING_SETTING,
         TIERED_REMOTE_INDEX_SETTING,
         PLUGGABLE_CACHE_SETTING,
-        REMOTE_PUBLICATION_EXPERIMENTAL_SETTING,
         STAR_TREE_INDEX_SETTING,
         APPLICATION_BASED_CONFIGURATION_TEMPLATES_SETTING
     );

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

@@ -39,7 +39,6 @@
 import org.opensearch.common.settings.Setting.Property;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
-import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.common.util.io.IOUtils;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
@@ -89,7 +88,6 @@
 
 import static java.util.Collections.emptyList;
 import static java.util.Collections.emptyMap;
-import static org.opensearch.common.util.FeatureFlags.REMOTE_PUBLICATION_EXPERIMENTAL;
 import static org.opensearch.gateway.PersistedClusterStateService.SLOW_WRITE_LOGGING_THRESHOLD;
 import static org.opensearch.gateway.remote.ClusterMetadataManifest.CODEC_V2;
 import static org.opensearch.gateway.remote.RemoteClusterStateAttributesManager.CLUSTER_BLOCKS;
@@ -118,6 +116,18 @@ public class RemoteClusterStateService implements Closeable {
 
     private static final Logger logger = LogManager.getLogger(RemoteClusterStateService.class);
 
+    /**
+     * Gates the functionality of remote publication.
+     */
+    public static final String REMOTE_PUBLICATION_SETTING_KEY = "opensearch.feature.remote_store.publication.enabled";
+
+    public static final Setting<Boolean> REMOTE_PUBLICATION_SETTING = Setting.boolSetting(
+        REMOTE_PUBLICATION_SETTING_KEY,
+        false,
+        Property.NodeScope,
+        Property.Final
+    );
+
     /**
      * Used to specify if cluster state metadata should be published to remote store
      */
@@ -161,7 +171,8 @@ public class RemoteClusterStateService implements Closeable {
     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 boolean shouldWriteFullStateForPublication;
 
     // ToXContent Params with gateway mode.
     // We are using gateway context mode to persist all custom metadata.
@@ -197,7 +208,7 @@ public RemoteClusterStateService(
         this.remoteStateStats = new RemotePersistenceStats();
         this.namedWriteableRegistry = namedWriteableRegistry;
         this.indexMetadataUploadListeners = indexMetadataUploadListeners;
-        this.isPublicationEnabled = FeatureFlags.isEnabled(REMOTE_PUBLICATION_EXPERIMENTAL)
+        this.isPublicationEnabled = REMOTE_PUBLICATION_SETTING.get(settings)
             && RemoteStoreNodeAttribute.isRemoteStoreClusterStateEnabled(settings)
             && RemoteStoreNodeAttribute.isRemoteRoutingTableEnabled(settings);
         this.remoteRoutingTableService = RemoteRoutingTableServiceFactory.getService(

server/src/main/java/org/opensearch/node/remotestore/RemoteStoreNodeAttribute.java

@@ -13,7 +13,6 @@
 import org.opensearch.cluster.metadata.RepositoryMetadata;
 import org.opensearch.cluster.node.DiscoveryNode;
 import org.opensearch.common.settings.Settings;
-import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.gateway.remote.RemoteClusterStateService;
 import org.opensearch.node.Node;
 import org.opensearch.repositories.blobstore.BlobStoreRepository;
@@ -29,7 +28,7 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 
-import static org.opensearch.common.util.FeatureFlags.REMOTE_PUBLICATION_EXPERIMENTAL;
+import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_PUBLICATION_SETTING;
 
 /**
  * This is an abstraction for validating and storing information specific to remote backed storage nodes.
@@ -202,7 +201,7 @@ private static boolean isRemoteRoutingTableAttributePresent(Settings settings) {
     }
 
     public static boolean isRemoteRoutingTableEnabled(Settings settings) {
-        return FeatureFlags.isEnabled(REMOTE_PUBLICATION_EXPERIMENTAL) && isRemoteRoutingTableAttributePresent(settings);
+        return REMOTE_PUBLICATION_SETTING.get(settings) && isRemoteRoutingTableAttributePresent(settings);
     }
 
     public RepositoriesMetadata getRepositoriesMetadata() {

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

@@ -66,8 +66,8 @@
 
 import static java.util.Collections.emptyMap;
 import static java.util.Collections.emptySet;
-import static org.opensearch.common.util.FeatureFlags.REMOTE_PUBLICATION_EXPERIMENTAL;
 import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_PUBLICATION_SETTING_KEY;
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_CLUSTER_STATE_REPOSITORY_NAME_ATTRIBUTE_KEY;
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_REPOSITORY_SETTINGS_ATTRIBUTE_KEY_PREFIX;
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_REPOSITORY_TYPE_ATTRIBUTE_KEY_FORMAT;
@@ -1008,7 +1008,7 @@ public void testIsRemotePublicationEnabled_WithInconsistentSettings() {
         // create settings with remote state disabled but publication enabled
         Settings settings = Settings.builder()
             .put(REMOTE_CLUSTER_STATE_ENABLED_SETTING.getKey(), false)
-            .put(REMOTE_PUBLICATION_EXPERIMENTAL, true)
+            .put(REMOTE_PUBLICATION_SETTING_KEY, true)
             .build();
         CoordinationState coordinationState = createCoordinationState(psr1, node1, settings);
         assertFalse(coordinationState.isRemotePublicationEnabled());

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

@@ -133,7 +133,12 @@ public void writeTo(StreamOutput out) throws IOException {
 
         OpenSearchException e = expectThrows(
             OpenSearchException.class,
-            () -> handler.newPublicationContext(new ClusterChangedEvent("test", unserializableClusterState, clusterState), false, null)
+            () -> handler.newPublicationContext(
+                new ClusterChangedEvent("test", unserializableClusterState, clusterState),
+                false,
+                null,
+                false
+            )
         );
         assertNotNull(e.getCause());
         assertThat(e.getCause(), instanceOf(IOException.class));

server/src/test/java/org/opensearch/cluster/routing/remote/RemoteRoutingTableServiceFactoryTests.java

@@ -20,7 +20,7 @@
 
 import java.util.function.Supplier;
 
-import static org.opensearch.common.util.FeatureFlags.REMOTE_PUBLICATION_EXPERIMENTAL;
+import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_PUBLICATION_SETTING_KEY;
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY;
 
 public class RemoteRoutingTableServiceFactoryTests extends OpenSearchTestCase {
@@ -51,7 +51,7 @@ public void testGetServiceWhenRemoteRoutingEnabled() {
             .put("node.attr." + REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY, "routing_repository")
             .put(FsRepository.REPOSITORIES_COMPRESS_SETTING.getKey(), false)
             .build();
-        Settings nodeSettings = Settings.builder().put(REMOTE_PUBLICATION_EXPERIMENTAL, "true").build();
+        Settings nodeSettings = Settings.builder().put(REMOTE_PUBLICATION_SETTING_KEY, "true").build();
         FeatureFlags.initializeFeatureFlags(nodeSettings);
         RemoteRoutingTableService service = RemoteRoutingTableServiceFactory.getService(
             repositoriesService,