[Draft] Original changes against main

libs/core/src/main/java/org/opensearch/core/common/transport/TransportAddress.java

@@ -162,4 +162,14 @@ public String toString() {
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         return builder.value(toString());
     }
+
+    public static TransportAddress fromString(String address) throws UnknownHostException {
+        String[] addressSplit = address.split(":");
+        if (addressSplit.length != 2) {
+            throw new IllegalArgumentException("address must be of the form [hostname/ip]:[port]");
+        }
+        String hostname = addressSplit[0];
+        int port = Integer.parseInt(addressSplit[1]);
+        return new TransportAddress(InetAddress.getByName(hostname), port);
+    }
 }

libs/core/src/main/java/org/opensearch/core/xcontent/XContentParserUtils.java

@@ -38,6 +38,8 @@
 import org.opensearch.core.xcontent.XContentParser.Token;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Locale;
 import java.util.function.Consumer;
 
@@ -178,4 +180,14 @@ public static <T> void parseTypedKeysObject(XContentParser parser, String delimi
             throw new ParsingException(parser.getTokenLocation(), "Failed to parse object: empty key");
         }
     }
+
+    public static List<String> parseStringList(XContentParser parser) throws IOException {
+         List<String> valueList = new ArrayList<>();
+         ensureExpectedToken(Token.START_ARRAY, parser.currentToken(), parser);
+         while (parser.nextToken() != Token.END_ARRAY) {
+             ensureExpectedToken(Token.VALUE_STRING, parser.currentToken(), parser);
+             valueList.add(parser.text());
+         }
+         return valueList;
+    }
 }

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

