-
Notifications
You must be signed in to change notification settings - Fork 6
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
NXP-24208: use S3 copy when writing a blob from another S3 binary man…
…ager
- Loading branch information
Luís Duarte
authored and
Florent Guillaume
committed
Mar 8, 2018
1 parent
fcd823b
commit a25ed36
Showing
4 changed files
with
323 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
167 changes: 167 additions & 0 deletions
167
nuxeo-core-binarymanager-s3/src/main/java/org/nuxeo/ecm/core/storage/sql/S3Utils.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,167 @@ | ||
| /* | ||
| * (C) Copyright 2011-2018 Nuxeo (http://nuxeo.com/) and others. | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| * | ||
| * Contributors: | ||
| * Luís Duarte | ||
| * Florent Guillaume | ||
| */ | ||
| package org.nuxeo.ecm.core.storage.sql; | ||
|
|
||
| import static java.lang.Math.min; | ||
| import static org.apache.commons.lang3.StringUtils.isBlank; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| import org.nuxeo.ecm.core.api.NuxeoException; | ||
|
|
||
| import com.amazonaws.AmazonClientException; | ||
| import com.amazonaws.auth.AWSCredentialsProvider; | ||
| import com.amazonaws.auth.AWSStaticCredentialsProvider; | ||
| import com.amazonaws.auth.BasicAWSCredentials; | ||
| import com.amazonaws.auth.InstanceProfileCredentialsProvider; | ||
| import com.amazonaws.services.s3.AmazonS3; | ||
| import com.amazonaws.services.s3.model.CompleteMultipartUploadRequest; | ||
| import com.amazonaws.services.s3.model.CopyObjectRequest; | ||
| import com.amazonaws.services.s3.model.CopyPartRequest; | ||
| import com.amazonaws.services.s3.model.CopyPartResult; | ||
| import com.amazonaws.services.s3.model.InitiateMultipartUploadRequest; | ||
| import com.amazonaws.services.s3.model.InitiateMultipartUploadResult; | ||
| import com.amazonaws.services.s3.model.ObjectMetadata; | ||
| import com.amazonaws.services.s3.model.PartETag; | ||
|
|
||
| /** | ||
| * AWS S3 utilities. | ||
| * | ||
| * @since 10.1 | ||
| */ | ||
| public class S3Utils { | ||
|
|
||
| /** The maximum size of a file that can be copied without using multipart: 5 GB */ | ||
| public static final long NON_MULTIPART_COPY_MAX_SIZE = 5L * 1024 * 1024 * 1024; | ||
|
|
||
| /** The size of the parts that we use for multipart copy. */ | ||
| public static final long PART_SIZE = 5L * 1024 * 1024; // 5 MB | ||
|
|
||
| private S3Utils() { | ||
| // utility class | ||
| } | ||
|
|
||
| /** | ||
| * Represents an operation that accepts a slice number and a slice begin and end position. | ||
| */ | ||
| @FunctionalInterface | ||
| public static interface SliceConsumer { | ||
| /** | ||
| * Performs this operation on the arguments. | ||
| * | ||
| * @param num the slice number, starting at 0 | ||
| * @param begin the begin position | ||
| * @param end the end position + 1 | ||
| */ | ||
| public void accept(int num, long begin, long end); | ||
| } | ||
|
|
||
| /** | ||
| * Calls the consumer on all slices. | ||
| * | ||
| * @param slice the slice size | ||
| * @param length the total length | ||
| * @param consumer the slice consumer | ||
| */ | ||
| public static void processSlices(long slice, long length, SliceConsumer consumer) { | ||
| if (slice <= 0) { | ||
| throw new IllegalArgumentException("Invalid slice length: " + slice); | ||
| } | ||
| long begin = 0; | ||
| for (int num = 0; begin < length; num++) { | ||
| long end = min(begin + slice, length); | ||
| consumer.accept(num, begin, end); | ||
| begin += slice; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Copies a file using multipart upload. | ||
| * | ||
| * @param amazonS3 the S3 client | ||
| * @param objectMetadata the metadata of the object being copied | ||
| * @param sourceBucket the source bucket | ||
| * @param sourceKey the source key | ||
| * @param targetBucket the target bucket | ||
| * @param targetKey the target key | ||
| * @param deleteSource whether to delete the source object if the copy is successful | ||
| */ | ||
| public static ObjectMetadata copyFileMultipart(AmazonS3 amazonS3, ObjectMetadata objectMetadata, | ||
| String sourceBucket, String sourceKey, String targetBucket, String targetKey, boolean deleteSource) { | ||
| InitiateMultipartUploadRequest initiateMultipartUploadRequest = new InitiateMultipartUploadRequest(sourceBucket, | ||
| targetKey); | ||
| InitiateMultipartUploadResult initiateMultipartUploadResult = amazonS3.initiateMultipartUpload( | ||
| initiateMultipartUploadRequest); | ||
|
|
||
| String uploadId = initiateMultipartUploadResult.getUploadId(); | ||
| long objectSize = objectMetadata.getContentLength(); | ||
| List<CopyPartResult> copyResponses = new ArrayList<>(); | ||
|
|
||
| SliceConsumer partCopy = (num, begin, end) -> { | ||
| CopyPartRequest copyRequest = new CopyPartRequest().withSourceBucketName(sourceBucket) | ||
| .withSourceKey(sourceKey) | ||
| .withDestinationBucketName(targetBucket) | ||
| .withDestinationKey(targetKey) | ||
| .withFirstByte(begin) | ||
| .withLastByte(end - 1) | ||
| .withUploadId(uploadId) | ||
| .withPartNumber(num + 1); | ||
| copyResponses.add(amazonS3.copyPart(copyRequest)); | ||
| }; | ||
| processSlices(PART_SIZE, objectSize, partCopy); | ||
|
|
||
| CompleteMultipartUploadRequest completeRequest = new CompleteMultipartUploadRequest(targetBucket, targetKey, | ||
| uploadId, responsesToETags(copyResponses)); | ||
| amazonS3.completeMultipartUpload(completeRequest); | ||
| if (deleteSource) { | ||
| amazonS3.deleteObject(sourceBucket, sourceKey); | ||
| } | ||
| return amazonS3.getObjectMetadata(targetBucket, targetKey); | ||
| } | ||
|
|
||
| protected static List<PartETag> responsesToETags(List<CopyPartResult> responses) { | ||
| return responses.stream().map(response -> new PartETag(response.getPartNumber(), response.getETag())).collect( | ||
| Collectors.toList()); | ||
| } | ||
|
|
||
| /** | ||
| * Copies a file without using multipart upload. | ||
| * | ||
| * @param amazonS3 the S3 client | ||
| * @param objectMetadata the metadata of the object being copied | ||
| * @param sourceBucket the source bucket | ||
| * @param sourceKey the source key | ||
| * @param targetBucket the target bucket | ||
| * @param targetKey the target key | ||
| * @param deleteSource whether to delete the source object if the copy is successful | ||
| */ | ||
| public static ObjectMetadata copyFile(AmazonS3 amazonS3, ObjectMetadata objectMetadata, String sourceBucket, | ||
| String sourceKey, String targetBucket, String targetKey, boolean deleteSource) { | ||
| CopyObjectRequest copyObjectRequest = new CopyObjectRequest(sourceBucket, sourceKey, targetBucket, targetKey); | ||
| amazonS3.copyObject(copyObjectRequest); | ||
| if (deleteSource) { | ||
| amazonS3.deleteObject(sourceBucket, sourceKey); | ||
| } | ||
| return amazonS3.getObjectMetadata(targetBucket, targetKey); | ||
| } | ||
|
|
||
| } |
66 changes: 66 additions & 0 deletions
66
nuxeo-core-binarymanager-s3/src/test/java/org/nuxeo/ecm/core/storage/sql/TestS3Utils.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,66 @@ | ||
| /* | ||
| * (C) Copyright 2018 Nuxeo (http://nuxeo.com/) and others. | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| * | ||
| * Contributors: | ||
| * Florent Guillaume | ||
| */ | ||
| package org.nuxeo.ecm.core.storage.sql; | ||
|
|
||
| import static org.junit.Assert.assertEquals; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.Arrays; | ||
| import java.util.Collections; | ||
| import java.util.List; | ||
|
|
||
| import org.junit.Test; | ||
| import org.nuxeo.ecm.core.storage.sql.S3Utils.SliceConsumer; | ||
|
|
||
| public class TestS3Utils { | ||
|
|
||
| @Test | ||
| public void testProcessSlices() { | ||
| List<String> list = new ArrayList<>(); | ||
| SliceConsumer recorder = (num, begin, end) -> { | ||
| list.add(num + ":" + begin + "-" + end); | ||
| }; | ||
|
|
||
| // typical case | ||
| list.clear(); | ||
| S3Utils.processSlices(10, 25, recorder); | ||
| assertEquals(Arrays.asList("0:0-10", "1:10-20", "2:20-25"), list); | ||
|
|
||
| // exactly at the end | ||
| list.clear(); | ||
| S3Utils.processSlices(10, 30, recorder); | ||
| assertEquals(Arrays.asList("0:0-10", "1:10-20", "2:20-30"), list); | ||
|
|
||
| // exactly one slice | ||
| list.clear(); | ||
| S3Utils.processSlices(10, 10, recorder); | ||
| assertEquals(Arrays.asList("0:0-10"), list); | ||
|
|
||
| // slice smaller than total length | ||
| list.clear(); | ||
| S3Utils.processSlices(10, 5, recorder); | ||
| assertEquals(Arrays.asList("0:0-5"), list); | ||
|
|
||
| // degenerate case | ||
| list.clear(); | ||
| S3Utils.processSlices(10, 0, recorder); | ||
| assertEquals(Collections.emptyList(), list); | ||
| } | ||
|
|
||
| } |