Use Azure blob batch API to delete blobs in batches

gradle/verification-metadata.xml

@@ -144,6 +144,11 @@
             <sha256 value="31915426834400cac854f48441c168d55aa6fc054527f28f1d242a7067affd14" origin="Generated by Gradle"/>
          </artifact>
       </component>
+      <component group="com.azure" name="azure-storage-blob-batch" version="12.23.1">
+         <artifact name="azure-storage-blob-batch-12.23.1.jar">
+            <sha256 value="8c11749c783222873f63f22575aa5ae7ee8f285388183b82d1a18db21f4d2eba" origin="Generated by Gradle"/>
+         </artifact>
+      </component>
       <component group="com.azure" name="azure-storage-common" version="12.26.1">
          <artifact name="azure-storage-common-12.26.1.jar">
             <sha256 value="b0297ac1a9017ccd8a1e5cf41fb8d00ff0adbdd06849f6c5aafb3208708264dd" origin="Generated by Gradle"/>

modules/repository-azure/build.gradle

@@ -30,6 +30,7 @@ dependencies {
   api "com.azure:azure-identity:1.13.2"
   api "com.azure:azure-json:1.2.0"
   api "com.azure:azure-storage-blob:12.27.1"
+  api "com.azure:azure-storage-blob-batch:12.23.1"
   api "com.azure:azure-storage-common:12.26.1"
   api "com.azure:azure-storage-internal-avro:12.12.1"
   api "com.azure:azure-xml:1.1.0"

modules/repository-azure/src/internalClusterTest/java/org/elasticsearch/repositories/azure/AzureBlobStoreRepositoryTests.java

@@ -249,6 +249,8 @@ protected void maybeTrack(String request, Headers headers) {
                 trackRequest("PutBlockList");
             } else if (Regex.simpleMatch("PUT /*/*", request)) {
                 trackRequest("PutBlob");
+            } else if (Regex.simpleMatch("POST /*?*comp=batch*", request)) {
+                trackRequest("BatchDelete");
             }
         }
 

modules/repository-azure/src/main/java/module-info.java

@@ -18,18 +18,15 @@
     requires org.apache.logging.log4j;
     requires org.apache.logging.log4j.core;
 
-    requires com.azure.core;
     requires com.azure.http.netty;
-    requires com.azure.storage.blob;
-    requires com.azure.storage.common;
     requires com.azure.identity;
 
     requires io.netty.buffer;
     requires io.netty.transport;
     requires io.netty.resolver;
     requires io.netty.common;
 
-    requires reactor.core;
     requires reactor.netty.core;
     requires reactor.netty.http;
+    requires com.azure.storage.blob.batch;
 }

modules/repository-azure/src/main/java/org/elasticsearch/repositories/azure/AzureBlobStore.java

@@ -21,10 +21,13 @@
 import com.azure.core.util.BinaryData;
 import com.azure.storage.blob.BlobAsyncClient;
 import com.azure.storage.blob.BlobClient;
-import com.azure.storage.blob.BlobContainerAsyncClient;
 import com.azure.storage.blob.BlobContainerClient;
 import com.azure.storage.blob.BlobServiceAsyncClient;
 import com.azure.storage.blob.BlobServiceClient;
+import com.azure.storage.blob.batch.BlobBatch;
+import com.azure.storage.blob.batch.BlobBatchAsyncClient;
+import com.azure.storage.blob.batch.BlobBatchClientBuilder;
+import com.azure.storage.blob.batch.BlobBatchStorageException;
 import com.azure.storage.blob.models.BlobErrorCode;
 import com.azure.storage.blob.models.BlobItem;
 import com.azure.storage.blob.models.BlobItemProperties;
@@ -84,21 +87,20 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.Spliterator;
-import java.util.Spliterators;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.function.BiPredicate;
 import java.util.stream.Collectors;
-import java.util.stream.StreamSupport;
 
 import static org.elasticsearch.core.Strings.format;
 
 public class AzureBlobStore implements BlobStore {
     private static final Logger logger = LogManager.getLogger(AzureBlobStore.class);
+    // See https://learn.microsoft.com/en-us/rest/api/storageservices/blob-batch#request-body
+    private static final int MAX_ELEMENTS_PER_BATCH = 256;
     private static final long DEFAULT_READ_CHUNK_SIZE = new ByteSizeValue(32, ByteSizeUnit.MB).getBytes();
     private static final int DEFAULT_UPLOAD_BUFFERS_SIZE = (int) new ByteSizeValue(64, ByteSizeUnit.KB).getBytes();
 
@@ -147,7 +149,8 @@ public AzureBlobStore(
                     && isPutBlockRequest(httpMethod, url) == false
                     && isPutBlockListRequest(httpMethod, url) == false,
                 Operation.PUT_BLOB
-            )
+            ),
+            new RequestMatcher(AzureBlobStore::isBatchDelete, Operation.BATCH_DELETE)
         );
 
         this.requestMetricsHandler = (purpose, method, url, metrics) -> {
@@ -172,6 +175,10 @@ && isPutBlockListRequest(httpMethod, url) == false,
         };
     }
 
