Merge pull request #112 from EMCECS/feature-SDK-649

xiaoxin-ren · web-flow · commit 7eba748166c0 · 2024-05-24T08:39:49.000+01:00
[SDK-649] Initial commit for LargeFileUploader enhancement to support…
diff --git a/src/main/java/com/emc/object/s3/LargeFileUploader.java b/src/main/java/com/emc/object/s3/LargeFileUploader.java
@@ -81,6 +81,10 @@ public static String getMpuETag(List<MultipartPartETag> partETags) {
     private final String bucket;
     private final String key;
 
+    private final String srcBucket;
+    private final String srcKey;
+    private String sourceVersionId;
+
     private final InputStream stream;
     private final LargeFileMultipartSource multipartSource;
 
@@ -125,6 +129,8 @@ public LargeFileUploader(S3Client s3Client, String bucket, String key, InputStre
         this.stream = stream;
         this.fullSize = size;
         this.multipartSource = null;
+        this.srcBucket = null;
+        this.srcKey = null;
     }
 
     /**
@@ -139,6 +145,23 @@ public LargeFileUploader(S3Client s3Client, String bucket, String key, LargeFile
         this.key = key;
         this.multipartSource = multipartSource;
         this.stream = null;
+        this.srcBucket = null;
+        this.srcKey = null;
+    }
+
+    /**
+     * Creates a new LargeFileUpload instance using the specified <code>s3Client</code> to copy
+     * from <code>srcBucket/srcKey</code> to <code>dstBucket/dstKey</code> without streaming
+     * data between the client and ECS server.
+     */
+    public LargeFileUploader(S3Client s3Client, String srcBucket, String srcKey, String dstBucket, String dstKey) {
+        this.s3Client = s3Client;
+        this.bucket = dstBucket;
+        this.key = dstKey;
+        this.srcBucket = srcBucket;
+        this.srcKey = srcKey;
+        this.multipartSource = null;
+        this.stream = null;
     }
 
     @Override
@@ -282,6 +305,8 @@ public void upload() {
 
         if (fullSize >= mpuThreshold)
             doMultipartUpload();
+        else if (srcKey != null && srcBucket != null)
+            doSingleCopy();
         else
             doSinglePut();
     }
@@ -291,6 +316,8 @@ private InputStream getSourceCompleteDataStream() throws IOException {
         if (multipartSource != null) {
             is = multipartSource.getCompleteDataStream();
         } else {
+            if (stream == null)
+                throw new IOException("null source stream");
             // only close the source stream if configured to do so
             is = new FilterInputStream(stream) {
                 @Override
@@ -336,6 +363,17 @@ public void doSinglePut() {
         }
     }
 
+    public void doSingleCopy() {
+        configure();
+        if (srcKey == null || srcBucket == null)
+            throw new IllegalArgumentException("must specify source bucket and key for the copy");
+        CopyObjectRequest copyObjectRequest = new CopyObjectRequest(srcBucket, srcKey, bucket, key)
+                .withSourceVersionId(sourceVersionId).withObjectMetadata(objectMetadata)
+                .withAcl(acl).withCannedAcl(cannedAcl);
+        CopyObjectResult result = s3Client.copyObject(copyObjectRequest);
+        eTag = result.getRawETag();
+    }
+
     /*
      * get a map of existing MPU parts from which we can resume an MPU. we can only resume an MPU if the existing
      * part sizes and count are exactly the same as configured in this LFU instance
@@ -426,7 +464,11 @@ public void doMultipartUpload() {
 
                     // no existing part to use, so upload this part
                 } else {
-                    futures.add(executorService.submit(new UploadPartTask(resumeContext.getUploadId(), partNumber, offset, length)));
+                    if (srcKey != null && srcBucket != null) {
+                        futures.add(executorService.submit(new CopyPartTask(resumeContext.getUploadId(), partNumber, offset, length)));
+                    }else {
+                        futures.add(executorService.submit(new UploadPartTask(resumeContext.getUploadId(), partNumber, offset, length)));
+                    }
                 }
             }
 
@@ -565,6 +607,10 @@ protected void configure() {
             // must read stream sequentially
             executorService = null;
             threads = 1;
+        } else if (srcKey != null && srcBucket != null) {
+            // If resuming from copied parts, no need to verify the parts found in target
+            if (resumeContext != null) resumeContext.setVerifyPartsFoundInTarget(false);
+            fullSize = s3Client.getObjectMetadata(new GetObjectMetadataRequest(srcBucket, srcKey).withVersionId(sourceVersionId)).getContentLength();
         } else {
             throw new IllegalArgumentException("must specify a file, stream, or multipartSource to read");
         }
@@ -773,6 +819,14 @@ public void setAbortMpuOnFailure(boolean abortMpuOnFailure) {
         this.abortMpuOnFailure = abortMpuOnFailure;
     }
 
+    public String getSourceVersionId() {
+        return sourceVersionId;
+    }
+
+    public void setSourceVersionId(String sourceVersionId) {
+        this.sourceVersionId = sourceVersionId;
+    }
+
     public LargeFileUploader withObjectMetadata(S3ObjectMetadata objectMetadata) {
         setObjectMetadata(objectMetadata);
         return this;
@@ -834,6 +888,44 @@ public LargeFileUploader withAbortMpuOnFailure(boolean abortMpuOnFailure) {
         return this;
     }
 
+    public LargeFileUploader withSourceVersionId(String sourceVersionId) {
+        setSourceVersionId(sourceVersionId);
+        return this;
+    }
+
+    private class CopyPartTask implements Callable<MultipartPartETag> {
+        private final String uploadId;
+        private final int partNumber;
+        private final long offset;
+        private final long length;
+        public CopyPartTask(String uploadId, int partNumber, long offset, long length) {
+            this.uploadId = uploadId;
+            this.partNumber = partNumber;
+            this.offset = offset;
+            this.length = length;
+        }
+
+        @Override
+        public MultipartPartETag call() {
+            if (!active.get()) {
+                // we were paused or aborted, so should not start any more tasks
+                throw new CancellationException();
+            } else {
+                log.debug("copying {}/{}, uploadId: {}, partNumber {} (offset: {}, length: {}), versionId: {}",
+                        bucket, key, uploadId, partNumber, offset, length, sourceVersionId);
+                CopyPartRequest copyPartRequest = new CopyPartRequest(srcBucket, srcKey, bucket, key, resumeContext.getUploadId(), partNumber)
+                        .withSourceRange(new Range(offset, offset + length - 1))
+                        .withSourceVersionId(sourceVersionId);
+                try {
+                    CopyPartResult result = s3Client.copyPart(copyPartRequest);
+                    return new MultipartPartETag(result.getPartNumber(), result.getRawETag());
+                } catch (S3Exception e) {
+                    throw new RuntimeException(e);
+                }
+            }
+        }
+    }
+
     private class UploadPartTask implements Callable<MultipartPartETag> {
         private final String uploadId;
         private final int partNumber;
diff --git a/src/main/java/com/emc/object/s3/request/CopyPartRequest.java b/src/main/java/com/emc/object/s3/request/CopyPartRequest.java
@@ -79,4 +79,10 @@ public CopyPartRequest withSourceRange(Range sourceRange) {
         setSourceRange(sourceRange);
         return this;
     }
+
+    @Override
+    public CopyPartRequest withSourceVersionId(String sourceVersionId) {
+        super.setSourceVersionId(sourceVersionId);
+        return this;
+    }
 }
diff --git a/src/test/java/com/emc/object/s3/LargeFileUploaderTest.java b/src/test/java/com/emc/object/s3/LargeFileUploaderTest.java
@@ -26,19 +26,13 @@
  */
 package com.emc.object.s3;
 
-import com.emc.object.s3.bean.GetObjectResult;
-import com.emc.object.s3.bean.MultipartPart;
-import com.emc.object.s3.bean.MultipartPartETag;
-import com.emc.object.s3.bean.Upload;
+import com.emc.object.s3.bean.*;
 import com.emc.object.s3.jersey.S3JerseyClient;
 import com.emc.object.s3.lfu.LargeFileMultipartSource;
 import com.emc.object.s3.lfu.LargeFileUpload;
 import com.emc.object.s3.lfu.LargeFileUploaderResumeContext;
 import com.emc.object.s3.lfu.PartMismatchException;
-import com.emc.object.s3.request.AbortMultipartUploadRequest;
-import com.emc.object.s3.request.GetObjectRequest;
-import com.emc.object.s3.request.ListMultipartUploadsRequest;
-import com.emc.object.s3.request.UploadPartRequest;
+import com.emc.object.s3.request.*;
 import com.emc.object.util.ProgressListener;
 import com.emc.rest.util.StreamUtil;
 import com.emc.util.RandomInputStream;
@@ -52,6 +46,7 @@
 import java.security.DigestInputStream;
 import java.security.MessageDigest;
 import java.util.*;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicLong;
@@ -168,6 +163,126 @@ public void testLargeFileUploaderProgressListener() throws Exception {
         client.deleteObject(getTestBucket(), key);
     }
 
+    @Test
+    public void testLFUCopy() throws Exception {
+        // enable versioning on the bucket
+        client.setBucketVersioning(getTestBucket(), new VersioningConfiguration().withStatus(VersioningConfiguration.Status.Enabled));
+
+        String srcKey = "lfu-source.bin";
+        String dstKey = "lfu-single-copy.bin";
+        byte[] data = new byte[(int) MockMultipartSource.totalSize];
+        new Random().nextBytes(data);
+        String srcMD5Sum = DigestUtils.md5Hex(data);
+        client.putObject(getTestBucket(), srcKey, data, null);
+        String versionId0 = client.listVersions(getTestBucket(), null).getVersions().get(0).getVersionId();
+        S3ObjectMetadata sourceMetadata0 = client.getObjectMetadata(new GetObjectMetadataRequest(getTestBucket(), srcKey).withVersionId(versionId0));
+        Assert.assertEquals(MockMultipartSource.totalSize, sourceMetadata0.getContentLength().longValue());
+        Assert.assertEquals(srcMD5Sum, sourceMetadata0.getETag());
+
+        //Upload new Version Object
+        new Random().nextBytes(data);
+        String srcMD5SumNew = DigestUtils.md5Hex(data);
+        client.putObject(getTestBucket(), srcKey, data, null);
+        String versionId1 = client.listVersions(getTestBucket(), null).getVersions().get(0).getVersionId();
+        S3ObjectMetadata sourceMetadata1 = client.getObjectMetadata(new GetObjectMetadataRequest(getTestBucket(), srcKey).withVersionId(versionId1));
+        Assert.assertEquals(MockMultipartSource.totalSize, sourceMetadata1.getContentLength().longValue());
+        Assert.assertEquals(srcMD5SumNew, sourceMetadata1.getETag());
+
+        //Single Copy from old version object
+        S3ObjectMetadata objectMetadata = new S3ObjectMetadata().addUserMetadata("key", "value");
+        LargeFileUploader uploader = new TestLargeFileUploader(client, getTestBucket(), srcKey, getTestBucket(), dstKey)
+                .withObjectMetadata(objectMetadata)
+                .withSourceVersionId(versionId0);
+        uploader.upload();
+        Assert.assertEquals(0, uploader.getBytesTransferred());
+        Assert.assertEquals(srcMD5Sum, uploader.getETag());
+        Assert.assertEquals(MockMultipartSource.totalSize, client.getObjectMetadata(getTestBucket(), dstKey).getContentLength().longValue());
+        Assert.assertEquals(objectMetadata.getUserMetadata(), client.getObjectMetadata(getTestBucket(), dstKey).getUserMetadata());
+
+        //Single Copy from new version object (without providing version Id)
+        uploader = new TestLargeFileUploader(client, getTestBucket(), srcKey, getTestBucket(), dstKey)
+                .withObjectMetadata(objectMetadata);
+        uploader.upload();
+        Assert.assertEquals(0, uploader.getBytesTransferred());
+        Assert.assertEquals(srcMD5SumNew, uploader.getETag());
+        Assert.assertEquals(MockMultipartSource.totalSize, client.getObjectMetadata(getTestBucket(), dstKey).getContentLength().longValue());
+        Assert.assertEquals(objectMetadata.getUserMetadata(), client.getObjectMetadata(getTestBucket(), dstKey).getUserMetadata());
+
+        //Multi Part Copy from old version object
+        uploader = new TestLargeFileUploader(client, getTestBucket(), srcKey, getTestBucket(), dstKey)
+                .withObjectMetadata(objectMetadata)
+                .withPartSize(MockMultipartSource.partSize)
+                .withMpuThreshold(MockMultipartSource.partSize)
+                .withSourceVersionId(versionId0);
+        uploader.upload();
+        Assert.assertEquals(0, uploader.getBytesTransferred());
+        Assert.assertTrue(uploader.getETag().contains("-")); // hyphen signifies multipart / updated object
+        Assert.assertEquals(MockMultipartSource.totalSize, client.getObjectMetadata(getTestBucket(), dstKey).getContentLength().longValue());
+        Assert.assertEquals(objectMetadata.getUserMetadata(), client.getObjectMetadata(getTestBucket(), dstKey).getUserMetadata());
+        Assert.assertEquals(srcMD5Sum, DigestUtils.md5Hex(client.getObject(getTestBucket(), dstKey).getObject()));
+
+        //Multi Part Copy from new version object
+        uploader = new TestLargeFileUploader(client, getTestBucket(), srcKey, getTestBucket(), dstKey)
+                .withObjectMetadata(objectMetadata)
+                .withPartSize(MockMultipartSource.partSize)
+                .withMpuThreshold(MockMultipartSource.partSize)
+                .withSourceVersionId(versionId1);
+        uploader.upload();
+        Assert.assertEquals(0, uploader.getBytesTransferred());
+        Assert.assertTrue(uploader.getETag().contains("-")); // hyphen signifies multipart / updated object
+        Assert.assertEquals(MockMultipartSource.totalSize, client.getObjectMetadata(getTestBucket(), dstKey).getContentLength().longValue());
+        Assert.assertEquals(objectMetadata.getUserMetadata(), client.getObjectMetadata(getTestBucket(), dstKey).getUserMetadata());
+        Assert.assertEquals(srcMD5SumNew, DigestUtils.md5Hex(client.getObject(getTestBucket(), dstKey).getObject()));
+    }
+
+    @Test
+    public void testLFUCopyPauseResume() throws Exception {
+        String srcKey = "mpu-copy-pause.source";
+        String dstKey = "mpu-copy-pause.target";
+
+        MockMultipartSource mockMultipartSource = new MockMultipartSource();
+        LargeFileUploader lfu = new TestLargeFileUploader(client, getTestBucket(), srcKey, mockMultipartSource)
+                .withPartSize(mockMultipartSource.getPartSize()).withMpuThreshold((int) mockMultipartSource.getTotalSize());
+        lfu.doMultipartUpload();
+
+        lfu = new TestLargeFileUploader(client, getTestBucket(), srcKey, getTestBucket(), dstKey)
+                .withPartSize(mockMultipartSource.getPartSize()).withMpuThreshold((int) mockMultipartSource.getTotalSize())
+                .withThreads(1);
+        LargeFileUpload upload = lfu.uploadAsync();
+
+        // wait for first a few parts to start
+        while (lfu.getExecutorService() == null || ((ThreadPoolExecutor) lfu.getExecutorService()).getCompletedTaskCount() == 0) {
+            Thread.sleep(100);
+        }
+        LargeFileUploaderResumeContext resumeContext = upload.pause();
+
+        // object should not exist
+        try {
+            Assert.assertNull(client.getObjectMetadata(getTestBucket(), dstKey));
+        } catch (S3Exception e) {
+            Assert.assertEquals(404, e.getHttpCode());
+            Assert.assertEquals("NoSuchKey", e.getErrorCode());
+        }
+
+        // check resume context accuracy
+        Assert.assertNotNull(resumeContext.getUploadId());
+        Assert.assertNotNull(resumeContext.getUploadedParts());
+        Assert.assertFalse(resumeContext.getUploadedParts().isEmpty());
+        List<MultipartPart> parts = client.listParts(getTestBucket(), dstKey, resumeContext.getUploadId()).getParts();
+        Assert.assertNotNull(parts);
+
+        lfu = new TestLargeFileUploader(client, getTestBucket(), srcKey, getTestBucket(), dstKey)
+                .withPartSize(mockMultipartSource.getPartSize()).withMpuThreshold((int) mockMultipartSource.getTotalSize())
+                .withResumeContext(resumeContext);
+        lfu.doMultipartUpload();
+
+        // Unfortunately, MPU ETag is not preserved even if copy parts matches the original multipart upload.
+        // So the content verification is done by calculating md5sum instead of checking ETag.
+        //Assert.assertEquals(mockMultipartSource.getMpuETag(), client.getObjectMetadata(getTestBucket(), dstKey).getETag());
+        Assert.assertEquals(DigestUtils.md5Hex(client.getObject(getTestBucket(), srcKey).getObject()), DigestUtils.md5Hex(client.getObject(getTestBucket(), dstKey).getObject()));
+        Assert.assertEquals(mockMultipartSource.getTotalSize(), client.getObjectMetadata(getTestBucket(), dstKey).getContentLength().longValue());
+    }
+
     @Test
     public void testLargeFileUploaderStream() {
         String key = "large-file-uploader-stream.bin";
diff --git a/src/test/java/com/emc/object/s3/TestLargeFileUploader.java b/src/test/java/com/emc/object/s3/TestLargeFileUploader.java
@@ -45,6 +45,9 @@ public TestLargeFileUploader(S3Client s3Client, String bucket, String key, Large
         super(s3Client, bucket, key, multipartSource);
     }
 
+    public TestLargeFileUploader(S3Client s3Client, String srcBucket, String srcKey, String dstBucket, String dstKey){
+        super(s3Client, srcBucket, srcKey, dstBucket, dstKey);
+    }
     @Override
     protected long getMinPartSize() {
         return 100 * 1024; // 100 KiB for testing

Original file line number	Diff line number	Diff line change
`@@ -79,4 +79,10 @@ public CopyPartRequest withSourceRange(Range sourceRange) {`
`79`	`79`	`setSourceRange(sourceRange);`
`80`	`80`	`return this;`
`81`	`81`	`}`
	`82`	`+`
	`83`	`+ @Override`
	`84`	`+ public CopyPartRequest withSourceVersionId(String sourceVersionId) {`
	`85`	`+ super.setSourceVersionId(sourceVersionId);`
	`86`	`+ return this;`
	`87`	`+ }`
`82`	`88`	`}`
Original file line number	Diff line number	Diff line change
`@@ -45,6 +45,9 @@ public TestLargeFileUploader(S3Client s3Client, String bucket, String key, Large`
`45`	`45`	`super(s3Client, bucket, key, multipartSource);`
`46`	`46`	`}`
`47`	`47`
	`48`	`+ public TestLargeFileUploader(S3Client s3Client, String srcBucket, String srcKey, String dstBucket, String dstKey){`
	`49`	`+ super(s3Client, srcBucket, srcKey, dstBucket, dstKey);`
	`50`	`+ }`
`48`	`51`	`@Override`
`49`	`52`	`protected long getMinPartSize() {`
`50`	`53`	`return 100 * 1024; // 100 KiB for testing`