From 0db6bf511633b61fe8875be3eed4af0ff9b8ee05 Mon Sep 17 00:00:00 2001 From: Pranshu Shukla Date: Fri, 13 Feb 2026 14:16:33 +0530 Subject: [PATCH 01/12] Implementing batched deletions of stale ClusterMetadataManifests in RemoteClusterStateCleanupManager Signed-off-by: Pranshu Shukla --- CHANGELOG.md | 2 + .../RemoteClusterStateCleanupManagerIT.java | 168 ++++++ .../InternalRemoteRoutingTableService.java | 38 +- .../common/settings/ClusterSettings.java | 2 + .../RemoteClusterStateCleanupManager.java | 251 ++++++-- ...RemoteClusterStateCleanupManagerTests.java | 541 ++++++++++++++++-- 6 files changed, 915 insertions(+), 87 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2916db7cdc792..8cd1257a96f44 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), - Update EncryptedBlobContainer to adhere limits while listing blobs in specific sort order if wrapped blob container supports ([#20514](https://github.com/opensearch-project/OpenSearch/pull/20514)) - [segment replication] Fix segment replication infinite retry due to stale metadata checkpoint ([#20551](https://github.com/opensearch-project/OpenSearch/pull/20551)) - Changing opensearch.cgroups.hierarchy.override causes java.lang.SecurityException exception ([#20565](https://github.com/opensearch-project/OpenSearch/pull/20565)) +- Fix CriteriaBasedCodec to work with delegate codec. ([20442](https://github.com/opensearch-project/OpenSearch/pull/20442)) +- Update RemoteClusterStateCleanupManager to performed batched deletions of stale ClusterMetadataManifests and address deletion timeout issues ([#20566](https://github.com/opensearch-project/OpenSearch/pull/20566)) - Fix CriteriaBasedCodec to work with delegate codec. ([#20442](https://github.com/opensearch-project/OpenSearch/pull/20442)) - 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)) diff --git a/server/src/internalClusterTest/java/org/opensearch/gateway/remote/RemoteClusterStateCleanupManagerIT.java b/server/src/internalClusterTest/java/org/opensearch/gateway/remote/RemoteClusterStateCleanupManagerIT.java index cf17a58d937de..b52e6fc521279 100644 --- a/server/src/internalClusterTest/java/org/opensearch/gateway/remote/RemoteClusterStateCleanupManagerIT.java +++ b/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 initialManifests = repository.blobStore() + .blobContainer(manifestContainerPath) + .listBlobsByPrefix("manifest") + .keySet() + .stream() + .sorted() + .collect(Collectors.toList()); + assertTrue(initialManifests.size() >= 50); + + List 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 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 initialManifests = repository.blobStore() + .blobContainer(manifestContainerPath) + .listBlobsByPrefix("manifest") + .keySet() + .stream() + .sorted() + .toList(); + + List last10Initial = initialManifests.subList(0, RETAINED_MANIFESTS); + int expectedManifestsToBeDeleted = (batchSizePerDeletion - RETAINED_MANIFESTS) * maxBatchesToDelete; + List 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 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 + ); + } } diff --git a/server/src/main/java/org/opensearch/cluster/routing/remote/InternalRemoteRoutingTableService.java b/server/src/main/java/org/opensearch/cluster/routing/remote/InternalRemoteRoutingTableService.java index 31d7614f6ad0a..dd1974017a0f0 100644 --- a/server/src/main/java/org/opensearch/cluster/routing/remote/InternalRemoteRoutingTableService.java +++ b/server/src/main/java/org/opensearch/cluster/routing/remote/InternalRemoteRoutingTableService.java @@ -13,17 +13,21 @@ 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.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 +47,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 +76,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, @@ -280,10 +289,37 @@ public void deleteStaleIndexRoutingPaths(List stalePaths) throws IOExcep public void deleteStaleIndexRoutingDiffPaths(List stalePaths) throws IOException { try { logger.debug(() -> "Deleting stale index routing diff files from remote - " + stalePaths); - blobStoreRepository.blobStore().blobContainer(BlobPath.cleanPath()).deleteBlobsIgnoringIfNotExists(stalePaths); + if (blobStoreRepository.blobStore().blobContainer(BlobPath.cleanPath()) instanceof AsyncMultiStreamBlobContainer) { + deleteAsyncInternal(stalePaths); + } else { + blobStoreRepository.blobStore().blobContainer(BlobPath.cleanPath()).deleteBlobsIgnoringIfNotExists(stalePaths); + } } catch (IOException e) { logger.error(() -> new ParameterizedMessage("Failed to delete some stale index routing diff paths from {}", stalePaths), e); throw e; } } + + private void deleteAsyncInternal(List fileNames) throws IOException { + PlainActionFuture future = new PlainActionFuture<>(); + try { + ((AsyncMultiStreamBlobContainer) blobStoreRepository.blobStore().blobContainer(BlobPath.cleanPath())) + .deleteBlobsAsyncIgnoringIfNotExists(fileNames, future); + future.get(DEFAULT_DELETION_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", DEFAULT_DELETION_TIMEOUT.seconds()), + e + ); + } + } } diff --git a/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java b/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java index 04ccfcde38850..f2c363bbf5b61 100644 --- a/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java +++ b/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java @@ -781,6 +781,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, diff --git a/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateCleanupManager.java b/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateCleanupManager.java index 559b7422312c0..7278cec53d2d3 100644 --- a/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateCleanupManager.java +++ b/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateCleanupManager.java @@ -11,16 +11,19 @@ import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.message.ParameterizedMessage; +import org.opensearch.action.support.PlainActionFuture; import org.opensearch.cluster.ClusterState; import org.opensearch.cluster.routing.remote.RemoteRoutingTableService; import org.opensearch.cluster.service.ClusterApplierService; import org.opensearch.cluster.service.ClusterService; +import org.opensearch.common.blobstore.AsyncMultiStreamBlobContainer; import org.opensearch.common.blobstore.BlobMetadata; import org.opensearch.common.blobstore.BlobPath; import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.settings.Setting; import org.opensearch.common.unit.TimeValue; import org.opensearch.common.util.concurrent.AbstractAsyncTask; +import org.opensearch.common.util.concurrent.FutureUtils; import org.opensearch.core.action.ActionListener; import org.opensearch.core.common.Strings; import org.opensearch.index.translog.transfer.BlobStoreTransferService; @@ -32,7 +35,12 @@ import java.util.HashSet; import java.util.List; import java.util.Locale; +import java.util.Map; +import java.util.Objects; import java.util.Set; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import static org.opensearch.gateway.remote.RemoteClusterStateUtils.GLOBAL_METADATA_PATH_TOKEN; @@ -48,9 +56,15 @@ public class RemoteClusterStateCleanupManager implements Closeable { public static final int RETAINED_MANIFESTS = 10; public static final int SKIP_CLEANUP_STATE_CHANGES = 10; + public static final int MANIFEST_CLEANUP_BATCH_SIZE_DEFAULT = 1000; + public static final int MANIFEST_CLEANUP_MAX_BATCHES_DEFAULT = 3; public static final TimeValue CLUSTER_STATE_CLEANUP_INTERVAL_DEFAULT = TimeValue.timeValueMinutes(5); public static final TimeValue CLUSTER_STATE_CLEANUP_INTERVAL_MINIMUM = TimeValue.MINUS_ONE; + public static final String REMOTE_CLUSTER_STATE_CLEANUP_BATCH_SIZE_SETTING_NAME = "cluster.remote_store.state.cleanup.batch_size"; + public static final String REMOTE_CLUSTER_STATE_CLEANUP_MAX_BATCHES_SETTING_NAME = "cluster.remote_store.state.cleanup.max_batches"; + private static final TimeValue DEFAULT_DELETION_TIMEOUT = TimeValue.timeValueSeconds(300); + /** * Setting to specify the interval to do run stale file cleanup job * Min value -1 indicates that the stale file cleanup job should be disabled @@ -62,11 +76,36 @@ public class RemoteClusterStateCleanupManager implements Closeable { Setting.Property.NodeScope, Setting.Property.Dynamic ); + + /** + * Setting to specify the batch size for manifest cleanup operations + */ + public static final Setting REMOTE_CLUSTER_STATE_CLEANUP_BATCH_SIZE_SETTING = Setting.intSetting( + REMOTE_CLUSTER_STATE_CLEANUP_BATCH_SIZE_SETTING_NAME, + MANIFEST_CLEANUP_BATCH_SIZE_DEFAULT, + 0, + new RemoteClusterStateCleanupBatchSizeValidator(), + Setting.Property.NodeScope, + Setting.Property.Dynamic + ); + + /** + * Setting to specify the maximum number of batches to process during cleanup + */ + public static final Setting REMOTE_CLUSTER_STATE_CLEANUP_MAX_BATCHES_SETTING = Setting.intSetting( + REMOTE_CLUSTER_STATE_CLEANUP_MAX_BATCHES_SETTING_NAME, + MANIFEST_CLEANUP_MAX_BATCHES_DEFAULT, + 1, + Setting.Property.NodeScope, + Setting.Property.Dynamic + ); private static final Logger logger = LogManager.getLogger(RemoteClusterStateCleanupManager.class); private final RemoteClusterStateService remoteClusterStateService; private final RemotePersistenceStats remoteStateStats; private BlobStoreTransferService blobStoreTransferService; private TimeValue staleFileCleanupInterval; + private volatile int cleanupBatchSize; + private volatile int cleanupMaxBatches; private final AtomicBoolean deleteStaleMetadataRunning = new AtomicBoolean(false); private volatile AsyncStaleFileDeletion staleFileDeletionTask; private long lastCleanupAttemptStateVersion; @@ -85,10 +124,14 @@ public RemoteClusterStateCleanupManager( ClusterSettings clusterSettings = clusterService.getClusterSettings(); this.clusterApplierService = clusterService.getClusterApplierService(); this.staleFileCleanupInterval = clusterSettings.get(REMOTE_CLUSTER_STATE_CLEANUP_INTERVAL_SETTING); + this.cleanupBatchSize = clusterSettings.get(REMOTE_CLUSTER_STATE_CLEANUP_BATCH_SIZE_SETTING); + this.cleanupMaxBatches = clusterSettings.get(REMOTE_CLUSTER_STATE_CLEANUP_MAX_BATCHES_SETTING); this.threadpool = remoteClusterStateService.getThreadpool(); // initialize with 0, a cleanup will be done when this node is elected master node and version is incremented more than threshold this.lastCleanupAttemptStateVersion = 0; clusterSettings.addSettingsUpdateConsumer(REMOTE_CLUSTER_STATE_CLEANUP_INTERVAL_SETTING, this::updateCleanupInterval); + clusterSettings.addSettingsUpdateConsumer(REMOTE_CLUSTER_STATE_CLEANUP_BATCH_SIZE_SETTING, this::updateCleanupBatchSize); + clusterSettings.addSettingsUpdateConsumer(REMOTE_CLUSTER_STATE_CLEANUP_MAX_BATCHES_SETTING, this::updateCleanupMaxBatches); this.remoteRoutingTableService = remoteRoutingTableService; } @@ -104,7 +147,8 @@ public void close() throws IOException { } } - private BlobStoreTransferService getBlobStoreTransferService() { + // package private for testing + BlobStoreTransferService getBlobStoreTransferService() { if (blobStoreTransferService == null) { blobStoreTransferService = new BlobStoreTransferService(remoteClusterStateService.getBlobStore(), threadpool); } @@ -120,6 +164,16 @@ private void updateCleanupInterval(TimeValue updatedInterval) { } } + private void updateCleanupBatchSize(Integer updatedBatchSize) { + this.cleanupBatchSize = updatedBatchSize; + logger.info("updated remote state cleanup batch size to {}", updatedBatchSize); + } + + private void updateCleanupMaxBatches(Integer updatedMaxBatches) { + this.cleanupMaxBatches = updatedMaxBatches; + logger.info("updated remote state cleanup max batches to {}", updatedMaxBatches); + } + // visible for testing void cleanUpStaleFiles() { ClusterState currentAppliedState = clusterApplierService.state(); @@ -137,9 +191,9 @@ void cleanUpStaleFiles() { this.deleteStaleClusterMetadata( currentAppliedState.getClusterName().value(), currentAppliedState.metadata().clusterUUID(), - RETAINED_MANIFESTS + RETAINED_MANIFESTS, + cleanUpAttemptStateVersion ); - lastCleanupAttemptStateVersion = cleanUpAttemptStateVersion; } else { logger.debug( "Skipping cleanup of stale remote state files for cluster [{}] with uuid [{}]. Last clean was done before {} updates, which is less than threshold {}", @@ -171,7 +225,7 @@ void deleteClusterMetadata( String clusterUUID, List activeManifestBlobMetadata, List staleManifestBlobMetadata - ) { + ) throws IOException { try { Set filesToKeep = new HashSet<>(); Set staleManifestPaths = new HashSet<>(); @@ -180,6 +234,9 @@ void deleteClusterMetadata( Set staleEphemeralAttributePaths = new HashSet<>(); Set staleIndexRoutingPaths = new HashSet<>(); Set staleIndexRoutingDiffPaths = new HashSet<>(); + + // todo: Avoid repetitive fetch of manifestsToRetain across batches if they were fetched earlier and are the same + // (for example the first 10 if not new manifests are uploaded in between) activeManifestBlobMetadata.forEach(blobMetadata -> { ClusterMetadataManifest clusterMetadataManifest = remoteManifestManager.fetchRemoteClusterMetadataManifest( clusterName, @@ -260,7 +317,7 @@ void deleteClusterMetadata( clusterMetadataManifest.getIndicesRouting().forEach(uploadedIndicesRouting -> { if (!filesToKeep.contains(uploadedIndicesRouting.getUploadedFilename())) { staleIndexRoutingPaths.add(uploadedIndicesRouting.getUploadedFilename()); - logger.debug( + logger.trace( () -> new ParameterizedMessage( "Indices routing paths in stale manifest: {}", uploadedIndicesRouting.getUploadedFilename() @@ -273,7 +330,7 @@ void deleteClusterMetadata( && clusterMetadataManifest.getDiffManifest().getIndicesRoutingDiffPath() != null) { if (!filesToKeep.contains(clusterMetadataManifest.getDiffManifest().getIndicesRoutingDiffPath())) { staleIndexRoutingDiffPaths.add(clusterMetadataManifest.getDiffManifest().getIndicesRoutingDiffPath()); - logger.debug( + logger.trace( () -> new ParameterizedMessage( "Indices routing diff paths in stale manifest: {}", clusterMetadataManifest.getDiffManifest().getIndicesRoutingDiffPath() @@ -320,10 +377,22 @@ void deleteClusterMetadata( return; } + logger.info( + "Processed [{}] manifests, Deleting [{}] stale Global Metadata files, " + + "[{}] stale Index Metadata files, [{}] stale Ephemeral Metadata files, " + + "[{}] stale Index Routing files and [{}] stale Index routing diff files", + staleManifestPaths.size(), + staleGlobalMetadataPaths.size(), + staleIndexMetadataPaths.size(), + staleEphemeralAttributePaths.size(), + staleIndexRoutingPaths.size(), + staleIndexRoutingDiffPaths.size() + ); + deleteStalePaths(new ArrayList<>(staleGlobalMetadataPaths)); deleteStalePaths(new ArrayList<>(staleIndexMetadataPaths)); deleteStalePaths(new ArrayList<>(staleEphemeralAttributePaths)); - deleteStalePaths(new ArrayList<>(staleManifestPaths)); + try { remoteRoutingTableService.deleteStaleIndexRoutingPaths(new ArrayList<>(staleIndexRoutingPaths)); } catch (IOException e) { @@ -332,7 +401,10 @@ void deleteClusterMetadata( e ); remoteStateStats.indexRoutingFilesCleanupAttemptFailed(); + // throw exception as we do not want to fail repeatedly on all batches + throw e; } + try { remoteRoutingTableService.deleteStaleIndexRoutingDiffPaths(new ArrayList<>(staleIndexRoutingDiffPaths)); } catch (IOException e) { @@ -341,15 +413,28 @@ void deleteClusterMetadata( e ); remoteStateStats.indicesRoutingDiffFileCleanupAttemptFailed(); + // throw exception as we do not want to fail repeatedly on all batches + throw e; } + + // Delete Manifests in the very end to avoid dangling routing files in-case deletion of stale index routing + // files after deleting manifests + deleteStalePaths(new ArrayList<>(staleManifestPaths)); + } catch (IllegalStateException e) { logger.error("Error while fetching Remote Cluster Metadata manifests", e); + // throw exception as we do not want to fail repeatedly on all batches + throw e; } catch (IOException e) { logger.error("Error while deleting stale Remote Cluster Metadata files", e); remoteStateStats.cleanUpAttemptFailed(); + // throw exception as we do not want to fail repeatedly on all batches + throw e; } catch (Exception e) { logger.error("Unexpected error while deleting stale Remote Cluster Metadata files", e); remoteStateStats.cleanUpAttemptFailed(); + // throw exception as we do not want to fail repeatedly on all batches + throw e; } } @@ -359,49 +444,73 @@ void deleteClusterMetadata( * @param clusterName name of the cluster * @param clusterUUID uuid of cluster state to refer to in remote * @param manifestsToRetain no of latest manifest files to keep in remote + * @param cleanUpAttemptStateVersion the state version for this cleanup attempt */ // package private for testing - void deleteStaleClusterMetadata(String clusterName, String clusterUUID, int manifestsToRetain) { + void deleteStaleClusterMetadata(String clusterName, String clusterUUID, int manifestsToRetain, long cleanUpAttemptStateVersion) { if (deleteStaleMetadataRunning.compareAndSet(false, true) == false) { logger.info("Delete stale cluster metadata task is already in progress."); return; } - try { - getBlobStoreTransferService().listAllInSortedOrderAsync( - ThreadPool.Names.REMOTE_PURGE, - remoteManifestManager.getManifestFolderPath(clusterName, clusterUUID), - MANIFEST, - Integer.MAX_VALUE, - new ActionListener<>() { - @Override - public void onResponse(List blobMetadata) { - if (blobMetadata.size() > manifestsToRetain) { - deleteClusterMetadata( - clusterName, - clusterUUID, - blobMetadata.subList(0, manifestsToRetain), - blobMetadata.subList(manifestsToRetain, blobMetadata.size()) - ); - } - deleteStaleMetadataRunning.set(false); - } - @Override - public void onFailure(Exception e) { - logger.error( - new ParameterizedMessage( - "Exception occurred while deleting Remote Cluster Metadata for clusterUUIDs {}", - clusterUUID - ) + logger.info( + "Starting batched cleanup for cluster [{}] with batch size [{}] and retaining [{}] manifests. Maximum batches to be attempted will be [{}]", + clusterName, + cleanupBatchSize, + manifestsToRetain, + cleanupMaxBatches + ); + + threadpool.executor(ThreadPool.Names.REMOTE_PURGE).execute(() -> { + try { + int batchesProcessed = 0; + + while (batchesProcessed < cleanupMaxBatches) { + batchesProcessed++; + + // todo: To avoid repetitive fetch, we should use a paginated listener for each response page and act upon it + List batchManifests = getBlobStoreTransferService().listAllInSortedOrder( + remoteManifestManager.getManifestFolderPath(clusterName, clusterUUID), + MANIFEST, + cleanupBatchSize + ); + + if (Objects.nonNull(batchManifests) && batchManifests.size() > manifestsToRetain) { + List manifestsToDeletes = batchManifests.subList(manifestsToRetain, batchManifests.size()); + logger.debug("[Batch {}] Deleting [{}] stale manifests", batchesProcessed, manifestsToDeletes.size()); + + deleteClusterMetadata(clusterName, clusterUUID, batchManifests.subList(0, manifestsToRetain), manifestsToDeletes); + } else { + logger.debug( + "Number of manifests [{}] are less than or equal to manifests to retain [{}]. Skipping deletion", + batchManifests.size(), + manifestsToRetain ); - deleteStaleMetadataRunning.set(false); + break; } } - ); - } catch (Exception e) { - deleteStaleMetadataRunning.set(false); - throw e; - } + + if (batchesProcessed == cleanupMaxBatches) { + logger.warn("Exhausted batch limit for deleting entities. Attempted [{}] batches", batchesProcessed); + } else { + // Update version only after successful completion + logger.info("Completed cleaning up all stale cluster-state metadata files in [{}] batches", batchesProcessed); + lastCleanupAttemptStateVersion = cleanUpAttemptStateVersion; + } + } catch (Exception e) { + logger.error( + new ParameterizedMessage( + "Exception occurred while deleting Remote Cluster Metadata for clusterUUID [{}] for attempted cluster-state version [{}]", + clusterUUID, + cleanUpAttemptStateVersion + ), + e + ); + } finally { + deleteStaleMetadataRunning.set(false); + logger.debug("Released cleanup lock for cluster [{}]", clusterName); + } + }); } /** @@ -444,7 +553,34 @@ public void onFailure(Exception e) { // package private for testing void deleteStalePaths(List stalePaths) throws IOException { logger.debug(String.format(Locale.ROOT, "Deleting stale files from remote - %s", stalePaths)); - getBlobStoreTransferService().deleteBlobs(BlobPath.cleanPath(), stalePaths); + if (remoteClusterStateService.getBlobStore().blobContainer(BlobPath.cleanPath()) instanceof AsyncMultiStreamBlobContainer) { + deleteAsyncInternal(stalePaths); + } else { + getBlobStoreTransferService().deleteBlobs(BlobPath.cleanPath(), stalePaths); + } + } + + private void deleteAsyncInternal(List fileNames) throws IOException { + PlainActionFuture future = new PlainActionFuture<>(); + try { + ((AsyncMultiStreamBlobContainer) remoteClusterStateService.getBlobStore().blobContainer(BlobPath.cleanPath())) + .deleteBlobsAsyncIgnoringIfNotExists(fileNames, future); + future.get(DEFAULT_DELETION_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", DEFAULT_DELETION_TIMEOUT.seconds()), + e + ); + } } /** @@ -484,6 +620,16 @@ RemotePersistenceStats getStats() { return this.remoteStateStats; } + // visible for testing + long getLastCleanupAttemptStateVersion() { + return lastCleanupAttemptStateVersion; + } + + // visible for testing + AtomicBoolean isDeleteStaleMetadataRunning() { + return deleteStaleMetadataRunning; + } + static final class AsyncStaleFileDeletion extends AbstractAsyncTask { private final RemoteClusterStateCleanupManager remoteClusterStateCleanupManager; @@ -513,4 +659,27 @@ protected String getThreadPool() { return ThreadPool.Names.REMOTE_PURGE; } } + + /** + * Validates the batch size setting for cleaning up stale manifest files + */ + static final class RemoteClusterStateCleanupBatchSizeValidator implements Setting.Validator { + + @Override + public void validate(Integer value) {} + + @Override + public void validate(final Integer cleanupBatchSize, final Map, Object> settings) { + if (cleanupBatchSize <= RETAINED_MANIFESTS) { + throw new IllegalArgumentException( + String.format( + Locale.ROOT, + "Cleanup Batch Size should be greater than [%s] to as we " + "retain [%s] latest manifests in remote", + RETAINED_MANIFESTS, + RETAINED_MANIFESTS + ) + ); + } + } + } } diff --git a/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateCleanupManagerTests.java b/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateCleanupManagerTests.java index 753e791ca915c..34aa9ed8169a9 100644 --- a/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateCleanupManagerTests.java +++ b/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateCleanupManagerTests.java @@ -17,6 +17,7 @@ import org.opensearch.cluster.routing.remote.RemoteRoutingTableService; import org.opensearch.cluster.service.ClusterApplierService; import org.opensearch.cluster.service.ClusterService; +import org.opensearch.common.blobstore.AsyncMultiStreamBlobContainer; import org.opensearch.common.blobstore.BlobContainer; import org.opensearch.common.blobstore.BlobMetadata; import org.opensearch.common.blobstore.BlobPath; @@ -24,8 +25,10 @@ import org.opensearch.common.blobstore.support.PlainBlobMetadata; import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.settings.Settings; +import org.opensearch.common.unit.TimeValue; import org.opensearch.common.util.concurrent.AbstractAsyncTask; import org.opensearch.core.action.ActionListener; +import org.opensearch.index.translog.transfer.BlobStoreTransferService; import org.opensearch.repositories.RepositoriesService; import org.opensearch.repositories.blobstore.BlobStoreRepository; import org.opensearch.repositories.fs.FsRepository; @@ -39,6 +42,7 @@ import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Locale; @@ -56,7 +60,11 @@ import static org.opensearch.gateway.remote.ClusterMetadataManifest.UploadedMetadataAttribute; import static org.opensearch.gateway.remote.RemoteClusterStateCleanupManager.AsyncStaleFileDeletion; import static org.opensearch.gateway.remote.RemoteClusterStateCleanupManager.CLUSTER_STATE_CLEANUP_INTERVAL_DEFAULT; +import static org.opensearch.gateway.remote.RemoteClusterStateCleanupManager.MANIFEST_CLEANUP_BATCH_SIZE_DEFAULT; +import static org.opensearch.gateway.remote.RemoteClusterStateCleanupManager.MANIFEST_CLEANUP_MAX_BATCHES_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.RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN; @@ -74,6 +82,8 @@ import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doNothing; @@ -82,6 +92,7 @@ import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; import static org.mockito.Mockito.when; public class RemoteClusterStateCleanupManagerTests extends OpenSearchTestCase { @@ -94,6 +105,7 @@ public class RemoteClusterStateCleanupManagerTests extends OpenSearchTestCase { private ClusterApplierService clusterApplierService; private ClusterState clusterState; private Metadata metadata; + private DiscoveryNodes nodes; private RemoteClusterStateService remoteClusterStateService; private RemoteManifestManager remoteManifestManager; private RemoteRoutingTableService remoteRoutingTableService; @@ -129,11 +141,13 @@ public void setup() { clusterApplierService = mock(ClusterApplierService.class); clusterState = mock(ClusterState.class); metadata = mock(Metadata.class); + nodes = mock(DiscoveryNodes.class); clusterService = mock(ClusterService.class); when(clusterService.getClusterSettings()).thenReturn(clusterSettings); when(clusterState.getClusterName()).thenReturn(new ClusterName("test")); when(metadata.clusterUUID()).thenReturn("testUUID"); when(clusterState.metadata()).thenReturn(metadata); + when(clusterState.nodes()).thenReturn(nodes); when(clusterApplierService.state()).thenReturn(clusterState); when(clusterService.getClusterApplierService()).thenReturn(clusterApplierService); @@ -276,21 +290,32 @@ public void testDeleteClusterMetadata() throws IOException { when(remoteManifestManager.getManifestFolderPath(eq(clusterName), eq(clusterUUID))).thenReturn( new BlobPath().add(encodeString(clusterName)).add(CLUSTER_STATE_PATH_TOKEN).add(clusterUUID).add(MANIFEST) ); - BlobContainer container = mock(BlobContainer.class); + AsyncMultiStreamBlobContainer container = mock(AsyncMultiStreamBlobContainer.class); when(blobStore.blobContainer(any())).thenReturn(container); - doNothing().when(container).deleteBlobsIgnoringIfNotExists(any()); + doAnswer(invocation -> { + ActionListener listener = invocation.getArgument(1); + listener.onResponse(null); + return null; + }).when(container).deleteBlobsAsyncIgnoringIfNotExists(any(), any()); + remoteClusterStateCleanupManager.start(); remoteClusterStateCleanupManager.deleteClusterMetadata(clusterName, clusterUUID, activeBlobs, inactiveBlobs); - verify(container).deleteBlobsIgnoringIfNotExists( - List.of( - // coordination/setting metadata is from CODEC_V2, the uploaded filename with contain the complete path - coordinationMetadata.getUploadedFilename(), - settingMetadata.getUploadedFilename(), - new BlobPath().add(GLOBAL_METADATA_PATH_TOKEN).buildAsString() + "global_metadata.dat" - ) + verify(container).deleteBlobsAsyncIgnoringIfNotExists( + eq( + List.of( + // coordination/setting metadata is from CODEC_V2, the uploaded filename with contain the complete path + coordinationMetadata.getUploadedFilename(), + settingMetadata.getUploadedFilename(), + new BlobPath().add(GLOBAL_METADATA_PATH_TOKEN).buildAsString() + "global_metadata.dat" + ) + ), + any() + ); + verify(container).deleteBlobsAsyncIgnoringIfNotExists( + eq(List.of(getFormattedIndexFileName(index1Metadata.getUploadedFilePath()))), + any() ); - verify(container).deleteBlobsIgnoringIfNotExists(List.of(getFormattedIndexFileName(index1Metadata.getUploadedFilePath()))); - verify(container).deleteBlobsIgnoringIfNotExists(List.of("restore_file1", "snapshot_file1")); + verify(container).deleteBlobsAsyncIgnoringIfNotExists(eq(List.of("restore_file1", "snapshot_file1")), any()); Set staleManifest = new HashSet<>(); inactiveBlobs.forEach( blob -> staleManifest.add( @@ -298,8 +323,9 @@ public void testDeleteClusterMetadata() throws IOException { .name() ) ); - verify(container).deleteBlobsIgnoringIfNotExists(new ArrayList<>(staleManifest)); + verify(container).deleteBlobsAsyncIgnoringIfNotExists(eq(new ArrayList<>(staleManifest)), any()); verify(remoteRoutingTableService).deleteStaleIndexRoutingPaths(List.of(index3Metadata.getUploadedFilename())); + verifyNoMoreInteractions(container); } public void testDeleteStaleIndicesRoutingDiffFile() throws IOException { @@ -347,8 +373,13 @@ public void testDeleteStaleIndicesRoutingDiffFile() throws IOException { .diffManifest(diffManifest2) .build(); - BlobContainer blobContainer = mock(BlobContainer.class); - doThrow(IOException.class).when(blobContainer).delete(); + AsyncMultiStreamBlobContainer blobContainer = mock(AsyncMultiStreamBlobContainer.class); + doAnswer(invocation -> { + ActionListener listener = invocation.getArgument(1); + listener.onResponse(null); + return null; + }).when(blobContainer).deleteBlobsAsyncIgnoringIfNotExists(any(), any()); + when(blobStore.blobContainer(any())).thenReturn(blobContainer); BlobPath blobPath = new BlobPath().add("random-path"); when((blobStoreRepository.basePath())).thenReturn(blobPath); @@ -411,8 +442,13 @@ public void testDeleteClusterMetadataNoOpsRoutingTableService() throws IOExcepti .indicesRouting(indicesRouting2) .build(); - BlobContainer blobContainer = mock(BlobContainer.class); - doThrow(IOException.class).when(blobContainer).delete(); + AsyncMultiStreamBlobContainer blobContainer = mock(AsyncMultiStreamBlobContainer.class); + doAnswer(invocation -> { + ActionListener listener = invocation.getArgument(1); + listener.onResponse(null); + return null; + }).when(blobContainer).deleteBlobsAsyncIgnoringIfNotExists(any(), any()); + when(blobStore.blobContainer(any())).thenReturn(blobContainer); BlobPath blobPath = new BlobPath().add("random-path"); when((blobStoreRepository.basePath())).thenReturn(blobPath); @@ -557,9 +593,13 @@ public void testIndexRoutingFilesCleanupFailureStats() throws Exception { .indicesRouting(indicesRouting2) .build(); - BlobContainer blobContainer = mock(BlobContainer.class); - doThrow(IOException.class).when(blobContainer).delete(); + AsyncMultiStreamBlobContainer blobContainer = mock(AsyncMultiStreamBlobContainer.class); when(blobStore.blobContainer(any())).thenReturn(blobContainer); + doAnswer(invocation -> { + ActionListener listener = invocation.getArgument(1); + listener.onResponse(null); + return null; + }).when(blobContainer).deleteBlobsAsyncIgnoringIfNotExists(any(), any()); BlobPath blobPath = new BlobPath().add("random-path"); when((blobStoreRepository.basePath())).thenReturn(blobPath); @@ -580,8 +620,26 @@ public void testIndexRoutingFilesCleanupFailureStats() throws Exception { assertEquals(0, remoteClusterStateCleanupManager.getStats().getIndexRoutingFilesCleanupAttemptFailedCount()); }); + doAnswer(invocation -> { + List filesToDelete = invocation.getArgument(0); + if (!filesToDelete.isEmpty()) { + assertFalse( + "Attempted to delete manifests before index routing files " + filesToDelete, + filesToDelete.getFirst().contains("manifest") + ); + } + ActionListener listener = invocation.getArgument(1); + listener.onResponse(null); + return null; + }).when(blobContainer).deleteBlobsAsyncIgnoringIfNotExists(any(), any()); + doThrow(IOException.class).when(remoteRoutingTableService).deleteStaleIndexRoutingPaths(any()); - remoteClusterStateCleanupManager.deleteClusterMetadata(clusterName, clusterUUID, activeBlobs, inactiveBlobs); + + assertThrows( + IOException.class, + () -> remoteClusterStateCleanupManager.deleteClusterMetadata(clusterName, clusterUUID, activeBlobs, inactiveBlobs) + ); + assertBusy(() -> { // wait for stats to get updated assertNotNull(remoteClusterStateCleanupManager.getStats()); @@ -634,8 +692,12 @@ public void testIndicesRoutingDiffFilesCleanupFailureStats() throws Exception { .diffManifest(diffManifest2) .build(); - BlobContainer blobContainer = mock(BlobContainer.class); - doThrow(IOException.class).when(blobContainer).delete(); + AsyncMultiStreamBlobContainer blobContainer = mock(AsyncMultiStreamBlobContainer.class); + doAnswer(invocation -> { + ActionListener listener = invocation.getArgument(1); + listener.onResponse(null); + return null; + }).when(blobContainer).deleteBlobsAsyncIgnoringIfNotExists(any(), any()); when(blobStore.blobContainer(any())).thenReturn(blobContainer); BlobPath blobPath = new BlobPath().add("random-path"); @@ -658,7 +720,25 @@ public void testIndicesRoutingDiffFilesCleanupFailureStats() throws Exception { }); doThrow(IOException.class).when(remoteRoutingTableService).deleteStaleIndexRoutingDiffPaths(any()); - remoteClusterStateCleanupManager.deleteClusterMetadata(clusterName, clusterUUID, activeBlobs, inactiveBlobs); + + doAnswer(invocation -> { + List filesToDelete = invocation.getArgument(0); + if (!filesToDelete.isEmpty()) { + assertFalse( + "Attempted to delete manifests before index routing files " + filesToDelete, + filesToDelete.getFirst().contains("manifest") + ); + } + ActionListener listener = invocation.getArgument(1); + listener.onResponse(null); + return null; + }).when(blobContainer).deleteBlobsAsyncIgnoringIfNotExists(any(), any()); + + assertThrows( + IOException.class, + () -> remoteClusterStateCleanupManager.deleteClusterMetadata(clusterName, clusterUUID, activeBlobs, inactiveBlobs) + ); + assertBusy(() -> { // wait for stats to get updated assertNotNull(remoteClusterStateCleanupManager.getStats()); @@ -667,7 +747,7 @@ public void testIndicesRoutingDiffFilesCleanupFailureStats() throws Exception { } public void testSingleConcurrentExecutionOfStaleManifestCleanup() throws Exception { - BlobContainer blobContainer = mock(BlobContainer.class); + AsyncMultiStreamBlobContainer blobContainer = mock(AsyncMultiStreamBlobContainer.class); when(blobStore.blobContainer(any())).thenReturn(blobContainer); CountDownLatch latch = new CountDownLatch(1); @@ -677,18 +757,13 @@ public void testSingleConcurrentExecutionOfStaleManifestCleanup() throws Excepti if (latch.await(5000, TimeUnit.SECONDS) == false) { throw new Exception("Timed out waiting for delete task queuing to complete"); } - return null; + return Collections.emptyList(); }).when(blobContainer) - .listBlobsByPrefixInSortedOrder( - any(String.class), - any(int.class), - any(BlobContainer.BlobNameSortOrder.class), - any(ActionListener.class) - ); + .listBlobsByPrefixInSortedOrder(any(String.class), any(int.class), any(BlobContainer.BlobNameSortOrder.class)); remoteClusterStateCleanupManager.start(); - remoteClusterStateCleanupManager.deleteStaleClusterMetadata("cluster-name", "cluster-uuid", RETAINED_MANIFESTS); - remoteClusterStateCleanupManager.deleteStaleClusterMetadata("cluster-name", "cluster-uuid", RETAINED_MANIFESTS); + remoteClusterStateCleanupManager.deleteStaleClusterMetadata("cluster-name", "cluster-uuid", RETAINED_MANIFESTS, 10); + remoteClusterStateCleanupManager.deleteStaleClusterMetadata("cluster-name", "cluster-uuid", RETAINED_MANIFESTS, 10); latch.countDown(); assertBusy(() -> assertEquals(1, callCount.get())); @@ -721,19 +796,33 @@ public void testRemoteCleanupTaskScheduled() { assertFalse(remoteClusterStateCleanupManager.getStaleFileDeletionTask().isClosed()); } + public void testUpdateCleanupIntervalDoesNotRescheduleWhenTaskScheduled() { + remoteClusterStateCleanupManager.start(); + + assertTrue(remoteClusterStateCleanupManager.getStaleFileDeletionTask().isScheduled()); + + TimeValue newInterval = TimeValue.timeValueMinutes(10); + Settings newSettings = Settings.builder().put("cluster.remote_store.state.cleanup_interval", newInterval).build(); + clusterSettings.applySettings(newSettings); + + assertTrue(remoteClusterStateCleanupManager.getStaleFileDeletionTask().isScheduled()); + assertEquals(newInterval, remoteClusterStateCleanupManager.getStaleFileDeletionTask().getInterval()); + } + public void testRemoteCleanupSkipsOnOnlyElectedClusterManager() { DiscoveryNodes nodes = mock(DiscoveryNodes.class); when(nodes.isLocalNodeElectedClusterManager()).thenReturn(false); when(clusterState.nodes()).thenReturn(nodes); - RemoteClusterStateCleanupManager spyManager = spy(remoteClusterStateCleanupManager); + RemoteClusterStateCleanupManager cleanUpManager = spy(remoteClusterStateCleanupManager); AtomicInteger callCount = new AtomicInteger(0); - doAnswer(invocation -> callCount.incrementAndGet()).when(spyManager).deleteStaleClusterMetadata(any(), any(), anyInt()); - spyManager.cleanUpStaleFiles(); + doAnswer(invocation -> callCount.incrementAndGet()).when(cleanUpManager) + .deleteStaleClusterMetadata(any(), any(), anyInt(), anyLong()); + cleanUpManager.cleanUpStaleFiles(); assertEquals(0, callCount.get()); when(nodes.isLocalNodeElectedClusterManager()).thenReturn(true); when(clusterState.version()).thenReturn(randomLongBetween(11, 20)); - spyManager.cleanUpStaleFiles(); + cleanUpManager.cleanUpStaleFiles(); assertEquals(1, callCount.get()); } @@ -745,11 +834,12 @@ public void testRemoteCleanupSkipsIfVersionIncrementLessThanThreshold() { when(clusterState.nodes()).thenReturn(nodes); when(clusterState.version()).thenReturn(version); - RemoteClusterStateCleanupManager spyManager = spy(remoteClusterStateCleanupManager); + RemoteClusterStateCleanupManager cleanUpManager = spy(remoteClusterStateCleanupManager); AtomicInteger callCount = new AtomicInteger(0); - doAnswer(invocation -> callCount.incrementAndGet()).when(spyManager).deleteStaleClusterMetadata(any(), any(), anyInt()); + doAnswer(invocation -> callCount.incrementAndGet()).when(cleanUpManager) + .deleteStaleClusterMetadata(any(), any(), anyInt(), anyLong()); - remoteClusterStateCleanupManager.cleanUpStaleFiles(); + cleanUpManager.cleanUpStaleFiles(); assertEquals(0, callCount.get()); } @@ -761,24 +851,25 @@ public void testRemoteCleanupCallsDeleteIfVersionIncrementGreaterThanThreshold() when(clusterState.nodes()).thenReturn(nodes); when(clusterState.version()).thenReturn(version); - RemoteClusterStateCleanupManager spyManager = spy(remoteClusterStateCleanupManager); + RemoteClusterStateCleanupManager cleanUpManager = spy(remoteClusterStateCleanupManager); AtomicInteger callCount = new AtomicInteger(0); - doAnswer(invocation -> callCount.incrementAndGet()).when(spyManager).deleteStaleClusterMetadata(any(), any(), anyInt()); + doAnswer(invocation -> callCount.incrementAndGet()).when(cleanUpManager) + .deleteStaleClusterMetadata(any(), any(), anyInt(), anyLong()); // using spied cleanup manager so that stubbed deleteStaleClusterMetadata is called - spyManager.cleanUpStaleFiles(); + cleanUpManager.cleanUpStaleFiles(); assertEquals(1, callCount.get()); } public void testRemoteCleanupSchedulesEvenAfterFailure() { remoteClusterStateCleanupManager.start(); - RemoteClusterStateCleanupManager spyManager = spy(remoteClusterStateCleanupManager); + RemoteClusterStateCleanupManager cleanUpManager = spy(remoteClusterStateCleanupManager); AtomicInteger callCount = new AtomicInteger(0); doAnswer(invocationOnMock -> { callCount.incrementAndGet(); throw new RuntimeException("Test exception"); - }).when(spyManager).cleanUpStaleFiles(); - AsyncStaleFileDeletion task = new AsyncStaleFileDeletion(spyManager); + }).when(cleanUpManager).cleanUpStaleFiles(); + AsyncStaleFileDeletion task = new AsyncStaleFileDeletion(cleanUpManager); assertTrue(task.isScheduled()); task.run(); // Task is still scheduled after the failure @@ -790,4 +881,364 @@ public void testRemoteCleanupSchedulesEvenAfterFailure() { assertTrue(task.isScheduled()); assertEquals(2, callCount.get()); } + + public void testRemoteClusterStateCleanupMaxBatchesSetting() { + remoteClusterStateCleanupManager.start(); + + assertEquals( + RemoteClusterStateCleanupManager.MANIFEST_CLEANUP_MAX_BATCHES_DEFAULT, + clusterSettings.get(RemoteClusterStateCleanupManager.REMOTE_CLUSTER_STATE_CLEANUP_MAX_BATCHES_SETTING).intValue() + ); + + int maxBatches = randomIntBetween(10, 500); + Settings newSettings = Settings.builder().put("cluster.remote_store.state.cleanup.max_batches", maxBatches).build(); + clusterSettings.applySettings(newSettings); + assertEquals( + maxBatches, + clusterSettings.get(RemoteClusterStateCleanupManager.REMOTE_CLUSTER_STATE_CLEANUP_MAX_BATCHES_SETTING).intValue() + ); + } + + public void testBatchSizeSettingUpdate() { + remoteClusterStateCleanupManager.start(); + + assertEquals(MANIFEST_CLEANUP_BATCH_SIZE_DEFAULT, clusterSettings.get(REMOTE_CLUSTER_STATE_CLEANUP_BATCH_SIZE_SETTING).intValue()); + + int newBatchSize = randomIntBetween(100, 5000); + Settings newSettings = Settings.builder().put(REMOTE_CLUSTER_STATE_CLEANUP_BATCH_SIZE_SETTING.getKey(), newBatchSize).build(); + clusterSettings.applySettings(newSettings); + + assertEquals(newBatchSize, clusterSettings.get(REMOTE_CLUSTER_STATE_CLEANUP_BATCH_SIZE_SETTING).intValue()); + } + + public void testMaxBatchesSettingUpdate() { + remoteClusterStateCleanupManager.start(); + + assertEquals( + MANIFEST_CLEANUP_MAX_BATCHES_DEFAULT, + clusterSettings.get(REMOTE_CLUSTER_STATE_CLEANUP_MAX_BATCHES_SETTING).intValue() + ); + + int newMaxBatches = randomIntBetween(10, 500); + Settings newSettings = Settings.builder().put(REMOTE_CLUSTER_STATE_CLEANUP_MAX_BATCHES_SETTING.getKey(), newMaxBatches).build(); + clusterSettings.applySettings(newSettings); + + assertEquals(newMaxBatches, clusterSettings.get(REMOTE_CLUSTER_STATE_CLEANUP_MAX_BATCHES_SETTING).intValue()); + } + + public void testBatchedDeletionWithSingleBatch() throws Exception { + String clusterName = "test-cluster"; + String clusterUUID = "test-uuid"; + int manifestsToRetain = RETAINED_MANIFESTS; + + List manifests = new ArrayList<>(); + for (int i = 0; i < 15; i++) { + manifests.add(new PlainBlobMetadata("manifest" + i + ".dat", 1L)); + } + + when(remoteManifestManager.getManifestFolderPath(eq(clusterName), eq(clusterUUID))).thenReturn( + new BlobPath().add(encodeString(clusterName)).add(CLUSTER_STATE_PATH_TOKEN).add(clusterUUID).add(MANIFEST) + ); + + RemoteClusterStateCleanupManager cleanUpManager = spy(remoteClusterStateCleanupManager); + BlobStoreTransferService mockTransferService = mock(BlobStoreTransferService.class); + when(mockTransferService.listAllInSortedOrder(any(), eq(MANIFEST), anyInt())).thenReturn( + manifests, + manifests.subList(0, manifestsToRetain) + ); + + CountDownLatch latch = new CountDownLatch(1); + doAnswer(invocation -> { + latch.countDown(); + return null; + }).when(cleanUpManager).deleteClusterMetadata(anyString(), anyString(), any(), any()); + + doAnswer(invocation -> mockTransferService).when(cleanUpManager).getBlobStoreTransferService(); + + cleanUpManager.start(); + cleanUpManager.deleteStaleClusterMetadata(clusterName, clusterUUID, manifestsToRetain, 10); + + assertTrue(latch.await(5, TimeUnit.SECONDS)); + + verify(cleanUpManager, times(1)).deleteClusterMetadata( + eq(clusterName), + eq(clusterUUID), + eq(manifests.subList(0, manifestsToRetain)), + eq(manifests.subList(manifestsToRetain, manifests.size())) + ); + } + + public void testBatchedDeletionWithMultipleBatches() throws Exception { + String clusterName = "test-cluster"; + String clusterUUID = "test-uuid"; + int manifestsToRetain = RETAINED_MANIFESTS; + int batchSize = 50; + int maxBatches = 3; + + Settings newSettings = Settings.builder() + .put(REMOTE_CLUSTER_STATE_CLEANUP_BATCH_SIZE_SETTING.getKey(), batchSize) + .put(REMOTE_CLUSTER_STATE_CLEANUP_MAX_BATCHES_SETTING.getKey(), maxBatches) + .build(); + clusterSettings.applySettings(newSettings); + + when(remoteManifestManager.getManifestFolderPath(eq(clusterName), eq(clusterUUID))).thenReturn( + new BlobPath().add(encodeString(clusterName)).add(CLUSTER_STATE_PATH_TOKEN).add(clusterUUID).add(MANIFEST) + ); + + RemoteClusterStateCleanupManager cleanUpManager = spy(remoteClusterStateCleanupManager); + BlobStoreTransferService mockTransferService = mock(BlobStoreTransferService.class); + + List batch1 = new ArrayList<>(); + List batch2 = new ArrayList<>(); + List batch3 = new ArrayList<>(); + + for (int i = 0; i < batchSize; i++) { + batch1.add(new PlainBlobMetadata("manifest" + i + ".dat", 1L)); + batch2.add(new PlainBlobMetadata("manifest" + (i + batchSize) + ".dat", 1L)); + batch3.add(new PlainBlobMetadata("manifest" + (i + 2 * batchSize) + ".dat", 1L)); + } + + when(mockTransferService.listAllInSortedOrder(any(), eq(MANIFEST), eq(batchSize))).thenReturn(batch1, batch2, batch3); + + CountDownLatch latch = new CountDownLatch(maxBatches); + doAnswer(invocation -> { + latch.countDown(); + return null; + }).when(cleanUpManager).deleteClusterMetadata(anyString(), anyString(), any(), any()); + + doAnswer(invocation -> mockTransferService).when(cleanUpManager).getBlobStoreTransferService(); + + cleanUpManager.start(); + cleanUpManager.deleteStaleClusterMetadata(clusterName, clusterUUID, manifestsToRetain, 10); + + assertTrue(latch.await(5, TimeUnit.SECONDS)); + + verify(cleanUpManager, times(1)).deleteClusterMetadata( + eq(clusterName), + eq(clusterUUID), + eq(batch1.subList(0, manifestsToRetain)), + eq(batch1.subList(manifestsToRetain, batch1.size())) + ); + verify(cleanUpManager, times(1)).deleteClusterMetadata( + eq(clusterName), + eq(clusterUUID), + eq(batch2.subList(0, manifestsToRetain)), + eq(batch2.subList(manifestsToRetain, batch2.size())) + ); + verify(cleanUpManager, times(1)).deleteClusterMetadata( + eq(clusterName), + eq(clusterUUID), + eq(batch3.subList(0, manifestsToRetain)), + eq(batch3.subList(manifestsToRetain, batch3.size())) + ); + } + + public void testBatchedDeletionStopsWhenManifestsLessThanRetained() throws Exception { + String clusterName = "test-cluster"; + String clusterUUID = "test-uuid"; + int manifestsToRetain = RETAINED_MANIFESTS; + + List manifests = new ArrayList<>(); + for (int i = 0; i < manifestsToRetain - 2; i++) { + manifests.add(new PlainBlobMetadata("manifest" + i + ".dat", 1L)); + } + + when(remoteManifestManager.getManifestFolderPath(eq(clusterName), eq(clusterUUID))).thenReturn( + new BlobPath().add(encodeString(clusterName)).add(CLUSTER_STATE_PATH_TOKEN).add(clusterUUID).add(MANIFEST) + ); + + RemoteClusterStateCleanupManager cleanUpManager = spy(remoteClusterStateCleanupManager); + BlobStoreTransferService mockTransferService = mock(BlobStoreTransferService.class); + when(mockTransferService.listAllInSortedOrder(any(), eq(MANIFEST), anyInt())).thenReturn(manifests); + + CountDownLatch completionLatch = new CountDownLatch(1); + doAnswer(invocation -> { + fail("deleteClusterMetadata should not be called when manifests < retained"); + return null; + }).when(cleanUpManager).deleteClusterMetadata(anyString(), anyString(), any(), any()); + + doAnswer(invocation -> mockTransferService).when(cleanUpManager).getBlobStoreTransferService(); + + threadPool.executor(ThreadPool.Names.REMOTE_PURGE).execute(() -> { + try { + cleanUpManager.deleteStaleClusterMetadata(clusterName, clusterUUID, manifestsToRetain, 10); + } finally { + completionLatch.countDown(); + } + }); + + cleanUpManager.start(); + + assertTrue(completionLatch.await(5, TimeUnit.SECONDS)); + + verify(cleanUpManager, times(0)).deleteClusterMetadata(anyString(), anyString(), any(), any()); + } + + public void testBatchedDeletionExhaustsMaxBatches() throws Exception { + String clusterName = "test-cluster"; + String clusterUUID = "test-uuid"; + int manifestsToRetain = RETAINED_MANIFESTS; + int batchSize = 20; + int maxBatches = 2; + + Settings newSettings = Settings.builder() + .put(REMOTE_CLUSTER_STATE_CLEANUP_BATCH_SIZE_SETTING.getKey(), batchSize) + .put(REMOTE_CLUSTER_STATE_CLEANUP_MAX_BATCHES_SETTING.getKey(), maxBatches) + .build(); + clusterSettings.applySettings(newSettings); + + when(remoteManifestManager.getManifestFolderPath(eq(clusterName), eq(clusterUUID))).thenReturn( + new BlobPath().add(encodeString(clusterName)).add(CLUSTER_STATE_PATH_TOKEN).add(clusterUUID).add(MANIFEST) + ); + + RemoteClusterStateCleanupManager cleanUpManager = spy(remoteClusterStateCleanupManager); + BlobStoreTransferService mockTransferService = mock(BlobStoreTransferService.class); + + List batch = new ArrayList<>(); + for (int i = 0; i < batchSize; i++) { + batch.add(new PlainBlobMetadata("manifest" + i + ".dat", 1L)); + } + + when(mockTransferService.listAllInSortedOrder(any(), eq(MANIFEST), eq(batchSize))).thenReturn(batch); + + CountDownLatch latch = new CountDownLatch(maxBatches); + doAnswer(invocation -> { + latch.countDown(); + return null; + }).when(cleanUpManager).deleteClusterMetadata(anyString(), anyString(), any(), any()); + + doAnswer(invocation -> mockTransferService).when(cleanUpManager).getBlobStoreTransferService(); + + cleanUpManager.start(); + cleanUpManager.deleteStaleClusterMetadata(clusterName, clusterUUID, manifestsToRetain, 10); + + assertTrue(latch.await(5, TimeUnit.SECONDS)); + + verify(cleanUpManager, times(maxBatches)).deleteClusterMetadata( + eq(clusterName), + eq(clusterUUID), + eq(batch.subList(0, manifestsToRetain)), + eq(batch.subList(manifestsToRetain, batch.size())) + ); + } + + public void testBatchedDeletionWithOneNewManifest() throws Exception { + String clusterName = "test-cluster"; + String clusterUUID = "test-uuid"; + int manifestsToRetain = RETAINED_MANIFESTS; + int batchSize = 100; + + Settings newSettings = Settings.builder().put(REMOTE_CLUSTER_STATE_CLEANUP_BATCH_SIZE_SETTING.getKey(), batchSize).build(); + clusterSettings.applySettings(newSettings); + + List batch1 = new ArrayList<>(); + List batch2 = new ArrayList<>(); + for (int i = 100; i >= 1; i--) { + batch1.add(new PlainBlobMetadata("manifest-" + i + ".dat", 1L)); + } + for (int i = 101; i >= 91; i--) { + batch2.add(new PlainBlobMetadata("manifest-" + i + ".dat", 1L)); + } + + when(remoteManifestManager.getManifestFolderPath(eq(clusterName), eq(clusterUUID))).thenReturn( + new BlobPath().add(encodeString(clusterName)).add(CLUSTER_STATE_PATH_TOKEN).add(clusterUUID).add(MANIFEST) + ); + + RemoteClusterStateCleanupManager cleanUpManager = spy(remoteClusterStateCleanupManager); + BlobStoreTransferService mockTransferService = mock(BlobStoreTransferService.class); + when(mockTransferService.listAllInSortedOrder(any(), eq(MANIFEST), eq(batchSize))).thenReturn( + batch1, + batch2, + batch2.subList(1, batch2.size()) + ); + + doAnswer(invocation -> mockTransferService).when(cleanUpManager).getBlobStoreTransferService(); + + CountDownLatch latch = new CountDownLatch(2); + doAnswer(invocation -> { + latch.countDown(); + return null; + }).when(cleanUpManager).deleteClusterMetadata(anyString(), anyString(), any(), any()); + + cleanUpManager.start(); + cleanUpManager.deleteStaleClusterMetadata(clusterName, clusterUUID, manifestsToRetain, 20); + + assertTrue(latch.await(5, TimeUnit.SECONDS)); + + verify(cleanUpManager, times(1)).deleteClusterMetadata( + eq(clusterName), + eq(clusterUUID), + eq(batch1.subList(0, manifestsToRetain)), + eq(batch1.subList(manifestsToRetain, batch1.size())) + ); + + verify(cleanUpManager, times(1)).deleteClusterMetadata( + anyString(), + anyString(), + eq(batch2.subList(0, manifestsToRetain)), + eq(batch2.subList(manifestsToRetain, batch2.size())) + ); + } + + public void testLastCleanupAttemptStateVersionNotUpdatedOnFailure() throws Exception { + when(nodes.isLocalNodeElectedClusterManager()).thenReturn(true); + when(clusterState.version()).thenReturn(20L); + + RemoteClusterStateCleanupManager spyManager = spy(remoteClusterStateCleanupManager); + doThrow(new IOException("Test failure")).when(spyManager).deleteClusterMetadata(any(), any(), any(), any()); + + assertEquals(0L, spyManager.getLastCleanupAttemptStateVersion()); + spyManager.cleanUpStaleFiles(); + assertEquals(0L, spyManager.getLastCleanupAttemptStateVersion()); + } + + public void testLastCleanupAttemptStateVersionUpdatedOnlyAfterCompletion() throws Exception { + String clusterName = "test-cluster"; + String clusterUUID = "test-uuid"; + long stateVersion = 20L; + + List manifests = new ArrayList<>(); + for (int i = 0; i < 15; i++) { + manifests.add(new PlainBlobMetadata("manifest" + i + ".dat", 1L)); + } + + when(remoteManifestManager.getManifestFolderPath(eq(clusterName), eq(clusterUUID))).thenReturn(new BlobPath().add("test")); + + RemoteClusterStateCleanupManager spyManager = spy(remoteClusterStateCleanupManager); + BlobStoreTransferService mockTransferService = mock(BlobStoreTransferService.class); + when(mockTransferService.listAllInSortedOrder(any(), eq(MANIFEST), anyInt())).thenReturn( + manifests, + manifests.subList(0, RETAINED_MANIFESTS) + ); + + when(spyManager.getBlobStoreTransferService()).thenReturn(mockTransferService); + + CountDownLatch latch = new CountDownLatch(1); + doAnswer(invocation -> { + assertEquals(0L, spyManager.getLastCleanupAttemptStateVersion()); + latch.countDown(); + return null; + }).when(spyManager).deleteClusterMetadata(any(), any(), any(), any()); + + assertEquals(0L, spyManager.getLastCleanupAttemptStateVersion()); + spyManager.start(); + spyManager.deleteStaleClusterMetadata(clusterName, clusterUUID, RETAINED_MANIFESTS, stateVersion); + + assertTrue(latch.await(5, TimeUnit.SECONDS)); + assertBusy(() -> assertEquals(stateVersion, spyManager.getLastCleanupAttemptStateVersion())); + } + + public void testBatchSizeValidation() { + int invalidBatchSize = randomIntBetween(1, RETAINED_MANIFESTS); + Settings invalidSettings = Settings.builder() + .put(REMOTE_CLUSTER_STATE_CLEANUP_BATCH_SIZE_SETTING.getKey(), invalidBatchSize) + .build(); + + IllegalArgumentException exception = expectThrows( + IllegalArgumentException.class, + () -> { clusterSettings.applySettings(invalidSettings); } + ); + + assertTrue(exception.getCause().getMessage().contains("Cleanup Batch Size should be greater than [" + RETAINED_MANIFESTS + "]")); + } } From a5f5bc79508c4eda949ea137e7bc4f0b866b2422 Mon Sep 17 00:00:00 2001 From: Pranshu Shukla Date: Fri, 13 Feb 2026 14:20:50 +0530 Subject: [PATCH 02/12] Correcting CHANGELOG.md Signed-off-by: Pranshu Shukla --- CHANGELOG.md | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8cd1257a96f44..8aad9ac5ac2b5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,12 +21,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), - Update EncryptedBlobContainer to adhere limits while listing blobs in specific sort order if wrapped blob container supports ([#20514](https://github.com/opensearch-project/OpenSearch/pull/20514)) - [segment replication] Fix segment replication infinite retry due to stale metadata checkpoint ([#20551](https://github.com/opensearch-project/OpenSearch/pull/20551)) - Changing opensearch.cgroups.hierarchy.override causes java.lang.SecurityException exception ([#20565](https://github.com/opensearch-project/OpenSearch/pull/20565)) -- Fix CriteriaBasedCodec to work with delegate codec. ([20442](https://github.com/opensearch-project/OpenSearch/pull/20442)) -- Update RemoteClusterStateCleanupManager to performed batched deletions of stale ClusterMetadataManifests and address deletion timeout issues ([#20566](https://github.com/opensearch-project/OpenSearch/pull/20566)) - Fix CriteriaBasedCodec to work with delegate codec. ([#20442](https://github.com/opensearch-project/OpenSearch/pull/20442)) - 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)) ### Dependencies - Bump `ch.qos.logback:logback-core` and `ch.qos.logback:logback-classic` from 1.5.24 to 1.5.27 ([#20525](https://github.com/opensearch-project/OpenSearch/pull/20525)) From a4ac4cc818900be362db84bf1266360fb19f1eb2 Mon Sep 17 00:00:00 2001 From: Pranshu Shukla Date: Fri, 13 Feb 2026 15:18:38 +0530 Subject: [PATCH 03/12] Addressing comments Signed-off-by: Pranshu Shukla --- .../InternalRemoteRoutingTableService.java | 22 +++++++++++++------ .../RemoteClusterStateCleanupManager.java | 15 ++++++++----- 2 files changed, 24 insertions(+), 13 deletions(-) diff --git a/server/src/main/java/org/opensearch/cluster/routing/remote/InternalRemoteRoutingTableService.java b/server/src/main/java/org/opensearch/cluster/routing/remote/InternalRemoteRoutingTableService.java index dd1974017a0f0..5b0e22283722d 100644 --- a/server/src/main/java/org/opensearch/cluster/routing/remote/InternalRemoteRoutingTableService.java +++ b/server/src/main/java/org/opensearch/cluster/routing/remote/InternalRemoteRoutingTableService.java @@ -20,6 +20,7 @@ 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; @@ -279,7 +280,13 @@ protected void doStop() {} public void deleteStaleIndexRoutingPaths(List 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; @@ -289,10 +296,11 @@ public void deleteStaleIndexRoutingPaths(List stalePaths) throws IOExcep public void deleteStaleIndexRoutingDiffPaths(List stalePaths) throws IOException { try { logger.debug(() -> "Deleting stale index routing diff files from remote - " + stalePaths); + BlobContainer blobContainerForDeletion = blobStoreRepository.blobStore().blobContainer(BlobPath.cleanPath()); if (blobStoreRepository.blobStore().blobContainer(BlobPath.cleanPath()) instanceof AsyncMultiStreamBlobContainer) { - deleteAsyncInternal(stalePaths); + deleteAsyncInternal((AsyncMultiStreamBlobContainer) blobContainerForDeletion, stalePaths, DEFAULT_DELETION_TIMEOUT); } else { - blobStoreRepository.blobStore().blobContainer(BlobPath.cleanPath()).deleteBlobsIgnoringIfNotExists(stalePaths); + blobContainerForDeletion.deleteBlobsIgnoringIfNotExists(stalePaths); } } catch (IOException e) { logger.error(() -> new ParameterizedMessage("Failed to delete some stale index routing diff paths from {}", stalePaths), e); @@ -300,12 +308,12 @@ public void deleteStaleIndexRoutingDiffPaths(List stalePaths) throws IOE } } - private void deleteAsyncInternal(List fileNames) throws IOException { + private void deleteAsyncInternal(AsyncMultiStreamBlobContainer blobContainerForDeletion, List fileNames, TimeValue timeout) + throws IOException { PlainActionFuture future = new PlainActionFuture<>(); try { - ((AsyncMultiStreamBlobContainer) blobStoreRepository.blobStore().blobContainer(BlobPath.cleanPath())) - .deleteBlobsAsyncIgnoringIfNotExists(fileNames, future); - future.get(DEFAULT_DELETION_TIMEOUT.seconds(), TimeUnit.SECONDS); + blobContainerForDeletion.deleteBlobsAsyncIgnoringIfNotExists(fileNames, future); + future.get(timeout.seconds(), TimeUnit.SECONDS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IOException("Future got interrupted", e); diff --git a/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateCleanupManager.java b/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateCleanupManager.java index 7278cec53d2d3..9fba0817a641c 100644 --- a/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateCleanupManager.java +++ b/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateCleanupManager.java @@ -17,6 +17,7 @@ import org.opensearch.cluster.service.ClusterApplierService; import org.opensearch.cluster.service.ClusterService; import org.opensearch.common.blobstore.AsyncMultiStreamBlobContainer; +import org.opensearch.common.blobstore.BlobContainer; import org.opensearch.common.blobstore.BlobMetadata; import org.opensearch.common.blobstore.BlobPath; import org.opensearch.common.settings.ClusterSettings; @@ -553,19 +554,21 @@ public void onFailure(Exception e) { // package private for testing void deleteStalePaths(List stalePaths) throws IOException { logger.debug(String.format(Locale.ROOT, "Deleting stale files from remote - %s", stalePaths)); - if (remoteClusterStateService.getBlobStore().blobContainer(BlobPath.cleanPath()) instanceof AsyncMultiStreamBlobContainer) { - deleteAsyncInternal(stalePaths); + BlobContainer blobContainerForDeletion = remoteClusterStateService.getBlobStore().blobContainer(BlobPath.cleanPath()); + assert blobContainerForDeletion != null; + if (blobContainerForDeletion instanceof AsyncMultiStreamBlobContainer) { + deleteAsyncInternal((AsyncMultiStreamBlobContainer) blobContainerForDeletion, stalePaths, DEFAULT_DELETION_TIMEOUT); } else { getBlobStoreTransferService().deleteBlobs(BlobPath.cleanPath(), stalePaths); } } - private void deleteAsyncInternal(List fileNames) throws IOException { + private void deleteAsyncInternal(AsyncMultiStreamBlobContainer blobContainerForDeletion, List fileNames, TimeValue timeout) + throws IOException { PlainActionFuture future = new PlainActionFuture<>(); try { - ((AsyncMultiStreamBlobContainer) remoteClusterStateService.getBlobStore().blobContainer(BlobPath.cleanPath())) - .deleteBlobsAsyncIgnoringIfNotExists(fileNames, future); - future.get(DEFAULT_DELETION_TIMEOUT.seconds(), TimeUnit.SECONDS); + blobContainerForDeletion.deleteBlobsAsyncIgnoringIfNotExists(fileNames, future); + future.get(timeout.seconds(), TimeUnit.SECONDS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IOException("Future got interrupted", e); From 2b29ab4ce400c3646534127fe109ecfd2e759d37 Mon Sep 17 00:00:00 2001 From: Pranshu Shukla Date: Fri, 13 Feb 2026 16:04:03 +0530 Subject: [PATCH 04/12] Retry Build Signed-off-by: Pranshu Shukla From 860af16114df8ec8da73c41ec2d61bea19554054 Mon Sep 17 00:00:00 2001 From: Pranshu Shukla Date: Fri, 13 Feb 2026 16:49:56 +0530 Subject: [PATCH 05/12] Retry Build Signed-off-by: Pranshu Shukla From c6734b2fce3459486a5d094d5758b807625ac43d Mon Sep 17 00:00:00 2001 From: Pranshu Shukla Date: Fri, 13 Feb 2026 18:46:01 +0530 Subject: [PATCH 06/12] Retry Build Signed-off-by: Pranshu Shukla From 0ee5ee37518b139d93ee60d20c12b5030bcc3c89 Mon Sep 17 00:00:00 2001 From: Pranshu Shukla Date: Fri, 13 Feb 2026 20:00:39 +0530 Subject: [PATCH 07/12] Retry Build Signed-off-by: Pranshu Shukla From be3ca40118073dc976506b9b8b13105f77e410e2 Mon Sep 17 00:00:00 2001 From: Pranshu Shukla Date: Sun, 15 Feb 2026 17:05:49 +0530 Subject: [PATCH 08/12] Increasing code coverage Signed-off-by: Pranshu Shukla --- .../InternalRemoteRoutingTableService.java | 7 +- .../RemoteRoutingTableServiceTests.java | 118 ++++++++++++++++++ 2 files changed, 120 insertions(+), 5 deletions(-) diff --git a/server/src/main/java/org/opensearch/cluster/routing/remote/InternalRemoteRoutingTableService.java b/server/src/main/java/org/opensearch/cluster/routing/remote/InternalRemoteRoutingTableService.java index 5b0e22283722d..40ca2ba83fb28 100644 --- a/server/src/main/java/org/opensearch/cluster/routing/remote/InternalRemoteRoutingTableService.java +++ b/server/src/main/java/org/opensearch/cluster/routing/remote/InternalRemoteRoutingTableService.java @@ -308,7 +308,7 @@ public void deleteStaleIndexRoutingDiffPaths(List stalePaths) throws IOE } } - private void deleteAsyncInternal(AsyncMultiStreamBlobContainer blobContainerForDeletion, List fileNames, TimeValue timeout) + protected void deleteAsyncInternal(AsyncMultiStreamBlobContainer blobContainerForDeletion, List fileNames, TimeValue timeout) throws IOException { PlainActionFuture future = new PlainActionFuture<>(); try { @@ -324,10 +324,7 @@ private void deleteAsyncInternal(AsyncMultiStreamBlobContainer blobContainerForD 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", DEFAULT_DELETION_TIMEOUT.seconds()), - e - ); + throw new IOException(String.format(Locale.ROOT, "Delete operation timed out after %s seconds", timeout.seconds()), e); } } } diff --git a/server/src/test/java/org/opensearch/cluster/routing/remote/RemoteRoutingTableServiceTests.java b/server/src/test/java/org/opensearch/cluster/routing/remote/RemoteRoutingTableServiceTests.java index 3564e10935722..f9c6ffb72d729 100644 --- a/server/src/test/java/org/opensearch/cluster/routing/remote/RemoteRoutingTableServiceTests.java +++ b/server/src/test/java/org/opensearch/cluster/routing/remote/RemoteRoutingTableServiceTests.java @@ -21,6 +21,7 @@ import org.opensearch.cluster.routing.RoutingTableIncrementalDiff; import org.opensearch.cluster.routing.StringKeyDiffProvider; import org.opensearch.cluster.service.ClusterService; +import org.opensearch.common.blobstore.AsyncMultiStreamBlobContainer; import org.opensearch.common.blobstore.BlobContainer; import org.opensearch.common.blobstore.BlobPath; import org.opensearch.common.blobstore.BlobStore; @@ -28,6 +29,7 @@ import org.opensearch.common.compress.DeflateCompressor; import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.settings.Settings; +import org.opensearch.common.unit.TimeValue; import org.opensearch.common.util.TestCapturingListener; import org.opensearch.core.action.ActionListener; import org.opensearch.core.compress.Compressor; @@ -835,4 +837,120 @@ public void testDeleteStaleIndexRoutingDiffPathsThrowsIOException() throws IOExc assertEquals("test exception", thrown.getMessage()); verify(blobContainer).deleteBlobsIgnoringIfNotExists(stalePaths); } + + public void testDeleteStaleIndexRoutingPathsWithMultiStreamBlobContainer() throws IOException { + AsyncMultiStreamBlobContainer container = mock(AsyncMultiStreamBlobContainer.class); + when(blobStore.blobContainer(any())).thenReturn(container); + doAnswer(invocation -> { + ActionListener listener = invocation.getArgument(1); + listener.onResponse(null); + return null; + }).when(container).deleteBlobsAsyncIgnoringIfNotExists(any(), any()); + + when(blobStore.blobContainer(any())).thenReturn(container); + List stalePaths = Arrays.asList("path1", "path2"); + remoteRoutingTableService.doStart(); + remoteRoutingTableService.deleteStaleIndexRoutingPaths(stalePaths); + verify(container).deleteBlobsAsyncIgnoringIfNotExists(eq(stalePaths), any()); + } + + public void testDeleteStaleIndexRoutingDiffPathsWithMultiStreamBlobContainer() throws IOException { + AsyncMultiStreamBlobContainer container = mock(AsyncMultiStreamBlobContainer.class); + when(blobStore.blobContainer(any())).thenReturn(container); + doAnswer(invocation -> { + ActionListener listener = invocation.getArgument(1); + listener.onResponse(null); + return null; + }).when(container).deleteBlobsAsyncIgnoringIfNotExists(any(), any()); + + List stalePaths = Arrays.asList("path1", "path2"); + remoteRoutingTableService.doStart(); + remoteRoutingTableService.deleteStaleIndexRoutingDiffPaths(stalePaths); + verify(container).deleteBlobsAsyncIgnoringIfNotExists(eq(stalePaths), any()); + } + + public void testDeleteAsyncInternalSuccess() throws Exception { + AsyncMultiStreamBlobContainer container = mock(AsyncMultiStreamBlobContainer.class); + List fileNames = Arrays.asList("file1", "file2"); + + doAnswer(invocation -> { + ActionListener listener = invocation.getArgument(1); + listener.onResponse(null); + return null; + }).when(container).deleteBlobsAsyncIgnoringIfNotExists(eq(fileNames), any()); + + Settings settings = Settings.builder() + .put("node.attr." + REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY, "routing_repository") + .put(REMOTE_PUBLICATION_SETTING_KEY, "true") + .build(); + InternalRemoteRoutingTableService testRemoteRoutingTableService = new InternalRemoteRoutingTableService( + repositoriesServiceSupplier, + settings, + new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS), + threadPool, + "test-cluster" + ); + + testRemoteRoutingTableService.doStart(); + testRemoteRoutingTableService.deleteAsyncInternal(container, fileNames, TimeValue.timeValueSeconds(10)); + + verify(container).deleteBlobsAsyncIgnoringIfNotExists(eq(fileNames), any()); + } + + public void testDeleteAsyncInternalExecutionException() throws Exception { + AsyncMultiStreamBlobContainer container = mock(AsyncMultiStreamBlobContainer.class); + List fileNames = Arrays.asList("file1", "file2"); + + doAnswer(invocation -> { + ActionListener listener = invocation.getArgument(1); + listener.onFailure(new IOException("deletion failed")); + return null; + }).when(container).deleteBlobsAsyncIgnoringIfNotExists(eq(fileNames), any()); + + Settings settings = Settings.builder() + .put("node.attr." + REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY, "routing_repository") + .put(REMOTE_PUBLICATION_SETTING_KEY, "true") + .build(); + InternalRemoteRoutingTableService testRemoteRoutingTableService = new InternalRemoteRoutingTableService( + repositoriesServiceSupplier, + settings, + new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS), + threadPool, + "test-cluster" + ); + + testRemoteRoutingTableService.doStart(); + IOException thrown = assertThrows(IOException.class, () -> { + testRemoteRoutingTableService.deleteAsyncInternal(container, fileNames, TimeValue.timeValueSeconds(10)); + }); + + assertEquals("deletion failed", thrown.getMessage()); + } + + public void testDeleteAsyncInternalTimeout() throws Exception { + AsyncMultiStreamBlobContainer container = mock(AsyncMultiStreamBlobContainer.class); + List fileNames = Arrays.asList("file1", "file2"); + + // Don't call listener to simulate timeout + doNothing().when(container).deleteBlobsAsyncIgnoringIfNotExists(eq(fileNames), any()); + + Settings settings = Settings.builder() + .put("node.attr." + REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY, "routing_repository") + .put(REMOTE_PUBLICATION_SETTING_KEY, "true") + .build(); + InternalRemoteRoutingTableService testRemoteRoutingTableService = new InternalRemoteRoutingTableService( + repositoriesServiceSupplier, + settings, + new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS), + threadPool, + "test-cluster" + ); + + testRemoteRoutingTableService.doStart(); + IOException thrown = assertThrows(IOException.class, () -> { + testRemoteRoutingTableService.deleteAsyncInternal(container, fileNames, TimeValue.timeValueMillis(1)); + }); + + assertTrue(thrown.getMessage().contains("Delete operation timed out")); + } } From a7d054d8c340dd2bf93e6495613759b02b58aeba Mon Sep 17 00:00:00 2001 From: Pranshu Shukla Date: Sun, 15 Feb 2026 18:43:38 +0530 Subject: [PATCH 09/12] Retry Build Signed-off-by: Pranshu Shukla From 9912dd1112ca57801790303e4e86840d27951a90 Mon Sep 17 00:00:00 2001 From: Pranshu Shukla Date: Sun, 15 Feb 2026 20:34:22 +0530 Subject: [PATCH 10/12] Retry Build Signed-off-by: Pranshu Shukla From 33f7d63d9a48aaeffe1bdd52d1ba6eeaa5b86956 Mon Sep 17 00:00:00 2001 From: Pranshu Shukla Date: Sun, 15 Feb 2026 21:32:31 +0530 Subject: [PATCH 11/12] Retry Build Signed-off-by: Pranshu Shukla From c04e8105cbbbebfa82e2698af874ee7252bdb515 Mon Sep 17 00:00:00 2001 From: Pranshu Shukla Date: Mon, 16 Feb 2026 08:13:38 +0530 Subject: [PATCH 12/12] Retry Build Signed-off-by: Pranshu Shukla