Use Azure blob batch API to delete blobs in batches (elastic#114566)

Closes ES-9777

Author: nicktindall

docs/changelog/114566.yaml

@@ -0,0 +1,5 @@
+pr: 114566
+summary: Use Azure blob batch API to delete blobs in batches
+area: Distributed
+type: enhancement
+issues: []

docs/reference/snapshot-restore/repository-azure.asciidoc

@@ -259,6 +259,15 @@ include::repository-shared-settings.asciidoc[]
     `primary_only` or `secondary_only`. Defaults to `primary_only`. Note that if you set it
     to `secondary_only`, it will force `readonly` to true.
 
+`delete_objects_max_size`::
+
+    (integer) Sets the maxmimum batch size, betewen 1 and 256, used for `BlobBatch` requests. Defaults to 256 which is the maximum
+    number supported by the https://learn.microsoft.com/en-us/rest/api/storageservices/blob-batch#remarks[Azure blob batch API].
+
+`max_concurrent_batch_deletes`::
+
+    (integer) Sets the maximum number of concurrent batch delete requests that will be submitted for any individual bulk delete with `BlobBatch`. Note that the effective number of concurrent deletes is further limited by the Azure client connection and event loop thread limits. Defaults to 10, minimum is 1, maximum is 100.
+
 [[repository-azure-validation]]
 ==== Repository validation rules
 

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/AzureBlobStoreRepositoryMetricsTests.java

@@ -9,14 +9,18 @@
 
 package org.elasticsearch.repositories.azure;
 
+import com.sun.net.httpserver.Headers;
 import com.sun.net.httpserver.HttpExchange;
 import com.sun.net.httpserver.HttpHandler;
 
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.Randomness;
 import org.elasticsearch.common.blobstore.BlobContainer;
 import org.elasticsearch.common.blobstore.BlobPath;
 import org.elasticsearch.common.blobstore.OperationPurpose;
+import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.SuppressForbidden;
 import org.elasticsearch.plugins.PluginsService;
 import org.elasticsearch.repositories.RepositoriesMetrics;
@@ -31,6 +35,7 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Queue;
@@ -43,6 +48,7 @@
 import java.util.stream.IntStream;
 
 import static org.elasticsearch.repositories.azure.AbstractAzureServerTestCase.randomBlobContent;
+import static org.elasticsearch.repositories.blobstore.BlobStoreTestUtil.randomPurpose;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
@@ -225,6 +231,91 @@ public void testRequestTimeIsAccurate() throws IOException {
         assertThat(recordedRequestTime, lessThanOrEqualTo(elapsedTimeMillis));
     }
 
+    public void testBatchDeleteFailure() throws IOException {
+        final int deleteBatchSize = randomIntBetween(1, 30);
+        final String repositoryName = randomRepositoryName();
+        final String repository = createRepository(
+            repositoryName,
+            Settings.builder()
+                .put(repositorySettings(repositoryName))
+                .put(AzureRepository.Repository.DELETION_BATCH_SIZE_SETTING.getKey(), deleteBatchSize)
+                .build(),
+            true
+        );
+        final String dataNodeName = internalCluster().getNodeNameThat(DiscoveryNode::canContainData);
+        final BlobContainer container = getBlobContainer(dataNodeName, repository);
+
+        final List<String> blobsToDelete = new ArrayList<>();
+        final int numberOfBatches = randomIntBetween(3, 20);
+        final int numberOfBlobs = numberOfBatches * deleteBatchSize;
+        final int failedBatches = randomIntBetween(1, numberOfBatches);
+        for (int i = 0; i < numberOfBlobs; i++) {
+            byte[] bytes = randomBytes(randomInt(100));
+            String blobName = "index-" + randomAlphaOfLength(10);
+            container.writeBlob(randomPurpose(), blobName, new BytesArray(bytes), false);
+            blobsToDelete.add(blobName);
+        }
+        Randomness.shuffle(blobsToDelete);
+        clearMetrics(dataNodeName);
+
+        // Handler will fail one or more of the batch requests
+        final RequestHandler failNRequestRequestHandler = createFailNRequestsHandler(failedBatches);
+
+        // Exhaust the retries
+        IntStream.range(0, (numberOfBatches - failedBatches) + (failedBatches * (MAX_RETRIES + 1)))
+            .forEach(i -> requestHandlers.offer(failNRequestRequestHandler));
+
+        logger.info("--> Failing {} of {} batches", failedBatches, numberOfBatches);
+
+        final IOException exception = assertThrows(
+            IOException.class,
+            () -> container.deleteBlobsIgnoringIfNotExists(randomPurpose(), blobsToDelete.iterator())
+        );
+        assertEquals(Math.min(failedBatches, 10), exception.getSuppressed().length);
+        assertEquals(
+            (numberOfBatches - failedBatches) + (failedBatches * (MAX_RETRIES + 1L)),
+            getLongCounterTotal(dataNodeName, RepositoriesMetrics.METRIC_REQUESTS_TOTAL)
+        );
+        assertEquals((failedBatches * (MAX_RETRIES + 1L)), getLongCounterTotal(dataNodeName, RepositoriesMetrics.METRIC_EXCEPTIONS_TOTAL));
+        assertEquals(failedBatches * deleteBatchSize, container.listBlobs(randomPurpose()).size());
+    }
+
+    private long getLongCounterTotal(String dataNodeName, String metricKey) {
+        return getTelemetryPlugin(dataNodeName).getLongCounterMeasurement(metricKey)
+            .stream()
+            .mapToLong(Measurement::getLong)
+            .reduce(0L, Long::sum);
+    }
+
+    /**
+     * Creates a {@link RequestHandler} that will persistently fail the first <code>numberToFail</code> distinct requests
+     * it sees. Any other requests are passed through to the delegate.
+     *
+     * @param numberToFail The number of requests to fail
+     * @return the handler
+     */
+    private static RequestHandler createFailNRequestsHandler(int numberToFail) {
+        final List<String> requestsToFail = new ArrayList<>(numberToFail);
+        return (exchange, delegate) -> {
+            final Headers requestHeaders = exchange.getRequestHeaders();
+            final String requestId = requestHeaders.get("X-ms-client-request-id").get(0);
+            boolean failRequest = false;
+            synchronized (requestsToFail) {
+                if (requestsToFail.contains(requestId)) {
+                    failRequest = true;
+                } else if (requestsToFail.size() < numberToFail) {
+                    requestsToFail.add(requestId);
+                    failRequest = true;
+                }
+            }
+            if (failRequest) {
+                exchange.sendResponseHeaders(500, -1);
+            } else {
+                delegate.handle(exchange);
+            }
+        };
+    }
+
     private void clearMetrics(String discoveryNode) {
         internalCluster().getInstance(PluginsService.class, discoveryNode)
             .filterPlugins(TestTelemetryPlugin.class)

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

@@ -89,7 +89,9 @@ protected Settings repositorySettings(String repoName) {
             .put(super.repositorySettings(repoName))
             .put(AzureRepository.Repository.MAX_SINGLE_PART_UPLOAD_SIZE_SETTING.getKey(), new ByteSizeValue(1, ByteSizeUnit.MB))
             .put(AzureRepository.Repository.CONTAINER_SETTING.getKey(), "container")
-            .put(AzureStorageSettings.ACCOUNT_SETTING.getKey(), "test");
+            .put(AzureStorageSettings.ACCOUNT_SETTING.getKey(), "test")
+            .put(AzureRepository.Repository.DELETION_BATCH_SIZE_SETTING.getKey(), randomIntBetween(5, 256))
+            .put(AzureRepository.Repository.MAX_CONCURRENT_BATCH_DELETES_SETTING.getKey(), randomIntBetween(1, 10));
         if (randomBoolean()) {
             settingsBuilder.put(AzureRepository.Repository.BASE_PATH_SETTING.getKey(), randomFrom("test", "test/1"));
         }
@@ -249,6 +251,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("BlobBatch");
             }
         }
 
@@ -279,10 +283,22 @@ public void testLargeBlobCountDeletion() throws Exception {
     }
 
     public void testDeleteBlobsIgnoringIfNotExists() throws Exception {
-        try (BlobStore store = newBlobStore()) {
+        // Test with a smaller batch size here
+        final int deleteBatchSize = randomIntBetween(1, 30);
+        final String repositoryName = randomRepositoryName();
+        createRepository(
+            repositoryName,
+            Settings.builder()
+                .put(repositorySettings(repositoryName))
+                .put(AzureRepository.Repository.DELETION_BATCH_SIZE_SETTING.getKey(), deleteBatchSize)
+                .build(),
+            true
+        );
+        try (BlobStore store = newBlobStore(repositoryName)) {
             final BlobContainer container = store.blobContainer(BlobPath.EMPTY);
-            List<String> blobsToDelete = new ArrayList<>();
-            for (int i = 0; i < 10; i++) {
+            final int toDeleteCount = randomIntBetween(deleteBatchSize, 3 * deleteBatchSize);
+            final List<String> blobsToDelete = new ArrayList<>();
+            for (int i = 0; i < toDeleteCount; i++) {
                 byte[] bytes = randomBytes(randomInt(100));
                 String blobName = randomAlphaOfLength(10);
                 container.writeBlob(randomPurpose(), blobName, new BytesArray(bytes), false);

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

@@ -30,6 +30,8 @@
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.core.Booleans;
+import org.elasticsearch.logging.LogManager;
+import org.elasticsearch.logging.Logger;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.repositories.AbstractThirdPartyRepositoryTestCase;
 import org.elasticsearch.repositories.blobstore.BlobStoreRepository;
@@ -46,6 +48,7 @@
 import static org.hamcrest.Matchers.not;
 
 public class AzureStorageCleanupThirdPartyTests extends AbstractThirdPartyRepositoryTestCase {
+    private static final Logger logger = LogManager.getLogger(AzureStorageCleanupThirdPartyTests.class);
     private static final boolean USE_FIXTURE = Booleans.parseBoolean(System.getProperty("test.azure.fixture", "true"));
 
     private static final String AZURE_ACCOUNT = System.getProperty("test.azure.account");
@@ -89,8 +92,10 @@ protected SecureSettings credentials() {
         MockSecureSettings secureSettings = new MockSecureSettings();
         secureSettings.setString("azure.client.default.account", System.getProperty("test.azure.account"));
         if (hasSasToken) {
+            logger.info("--> Using SAS token authentication");
             secureSettings.setString("azure.client.default.sas_token", System.getProperty("test.azure.sas_token"));
         } else {
+            logger.info("--> Using key authentication");
             secureSettings.setString("azure.client.default.key", System.getProperty("test.azure.key"));
         }
         return secureSettings;

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/AzureBlobContainer.java

@@ -138,7 +138,7 @@ public void writeMetadataBlob(
     }
 
     @Override
-    public DeleteResult delete(OperationPurpose purpose) {
+    public DeleteResult delete(OperationPurpose purpose) throws IOException {
         return blobStore.deleteBlobDirectory(purpose, keyPath);
     }