[Draft] Original changes against main

CHANGELOG.md

@@ -8,6 +8,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add support for Azure Managed Identity in repository-azure ([#12423](https://github.com/opensearch-project/OpenSearch/issues/12423))
 - Add useCompoundFile index setting ([#13478](https://github.com/opensearch-project/OpenSearch/pull/13478))
 - Make outbound side of transport protocol dependent ([#13293](https://github.com/opensearch-project/OpenSearch/pull/13293))
+- [Remote State] Add async remote state deletion task running on a interval, configurable by a setting ([#13131](https://github.com/opensearch-project/OpenSearch/pull/13131))
 
 ### Dependencies
 - Bump `com.github.spullara.mustache.java:compiler` from 0.9.10 to 0.9.13 ([#13329](https://github.com/opensearch-project/OpenSearch/pull/13329), [#13559](https://github.com/opensearch-project/OpenSearch/pull/13559))

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

@@ -23,17 +23,20 @@
 import java.nio.charset.StandardCharsets;
 import java.util.Base64;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 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_CLEANUP_INTERVAL_SETTING;
 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.RemoteClusterStateService.RETAINED_MANIFESTS;
+import static org.opensearch.gateway.remote.RemoteClusterStateUtils.DELIMITER;
+import static org.opensearch.gateway.remote.RemoteClusterStateUtils.METADATA_FILE_PREFIX;
+import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.COORDINATION_METADATA;
+import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.CUSTOM_METADATA;
+import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.SETTING_METADATA;
+import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.TEMPLATES_METADATA;
 
 @OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0)
 public class RemoteClusterStateServiceIT extends RemoteStoreBaseIntegTestCase {
@@ -63,27 +66,59 @@ private Map<String, Long> initialTestSetup(int shardCount, int replicaCount, int
  return indexStats;
  }
 
- public void testFullClusterRestoreStaleDelete() throws Exception {
+ public void testRemoteCleanupTaskUpdated() {
  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
+ );
+
+ assertEquals(5, remoteClusterStateService.getStaleFileDeletionTask().getInterval().getMinutes());
+ assertTrue(remoteClusterStateService.getStaleFileDeletionTask().isScheduled());
+
+ // now disable
+ client().admin()
+ .cluster()
+ .prepareUpdateSettings()
+ .setPersistentSettings(Settings.builder().put(REMOTE_CLUSTER_STATE_CLEANUP_INTERVAL_SETTING.getKey(), -1))
+ .get();
+
+ assertEquals(-1, remoteClusterStateService.getStaleFileDeletionTask().getInterval().getMillis());
+ assertFalse(remoteClusterStateService.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, remoteClusterStateService.getStaleFileDeletionTask().getInterval().getMinutes());
+ }
+
+ public void testRemoteCleanupOnlyAfter10Updates() throws Exception {
+ int shardCount = randomIntBetween(1, 2);
+ int replicaCount = 1;
+ int dataNodeCount = shardCount * (replicaCount + 1);
+ int clusterManagerNodeCount = 1;
 
+ initialTestSetup(shardCount, replicaCount, dataNodeCount, clusterManagerNodeCount);
  RemoteClusterStateService remoteClusterStateService = internalCluster().getClusterManagerNodeInstance(
  RemoteClusterStateService.class
  );
 
+ // set cleanup interval to 1 min
+ client().admin()
+ .cluster()
+ .prepareUpdateSettings()
+ .setPersistentSettings(Settings.builder().put(REMOTE_CLUSTER_STATE_CLEANUP_INTERVAL_SETTING.getKey(), "1m"))
+ .get();
+
+ replicaCount = updateReplicaCountNTimes(9, replicaCount);
+
  RepositoriesService repositoriesService = internalCluster().getClusterManagerNodeInstance(RepositoriesService.class);
 
  BlobStoreRepository repository = (BlobStoreRepository) repositoriesService.repository(REPOSITORY_NAME);
@@ -95,14 +130,39 @@ public void testFullClusterRestoreStaleDelete() throws Exception {
  )
  .add("cluster-state")
  .add(getClusterState().metadata().clusterUUID());
+ BlobPath manifestContainerPath = baseMetadataPath.add("manifest");
+
+ assertBusy(() -> {
+ assertEquals(
+ RETAINED_MANIFESTS - 1,
+ repository.blobStore().blobContainer(manifestContainerPath).listBlobsByPrefix("manifest").size()
+ );
+ }, 1, TimeUnit.MINUTES);
+
+ replicaCount = updateReplicaCountNTimes(8, replicaCount);
+
+ // wait for 1 min, to ensure that clean up task ran and didn't clean up stale files because it was less than 10
+ Thread.sleep(60000);
+ assertNotEquals(
+ RETAINED_MANIFESTS - 1,
+ repository.blobStore().blobContainer(manifestContainerPath).listBlobsByPrefix("manifest").size()
+ );
+
+ // Do 2 more updates, now since the total successful state changes are more than 10, stale files will be cleaned up
+ replicaCount = updateReplicaCountNTimes(2, replicaCount);
 
- assertEquals(10, repository.blobStore().blobContainer(baseMetadataPath.add("manifest")).listBlobsByPrefix("manifest").size());
+ assertBusy(() -> {
+ assertEquals(
+ RETAINED_MANIFESTS - 1,
+ repository.blobStore().blobContainer(manifestContainerPath).listBlobsByPrefix("manifest").size()
+ );
+ }, 1, TimeUnit.MINUTES);
 
  Map<String, IndexMetadata> indexMetadataMap = remoteClusterStateService.getLatestClusterState(
  cluster().getClusterName(),
  getClusterState().metadata().clusterUUID()
  ).getMetadata().getIndices();
- assertEquals(0, indexMetadataMap.values().stream().findFirst().get().getNumberOfReplicas());
+ assertEquals(replicaCount, indexMetadataMap.values().stream().findFirst().get().getNumberOfReplicas());
  assertEquals(shardCount, indexMetadataMap.values().stream().findFirst().get().getNumberOfShards());
  }
 
@@ -243,4 +303,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

@@ -311,4 +311,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;
 
@@ -320,9 +321,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) {
@@ -348,7 +347,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"