@@ -0,0 +1,139 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.gateway.remote;
+
+import org.opensearch.action.admin.cluster.settings.ClusterUpdateSettingsResponse;
+import org.opensearch.common.blobstore.BlobPath;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.remotestore.RemoteStoreBaseIntegTestCase;
+import org.opensearch.repositories.RepositoriesService;
+import org.opensearch.repositories.blobstore.BlobStoreRepository;
+import org.opensearch.test.OpenSearchIntegTestCase;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+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.indices.IndicesService.CLUSTER_DEFAULT_INDEX_REFRESH_INTERVAL_SETTING;
+
+@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0)
+public class RemoteClusterStateCleanupManagerIT extends RemoteStoreBaseIntegTestCase {
+
+    private static final String INDEX_NAME = "test-index";
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal)).put(REMOTE_CLUSTER_STATE_ENABLED_SETTING.getKey(), true).build();
+    }
+
+    private Map<String, Long> initialTestSetup(int shardCount, int replicaCount, int dataNodeCount, int clusterManagerNodeCount) {
+        prepareCluster(clusterManagerNodeCount, dataNodeCount, INDEX_NAME, replicaCount, shardCount);
+        Map<String, Long> indexStats = indexData(1, false, INDEX_NAME);
+        assertEquals(shardCount * (replicaCount + 1), getNumShards(INDEX_NAME).totalNumShards);
+        ensureGreen(INDEX_NAME);
+        return indexStats;
+    }
+
+    public void testRemoteCleanupTaskUpdated() {
+        int shardCount = randomIntBetween(1, 2);
+        int replicaCount = 1;
+        int dataNodeCount = shardCount * (replicaCount + 1);
+        int clusterManagerNodeCount = 1;
+
+        initialTestSetup(shardCount, replicaCount, dataNodeCount, clusterManagerNodeCount);
+        RemoteClusterStateCleanupManager remoteClusterStateCleanupManager = internalCluster().getClusterManagerNodeInstance(
+            RemoteClusterStateCleanupManager.class
+        );
+
+        assertEquals(CLUSTER_STATE_CLEANUP_INTERVAL_DEFAULT, remoteClusterStateCleanupManager.getStaleFileDeletionTask().getInterval());
+        assertTrue(remoteClusterStateCleanupManager.getStaleFileDeletionTask().isScheduled());
+
+        // now disable
+        client().admin()
+            .cluster()
+            .prepareUpdateSettings()
+            .setPersistentSettings(Settings.builder().put(REMOTE_CLUSTER_STATE_CLEANUP_INTERVAL_SETTING.getKey(), -1))
+            .get();
+
+        assertEquals(-1, remoteClusterStateCleanupManager.getStaleFileDeletionTask().getInterval().getMillis());
+        assertFalse(remoteClusterStateCleanupManager.getStaleFileDeletionTask().isScheduled());
+
+        // now set Clean up interval to 1 min
+        client().admin()
+            .cluster()
+            .prepareUpdateSettings()
+            .setPersistentSettings(Settings.builder().put(REMOTE_CLUSTER_STATE_CLEANUP_INTERVAL_SETTING.getKey(), "1m"))
+            .get();
+        assertEquals(1, remoteClusterStateCleanupManager.getStaleFileDeletionTask().getInterval().getMinutes());
+    }
+
+    public void testRemoteCleanupDeleteStale() throws Exception {
+        int shardCount = randomIntBetween(1, 2);
+        int replicaCount = 1;
+        int dataNodeCount = shardCount * (replicaCount + 1);
+        int clusterManagerNodeCount = 1;
+
+        initialTestSetup(shardCount, replicaCount, dataNodeCount, clusterManagerNodeCount);
+
+        // set cleanup interval to 100 ms to make the test faster
+        ClusterUpdateSettingsResponse response = client().admin()
+            .cluster()
+            .prepareUpdateSettings()
+            .setPersistentSettings(Settings.builder().put(REMOTE_CLUSTER_STATE_CLEANUP_INTERVAL_SETTING.getKey(), "100ms"))
+            .get();
+
+        assertTrue(response.isAcknowledged());
+
+        // update cluster state 21 times to ensure that clean up has run after this will upload 42 manifest files
+        // to repository, if manifest files are less than that it means clean up has run
+        updateClusterStateNTimes(RETAINED_MANIFESTS + SKIP_CLEANUP_STATE_CHANGES + 1);
+
+        RepositoriesService repositoriesService = internalCluster().getClusterManagerNodeInstance(RepositoriesService.class);
+        BlobStoreRepository repository = (BlobStoreRepository) repositoriesService.repository(REPOSITORY_NAME);
+        BlobPath baseMetadataPath = repository.basePath()
+            .add(
+                Base64.getUrlEncoder()
+                    .withoutPadding()
+                    .encodeToString(getClusterState().getClusterName().value().getBytes(StandardCharsets.UTF_8))
+            )
+            .add("cluster-state")
+            .add(getClusterState().metadata().clusterUUID());
+        BlobPath manifestContainerPath = baseMetadataPath.add("manifest");
+
+        assertBusy(() -> {
+            int manifestFiles = repository.blobStore().blobContainer(manifestContainerPath).listBlobsByPrefix("manifest").size();
+            logger.info("number of current manifest file: {}", manifestFiles);
+            // we can't guarantee that we have same number of manifest as Retained manifest in our repo as there can be other queued task
+            // other than replica count change which can upload new manifest files, that's why we check that number of manifests is between
+            // Retained manifests and Retained manifests + 2 * Skip cleanup state changes (each cluster state update uploads 2 manifests)
+            assertTrue(
+                "Current number of manifest files: " + manifestFiles,
+                manifestFiles >= RETAINED_MANIFESTS && manifestFiles < RETAINED_MANIFESTS + 2 * SKIP_CLEANUP_STATE_CHANGES
+            );
+        }, 500, TimeUnit.MILLISECONDS);
+    }
+
+    private void updateClusterStateNTimes(int n) {
+        int newReplicaCount = randomIntBetween(0, 3);
+        for (int i = n; i > 0; i--) {
+            ClusterUpdateSettingsResponse response = client().admin()
+                .cluster()
+                .prepareUpdateSettings()
+                .setPersistentSettings(Settings.builder().put(CLUSTER_DEFAULT_INDEX_REFRESH_INTERVAL_SETTING.getKey(), i, TimeUnit.SECONDS))
+                .get();
+            assertTrue(response.isAcknowledged());
+        }
+    }
+}

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

@@ -10,7 +10,6 @@
 
 import org.opensearch.action.admin.cluster.node.stats.NodesStatsRequest;
 import org.opensearch.action.admin.cluster.node.stats.NodesStatsResponse;
-import org.opensearch.cluster.metadata.IndexMetadata;
 import org.opensearch.common.blobstore.BlobPath;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.discovery.DiscoveryStats;
