Implementing batched deletions of stale ClusterMetadataManifests in R…

CHANGELOG.md

@@ -26,6 +26,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Fix WLM workload group creation failing due to updated_at clock skew ([#20486](https://github.com/opensearch-project/OpenSearch/pull/20486))
 - Fix SLF4J component error ([#20587](https://github.com/opensearch-project/OpenSearch/pull/20587))
 - Service does not start on Windows with OpenJDK ([#20615](https://github.com/opensearch-project/OpenSearch/pull/20615))
+- Update RemoteClusterStateCleanupManager to performed batched deletions of stale ClusterMetadataManifests and address deletion timeout issues ([#20566](https://github.com/opensearch-project/OpenSearch/pull/20566))
 - Fix the regression of terms agg optimization at high cardinality ([#20623](https://github.com/opensearch-project/OpenSearch/pull/20623))
 
 ### Dependencies

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

@@ -34,9 +34,12 @@
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
 
 import static org.opensearch.gateway.remote.RemoteClusterStateCleanupManager.CLUSTER_STATE_CLEANUP_INTERVAL_DEFAULT;
+import static org.opensearch.gateway.remote.RemoteClusterStateCleanupManager.REMOTE_CLUSTER_STATE_CLEANUP_BATCH_SIZE_SETTING;
 import static org.opensearch.gateway.remote.RemoteClusterStateCleanupManager.REMOTE_CLUSTER_STATE_CLEANUP_INTERVAL_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateCleanupManager.REMOTE_CLUSTER_STATE_CLEANUP_MAX_BATCHES_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;
@@ -294,4 +297,169 @@ private BlobPath getIndexRoutingPath(BlobPath baseMetadataPath, String indexUUID
             RemoteStoreEnums.PathHashAlgorithm.FNV_1A_BASE64
         );
     }
+
+    public void testRemoteCleanupWithBatchSizeAs20() throws Exception {
+        performBatchedDeletionsGivenBatchSize(20);
+    }
+
+    public void testRemoteCleanupWithBatchSizeAs50() throws Exception {
+        performBatchedDeletionsGivenBatchSize(50);
+    }
+
+    public void testRemoteCleanupWithBatchSizeAs100() throws Exception {
+        performBatchedDeletionsGivenBatchSize(100);
+    }
+
+    private void performBatchedDeletionsGivenBatchSize(int batchSize) throws Exception {
+
+        int shardCount = randomIntBetween(1, 2);
+        int replicaCount = 1;
+        int dataNodeCount = shardCount * (replicaCount + 1);
+        int clusterManagerNodeCount = 1;
+
+        initialTestSetup(shardCount, replicaCount, dataNodeCount, clusterManagerNodeCount);
+        // To ensure we complete the cleanup entirely
+        int maxBatchesToAttempt = 20;
+
+        // disable auto cleanup as the test will trigger the deletion manually
+        ClusterUpdateSettingsResponse response = client().admin()
+            .cluster()
+            .prepareUpdateSettings()
+            .setPersistentSettings(
+                Settings.builder()
+                    .put(REMOTE_CLUSTER_STATE_CLEANUP_INTERVAL_SETTING.getKey(), -1)
+                    .put(REMOTE_CLUSTER_STATE_CLEANUP_MAX_BATCHES_SETTING.getKey(), maxBatchesToAttempt)
+                    .put(REMOTE_CLUSTER_STATE_CLEANUP_BATCH_SIZE_SETTING.getKey(), batchSize)
+            )
+            .get();
+        assertTrue(response.isAcknowledged());
+
+        updateClusterStateNTimes(50);
+
+        RepositoriesService repositoriesService = internalCluster().getClusterManagerNodeInstance(RepositoriesService.class);
+        BlobStoreRepository repository = (BlobStoreRepository) repositoriesService.repository(REPOSITORY_NAME);
+        BlobPath manifestContainerPath = getBaseMetadataPath(repository).add("manifest");
+
+        List<String> initialManifests = repository.blobStore()
+            .blobContainer(manifestContainerPath)
+            .listBlobsByPrefix("manifest")
+            .keySet()
+            .stream()
+            .sorted()
+            .collect(Collectors.toList());
+        assertTrue(initialManifests.size() >= 50);
+
+        List<String> last10Initial = initialManifests.subList(0, 10);
+
+        RemoteClusterStateCleanupManager cleanupManager = internalCluster().getClusterManagerNodeInstance(
+            RemoteClusterStateCleanupManager.class
+        );
+
+        long initialCleanupAttemptVersion = cleanupManager.getLastCleanupAttemptStateVersion();
+
+        // trigger deletion
+        cleanupManager.cleanUpStaleFiles();
+
+        // wait for completion
+        assertBusy(() -> { assertFalse(cleanupManager.isDeleteStaleMetadataRunning().get()); });
+
+        List<String> currentManifests = repository.blobStore()
+            .blobContainer(manifestContainerPath)
+            .listBlobsByPrefix("manifest")
+            .keySet()
+            .stream()
+            .sorted()
+            .collect(Collectors.toList());
+
+        // ensure cleanup is completed and successful
+        assertEquals("Last 10 manifest files should remain the same after cleanup", last10Initial, currentManifests);
+        // ensure we update the last successful cleanup attempt version once we update
+        assertNotEquals(
+            "lastCleanupAttemptStateVersion should be updated after successful cleanup",
+            initialCleanupAttemptVersion,
+            cleanupManager.getLastCleanupAttemptStateVersion()
+        );
+
+    }
+
+    public void testCleanupStopsAtMaxBatchesLimit() throws Exception {
+        int shardCount = randomIntBetween(1, 2);
+        int replicaCount = 1;
+        int dataNodeCount = shardCount * (replicaCount + 1);
+        int clusterManagerNodeCount = 1;
+
+        initialTestSetup(shardCount, replicaCount, dataNodeCount, clusterManagerNodeCount);
+
+        updateClusterStateNTimes(50);
+
+        RepositoriesService repositoriesService = internalCluster().getClusterManagerNodeInstance(RepositoriesService.class);
+        BlobStoreRepository repository = (BlobStoreRepository) repositoriesService.repository(REPOSITORY_NAME);
+        BlobPath manifestContainerPath = getBaseMetadataPath(repository).add("manifest");
+
+        // attempt only 1 batch to validate we stop after specified attempts
+        int maxBatchesToDelete = 1;
+        int batchSizePerDeletion = 30;
+
+        // disable auto cleanup as the test will trigger the deletion manually
+        ClusterUpdateSettingsResponse response = client().admin()
+            .cluster()
+            .prepareUpdateSettings()
+            .setPersistentSettings(
+                Settings.builder()
+                    .put(REMOTE_CLUSTER_STATE_CLEANUP_INTERVAL_SETTING.getKey(), -1)
+                    .put(REMOTE_CLUSTER_STATE_CLEANUP_MAX_BATCHES_SETTING.getKey(), maxBatchesToDelete)
+                    .put(REMOTE_CLUSTER_STATE_CLEANUP_BATCH_SIZE_SETTING.getKey(), batchSizePerDeletion)
+            )
+            .get();
+        assertTrue(response.isAcknowledged());
+
+        List<String> initialManifests = repository.blobStore()
+            .blobContainer(manifestContainerPath)
+            .listBlobsByPrefix("manifest")
+            .keySet()
+            .stream()
+            .sorted()
+            .toList();
+
+        List<String> last10Initial = initialManifests.subList(0, RETAINED_MANIFESTS);
+        int expectedManifestsToBeDeleted = (batchSizePerDeletion - RETAINED_MANIFESTS) * maxBatchesToDelete;
+        List<String> expectedListAfterDeletionAttempt = initialManifests.subList(0, initialManifests.size() - expectedManifestsToBeDeleted);
+
+        RemoteClusterStateCleanupManager cleanupManager = internalCluster().getClusterManagerNodeInstance(
+            RemoteClusterStateCleanupManager.class
+        );
+        long initialCleanupAttemptVersion = cleanupManager.getLastCleanupAttemptStateVersion();
+
+        // trigger deletion
+        cleanupManager.cleanUpStaleFiles();
+
+        // wait for completion
+        assertBusy(() -> { assertFalse(cleanupManager.isDeleteStaleMetadataRunning().get()); });
+
+        List<String> currentManifests = repository.blobStore()
+            .blobContainer(manifestContainerPath)
+            .listBlobsByPrefix("manifest")
+            .keySet()
+            .stream()
+            .sorted()
+            .toList();
+
+        assertEquals(
+            "Last 10 manifest files should remain the same after cleanup",
+            last10Initial,
+            currentManifests.subList(0, RETAINED_MANIFESTS)
+        );
+
+        assertEquals(
+            "Cleanup should have stopped after specified batch limit",
+            expectedListAfterDeletionAttempt.size(),
+            currentManifests.size()
+        );
+
+        assertEquals(
+            "As cleanup batch size is exhausted, we should not update last successful cleanup attempt version",
+            cleanupManager.getLastCleanupAttemptStateVersion(),
+            initialCleanupAttemptVersion
+        );
+    }
 }

server/src/main/java/org/opensearch/cluster/routing/remote/InternalRemoteRoutingTableService.java

@@ -13,17 +13,22 @@
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.opensearch.Version;
 import org.opensearch.action.LatchedActionListener;
+import org.opensearch.action.support.PlainActionFuture;
 import org.opensearch.cluster.Diff;
 import org.opensearch.cluster.routing.IndexRoutingTable;
 import org.opensearch.cluster.routing.RoutingTable;
 import org.opensearch.cluster.routing.RoutingTableIncrementalDiff;
 import org.opensearch.cluster.routing.StringKeyDiffProvider;
+import org.opensearch.common.blobstore.AsyncMultiStreamBlobContainer;
+import org.opensearch.common.blobstore.BlobContainer;
 import org.opensearch.common.blobstore.BlobPath;
 import org.opensearch.common.lifecycle.AbstractLifecycleComponent;
 import org.opensearch.common.remote.RemoteWritableEntityStore;
 import org.opensearch.common.remote.RemoteWriteableEntityBlobStore;
 import org.opensearch.common.settings.ClusterSettings;
 import org.opensearch.common.settings.Settings;
+import org.opensearch.common.unit.TimeValue;
+import org.opensearch.common.util.concurrent.FutureUtils;
 import org.opensearch.common.util.io.IOUtils;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.compress.Compressor;
@@ -43,8 +48,12 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
@@ -68,6 +77,7 @@ public class InternalRemoteRoutingTableService extends AbstractLifecycleComponen
     private BlobStoreRepository blobStoreRepository;
     private final ThreadPool threadPool;
     private final String clusterName;
+    private static final TimeValue DEFAULT_DELETION_TIMEOUT = TimeValue.timeValueSeconds(300);
 
     public InternalRemoteRoutingTableService(
         Supplier<RepositoriesService> repositoriesService,
@@ -270,7 +280,13 @@ protected void doStop() {}
     public void deleteStaleIndexRoutingPaths(List<String> stalePaths) throws IOException {
         try {
             logger.debug(() -> "Deleting stale index routing files from remote - " + stalePaths);
-            blobStoreRepository.blobStore().blobContainer(BlobPath.cleanPath()).deleteBlobsIgnoringIfNotExists(stalePaths);
+            BlobContainer blobContainerForDeletion = blobStoreRepository.blobStore().blobContainer(BlobPath.cleanPath());
+            assert blobContainerForDeletion != null;
+            if (blobContainerForDeletion instanceof AsyncMultiStreamBlobContainer) {
+                deleteAsyncInternal((AsyncMultiStreamBlobContainer) blobContainerForDeletion, stalePaths, DEFAULT_DELETION_TIMEOUT);
+            } else {
+                blobContainerForDeletion.deleteBlobsIgnoringIfNotExists(stalePaths);
+            }
         } catch (IOException e) {
             logger.error(() -> new ParameterizedMessage("Failed to delete some stale index routing paths from {}", stalePaths), e);
             throw e;
@@ -280,10 +296,35 @@ public void deleteStaleIndexRoutingPaths(List<String> stalePaths) throws IOExcep
     public void deleteStaleIndexRoutingDiffPaths(List<String> stalePaths) throws IOException {
         try {
             logger.debug(() -> "Deleting stale index routing diff files from remote - " + stalePaths);
-            blobStoreRepository.blobStore().blobContainer(BlobPath.cleanPath()).deleteBlobsIgnoringIfNotExists(stalePaths);
+            BlobContainer blobContainerForDeletion = blobStoreRepository.blobStore().blobContainer(BlobPath.cleanPath());
+            if (blobStoreRepository.blobStore().blobContainer(BlobPath.cleanPath()) instanceof AsyncMultiStreamBlobContainer) {
+                deleteAsyncInternal((AsyncMultiStreamBlobContainer) blobContainerForDeletion, stalePaths, DEFAULT_DELETION_TIMEOUT);
+            } else {
+                blobContainerForDeletion.deleteBlobsIgnoringIfNotExists(stalePaths);
+            }
         } catch (IOException e) {
             logger.error(() -> new ParameterizedMessage("Failed to delete some stale index routing diff paths from {}", stalePaths), e);
             throw e;
         }
     }
+
+    protected void deleteAsyncInternal(AsyncMultiStreamBlobContainer blobContainerForDeletion, List<String> fileNames, TimeValue timeout)
+        throws IOException {
+        PlainActionFuture<Void> future = new PlainActionFuture<>();
+        try {
+            blobContainerForDeletion.deleteBlobsAsyncIgnoringIfNotExists(fileNames, future);
+            future.get(timeout.seconds(), TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new IOException("Future got interrupted", e);
+        } catch (ExecutionException e) {
+            if (e.getCause() instanceof IOException) {
+                throw (IOException) e.getCause();
+            }
+            throw new RuntimeException(e.getCause());
+        } catch (TimeoutException e) {
+            FutureUtils.cancel(future);
+            throw new IOException(String.format(Locale.ROOT, "Delete operation timed out after %s seconds", timeout.seconds()), e);
+        }
+    }
 }

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

@@ -782,6 +782,8 @@ public void apply(Settings value, Settings current, Settings previous) {
                 RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING,
                 RemoteClusterStateService.REMOTE_PUBLICATION_SETTING,
                 RemoteClusterStateService.REMOTE_STATE_DOWNLOAD_TO_SERVE_READ_API,
+                RemoteClusterStateCleanupManager.REMOTE_CLUSTER_STATE_CLEANUP_BATCH_SIZE_SETTING,
+                RemoteClusterStateCleanupManager.REMOTE_CLUSTER_STATE_CLEANUP_MAX_BATCHES_SETTING,
 
                 INDEX_METADATA_UPLOAD_TIMEOUT_SETTING,
                 GLOBAL_METADATA_UPLOAD_TIMEOUT_SETTING,