+    private static boolean isBatchDelete(HttpMethod method, URL url) {
+        return method == HttpMethod.POST && url.getQuery() != null && url.getQuery().contains("comp=batch");
+    }
+
     private static boolean isListRequest(HttpMethod httpMethod, URL url) {
         return httpMethod == HttpMethod.GET && url.getQuery() != null && url.getQuery().contains("comp=list");
     }
@@ -231,30 +238,28 @@ public boolean blobExists(OperationPurpose purpose, String blob) throws IOExcept
         }
     }
 
-    // number of concurrent blob delete requests to use while bulk deleting
-    private static final int CONCURRENT_DELETES = 100;
-
     public DeleteResult deleteBlobDirectory(OperationPurpose purpose, String path) {
         final AtomicInteger blobsDeleted = new AtomicInteger(0);
         final AtomicLong bytesDeleted = new AtomicLong(0);
+        final List<String> blobNames = new ArrayList<>();
 
         SocketAccess.doPrivilegedVoidException(() -> {
-            final BlobContainerAsyncClient blobContainerAsyncClient = asyncClient(purpose).getBlobContainerAsyncClient(container);
-            final ListBlobsOptions options = new ListBlobsOptions().setPrefix(path)
-                .setDetails(new BlobListDetails().setRetrieveMetadata(true));
+            final AzureBlobServiceClient client = getAzureBlobServiceClientClient(purpose);
+            final BlobContainerClient blobContainerClient = client.getSyncClient().getBlobContainerClient(container);
             try {
-                blobContainerAsyncClient.listBlobs(options, null).flatMap(blobItem -> {
-                    if (blobItem.isPrefix() != null && blobItem.isPrefix()) {
-                        return Mono.empty();
-                    } else {
-                        final String blobName = blobItem.getName();
-                        BlobAsyncClient blobAsyncClient = blobContainerAsyncClient.getBlobAsyncClient(blobName);
-                        final Mono<Void> deleteTask = getDeleteTask(blobName, blobAsyncClient);
-                        bytesDeleted.addAndGet(blobItem.getProperties().getContentLength());
-                        blobsDeleted.incrementAndGet();
-                        return deleteTask;
+                final ListBlobsOptions options = new ListBlobsOptions().setPrefix(path)
+                    .setDetails(new BlobListDetails().setRetrieveMetadata(true));
+                for (BlobItem blobItem : blobContainerClient.listBlobs(options, null)) {
+                    if (blobItem.isPrefix()) {
+                        continue;
                     }
-                }, CONCURRENT_DELETES).then().block();
+                    blobNames.add(blobItem.getName());
+                    bytesDeleted.addAndGet(blobItem.getProperties().getContentLength());
+                    blobsDeleted.incrementAndGet();
+                }
+                if (blobNames.isEmpty() == false) {
+                    deleteListOfBlobs(client, blobNames.iterator());
+                }
             } catch (Exception e) {
                 filterDeleteExceptionsAndRethrow(e, new IOException("Deleting directory [" + path + "] failed"));
             }
@@ -278,48 +283,44 @@ private static void filterDeleteExceptionsAndRethrow(Exception e, IOException ex
         throw exception;
     }
 
-    /**
-     * {@inheritDoc}
-     * <p>
-     * Note that in this Azure implementation we issue a series of individual
-     * <a href="https://learn.microsoft.com/en-us/rest/api/storageservices/delete-blob">delete blob</a> calls rather than aggregating
-     * deletions into <a href="https://learn.microsoft.com/en-us/rest/api/storageservices/blob-batch">blob batch</a> calls.
-     * The reason for this is that the blob batch endpoint has limited support for SAS token authentication.
-     *
-     * @see <a href="https://learn.microsoft.com/en-us/rest/api/storageservices/blob-batch?tabs=shared-access-signatures#authorization">
-     *     API docs around SAS auth limitations</a>
-     * @see <a href="https://github.com/Azure/azure-storage-java/issues/538">Java SDK issue</a>
-     * @see <a href="https://github.com/elastic/elasticsearch/pull/65140#discussion_r528752070">Discussion on implementing PR</a>
-     */
     @Override
-    public void deleteBlobsIgnoringIfNotExists(OperationPurpose purpose, Iterator<String> blobs) {
-        if (blobs.hasNext() == false) {
+    public void deleteBlobsIgnoringIfNotExists(OperationPurpose purpose, Iterator<String> blobNames) throws IOException {
+        if (blobNames.hasNext() == false) {
             return;
         }
-
-        BlobServiceAsyncClient asyncClient = asyncClient(purpose);
-        SocketAccess.doPrivilegedVoidException(() -> {
-            final BlobContainerAsyncClient blobContainerClient = asyncClient.getBlobContainerAsyncClient(container);
-            try {
-                Flux.fromStream(StreamSupport.stream(Spliterators.spliteratorUnknownSize(blobs, Spliterator.ORDERED), false))
-                    .flatMap(blob -> getDeleteTask(blob, blobContainerClient.getBlobAsyncClient(blob)), CONCURRENT_DELETES)
-                    .then()
-                    .block();
-            } catch (Exception e) {
-                filterDeleteExceptionsAndRethrow(e, new IOException("Unable to delete blobs"));
-            }
-        });
+        SocketAccess.doPrivilegedVoidException(() -> deleteListOfBlobs(getAzureBlobServiceClientClient(purpose), blobNames));
     }
 
-    private static Mono<Void> getDeleteTask(String blobName, BlobAsyncClient blobAsyncClient) {
-        return blobAsyncClient.delete()
-            // Ignore not found blobs, as it's possible that due to network errors a request
-            // for an already deleted blob is retried, causing an error.
-            .onErrorResume(
-                e -> e instanceof BlobStorageException blobStorageException && blobStorageException.getStatusCode() == 404,
-                throwable -> Mono.empty()
-            )
-            .onErrorMap(throwable -> new IOException("Error deleting blob " + blobName, throwable));
+    private void deleteListOfBlobs(AzureBlobServiceClient azureBlobServiceClient, Iterator<String> blobNames) throws IOException {
+        // We need to use a container-scoped BlobBatchClient, so the restype=container parameter
+        // is sent, and we can support all SAS token types
+        // See https://learn.microsoft.com/en-us/rest/api/storageservices/blob-batch?tabs=shared-access-signatures#authorization
+        final BlobBatchAsyncClient batchAsyncClient = new BlobBatchClientBuilder(
+            azureBlobServiceClient.getAsyncClient().getBlobContainerAsyncClient(container)
+        ).buildAsyncClient();
+        final List<Mono<Void>> batchResponses = new ArrayList<>();
+        while (blobNames.hasNext()) {
+            final BlobBatch currentBatch = batchAsyncClient.getBlobBatch();
+            int counter = 0;
+            while (counter < MAX_ELEMENTS_PER_BATCH && blobNames.hasNext()) {
+                currentBatch.deleteBlob(container, blobNames.next());
+                counter++;
+            }
+            batchResponses.add(batchAsyncClient.submitBatch(currentBatch));
+        }
+        try {
+            Flux.merge(batchResponses).collectList().block();
+        } catch (BlobBatchStorageException bbse) {
+            final Iterable<BlobStorageException> batchExceptions = bbse.getBatchExceptions();
+            for (BlobStorageException bse : batchExceptions) {
+                // If one of the requests failed with something other than a BLOB_NOT_FOUND, throw the encompassing exception
+                if (BlobErrorCode.BLOB_NOT_FOUND.equals(bse.getErrorCode()) == false) {
+                    throw new IOException("Failed to delete batch", bbse);
+                }
+            }
+        } catch (Exception e) {
+            throw new IOException("Unable to delete blobs");
+        }
     }
 
     public InputStream getInputStream(OperationPurpose purpose, String blob, long position, final @Nullable Long length) {
@@ -689,7 +690,8 @@ enum Operation {
         GET_BLOB_PROPERTIES("GetBlobProperties"),
         PUT_BLOB("PutBlob"),
         PUT_BLOCK("PutBlock"),
-        PUT_BLOCK_LIST("PutBlockList");
+        PUT_BLOCK_LIST("PutBlockList"),
+        BATCH_DELETE("BatchDelete");
 
         private final String key;
 

modules/repository-azure/src/main/java/org/elasticsearch/repositories/azure/AzureClientProvider.java

@@ -317,6 +317,11 @@ private enum RetryMetricsTracker implements HttpPipelinePolicy {
 
         @Override
         public Mono<HttpResponse> process(HttpPipelineCallContext context, HttpPipelineNextPolicy next) {
+            if (requestIsPartOfABatch(context)) {
+                // Batch deletes fire once for each of the constituent requests, and they have a null response. Ignore those, we'll track
+                // metrics at the bulk level.
+                return next.process();
+            }
             Optional<Object> metricsData = context.getData(RequestMetricsTracker.ES_REQUEST_METRICS_CONTEXT_KEY);
             if (metricsData.isPresent() == false) {
                 assert false : "No metrics object associated with request " + context.getHttpRequest();
@@ -361,6 +366,11 @@ private RequestMetricsTracker(OperationPurpose purpose, RequestMetricsHandler re
 
         @Override
         public Mono<HttpResponse> process(HttpPipelineCallContext context, HttpPipelineNextPolicy next) {
+            if (requestIsPartOfABatch(context)) {
+                // Batch deletes fire once for each of the constituent requests, and they have a null response. Ignore those, we'll track
+                // metrics at the bulk level.
+                return next.process();
+            }
             final RequestMetrics requestMetrics = new RequestMetrics();
             context.setData(ES_REQUEST_METRICS_CONTEXT_KEY, requestMetrics);
             return next.process().doOnSuccess((httpResponse) -> {
@@ -389,6 +399,10 @@ public HttpPipelinePosition getPipelinePosition() {
         }
     }
 
+    private static boolean requestIsPartOfABatch(HttpPipelineCallContext context) {
+        return context.getData("Batch-Operation-Info").isPresent();
+    }
+
     /**
      * The {@link RequestMetricsTracker} calls this when a request completes
      */

modules/repository-azure/src/test/java/org/elasticsearch/repositories/azure/AzureBlobContainerStatsTests.java

@@ -18,6 +18,7 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.List;
 import java.util.Map;
 
 public class AzureBlobContainerStatsTests extends AbstractAzureServerTestCase {
@@ -47,6 +48,8 @@ public void testOperationPurposeIsReflectedInBlobStoreStats() throws IOException
             os.write(blobContent);
             os.flush();
         });
+        // BATCH_DELETE
+        blobStore.deleteBlobsIgnoringIfNotExists(purpose, List.of(randomIdentifier(), randomIdentifier(), randomIdentifier()).iterator());
 
         Map<String, Long> stats = blobStore.stats();
         String statsMapString = stats.toString();
@@ -55,6 +58,7 @@ public void testOperationPurposeIsReflectedInBlobStoreStats() throws IOException
         assertEquals(statsMapString, Long.valueOf(1L), stats.get(statsKey(purpose, AzureBlobStore.Operation.GET_BLOB_PROPERTIES)));
         assertEquals(statsMapString, Long.valueOf(1L), stats.get(statsKey(purpose, AzureBlobStore.Operation.PUT_BLOCK)));
         assertEquals(statsMapString, Long.valueOf(1L), stats.get(statsKey(purpose, AzureBlobStore.Operation.PUT_BLOCK_LIST)));
+        assertEquals(statsMapString, Long.valueOf(1L), stats.get(statsKey(purpose, AzureBlobStore.Operation.BATCH_DELETE)));
     }
 
     public void testOperationPurposeIsNotReflectedInBlobStoreStatsWhenNotServerless() throws IOException {
@@ -79,6 +83,11 @@ public void testOperationPurposeIsNotReflectedInBlobStoreStatsWhenNotServerless(
                 os.write(blobContent);
                 os.flush();
             });
+            // BATCH_DELETE
+            blobStore.deleteBlobsIgnoringIfNotExists(
+                purpose,
+                List.of(randomIdentifier(), randomIdentifier(), randomIdentifier()).iterator()
+            );
         }
 
         Map<String, Long> stats = blobStore.stats();
@@ -88,6 +97,7 @@ public void testOperationPurposeIsNotReflectedInBlobStoreStatsWhenNotServerless(
         assertEquals(statsMapString, Long.valueOf(repeatTimes), stats.get(AzureBlobStore.Operation.GET_BLOB_PROPERTIES.getKey()));
         assertEquals(statsMapString, Long.valueOf(repeatTimes), stats.get(AzureBlobStore.Operation.PUT_BLOCK.getKey()));
         assertEquals(statsMapString, Long.valueOf(repeatTimes), stats.get(AzureBlobStore.Operation.PUT_BLOCK_LIST.getKey()));
+        assertEquals(statsMapString, Long.valueOf(repeatTimes), stats.get(AzureBlobStore.Operation.BATCH_DELETE.getKey()));
     }
 
     private static String statsKey(OperationPurpose purpose, AzureBlobStore.Operation operation) {