@@ -28,13 +27,13 @@
 import java.util.stream.Collectors;
 
 import static org.opensearch.cluster.metadata.IndexMetadata.SETTING_NUMBER_OF_REPLICAS;
-import static org.opensearch.gateway.remote.RemoteClusterStateService.COORDINATION_METADATA;
-import static org.opensearch.gateway.remote.RemoteClusterStateService.CUSTOM_METADATA;
-import static org.opensearch.gateway.remote.RemoteClusterStateService.DELIMITER;
-import static org.opensearch.gateway.remote.RemoteClusterStateService.METADATA_FILE_PREFIX;
 import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
-import static org.opensearch.gateway.remote.RemoteClusterStateService.SETTING_METADATA;
-import static org.opensearch.gateway.remote.RemoteClusterStateService.TEMPLATES_METADATA;
+import static org.opensearch.gateway.remote.RemoteClusterStateUtils.DELIMITER;
+import static org.opensearch.gateway.remote.RemoteClusterStateUtils.METADATA_FILE_PREFIX;
+import static org.opensearch.gateway.remote.model.RemoteCoordinationMetadata.COORDINATION_METADATA;
+import static org.opensearch.gateway.remote.model.RemoteCustomMetadata.CUSTOM_METADATA;
+import static org.opensearch.gateway.remote.model.RemotePersistentSettingsMetadata.SETTING_METADATA;
+import static org.opensearch.gateway.remote.model.RemoteTemplatesMetadata.TEMPLATES_METADATA;
 
 @OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0)
 public class RemoteClusterStateServiceIT extends RemoteStoreBaseIntegTestCase {
@@ -51,16 +50,6 @@ protected Settings nodeSettings(int nodeOrdinal) {
         return Settings.builder().put(super.nodeSettings(nodeOrdinal)).put(REMOTE_CLUSTER_STATE_ENABLED_SETTING.getKey(), true).build();
     }
 
-    private 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);
-        }
-    }
-
     private Map<String, Long> initialTestSetup(int shardCount, int replicaCount, int dataNodeCount, int clusterManagerNodeCount) {
         prepareCluster(clusterManagerNodeCount, dataNodeCount, INDEX_NAME, replicaCount, shardCount);
         Map<String, Long> indexStats = indexData(1, false, INDEX_NAME);
@@ -69,49 +58,6 @@ private Map<String, Long> initialTestSetup(int shardCount, int replicaCount, int
         return indexStats;
     }
 
-    public void testFullClusterRestoreStaleDelete() throws Exception {
-        int shardCount = randomIntBetween(1, 2);
-        int replicaCount = 1;
-        int dataNodeCount = shardCount * (replicaCount + 1);
-        int clusterManagerNodeCount = 1;
-
-        initialTestSetup(shardCount, replicaCount, dataNodeCount, clusterManagerNodeCount);
-        setReplicaCount(0);
-        setReplicaCount(2);
-        setReplicaCount(0);
-        setReplicaCount(1);
-        setReplicaCount(0);
-        setReplicaCount(1);
-        setReplicaCount(0);
-        setReplicaCount(2);
-        setReplicaCount(0);
-
-        RemoteClusterStateService remoteClusterStateService = internalCluster().getClusterManagerNodeInstance(
-            RemoteClusterStateService.class
-        );
-
-        RepositoriesService repositoriesService = internalCluster().getClusterManagerNodeInstance(RepositoriesService.class);
-
-        BlobStoreRepository repository = (BlobStoreRepository) repositoriesService.repository(REPOSITORY_NAME);
-        BlobPath baseMetadataPath = repository.basePath()
-            .add(
-                Base64.getUrlEncoder()
-                    .withoutPadding()
-                    .encodeToString(getClusterState().getClusterName().value().getBytes(StandardCharsets.UTF_8))
-            )
-            .add("cluster-state")
-            .add(getClusterState().metadata().clusterUUID());
-
-        assertEquals(10, repository.blobStore().blobContainer(baseMetadataPath.add("manifest")).listBlobsByPrefix("manifest").size());
-
-        Map<String, IndexMetadata> indexMetadataMap = remoteClusterStateService.getLatestClusterState(
-            cluster().getClusterName(),
-            getClusterState().metadata().clusterUUID()
-        ).getMetadata().getIndices();
-        assertEquals(0, indexMetadataMap.values().stream().findFirst().get().getNumberOfReplicas());
-        assertEquals(shardCount, indexMetadataMap.values().stream().findFirst().get().getNumberOfShards());
-    }
-
     public void testRemoteStateStats() {
         int shardCount = randomIntBetween(1, 2);
         int replicaCount = 1;
@@ -249,4 +195,17 @@ private void setReplicaCount(int replicaCount) {
             .setSettings(Settings.builder().put(SETTING_NUMBER_OF_REPLICAS, replicaCount))
             .get();
     }
+
+    private int updateReplicaCountNTimes(int n, int initialCount) {
+        int newReplicaCount = randomIntBetween(0, 3);
+        ;
+        for (int i = 0; i < n; i++) {
+            while (newReplicaCount == initialCount) {
+                newReplicaCount = randomIntBetween(0, 3);
+            }
+            setReplicaCount(newReplicaCount);
+            initialCount = newReplicaCount;
+        }
+        return newReplicaCount;
+    }
 }

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

@@ -350,4 +350,14 @@ protected void restore(boolean restoreAllShards, String... indices) {
                 PlainActionFuture.newFuture()
             );
     }
