-
Notifications
You must be signed in to change notification settings - Fork 1.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: Ashish Singh <ssashish@amazon.com>
- Loading branch information
Showing
2 changed files
with
146 additions
and
83 deletions.
There are no files selected for viewing
123 changes: 123 additions & 0 deletions
123
plugins/repository-s3/src/main/java/org/opensearch/repositories/s3/S3AsyncDeleteHelper.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,123 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| */ | ||
|
|
||
| package org.opensearch.repositories.s3; | ||
|
|
||
| import software.amazon.awssdk.services.s3.S3AsyncClient; | ||
| import software.amazon.awssdk.services.s3.model.Delete; | ||
| import software.amazon.awssdk.services.s3.model.DeleteObjectsRequest; | ||
| import software.amazon.awssdk.services.s3.model.DeleteObjectsResponse; | ||
| import software.amazon.awssdk.services.s3.model.ObjectIdentifier; | ||
|
|
||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
| import org.apache.logging.log4j.message.ParameterizedMessage; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.concurrent.CompletableFuture; | ||
| import java.util.concurrent.CompletionException; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| public class S3AsyncDeleteHelper { | ||
| private static final Logger logger = LogManager.getLogger(S3AsyncDeleteHelper.class); | ||
|
|
||
| static CompletableFuture<Void> executeDeleteChain( | ||
| S3AsyncClient s3AsyncClient, | ||
| S3BlobStore blobStore, | ||
| List<String> objectsToDelete, | ||
| CompletableFuture<Void> currentChain, | ||
| boolean ignoreIfNotExists, | ||
| Runnable afterDeleteAction | ||
| ) { | ||
| List<List<String>> batches = createDeleteBatches(objectsToDelete, blobStore.getBulkDeletesSize()); | ||
| CompletableFuture<Void> newChain = currentChain.thenCompose( | ||
| v -> executeDeleteBatches(s3AsyncClient, blobStore, batches, ignoreIfNotExists) | ||
| ); | ||
| if (afterDeleteAction != null) { | ||
| newChain = newChain.thenRun(afterDeleteAction); | ||
| } | ||
| return newChain; | ||
| } | ||
|
|
||
| static List<List<String>> createDeleteBatches(List<String> keys, int bulkDeleteSize) { | ||
| List<List<String>> batches = new ArrayList<>(); | ||
| for (int i = 0; i < keys.size(); i += bulkDeleteSize) { | ||
| batches.add(keys.subList(i, Math.min(keys.size(), i + bulkDeleteSize))); | ||
| } | ||
| return batches; | ||
| } | ||
|
|
||
| private static CompletableFuture<Void> executeDeleteBatches( | ||
| S3AsyncClient s3AsyncClient, | ||
| S3BlobStore blobStore, | ||
| List<List<String>> batches, | ||
| boolean ignoreIfNotExists | ||
| ) { | ||
| CompletableFuture<Void> allDeletesFuture = CompletableFuture.completedFuture(null); | ||
|
|
||
| for (List<String> batch : batches) { | ||
| allDeletesFuture = allDeletesFuture.thenCompose( | ||
| v -> executeSingleDeleteBatch(s3AsyncClient, blobStore, batch, ignoreIfNotExists) | ||
| ); | ||
| } | ||
|
|
||
| return allDeletesFuture; | ||
| } | ||
|
|
||
| private static CompletableFuture<Void> executeSingleDeleteBatch( | ||
| S3AsyncClient s3AsyncClient, | ||
| S3BlobStore blobStore, | ||
| List<String> batch, | ||
| boolean ignoreIfNotExists | ||
| ) { | ||
| DeleteObjectsRequest deleteRequest = bulkDelete(blobStore.bucket(), batch, blobStore); | ||
| return s3AsyncClient.deleteObjects(deleteRequest) | ||
| .thenApply(response -> processDeleteResponse(response, ignoreIfNotExists)) | ||
| .exceptionally(e -> { | ||
| if (!ignoreIfNotExists) { | ||
| throw new CompletionException(e); | ||
| } | ||
| logger.warn("Error during batch deletion", e); | ||
| return null; | ||
| }); | ||
| } | ||
|
|
||
| private static Void processDeleteResponse(DeleteObjectsResponse deleteObjectsResponse, boolean ignoreIfNotExists) { | ||
| if (!deleteObjectsResponse.errors().isEmpty()) { | ||
| if (ignoreIfNotExists) { | ||
| logger.warn( | ||
| () -> new ParameterizedMessage( | ||
| "Failed to delete some blobs {}", | ||
| deleteObjectsResponse.errors() | ||
| .stream() | ||
| .map(s3Error -> "[" + s3Error.key() + "][" + s3Error.code() + "][" + s3Error.message() + "]") | ||
| .collect(Collectors.toList()) | ||
| ) | ||
| ); | ||
| } else { | ||
| throw new CompletionException(new IOException("Failed to delete some blobs: " + deleteObjectsResponse.errors())); | ||
| } | ||
| } | ||
| return null; | ||
| } | ||
|
|
||
| private static DeleteObjectsRequest bulkDelete(String bucket, List<String> blobs, S3BlobStore blobStore) { | ||
| return DeleteObjectsRequest.builder() | ||
| .bucket(bucket) | ||
| .delete( | ||
| Delete.builder() | ||
| .objects(blobs.stream().map(blob -> ObjectIdentifier.builder().key(blob).build()).collect(Collectors.toList())) | ||
| .quiet(true) | ||
| .build() | ||
| ) | ||
| .overrideConfiguration(o -> o.addMetricPublisher(blobStore.getStatsMetricPublisher().deleteObjectsMetricPublisher)) | ||
| .build(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters