BlobContainer: add copyBlob method

.buildkite/pipelines/pull-request/s3.yml

@@ -0,0 +1,14 @@
+steps:
+  - label: third-party / s3
+    command: |
+      export amazon_s3_bucket=elasticsearch-ci.us-west-2
+      export amazon_s3_base_path=$BUILDKITE_BRANCH
+      .ci/scripts/run-gradle.sh s3ThirdPartyTest
+    env:
+      USE_3RD_PARTY_S3_CREDENTIALS: "true"
+    timeout_in_minutes: 30
+    agents:
+      provider: gcp
+      image: family/elasticsearch-ubuntu-2004
+      machineType: n2-standard-8
+      buildDirectory: /dev/shm/bk

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

@@ -48,6 +48,7 @@
 import java.nio.charset.StandardCharsets;
 import java.nio.file.NoSuchFileException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Base64;
 import java.util.Collection;
 import java.util.Collections;
@@ -241,7 +242,7 @@ protected String requestUniqueId(final HttpExchange exchange) {
     private static class AzureHTTPStatsCollectorHandler extends HttpStatsCollectorHandler {
 
         private AzureHTTPStatsCollectorHandler(HttpHandler delegate) {
-            super(delegate);
+            super(delegate, Arrays.stream(AzureBlobStore.Operation.values()).map(AzureBlobStore.Operation::getKey).toArray(String[]::new));
         }
 
         @Override

modules/repository-gcs/src/internalClusterTest/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobStoreRepositoryTests.java

@@ -364,7 +364,7 @@ protected boolean canFailRequest(final HttpExchange exchange) {
     private static class GoogleCloudStorageStatsCollectorHttpHandler extends HttpStatsCollectorHandler {
 
         GoogleCloudStorageStatsCollectorHttpHandler(final HttpHandler delegate) {
-            super(delegate);
+            super(delegate, Arrays.stream(StorageOperation.values()).map(StorageOperation::key).toArray(String[]::new));
         }
 
         @Override

modules/repository-s3/qa/third-party/src/test/java/org/elasticsearch/repositories/s3/S3RepositoryThirdPartyTests.java

@@ -228,4 +228,56 @@ public void testReadFromPositionLargerThanBlobLength() {
             e -> asInstanceOf(AmazonS3Exception.class, e.getCause()).getStatusCode() == RestStatus.REQUESTED_RANGE_NOT_SATISFIED.getStatus()
         );
     }
+
+    public void testCopy() {
+        final var sourceBlobName = randomIdentifier();
+        final var blobBytes = randomBytesReference(randomIntBetween(100, 2_000));
+        final var destinationBlobName = randomIdentifier();
+
+        final var repository = getRepository();
+
+        final var targetBytes = executeOnBlobStore(repository, sourceBlobContainer -> {
+            sourceBlobContainer.writeBlob(randomPurpose(), sourceBlobName, blobBytes, true);
+
+            final var destinationBlobContainer = repository.blobStore().blobContainer(repository.basePath().add("target"));
+            destinationBlobContainer.copyBlob(
+                randomPurpose(),
+                sourceBlobContainer,
+                sourceBlobName,
+                destinationBlobName,
+                blobBytes.length()
+            );
+
+            return destinationBlobContainer.readBlob(randomPurpose(), destinationBlobName).readAllBytes();
+        });
+
+        assertArrayEquals(BytesReference.toBytes(blobBytes), targetBytes);
+    }
+
+    public void testMultipartCopy() {
+        final var sourceBlobName = randomIdentifier();
+        // executeMultipart requires a minimum part size of 5 MiB
+        final var blobBytes = randomBytesReference(randomIntBetween(5 * 1024 * 1024, 10 * 1024 * 1024));
+        final var destinationBlobName = randomIdentifier();
+
+        final var repository = getRepository();
+
+        final var targetBytes = executeOnBlobStore(repository, sourceBlobContainer -> {
+            sourceBlobContainer.writeBlob(randomPurpose(), sourceBlobName, blobBytes, true);
+
+            final S3BlobContainer destinationBlobContainer = (S3BlobContainer) repository.blobStore()
+                .blobContainer(repository.basePath().add("target"));
+            destinationBlobContainer.executeMultipartCopy(
+                randomPurpose(),
+                (S3BlobContainer) sourceBlobContainer,
+                sourceBlobName,
+                destinationBlobName,
+                blobBytes.length()
+            );
+
+            return destinationBlobContainer.readBlob(randomPurpose(), destinationBlobName).readAllBytes();
+        });
+
+        assertArrayEquals(BytesReference.toBytes(blobBytes), targetBytes);
+    }
 }

modules/repository-s3/src/internalClusterTest/java/org/elasticsearch/repositories/s3/S3BlobStoreRepositoryTests.java

@@ -71,6 +71,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.EnumSet;
@@ -688,7 +689,7 @@ protected class S3StatsCollectorHttpHandler extends HttpStatsCollectorHandler {
         private final Map<S3BlobStore.StatsKey, AtomicLong> metricsCount = ConcurrentCollections.newConcurrentMap();
 
         S3StatsCollectorHttpHandler(final HttpHandler delegate) {
-            super(delegate);
+            super(delegate, Arrays.stream(S3BlobStore.Operation.values()).map(S3BlobStore.Operation::getKey).toArray(String[]::new));
         }
 
         private S3HttpHandler.S3Request parseRequest(HttpExchange exchange) {
@@ -736,9 +737,17 @@ public void maybeTrack(HttpExchange exchange) {
                     k -> new AtomicLong()
                 ).incrementAndGet();
             } else if (request.isPutObjectRequest()) {
-                trackRequest("PutObject");
-                metricsCount.computeIfAbsent(new S3BlobStore.StatsKey(S3BlobStore.Operation.PUT_OBJECT, purpose), k -> new AtomicLong())
-                    .incrementAndGet();
+                if (exchange.getRequestHeaders().containsKey(S3BlobStore.CUSTOM_QUERY_PARAMETER_COPY_SOURCE)) {
+                    trackRequest("CopyObject");
+                    metricsCount.computeIfAbsent(
+                        new S3BlobStore.StatsKey(S3BlobStore.Operation.COPY_OBJECT, purpose),
+                        k -> new AtomicLong()
+                    ).incrementAndGet();
+                } else {
+                    trackRequest("PutObject");
+                    metricsCount.computeIfAbsent(new S3BlobStore.StatsKey(S3BlobStore.Operation.PUT_OBJECT, purpose), k -> new AtomicLong())
+                        .incrementAndGet();
+                }
             } else if (request.isMultiObjectDeleteRequest()) {
                 trackRequest("DeleteObjects");
                 metricsCount.computeIfAbsent(new S3BlobStore.StatsKey(S3BlobStore.Operation.DELETE_OBJECTS, purpose), k -> new AtomicLong())