+
+    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);
+        }
+    }
 }

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

@@ -36,6 +36,7 @@
 import org.opensearch.gateway.remote.ClusterMetadataManifest;
 import org.opensearch.gateway.remote.ClusterMetadataManifest.UploadedIndexMetadata;
 import org.opensearch.gateway.remote.RemoteClusterStateService;
+import org.opensearch.gateway.remote.RemoteClusterStateUtils;
 import org.opensearch.test.InternalTestCluster;
 import org.opensearch.test.OpenSearchIntegTestCase;
 import org.junit.Before;
@@ -326,9 +327,7 @@ public void testFullClusterRestoreManifestFilePointsToInvalidIndexMetadataPathTh
         // Step - 3 Delete index metadata file in remote
         try {
             Files.move(
-                segmentRepoPath.resolve(
-                    RemoteClusterStateService.encodeString(clusterName) + "/cluster-state/" + prevClusterUUID + "/index"
-                ),
+                segmentRepoPath.resolve(RemoteClusterStateUtils.encodeString(clusterName) + "/cluster-state/" + prevClusterUUID + "/index"),
                 segmentRepoPath.resolve("cluster-state/")
             );
         } catch (IOException e) {
@@ -354,7 +353,7 @@ public void testRemoteStateFullRestart() throws Exception {
         try {
             Files.move(
                 segmentRepoPath.resolve(
-                    RemoteClusterStateService.encodeString(clusterService().state().getClusterName().value())
+                    RemoteClusterStateUtils.encodeString(clusterService().state().getClusterName().value())
                         + "/cluster-state/"
                         + prevClusterUUID
                         + "/manifest"

server/src/main/java/org/opensearch/cluster/ClusterModule.java

@@ -320,6 +320,35 @@ public static List<NamedXContentRegistry.Entry> getNamedXWriteables() {
                 DecommissionAttributeMetadata::fromXContent
             )
         );
+        // Cluster State
+        entries.add(
+            new NamedXContentRegistry.Entry(
+                ClusterState.Custom.class,
+                new ParseField(SnapshotsInProgress.TYPE),
+                SnapshotsInProgress::fromXContent
+            )
+        );
+        entries.add(
+            new NamedXContentRegistry.Entry(
+                ClusterState.Custom.class,
+                new ParseField(RestoreInProgress.TYPE),
+                RestoreInProgress::fromXContent
+            )
+        );
+        entries.add(
+            new NamedXContentRegistry.Entry(
+                ClusterState.Custom.class,
+                new ParseField(SnapshotDeletionsInProgress.TYPE),
+                SnapshotDeletionsInProgress::fromXContent
+            )
+        );
+        entries.add(
+            new NamedXContentRegistry.Entry(
+                ClusterState.Custom.class,
+                new ParseField(RepositoryCleanupInProgress.TYPE),
+                RepositoryCleanupInProgress::fromXContent
+            )
+        );
         return entries;
     }