Implement Bulk Deletes for GCS Repository

plugins/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobContainer.java

@@ -26,7 +26,9 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 class GoogleCloudStorageBlobContainer extends AbstractBlobContainer {
 
@@ -78,7 +80,12 @@ public void deleteBlob(String blobName) throws IOException {
         blobStore.deleteBlob(buildKey(blobName));
     }
 
-    protected String buildKey(String blobName) {
+    @Override
+    public void deleteBlobsIgnoringIfNotExists(List<String> blobNames) throws IOException {
+        blobStore.deleteBlobs(blobNames.stream().map(this::buildKey).collect(Collectors.toList()));
+    }
+
+    private String buildKey(String blobName) {
         assert blobName != null;
         return path + blobName;
     }

plugins/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobStore.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.repositories.gcs;
 
+import com.google.cloud.BatchResult;
 import com.google.cloud.ReadChannel;
 import com.google.cloud.WriteChannel;
 import com.google.cloud.storage.Blob;
@@ -27,10 +28,10 @@
 import com.google.cloud.storage.Bucket;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.Storage.BlobListOption;
+import com.google.cloud.storage.StorageBatch;
 import com.google.cloud.storage.StorageException;
 
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
+import com.google.common.collect.Lists;
 import org.elasticsearch.common.SuppressForbidden;
 import org.elasticsearch.common.blobstore.BlobContainer;
 import org.elasticsearch.common.blobstore.BlobMetaData;
@@ -51,6 +52,7 @@
 import java.nio.file.FileAlreadyExistsException;
 import java.nio.file.NoSuchFileException;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -59,8 +61,6 @@
 import static java.net.HttpURLConnection.HTTP_PRECON_FAILED;
 
 class GoogleCloudStorageBlobStore implements BlobStore {
-
-    private static final Logger logger = LogManager.getLogger(GoogleCloudStorageBlobStore.class);
 
     // The recommended maximum size of a blob that should be uploaded in a single
     // request. Larger files should be uploaded over multiple requests (this is
@@ -105,7 +105,7 @@ public void close() {
      * @param bucketName name of the bucket
      * @return true iff the bucket exists
      */
-    boolean doesBucketExist(String bucketName) {
+    private boolean doesBucketExist(String bucketName) {
         try {
             final Bucket bucket = SocketAccess.doPrivilegedIOException(() -> client().get(bucketName));
             return bucket != null;
@@ -295,7 +295,7 @@ void deleteBlob(String blobName) throws IOException {
      *
      * @param prefix prefix of the blobs to delete
      */
-    void deleteBlobsByPrefix(String prefix) throws IOException {
+    private void deleteBlobsByPrefix(String prefix) throws IOException {
         deleteBlobs(listBlobsByPrefix("", prefix).keySet());
     }
 
@@ -314,17 +314,39 @@ void deleteBlobs(Collection<String> blobNames) throws IOException {
             return;
         }
         final List<BlobId> blobIdsToDelete = blobNames.stream().map(blob -> BlobId.of(bucketName, blob)).collect(Collectors.toList());
-        final List<Boolean> deletedStatuses = SocketAccess.doPrivilegedIOException(() -> client().delete(blobIdsToDelete));
+        final List<Boolean> deletedStatuses = SocketAccess.doPrivilegedIOException(() -> {
+            final StorageBatch batch = client().batch();
+            final List<Boolean> results = Lists.newArrayList();
+            for (BlobId blob : blobIdsToDelete) {
+                batch.delete(blob).notify(
+                    new BatchResult.Callback<>() {
+                        @Override
+                        public void success(Boolean result) {
+                            results.add(result);
+                        }
+
+                        @Override
+                        public void error(StorageException exception) {
+                            if (exception.getCode() == 404) {
+                                results.add(Boolean.TRUE);
+                            } else {
+                                results.add(Boolean.FALSE);
+                            }
+                        }
+                    });
+            }
+            batch.submit();
+            return Collections.unmodifiableList(results);
+        });
         assert blobIdsToDelete.size() == deletedStatuses.size();
         boolean failed = false;
         for (int i = 0; i < blobIdsToDelete.size(); i++) {
             if (deletedStatuses.get(i) == false) {
-                logger.error("Failed to delete blob [{}] in bucket [{}]", blobIdsToDelete.get(i).getName(), bucketName);
                 failed = true;
             }
         }
         if (failed) {
-            throw new IOException("Failed to delete all [" + blobIdsToDelete.size() + "] blobs");
+            throw new IOException("Failed to delete the following blobs " + blobIdsToDelete);
         }
     }
 

plugins/repository-gcs/src/test/java/org/elasticsearch/repositories/gcs/MockStorage.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.repositories.gcs;
 
 import com.google.api.gax.paging.Page;
+import com.google.cloud.BatchResult;
 import com.google.cloud.Policy;
 import com.google.cloud.ReadChannel;
 import com.google.cloud.RestorableState;
@@ -34,6 +35,7 @@
 import com.google.cloud.storage.ServiceAccount;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.StorageBatch;
+import com.google.cloud.storage.StorageBatchResult;
 import com.google.cloud.storage.StorageException;
 import com.google.cloud.storage.StorageOptions;
 import com.google.cloud.storage.StorageRpcOptionUtils;
@@ -57,6 +59,12 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyVararg;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
 /**
  * {@link MockStorage} mocks a {@link Storage} client by storing all the blobs
  * in a given concurrent map.
@@ -356,8 +364,21 @@ public byte[] readAllBytes(BlobId blob, BlobSourceOption... options) {
     }
 
     @Override
+    @SuppressWarnings("unchecked")
     public StorageBatch batch() {
-        return null;
+        final StorageBatch batch = mock(StorageBatch.class);
+        StorageBatchResult<Boolean> result = mock(StorageBatchResult.class);
+        doAnswer(answer -> {
+            BatchResult.Callback<Boolean, Exception> callback  = (BatchResult.Callback<Boolean, Exception>) answer.getArguments()[0];
+            callback.success(true);
+            return null;
+        }).when(result).notify(any(BatchResult.Callback.class));
+        when(batch.delete(any(BlobId.class), anyVararg())).thenAnswer(invocation -> {
+            final BlobId blobId = (BlobId) invocation.getArguments()[0];
+            delete(blobId);
+            return result;
+        });
+        return batch;
     }
 
     @Override