Move to latest release of minio-go (#4886)
- Region handling can now use region endpoints directly. - All uploads are streaming no more large buffer needed. - Major API overhaul for CopyObject(dst, src) - Fixes bugs present in existing code for copying - metadata replace directive CopyObject - PutObjectPart doesn't require md5Sum and sha256master
Harshavardhana
7 years ago
committed by
Dee Koder
parent
72490bf8db
commit
cf479eb401
@ -0,0 +1,532 @@ |
||||
/* |
||||
* Minio Go Library for Amazon S3 Compatible Cloud Storage (C) 2017 Minio, Inc. |
||||
* |
||||
* 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. |
||||
*/ |
||||
|
||||
package minio |
||||
|
||||
import ( |
||||
"encoding/base64" |
||||
"fmt" |
||||
"net/http" |
||||
"net/url" |
||||
"strconv" |
||||
"strings" |
||||
"time" |
||||
|
||||
"github.com/minio/minio-go/pkg/s3utils" |
||||
) |
||||
|
||||
// SSEInfo - represents Server-Side-Encryption parameters specified by
|
||||
// a user.
|
||||
type SSEInfo struct { |
||||
key []byte |
||||
algo string |
||||
} |
||||
|
||||
// NewSSEInfo - specifies (binary or un-encoded) encryption key and
|
||||
// algorithm name. If algo is empty, it defaults to "AES256". Ref:
|
||||
// https://docs.aws.amazon.com/AmazonS3/latest/dev/ServerSideEncryptionCustomerKeys.html
|
||||
func NewSSEInfo(key []byte, algo string) SSEInfo { |
||||
if algo == "" { |
||||
algo = "AES256" |
||||
} |
||||
return SSEInfo{key, algo} |
||||
} |
||||
|
||||
// internal method that computes SSE-C headers
|
||||
func (s *SSEInfo) getSSEHeaders(isCopySource bool) map[string]string { |
||||
if s == nil { |
||||
return nil |
||||
} |
||||
|
||||
cs := "" |
||||
if isCopySource { |
||||
cs = "copy-source-" |
||||
} |
||||
return map[string]string{ |
||||
"x-amz-" + cs + "server-side-encryption-customer-algorithm": s.algo, |
||||
"x-amz-" + cs + "server-side-encryption-customer-key": base64.StdEncoding.EncodeToString(s.key), |
||||
"x-amz-" + cs + "server-side-encryption-customer-key-MD5": base64.StdEncoding.EncodeToString(sumMD5(s.key)), |
||||
} |
||||
} |
||||
|
||||
// GetSSEHeaders - computes and returns headers for SSE-C as key-value
|
||||
// pairs. They can be set as metadata in PutObject* requests (for
|
||||
// encryption) or be set as request headers in `Core.GetObject` (for
|
||||
// decryption).
|
||||
func (s *SSEInfo) GetSSEHeaders() map[string]string { |
||||
return s.getSSEHeaders(false) |
||||
} |
||||
|
||||
// DestinationInfo - type with information about the object to be
|
||||
// created via server-side copy requests, using the Compose API.
|
||||
type DestinationInfo struct { |
||||
bucket, object string |
||||
|
||||
// key for encrypting destination
|
||||
encryption *SSEInfo |
||||
|
||||
// if no user-metadata is provided, it is copied from source
|
||||
// (when there is only once source object in the compose
|
||||
// request)
|
||||
userMetadata map[string]string |
||||
} |
||||
|
||||
// NewDestinationInfo - creates a compose-object/copy-source
|
||||
// destination info object.
|
||||
//
|
||||
// `encSSEC` is the key info for server-side-encryption with customer
|
||||
// provided key. If it is nil, no encryption is performed.
|
||||
//
|
||||
// `userMeta` is the user-metadata key-value pairs to be set on the
|
||||
// destination. The keys are automatically prefixed with `x-amz-meta-`
|
||||
// if needed. If nil is passed, and if only a single source (of any
|
||||
// size) is provided in the ComposeObject call, then metadata from the
|
||||
// source is copied to the destination.
|
||||
func NewDestinationInfo(bucket, object string, encryptSSEC *SSEInfo, |
||||
userMeta map[string]string) (d DestinationInfo, err error) { |
||||
|
||||
// Input validation.
|
||||
if err = s3utils.CheckValidBucketName(bucket); err != nil { |
||||
return d, err |
||||
} |
||||
if err = s3utils.CheckValidObjectName(object); err != nil { |
||||
return d, err |
||||
} |
||||
|
||||
// Process custom-metadata to remove a `x-amz-meta-` prefix if
|
||||
// present and validate that keys are distinct (after this
|
||||
// prefix removal).
|
||||
m := make(map[string]string) |
||||
for k, v := range userMeta { |
||||
if strings.HasPrefix(strings.ToLower(k), "x-amz-meta-") { |
||||
k = k[len("x-amz-meta-"):] |
||||
} |
||||
if _, ok := m[k]; ok { |
||||
return d, fmt.Errorf("Cannot add both %s and x-amz-meta-%s keys as custom metadata", k, k) |
||||
} |
||||
m[k] = v |
||||
} |
||||
|
||||
return DestinationInfo{ |
||||
bucket: bucket, |
||||
object: object, |
||||
encryption: encryptSSEC, |
||||
userMetadata: m, |
||||
}, nil |
||||
} |
||||
|
||||
// getUserMetaHeadersMap - construct appropriate key-value pairs to send
|
||||
// as headers from metadata map to pass into copy-object request. For
|
||||
// single part copy-object (i.e. non-multipart object), enable the
|
||||
// withCopyDirectiveHeader to set the `x-amz-metadata-directive` to
|
||||
// `REPLACE`, so that metadata headers from the source are not copied
|
||||
// over.
|
||||
func (d *DestinationInfo) getUserMetaHeadersMap(withCopyDirectiveHeader bool) map[string]string { |
||||
if len(d.userMetadata) == 0 { |
||||
return nil |
||||
} |
||||
r := make(map[string]string) |
||||
if withCopyDirectiveHeader { |
||||
r["x-amz-metadata-directive"] = "REPLACE" |
||||
} |
||||
for k, v := range d.userMetadata { |
||||
r["x-amz-meta-"+k] = v |
||||
} |
||||
return r |
||||
} |
||||
|
||||
// SourceInfo - represents a source object to be copied, using
|
||||
// server-side copying APIs.
|
||||
type SourceInfo struct { |
||||
bucket, object string |
||||
|
||||
start, end int64 |
||||
|
||||
decryptKey *SSEInfo |
||||
// Headers to send with the upload-part-copy request involving
|
||||
// this source object.
|
||||
Headers http.Header |
||||
} |
||||
|
||||
// NewSourceInfo - create a compose-object/copy-object source info
|
||||
// object.
|
||||
//
|
||||
// `decryptSSEC` is the decryption key using server-side-encryption
|
||||
// with customer provided key. It may be nil if the source is not
|
||||
// encrypted.
|
||||
func NewSourceInfo(bucket, object string, decryptSSEC *SSEInfo) SourceInfo { |
||||
r := SourceInfo{ |
||||
bucket: bucket, |
||||
object: object, |
||||
start: -1, // range is unspecified by default
|
||||
decryptKey: decryptSSEC, |
||||
Headers: make(http.Header), |
||||
} |
||||
|
||||
// Set the source header
|
||||
r.Headers.Set("x-amz-copy-source", s3utils.EncodePath(bucket+"/"+object)) |
||||
|
||||
// Assemble decryption headers for upload-part-copy request
|
||||
for k, v := range decryptSSEC.getSSEHeaders(true) { |
||||
r.Headers.Set(k, v) |
||||
} |
||||
|
||||
return r |
||||
} |
||||
|
||||
// SetRange - Set the start and end offset of the source object to be
|
||||
// copied. If this method is not called, the whole source object is
|
||||
// copied.
|
||||
func (s *SourceInfo) SetRange(start, end int64) error { |
||||
if start > end || start < 0 { |
||||
return ErrInvalidArgument("start must be non-negative, and start must be at most end.") |
||||
} |
||||
// Note that 0 <= start <= end
|
||||
s.start, s.end = start, end |
||||
return nil |
||||
} |
||||
|
||||
// SetMatchETagCond - Set ETag match condition. The object is copied
|
||||
// only if the etag of the source matches the value given here.
|
||||
func (s *SourceInfo) SetMatchETagCond(etag string) error { |
||||
if etag == "" { |
||||
return ErrInvalidArgument("ETag cannot be empty.") |
||||
} |
||||
s.Headers.Set("x-amz-copy-source-if-match", etag) |
||||
return nil |
||||
} |
||||
|
||||
// SetMatchETagExceptCond - Set the ETag match exception
|
||||
// condition. The object is copied only if the etag of the source is
|
||||
// not the value given here.
|
||||
func (s *SourceInfo) SetMatchETagExceptCond(etag string) error { |
||||
if etag == "" { |
||||
return ErrInvalidArgument("ETag cannot be empty.") |
||||
} |
||||
s.Headers.Set("x-amz-copy-source-if-none-match", etag) |
||||
return nil |
||||
} |
||||
|
||||
// SetModifiedSinceCond - Set the modified since condition.
|
||||
func (s *SourceInfo) SetModifiedSinceCond(modTime time.Time) error { |
||||
if modTime.IsZero() { |
||||
return ErrInvalidArgument("Input time cannot be 0.") |
||||
} |
||||
s.Headers.Set("x-amz-copy-source-if-modified-since", modTime.Format(http.TimeFormat)) |
||||
return nil |
||||
} |
||||
|
||||
// SetUnmodifiedSinceCond - Set the unmodified since condition.
|
||||
func (s *SourceInfo) SetUnmodifiedSinceCond(modTime time.Time) error { |
||||
if modTime.IsZero() { |
||||
return ErrInvalidArgument("Input time cannot be 0.") |
||||
} |
||||
s.Headers.Set("x-amz-copy-source-if-unmodified-since", modTime.Format(http.TimeFormat)) |
||||
return nil |
||||
} |
||||
|
||||
// Helper to fetch size and etag of an object using a StatObject call.
|
||||
func (s *SourceInfo) getProps(c Client) (size int64, etag string, userMeta map[string]string, err error) { |
||||
// Get object info - need size and etag here. Also, decryption
|
||||
// headers are added to the stat request if given.
|
||||
var objInfo ObjectInfo |
||||
rh := NewGetReqHeaders() |
||||
for k, v := range s.decryptKey.getSSEHeaders(false) { |
||||
rh.Set(k, v) |
||||
} |
||||
objInfo, err = c.statObject(s.bucket, s.object, rh) |
||||
if err != nil { |
||||
err = fmt.Errorf("Could not stat object - %s/%s: %v", s.bucket, s.object, err) |
||||
} else { |
||||
size = objInfo.Size |
||||
etag = objInfo.ETag |
||||
userMeta = make(map[string]string) |
||||
for k, v := range objInfo.Metadata { |
||||
if strings.HasPrefix(k, "x-amz-meta-") { |
||||
if len(v) > 0 { |
||||
userMeta[k] = v[0] |
||||
} |
||||
} |
||||
} |
||||
} |
||||
return |
||||
} |
||||
|
||||
// uploadPartCopy - helper function to create a part in a multipart
|
||||
// upload via an upload-part-copy request
|
||||
// https://docs.aws.amazon.com/AmazonS3/latest/API/mpUploadUploadPartCopy.html
|
||||
func (c Client) uploadPartCopy(bucket, object, uploadID string, partNumber int, |
||||
headers http.Header) (p CompletePart, err error) { |
||||
|
||||
// Build query parameters
|
||||
urlValues := make(url.Values) |
||||
urlValues.Set("partNumber", strconv.Itoa(partNumber)) |
||||
urlValues.Set("uploadId", uploadID) |
||||
|
||||
// Send upload-part-copy request
|
||||
resp, err := c.executeMethod("PUT", requestMetadata{ |
||||
bucketName: bucket, |
||||
objectName: object, |
||||
customHeader: headers, |
||||
queryValues: urlValues, |
||||
}) |
||||
defer closeResponse(resp) |
||||
if err != nil { |
||||
return p, err |
||||
} |
||||
|
||||
// Check if we got an error response.
|
||||
if resp.StatusCode != http.StatusOK { |
||||
return p, httpRespToErrorResponse(resp, bucket, object) |
||||
} |
||||
|
||||
// Decode copy-part response on success.
|
||||
cpObjRes := copyObjectResult{} |
||||
err = xmlDecoder(resp.Body, &cpObjRes) |
||||
if err != nil { |
||||
return p, err |
||||
} |
||||
p.PartNumber, p.ETag = partNumber, cpObjRes.ETag |
||||
return p, nil |
||||
} |
||||
|
||||
// ComposeObject - creates an object using server-side copying of
|
||||
// existing objects. It takes a list of source objects (with optional
|
||||
// offsets) and concatenates them into a new object using only
|
||||
// server-side copying operations.
|
||||
func (c Client) ComposeObject(dst DestinationInfo, srcs []SourceInfo) error { |
||||
if len(srcs) < 1 || len(srcs) > maxPartsCount { |
||||
return ErrInvalidArgument("There must be as least one and up to 10000 source objects.") |
||||
} |
||||
|
||||
srcSizes := make([]int64, len(srcs)) |
||||
var totalSize, size, totalParts int64 |
||||
var srcUserMeta map[string]string |
||||
var etag string |
||||
var err error |
||||
for i, src := range srcs { |
||||
size, etag, srcUserMeta, err = src.getProps(c) |
||||
if err != nil { |
||||
return fmt.Errorf("Could not get source props for %s/%s: %v", src.bucket, src.object, err) |
||||
} |
||||
|
||||
// Error out if client side encryption is used in this source object when
|
||||
// more than one source objects are given.
|
||||
if len(srcs) > 1 && src.Headers.Get("x-amz-meta-x-amz-key") != "" { |
||||
return ErrInvalidArgument( |
||||
fmt.Sprintf("Client side encryption is used in source object %s/%s", src.bucket, src.object)) |
||||
} |
||||
|
||||
// Since we did a HEAD to get size, we use the ETag
|
||||
// value to make sure the object has not changed by
|
||||
// the time we perform the copy. This is done, only if
|
||||
// the user has not set their own ETag match
|
||||
// condition.
|
||||
if src.Headers.Get("x-amz-copy-source-if-match") == "" { |
||||
src.SetMatchETagCond(etag) |
||||
} |
||||
|
||||
// Check if a segment is specified, and if so, is the
|
||||
// segment within object bounds?
|
||||
if src.start != -1 { |
||||
// Since range is specified,
|
||||
// 0 <= src.start <= src.end
|
||||
// so only invalid case to check is:
|
||||
if src.end >= size { |
||||
return ErrInvalidArgument( |
||||
fmt.Sprintf("SourceInfo %d has invalid segment-to-copy [%d, %d] (size is %d)", |
||||
i, src.start, src.end, size)) |
||||
} |
||||
size = src.end - src.start + 1 |
||||
} |
||||
|
||||
// Only the last source may be less than `absMinPartSize`
|
||||
if size < absMinPartSize && i < len(srcs)-1 { |
||||
return ErrInvalidArgument( |
||||
fmt.Sprintf("SourceInfo %d is too small (%d) and it is not the last part", i, size)) |
||||
} |
||||
|
||||
// Is data to copy too large?
|
||||
totalSize += size |
||||
if totalSize > maxMultipartPutObjectSize { |
||||
return ErrInvalidArgument(fmt.Sprintf("Cannot compose an object of size %d (> 5TiB)", totalSize)) |
||||
} |
||||
|
||||
// record source size
|
||||
srcSizes[i] = size |
||||
|
||||
// calculate parts needed for current source
|
||||
totalParts += partsRequired(size) |
||||
// Do we need more parts than we are allowed?
|
||||
if totalParts > maxPartsCount { |
||||
return ErrInvalidArgument(fmt.Sprintf( |
||||
"Your proposed compose object requires more than %d parts", maxPartsCount)) |
||||
} |
||||
} |
||||
|
||||
// Single source object case (i.e. when only one source is
|
||||
// involved, it is being copied wholly and at most 5GiB in
|
||||
// size).
|
||||
if totalParts == 1 && srcs[0].start == -1 && totalSize <= maxPartSize { |
||||
h := srcs[0].Headers |
||||
// Add destination encryption headers
|
||||
for k, v := range dst.encryption.getSSEHeaders(false) { |
||||
h.Set(k, v) |
||||
} |
||||
|
||||
// If no user metadata is specified (and so, the
|
||||
// for-loop below is not entered), metadata from the
|
||||
// source is copied to the destination (due to
|
||||
// single-part copy-object PUT request behaviour).
|
||||
for k, v := range dst.getUserMetaHeadersMap(true) { |
||||
h.Set(k, v) |
||||
} |
||||
|
||||
// Send copy request
|
||||
resp, err := c.executeMethod("PUT", requestMetadata{ |
||||
bucketName: dst.bucket, |
||||
objectName: dst.object, |
||||
customHeader: h, |
||||
}) |
||||
defer closeResponse(resp) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
// Check if we got an error response.
|
||||
if resp.StatusCode != http.StatusOK { |
||||
return httpRespToErrorResponse(resp, dst.bucket, dst.object) |
||||
} |
||||
|
||||
// Return nil on success.
|
||||
return nil |
||||
} |
||||
|
||||
// Now, handle multipart-copy cases.
|
||||
|
||||
// 1. Initiate a new multipart upload.
|
||||
|
||||
// Set user-metadata on the destination object. If no
|
||||
// user-metadata is specified, and there is only one source,
|
||||
// (only) then metadata from source is copied.
|
||||
userMeta := dst.getUserMetaHeadersMap(false) |
||||
metaMap := userMeta |
||||
if len(userMeta) == 0 && len(srcs) == 1 { |
||||
metaMap = srcUserMeta |
||||
} |
||||
metaHeaders := make(map[string][]string) |
||||
for k, v := range metaMap { |
||||
metaHeaders[k] = append(metaHeaders[k], v) |
||||
} |
||||
uploadID, err := c.newUploadID(dst.bucket, dst.object, metaHeaders) |
||||
if err != nil { |
||||
return fmt.Errorf("Error creating new upload: %v", err) |
||||
} |
||||
|
||||
// 2. Perform copy part uploads
|
||||
objParts := []CompletePart{} |
||||
partIndex := 1 |
||||
for i, src := range srcs { |
||||
h := src.Headers |
||||
// Add destination encryption headers
|
||||
for k, v := range dst.encryption.getSSEHeaders(false) { |
||||
h.Set(k, v) |
||||
} |
||||
|
||||
// calculate start/end indices of parts after
|
||||
// splitting.
|
||||
startIdx, endIdx := calculateEvenSplits(srcSizes[i], src) |
||||
for j, start := range startIdx { |
||||
end := endIdx[j] |
||||
|
||||
// Add (or reset) source range header for
|
||||
// upload part copy request.
|
||||
h.Set("x-amz-copy-source-range", |
||||
fmt.Sprintf("bytes=%d-%d", start, end)) |
||||
|
||||
// make upload-part-copy request
|
||||
complPart, err := c.uploadPartCopy(dst.bucket, |
||||
dst.object, uploadID, partIndex, h) |
||||
if err != nil { |
||||
return fmt.Errorf("Error in upload-part-copy - %v", err) |
||||
} |
||||
objParts = append(objParts, complPart) |
||||
partIndex++ |
||||
} |
||||
} |
||||
|
||||
// 3. Make final complete-multipart request.
|
||||
_, err = c.completeMultipartUpload(dst.bucket, dst.object, uploadID, |
||||
completeMultipartUpload{Parts: objParts}) |
||||
if err != nil { |
||||
err = fmt.Errorf("Error in complete-multipart request - %v", err) |
||||
} |
||||
return err |
||||
} |
||||
|
||||
// partsRequired is ceiling(size / copyPartSize)
|
||||
func partsRequired(size int64) int64 { |
||||
r := size / copyPartSize |
||||
if size%copyPartSize > 0 { |
||||
r++ |
||||
} |
||||
return r |
||||
} |
||||
|
||||
// calculateEvenSplits - computes splits for a source and returns
|
||||
// start and end index slices. Splits happen evenly to be sure that no
|
||||
// part is less than 5MiB, as that could fail the multipart request if
|
||||
// it is not the last part.
|
||||
func calculateEvenSplits(size int64, src SourceInfo) (startIndex, endIndex []int64) { |
||||
if size == 0 { |
||||
return |
||||
} |
||||
|
||||
reqParts := partsRequired(size) |
||||
startIndex = make([]int64, reqParts) |
||||
endIndex = make([]int64, reqParts) |
||||
// Compute number of required parts `k`, as:
|
||||
//
|
||||
// k = ceiling(size / copyPartSize)
|
||||
//
|
||||
// Now, distribute the `size` bytes in the source into
|
||||
// k parts as evenly as possible:
|
||||
//
|
||||
// r parts sized (q+1) bytes, and
|
||||
// (k - r) parts sized q bytes, where
|
||||
//
|
||||
// size = q * k + r (by simple division of size by k,
|
||||
// so that 0 <= r < k)
|
||||
//
|
||||
start := src.start |
||||
if start == -1 { |
||||
start = 0 |
||||
} |
||||
quot, rem := size/reqParts, size%reqParts |
||||
nextStart := start |
||||
for j := int64(0); j < reqParts; j++ { |
||||
curPartSize := quot |
||||
if j < rem { |
||||
curPartSize++ |
||||
} |
||||
|
||||
cStart := nextStart |
||||
cEnd := cStart + curPartSize - 1 |
||||
nextStart = cEnd + 1 |
||||
|
||||
startIndex[j], endIndex[j] = cStart, cEnd |
||||
} |
||||
return |
||||
} |
||||
@ -0,0 +1,46 @@ |
||||
/* |
||||
* Minio Go Library for Amazon S3 Compatible Cloud Storage (C) 2015 Minio, Inc. |
||||
* |
||||
* 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. |
||||
*/ |
||||
|
||||
package minio |
||||
|
||||
import ( |
||||
"io" |
||||
|
||||
"github.com/minio/minio-go/pkg/encrypt" |
||||
) |
||||
|
||||
// PutEncryptedObject - Encrypt and store object.
|
||||
func (c Client) PutEncryptedObject(bucketName, objectName string, reader io.Reader, encryptMaterials encrypt.Materials, metadata map[string][]string, progress io.Reader) (n int64, err error) { |
||||
|
||||
if encryptMaterials == nil { |
||||
return 0, ErrInvalidArgument("Unable to recognize empty encryption properties") |
||||
} |
||||
|
||||
if err := encryptMaterials.SetupEncryptMode(reader); err != nil { |
||||
return 0, err |
||||
} |
||||
|
||||
if metadata == nil { |
||||
metadata = make(map[string][]string) |
||||
} |
||||
|
||||
// Set the necessary encryption headers, for future decryption.
|
||||
metadata[amzHeaderIV] = []string{encryptMaterials.GetIV()} |
||||
metadata[amzHeaderKey] = []string{encryptMaterials.GetKey()} |
||||
metadata[amzHeaderMatDesc] = []string{encryptMaterials.GetDesc()} |
||||
|
||||
return c.putObjectMultipartStreamNoLength(bucketName, objectName, encryptMaterials, metadata, progress) |
||||
} |
||||
@ -1,191 +0,0 @@ |
||||
/* |
||||
* Minio Go Library for Amazon S3 Compatible Cloud Storage (C) 2015 Minio, Inc. |
||||
* |
||||
* 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. |
||||
*/ |
||||
|
||||
package minio |
||||
|
||||
import ( |
||||
"io" |
||||
"strings" |
||||
|
||||
"github.com/minio/minio-go/pkg/credentials" |
||||
"github.com/minio/minio-go/pkg/encrypt" |
||||
"github.com/minio/minio-go/pkg/s3utils" |
||||
) |
||||
|
||||
// PutObjectWithProgress - with progress.
|
||||
func (c Client) PutObjectWithProgress(bucketName, objectName string, reader io.Reader, contentType string, progress io.Reader) (n int64, err error) { |
||||
metaData := make(map[string][]string) |
||||
metaData["Content-Type"] = []string{contentType} |
||||
return c.PutObjectWithMetadata(bucketName, objectName, reader, metaData, progress) |
||||
} |
||||
|
||||
// PutEncryptedObject - Encrypt and store object.
|
||||
func (c Client) PutEncryptedObject(bucketName, objectName string, reader io.Reader, encryptMaterials encrypt.Materials, metaData map[string][]string, progress io.Reader) (n int64, err error) { |
||||
|
||||
if encryptMaterials == nil { |
||||
return 0, ErrInvalidArgument("Unable to recognize empty encryption properties") |
||||
} |
||||
|
||||
if err := encryptMaterials.SetupEncryptMode(reader); err != nil { |
||||
return 0, err |
||||
} |
||||
|
||||
if metaData == nil { |
||||
metaData = make(map[string][]string) |
||||
} |
||||
|
||||
// Set the necessary encryption headers, for future decryption.
|
||||
metaData[amzHeaderIV] = []string{encryptMaterials.GetIV()} |
||||
metaData[amzHeaderKey] = []string{encryptMaterials.GetKey()} |
||||
metaData[amzHeaderMatDesc] = []string{encryptMaterials.GetDesc()} |
||||
|
||||
return c.PutObjectWithMetadata(bucketName, objectName, encryptMaterials, metaData, progress) |
||||
} |
||||
|
||||
// PutObjectWithMetadata - with metadata.
|
||||
func (c Client) PutObjectWithMetadata(bucketName, objectName string, reader io.Reader, metaData map[string][]string, progress io.Reader) (n int64, err error) { |
||||
// Input validation.
|
||||
if err := s3utils.CheckValidBucketName(bucketName); err != nil { |
||||
return 0, err |
||||
} |
||||
if err := s3utils.CheckValidObjectName(objectName); err != nil { |
||||
return 0, err |
||||
} |
||||
if reader == nil { |
||||
return 0, ErrInvalidArgument("Input reader is invalid, cannot be nil.") |
||||
} |
||||
|
||||
// Size of the object.
|
||||
var size int64 |
||||
|
||||
// Get reader size.
|
||||
size, err = getReaderSize(reader) |
||||
if err != nil { |
||||
return 0, err |
||||
} |
||||
|
||||
// Check for largest object size allowed.
|
||||
if size > int64(maxMultipartPutObjectSize) { |
||||
return 0, ErrEntityTooLarge(size, maxMultipartPutObjectSize, bucketName, objectName) |
||||
} |
||||
|
||||
// NOTE: Google Cloud Storage does not implement Amazon S3 Compatible multipart PUT.
|
||||
if s3utils.IsGoogleEndpoint(c.endpointURL) { |
||||
// Do not compute MD5 for Google Cloud Storage.
|
||||
return c.putObjectNoChecksum(bucketName, objectName, reader, size, metaData, progress) |
||||
} |
||||
|
||||
// putSmall object.
|
||||
if size < minPartSize && size >= 0 { |
||||
return c.putObjectSingle(bucketName, objectName, reader, size, metaData, progress) |
||||
} |
||||
|
||||
// For all sizes greater than 5MiB do multipart.
|
||||
n, err = c.putObjectMultipart(bucketName, objectName, reader, size, metaData, progress) |
||||
if err != nil { |
||||
errResp := ToErrorResponse(err) |
||||
// Verify if multipart functionality is not available, if not
|
||||
// fall back to single PutObject operation.
|
||||
if errResp.Code == "AccessDenied" && strings.Contains(errResp.Message, "Access Denied") { |
||||
// Verify if size of reader is greater than '5GiB'.
|
||||
if size > maxSinglePutObjectSize { |
||||
return 0, ErrEntityTooLarge(size, maxSinglePutObjectSize, bucketName, objectName) |
||||
} |
||||
// Fall back to uploading as single PutObject operation.
|
||||
return c.putObjectSingle(bucketName, objectName, reader, size, metaData, progress) |
||||
} |
||||
return n, err |
||||
} |
||||
return n, nil |
||||
} |
||||
|
||||
// PutObjectStreaming using AWS streaming signature V4
|
||||
func (c Client) PutObjectStreaming(bucketName, objectName string, reader io.Reader) (n int64, err error) { |
||||
return c.PutObjectStreamingWithProgress(bucketName, objectName, reader, nil, nil) |
||||
} |
||||
|
||||
// PutObjectStreamingWithMetadata using AWS streaming signature V4
|
||||
func (c Client) PutObjectStreamingWithMetadata(bucketName, objectName string, reader io.Reader, metadata map[string][]string) (n int64, err error) { |
||||
return c.PutObjectStreamingWithProgress(bucketName, objectName, reader, metadata, nil) |
||||
} |
||||
|
||||
// PutObjectStreamingWithProgress using AWS streaming signature V4
|
||||
func (c Client) PutObjectStreamingWithProgress(bucketName, objectName string, reader io.Reader, metadata map[string][]string, progress io.Reader) (n int64, err error) { |
||||
// NOTE: Streaming signature is not supported by GCS.
|
||||
if s3utils.IsGoogleEndpoint(c.endpointURL) { |
||||
return 0, ErrorResponse{ |
||||
Code: "NotImplemented", |
||||
Message: "AWS streaming signature v4 is not supported with Google Cloud Storage", |
||||
Key: objectName, |
||||
BucketName: bucketName, |
||||
} |
||||
} |
||||
|
||||
if c.overrideSignerType.IsV2() { |
||||
return 0, ErrorResponse{ |
||||
Code: "NotImplemented", |
||||
Message: "AWS streaming signature v4 is not supported with minio client initialized for AWS signature v2", |
||||
Key: objectName, |
||||
BucketName: bucketName, |
||||
} |
||||
} |
||||
|
||||
// Size of the object.
|
||||
var size int64 |
||||
|
||||
// Get reader size.
|
||||
size, err = getReaderSize(reader) |
||||
if err != nil { |
||||
return 0, err |
||||
} |
||||
|
||||
// Check for largest object size allowed.
|
||||
if size > int64(maxMultipartPutObjectSize) { |
||||
return 0, ErrEntityTooLarge(size, maxMultipartPutObjectSize, bucketName, objectName) |
||||
} |
||||
|
||||
// If size cannot be found on a stream, it is not possible
|
||||
// to upload using streaming signature, fall back to multipart.
|
||||
if size < 0 { |
||||
return c.putObjectMultipartStream(bucketName, objectName, reader, size, metadata, progress) |
||||
} |
||||
|
||||
// Set streaming signature.
|
||||
c.overrideSignerType = credentials.SignatureV4Streaming |
||||
|
||||
if size < minPartSize && size >= 0 { |
||||
return c.putObjectNoChecksum(bucketName, objectName, reader, size, metadata, progress) |
||||
} |
||||
|
||||
// For all sizes greater than 64MiB do multipart.
|
||||
n, err = c.putObjectMultipartStreamNoChecksum(bucketName, objectName, reader, size, metadata, progress) |
||||
if err != nil { |
||||
errResp := ToErrorResponse(err) |
||||
// Verify if multipart functionality is not available, if not
|
||||
// fall back to single PutObject operation.
|
||||
if errResp.Code == "AccessDenied" && strings.Contains(errResp.Message, "Access Denied") { |
||||
// Verify if size of reader is greater than '5GiB'.
|
||||
if size > maxSinglePutObjectSize { |
||||
return 0, ErrEntityTooLarge(size, maxSinglePutObjectSize, bucketName, objectName) |
||||
} |
||||
// Fall back to uploading as single PutObject operation.
|
||||
return c.putObjectNoChecksum(bucketName, objectName, reader, size, metadata, progress) |
||||
} |
||||
return n, err |
||||
} |
||||
|
||||
return n, nil |
||||
} |
||||
@ -1,219 +0,0 @@ |
||||
/* |
||||
* Minio Go Library for Amazon S3 Compatible Cloud Storage (C) 2015, 2016 Minio, Inc. |
||||
* |
||||
* 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. |
||||
*/ |
||||
|
||||
package minio |
||||
|
||||
import ( |
||||
"bytes" |
||||
"fmt" |
||||
"io" |
||||
"io/ioutil" |
||||
"sort" |
||||
|
||||
"github.com/minio/minio-go/pkg/s3utils" |
||||
) |
||||
|
||||
// uploadedPartRes - the response received from a part upload.
|
||||
type uploadedPartRes struct { |
||||
Error error // Any error encountered while uploading the part.
|
||||
PartNum int // Number of the part uploaded.
|
||||
Size int64 // Size of the part uploaded.
|
||||
Part *ObjectPart |
||||
} |
||||
|
||||
type uploadPartReq struct { |
||||
PartNum int // Number of the part uploaded.
|
||||
Part *ObjectPart // Size of the part uploaded.
|
||||
} |
||||
|
||||
// putObjectMultipartFromReadAt - Uploads files bigger than 5MiB. Supports reader
|
||||
// of type which implements io.ReaderAt interface (ReadAt method).
|
||||
//
|
||||
// NOTE: This function is meant to be used for all readers which
|
||||
// implement io.ReaderAt which allows us for resuming multipart
|
||||
// uploads but reading at an offset, which would avoid re-read the
|
||||
// data which was already uploaded. Internally this function uses
|
||||
// temporary files for staging all the data, these temporary files are
|
||||
// cleaned automatically when the caller i.e http client closes the
|
||||
// stream after uploading all the contents successfully.
|
||||
func (c Client) putObjectMultipartFromReadAt(bucketName, objectName string, reader io.ReaderAt, size int64, metaData map[string][]string, progress io.Reader) (n int64, err error) { |
||||
// Input validation.
|
||||
if err := s3utils.CheckValidBucketName(bucketName); err != nil { |
||||
return 0, err |
||||
} |
||||
if err := s3utils.CheckValidObjectName(objectName); err != nil { |
||||
return 0, err |
||||
} |
||||
|
||||
// Initiate a new multipart upload.
|
||||
uploadID, err := c.newUploadID(bucketName, objectName, metaData) |
||||
if err != nil { |
||||
return 0, err |
||||
} |
||||
|
||||
// Total data read and written to server. should be equal to 'size' at the end of the call.
|
||||
var totalUploadedSize int64 |
||||
|
||||
// Complete multipart upload.
|
||||
var complMultipartUpload completeMultipartUpload |
||||
|
||||
// Calculate the optimal parts info for a given size.
|
||||
totalPartsCount, partSize, lastPartSize, err := optimalPartInfo(size) |
||||
if err != nil { |
||||
return 0, err |
||||
} |
||||
|
||||
// Declare a channel that sends the next part number to be uploaded.
|
||||
// Buffered to 10000 because thats the maximum number of parts allowed
|
||||
// by S3.
|
||||
uploadPartsCh := make(chan uploadPartReq, 10000) |
||||
|
||||
// Declare a channel that sends back the response of a part upload.
|
||||
// Buffered to 10000 because thats the maximum number of parts allowed
|
||||
// by S3.
|
||||
uploadedPartsCh := make(chan uploadedPartRes, 10000) |
||||
|
||||
// Used for readability, lastPartNumber is always totalPartsCount.
|
||||
lastPartNumber := totalPartsCount |
||||
|
||||
// Initialize parts uploaded map.
|
||||
partsInfo := make(map[int]ObjectPart) |
||||
|
||||
// Send each part number to the channel to be processed.
|
||||
for p := 1; p <= totalPartsCount; p++ { |
||||
part, ok := partsInfo[p] |
||||
if ok { |
||||
uploadPartsCh <- uploadPartReq{PartNum: p, Part: &part} |
||||
} else { |
||||
uploadPartsCh <- uploadPartReq{PartNum: p, Part: nil} |
||||
} |
||||
} |
||||
close(uploadPartsCh) |
||||
|
||||
// Receive each part number from the channel allowing three parallel uploads.
|
||||
for w := 1; w <= totalWorkers; w++ { |
||||
go func() { |
||||
// Read defaults to reading at 5MiB buffer.
|
||||
readAtBuffer := make([]byte, optimalReadBufferSize) |
||||
|
||||
// Each worker will draw from the part channel and upload in parallel.
|
||||
for uploadReq := range uploadPartsCh { |
||||
// Declare a new tmpBuffer.
|
||||
tmpBuffer := new(bytes.Buffer) |
||||
|
||||
// If partNumber was not uploaded we calculate the missing
|
||||
// part offset and size. For all other part numbers we
|
||||
// calculate offset based on multiples of partSize.
|
||||
readOffset := int64(uploadReq.PartNum-1) * partSize |
||||
missingPartSize := partSize |
||||
|
||||
// As a special case if partNumber is lastPartNumber, we
|
||||
// calculate the offset based on the last part size.
|
||||
if uploadReq.PartNum == lastPartNumber { |
||||
readOffset = (size - lastPartSize) |
||||
missingPartSize = lastPartSize |
||||
} |
||||
|
||||
// Get a section reader on a particular offset.
|
||||
sectionReader := io.NewSectionReader(reader, readOffset, missingPartSize) |
||||
|
||||
// Choose the needed hash algorithms to be calculated by hashCopyBuffer.
|
||||
// Sha256 is avoided in non-v4 signature requests or HTTPS connections
|
||||
hashAlgos, hashSums := c.hashMaterials() |
||||
|
||||
var prtSize int64 |
||||
var err error |
||||
prtSize, err = hashCopyBuffer(hashAlgos, hashSums, tmpBuffer, sectionReader, readAtBuffer) |
||||
if err != nil { |
||||
// Send the error back through the channel.
|
||||
uploadedPartsCh <- uploadedPartRes{ |
||||
Size: 0, |
||||
Error: err, |
||||
} |
||||
// Exit the goroutine.
|
||||
return |
||||
} |
||||
|
||||
// Proceed to upload the part.
|
||||
var objPart ObjectPart |
||||
objPart, err = c.uploadPart(bucketName, objectName, uploadID, tmpBuffer, |
||||
uploadReq.PartNum, hashSums["md5"], hashSums["sha256"], prtSize) |
||||
if err != nil { |
||||
uploadedPartsCh <- uploadedPartRes{ |
||||
Size: 0, |
||||
Error: err, |
||||
} |
||||
// Exit the goroutine.
|
||||
return |
||||
} |
||||
|
||||
// Save successfully uploaded part metadata.
|
||||
uploadReq.Part = &objPart |
||||
|
||||
// Send successful part info through the channel.
|
||||
uploadedPartsCh <- uploadedPartRes{ |
||||
Size: missingPartSize, |
||||
PartNum: uploadReq.PartNum, |
||||
Part: uploadReq.Part, |
||||
Error: nil, |
||||
} |
||||
} |
||||
}() |
||||
} |
||||
|
||||
// Gather the responses as they occur and update any
|
||||
// progress bar.
|
||||
for u := 1; u <= totalPartsCount; u++ { |
||||
uploadRes := <-uploadedPartsCh |
||||
if uploadRes.Error != nil { |
||||
return totalUploadedSize, uploadRes.Error |
||||
} |
||||
// Retrieve each uploaded part and store it to be completed.
|
||||
// part, ok := partsInfo[uploadRes.PartNum]
|
||||
part := uploadRes.Part |
||||
if part == nil { |
||||
return 0, ErrInvalidArgument(fmt.Sprintf("Missing part number %d", uploadRes.PartNum)) |
||||
} |
||||
// Update the totalUploadedSize.
|
||||
totalUploadedSize += uploadRes.Size |
||||
// Update the progress bar if there is one.
|
||||
if progress != nil { |
||||
if _, err = io.CopyN(ioutil.Discard, progress, uploadRes.Size); err != nil { |
||||
return totalUploadedSize, err |
||||
} |
||||
} |
||||
// Store the parts to be completed in order.
|
||||
complMultipartUpload.Parts = append(complMultipartUpload.Parts, CompletePart{ |
||||
ETag: part.ETag, |
||||
PartNumber: part.PartNumber, |
||||
}) |
||||
} |
||||
|
||||
// Verify if we uploaded all the data.
|
||||
if totalUploadedSize != size { |
||||
return totalUploadedSize, ErrUnexpectedEOF(totalUploadedSize, size, bucketName, objectName) |
||||
} |
||||
|
||||
// Sort all completed parts.
|
||||
sort.Sort(completedParts(complMultipartUpload.Parts)) |
||||
_, err = c.completeMultipartUpload(bucketName, objectName, uploadID, complMultipartUpload) |
||||
if err != nil { |
||||
return totalUploadedSize, err |
||||
} |
||||
|
||||
// Return final size.
|
||||
return totalUploadedSize, nil |
||||
} |
||||
@ -0,0 +1,436 @@ |
||||
/* |
||||
* Minio Go Library for Amazon S3 Compatible Cloud Storage (C) 2017 Minio, Inc. |
||||
* |
||||
* 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. |
||||
*/ |
||||
|
||||
package minio |
||||
|
||||
import ( |
||||
"fmt" |
||||
"io" |
||||
"net/http" |
||||
"sort" |
||||
"strings" |
||||
|
||||
"github.com/minio/minio-go/pkg/s3utils" |
||||
) |
||||
|
||||
// PutObjectStreaming using AWS streaming signature V4
|
||||
func (c Client) PutObjectStreaming(bucketName, objectName string, reader io.Reader) (n int64, err error) { |
||||
return c.PutObjectWithProgress(bucketName, objectName, reader, nil, nil) |
||||
} |
||||
|
||||
// putObjectMultipartStream - upload a large object using
|
||||
// multipart upload and streaming signature for signing payload.
|
||||
// Comprehensive put object operation involving multipart uploads.
|
||||
//
|
||||
// Following code handles these types of readers.
|
||||
//
|
||||
// - *os.File
|
||||
// - *minio.Object
|
||||
// - Any reader which has a method 'ReadAt()'
|
||||
//
|
||||
func (c Client) putObjectMultipartStream(bucketName, objectName string, |
||||
reader io.Reader, size int64, metadata map[string][]string, progress io.Reader) (n int64, err error) { |
||||
|
||||
// Verify if reader is *minio.Object, *os.File or io.ReaderAt.
|
||||
// NOTE: Verification of object is kept for a specific purpose
|
||||
// while it is going to be duck typed similar to io.ReaderAt.
|
||||
// It is to indicate that *minio.Object implements io.ReaderAt.
|
||||
// and such a functionality is used in the subsequent code path.
|
||||
if isFile(reader) || !isObject(reader) && isReadAt(reader) { |
||||
n, err = c.putObjectMultipartStreamFromReadAt(bucketName, objectName, reader.(io.ReaderAt), size, metadata, progress) |
||||
} else { |
||||
n, err = c.putObjectMultipartStreamNoChecksum(bucketName, objectName, reader, size, metadata, progress) |
||||
} |
||||
if err != nil { |
||||
errResp := ToErrorResponse(err) |
||||
// Verify if multipart functionality is not available, if not
|
||||
// fall back to single PutObject operation.
|
||||
if errResp.Code == "AccessDenied" && strings.Contains(errResp.Message, "Access Denied") { |
||||
// Verify if size of reader is greater than '5GiB'.
|
||||
if size > maxSinglePutObjectSize { |
||||
return 0, ErrEntityTooLarge(size, maxSinglePutObjectSize, bucketName, objectName) |
||||
} |
||||
// Fall back to uploading as single PutObject operation.
|
||||
return c.putObjectNoChecksum(bucketName, objectName, reader, size, metadata, progress) |
||||
} |
||||
} |
||||
return n, err |
||||
} |
||||
|
||||
// uploadedPartRes - the response received from a part upload.
|
||||
type uploadedPartRes struct { |
||||
Error error // Any error encountered while uploading the part.
|
||||
PartNum int // Number of the part uploaded.
|
||||
Size int64 // Size of the part uploaded.
|
||||
Part *ObjectPart |
||||
} |
||||
|
||||
type uploadPartReq struct { |
||||
PartNum int // Number of the part uploaded.
|
||||
Part *ObjectPart // Size of the part uploaded.
|
||||
} |
||||
|
||||
// putObjectMultipartFromReadAt - Uploads files bigger than 64MiB.
|
||||
// Supports all readers which implements io.ReaderAt interface
|
||||
// (ReadAt method).
|
||||
//
|
||||
// NOTE: This function is meant to be used for all readers which
|
||||
// implement io.ReaderAt which allows us for resuming multipart
|
||||
// uploads but reading at an offset, which would avoid re-read the
|
||||
// data which was already uploaded. Internally this function uses
|
||||
// temporary files for staging all the data, these temporary files are
|
||||
// cleaned automatically when the caller i.e http client closes the
|
||||
// stream after uploading all the contents successfully.
|
||||
func (c Client) putObjectMultipartStreamFromReadAt(bucketName, objectName string, |
||||
reader io.ReaderAt, size int64, metadata map[string][]string, progress io.Reader) (n int64, err error) { |
||||
// Input validation.
|
||||
if err = s3utils.CheckValidBucketName(bucketName); err != nil { |
||||
return 0, err |
||||
} |
||||
if err = s3utils.CheckValidObjectName(objectName); err != nil { |
||||
return 0, err |
||||
} |
||||
|
||||
// Calculate the optimal parts info for a given size.
|
||||
totalPartsCount, partSize, lastPartSize, err := optimalPartInfo(size) |
||||
if err != nil { |
||||
return 0, err |
||||
} |
||||
|
||||
// Initiate a new multipart upload.
|
||||
uploadID, err := c.newUploadID(bucketName, objectName, metadata) |
||||
if err != nil { |
||||
return 0, err |
||||
} |
||||
|
||||
// Aborts the multipart upload in progress, if the
|
||||
// function returns any error, since we do not resume
|
||||
// we should purge the parts which have been uploaded
|
||||
// to relinquish storage space.
|
||||
defer func() { |
||||
if err != nil { |
||||
c.abortMultipartUpload(bucketName, objectName, uploadID) |
||||
} |
||||
}() |
||||
|
||||
// Total data read and written to server. should be equal to 'size' at the end of the call.
|
||||
var totalUploadedSize int64 |
||||
|
||||
// Complete multipart upload.
|
||||
var complMultipartUpload completeMultipartUpload |
||||
|
||||
// Declare a channel that sends the next part number to be uploaded.
|
||||
// Buffered to 10000 because thats the maximum number of parts allowed
|
||||
// by S3.
|
||||
uploadPartsCh := make(chan uploadPartReq, 10000) |
||||
|
||||
// Declare a channel that sends back the response of a part upload.
|
||||
// Buffered to 10000 because thats the maximum number of parts allowed
|
||||
// by S3.
|
||||
uploadedPartsCh := make(chan uploadedPartRes, 10000) |
||||
|
||||
// Used for readability, lastPartNumber is always totalPartsCount.
|
||||
lastPartNumber := totalPartsCount |
||||
|
||||
// Send each part number to the channel to be processed.
|
||||
for p := 1; p <= totalPartsCount; p++ { |
||||
uploadPartsCh <- uploadPartReq{PartNum: p, Part: nil} |
||||
} |
||||
close(uploadPartsCh) |
||||
|
||||
// Receive each part number from the channel allowing three parallel uploads.
|
||||
for w := 1; w <= totalWorkers; w++ { |
||||
go func(partSize int64) { |
||||
// Each worker will draw from the part channel and upload in parallel.
|
||||
for uploadReq := range uploadPartsCh { |
||||
|
||||
// If partNumber was not uploaded we calculate the missing
|
||||
// part offset and size. For all other part numbers we
|
||||
// calculate offset based on multiples of partSize.
|
||||
readOffset := int64(uploadReq.PartNum-1) * partSize |
||||
|
||||
// As a special case if partNumber is lastPartNumber, we
|
||||
// calculate the offset based on the last part size.
|
||||
if uploadReq.PartNum == lastPartNumber { |
||||
readOffset = (size - lastPartSize) |
||||
partSize = lastPartSize |
||||
} |
||||
|
||||
// Get a section reader on a particular offset.
|
||||
sectionReader := newHook(io.NewSectionReader(reader, readOffset, partSize), progress) |
||||
|
||||
// Proceed to upload the part.
|
||||
var objPart ObjectPart |
||||
objPart, err = c.uploadPart(bucketName, objectName, uploadID, |
||||
sectionReader, uploadReq.PartNum, |
||||
nil, nil, partSize, metadata) |
||||
if err != nil { |
||||
uploadedPartsCh <- uploadedPartRes{ |
||||
Size: 0, |
||||
Error: err, |
||||
} |
||||
// Exit the goroutine.
|
||||
return |
||||
} |
||||
|
||||
// Save successfully uploaded part metadata.
|
||||
uploadReq.Part = &objPart |
||||
|
||||
// Send successful part info through the channel.
|
||||
uploadedPartsCh <- uploadedPartRes{ |
||||
Size: objPart.Size, |
||||
PartNum: uploadReq.PartNum, |
||||
Part: uploadReq.Part, |
||||
Error: nil, |
||||
} |
||||
} |
||||
}(partSize) |
||||
} |
||||
|
||||
// Gather the responses as they occur and update any
|
||||
// progress bar.
|
||||
for u := 1; u <= totalPartsCount; u++ { |
||||
uploadRes := <-uploadedPartsCh |
||||
if uploadRes.Error != nil { |
||||
return totalUploadedSize, uploadRes.Error |
||||
} |
||||
// Retrieve each uploaded part and store it to be completed.
|
||||
// part, ok := partsInfo[uploadRes.PartNum]
|
||||
part := uploadRes.Part |
||||
if part == nil { |
||||
return 0, ErrInvalidArgument(fmt.Sprintf("Missing part number %d", uploadRes.PartNum)) |
||||
} |
||||
// Update the totalUploadedSize.
|
||||
totalUploadedSize += uploadRes.Size |
||||
// Store the parts to be completed in order.
|
||||
complMultipartUpload.Parts = append(complMultipartUpload.Parts, CompletePart{ |
||||
ETag: part.ETag, |
||||
PartNumber: part.PartNumber, |
||||
}) |
||||
} |
||||
|
||||
// Verify if we uploaded all the data.
|
||||
if totalUploadedSize != size { |
||||
return totalUploadedSize, ErrUnexpectedEOF(totalUploadedSize, size, bucketName, objectName) |
||||
} |
||||
|
||||
// Sort all completed parts.
|
||||
sort.Sort(completedParts(complMultipartUpload.Parts)) |
||||
_, err = c.completeMultipartUpload(bucketName, objectName, uploadID, complMultipartUpload) |
||||
if err != nil { |
||||
return totalUploadedSize, err |
||||
} |
||||
|
||||
// Return final size.
|
||||
return totalUploadedSize, nil |
||||
} |
||||
|
||||
func (c Client) putObjectMultipartStreamNoChecksum(bucketName, objectName string, |
||||
reader io.Reader, size int64, metadata map[string][]string, progress io.Reader) (n int64, err error) { |
||||
// Input validation.
|
||||
if err = s3utils.CheckValidBucketName(bucketName); err != nil { |
||||
return 0, err |
||||
} |
||||
if err = s3utils.CheckValidObjectName(objectName); err != nil { |
||||
return 0, err |
||||
} |
||||
|
||||
// Calculate the optimal parts info for a given size.
|
||||
totalPartsCount, partSize, lastPartSize, err := optimalPartInfo(size) |
||||
if err != nil { |
||||
return 0, err |
||||
} |
||||
|
||||
// Initiates a new multipart request
|
||||
uploadID, err := c.newUploadID(bucketName, objectName, metadata) |
||||
if err != nil { |
||||
return 0, err |
||||
} |
||||
|
||||
// Aborts the multipart upload if the function returns
|
||||
// any error, since we do not resume we should purge
|
||||
// the parts which have been uploaded to relinquish
|
||||
// storage space.
|
||||
defer func() { |
||||
if err != nil { |
||||
c.abortMultipartUpload(bucketName, objectName, uploadID) |
||||
} |
||||
}() |
||||
|
||||
// Total data read and written to server. should be equal to 'size' at the end of the call.
|
||||
var totalUploadedSize int64 |
||||
|
||||
// Initialize parts uploaded map.
|
||||
partsInfo := make(map[int]ObjectPart) |
||||
|
||||
// Part number always starts with '1'.
|
||||
var partNumber int |
||||
for partNumber = 1; partNumber <= totalPartsCount; partNumber++ { |
||||
// Update progress reader appropriately to the latest offset
|
||||
// as we read from the source.
|
||||
hookReader := newHook(reader, progress) |
||||
|
||||
// Proceed to upload the part.
|
||||
if partNumber == totalPartsCount { |
||||
partSize = lastPartSize |
||||
} |
||||
|
||||
var objPart ObjectPart |
||||
objPart, err = c.uploadPart(bucketName, objectName, uploadID, |
||||
io.LimitReader(hookReader, partSize), |
||||
partNumber, nil, nil, partSize, metadata) |
||||
if err != nil { |
||||
return totalUploadedSize, err |
||||
} |
||||
|
||||
// Save successfully uploaded part metadata.
|
||||
partsInfo[partNumber] = objPart |
||||
|
||||
// Save successfully uploaded size.
|
||||
totalUploadedSize += partSize |
||||
} |
||||
|
||||
// Verify if we uploaded all the data.
|
||||
if size > 0 { |
||||
if totalUploadedSize != size { |
||||
return totalUploadedSize, ErrUnexpectedEOF(totalUploadedSize, size, bucketName, objectName) |
||||
} |
||||
} |
||||
|
||||
// Complete multipart upload.
|
||||
var complMultipartUpload completeMultipartUpload |
||||
|
||||
// Loop over total uploaded parts to save them in
|
||||
// Parts array before completing the multipart request.
|
||||
for i := 1; i < partNumber; i++ { |
||||
part, ok := partsInfo[i] |
||||
if !ok { |
||||
return 0, ErrInvalidArgument(fmt.Sprintf("Missing part number %d", i)) |
||||
} |
||||
complMultipartUpload.Parts = append(complMultipartUpload.Parts, CompletePart{ |
||||
ETag: part.ETag, |
||||
PartNumber: part.PartNumber, |
||||
}) |
||||
} |
||||
|
||||
// Sort all completed parts.
|
||||
sort.Sort(completedParts(complMultipartUpload.Parts)) |
||||
_, err = c.completeMultipartUpload(bucketName, objectName, uploadID, complMultipartUpload) |
||||
if err != nil { |
||||
return totalUploadedSize, err |
||||
} |
||||
|
||||
// Return final size.
|
||||
return totalUploadedSize, nil |
||||
} |
||||
|
||||
// putObjectNoChecksum special function used Google Cloud Storage. This special function
|
||||
// is used for Google Cloud Storage since Google's multipart API is not S3 compatible.
|
||||
func (c Client) putObjectNoChecksum(bucketName, objectName string, reader io.Reader, size int64, metaData map[string][]string, progress io.Reader) (n int64, err error) { |
||||
// Input validation.
|
||||
if err := s3utils.CheckValidBucketName(bucketName); err != nil { |
||||
return 0, err |
||||
} |
||||
if err := s3utils.CheckValidObjectName(objectName); err != nil { |
||||
return 0, err |
||||
} |
||||
|
||||
// Size -1 is only supported on Google Cloud Storage, we error
|
||||
// out in all other situations.
|
||||
if size < 0 && !s3utils.IsGoogleEndpoint(c.endpointURL) { |
||||
return 0, ErrEntityTooSmall(size, bucketName, objectName) |
||||
} |
||||
if size > 0 { |
||||
if isReadAt(reader) && !isObject(reader) { |
||||
reader = io.NewSectionReader(reader.(io.ReaderAt), 0, size) |
||||
} |
||||
} |
||||
|
||||
// Update progress reader appropriately to the latest offset as we
|
||||
// read from the source.
|
||||
readSeeker := newHook(reader, progress) |
||||
|
||||
// This function does not calculate sha256 and md5sum for payload.
|
||||
// Execute put object.
|
||||
st, err := c.putObjectDo(bucketName, objectName, readSeeker, nil, nil, size, metaData) |
||||
if err != nil { |
||||
return 0, err |
||||
} |
||||
if st.Size != size { |
||||
return 0, ErrUnexpectedEOF(st.Size, size, bucketName, objectName) |
||||
} |
||||
return size, nil |
||||
} |
||||
|
||||
// putObjectDo - executes the put object http operation.
|
||||
// NOTE: You must have WRITE permissions on a bucket to add an object to it.
|
||||
func (c Client) putObjectDo(bucketName, objectName string, reader io.Reader, md5Sum []byte, sha256Sum []byte, size int64, metaData map[string][]string) (ObjectInfo, error) { |
||||
// Input validation.
|
||||
if err := s3utils.CheckValidBucketName(bucketName); err != nil { |
||||
return ObjectInfo{}, err |
||||
} |
||||
if err := s3utils.CheckValidObjectName(objectName); err != nil { |
||||
return ObjectInfo{}, err |
||||
} |
||||
|
||||
// Set headers.
|
||||
customHeader := make(http.Header) |
||||
|
||||
// Set metadata to headers
|
||||
for k, v := range metaData { |
||||
if len(v) > 0 { |
||||
customHeader.Set(k, v[0]) |
||||
} |
||||
} |
||||
|
||||
// If Content-Type is not provided, set the default application/octet-stream one
|
||||
if v, ok := metaData["Content-Type"]; !ok || len(v) == 0 { |
||||
customHeader.Set("Content-Type", "application/octet-stream") |
||||
} |
||||
|
||||
// Populate request metadata.
|
||||
reqMetadata := requestMetadata{ |
||||
bucketName: bucketName, |
||||
objectName: objectName, |
||||
customHeader: customHeader, |
||||
contentBody: reader, |
||||
contentLength: size, |
||||
contentMD5Bytes: md5Sum, |
||||
contentSHA256Bytes: sha256Sum, |
||||
} |
||||
|
||||
// Execute PUT an objectName.
|
||||
resp, err := c.executeMethod("PUT", reqMetadata) |
||||
defer closeResponse(resp) |
||||
if err != nil { |
||||
return ObjectInfo{}, err |
||||
} |
||||
if resp != nil { |
||||
if resp.StatusCode != http.StatusOK { |
||||
return ObjectInfo{}, httpRespToErrorResponse(resp, bucketName, objectName) |
||||
} |
||||
} |
||||
|
||||
var objInfo ObjectInfo |
||||
// Trim off the odd double quotes from ETag in the beginning and end.
|
||||
objInfo.ETag = strings.TrimPrefix(resp.Header.Get("ETag"), "\"") |
||||
objInfo.ETag = strings.TrimSuffix(objInfo.ETag, "\"") |
||||
// A success here means data was written to server successfully.
|
||||
objInfo.Size = size |
||||
|
||||
// Return here.
|
||||
return objInfo, nil |
||||
} |
||||
@ -1,99 +0,0 @@ |
||||
/* |
||||
* Minio Go Library for Amazon S3 Compatible Cloud Storage (C) 2016 Minio, Inc. |
||||
* |
||||
* 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. |
||||
*/ |
||||
|
||||
package minio |
||||
|
||||
import ( |
||||
"net/http" |
||||
"time" |
||||
) |
||||
|
||||
// copyCondition explanation:
|
||||
// http://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectCOPY.html
|
||||
//
|
||||
// Example:
|
||||
//
|
||||
// copyCondition {
|
||||
// key: "x-amz-copy-if-modified-since",
|
||||
// value: "Tue, 15 Nov 1994 12:45:26 GMT",
|
||||
// }
|
||||
//
|
||||
type copyCondition struct { |
||||
key string |
||||
value string |
||||
} |
||||
|
||||
// CopyConditions - copy conditions.
|
||||
type CopyConditions struct { |
||||
conditions []copyCondition |
||||
} |
||||
|
||||
// NewCopyConditions - Instantiate new list of conditions. This
|
||||
// function is left behind for backward compatibility. The idiomatic
|
||||
// way to set an empty set of copy conditions is,
|
||||
// ``copyConditions := CopyConditions{}``.
|
||||
//
|
||||
func NewCopyConditions() CopyConditions { |
||||
return CopyConditions{} |
||||
} |
||||
|
||||
// SetMatchETag - set match etag.
|
||||
func (c *CopyConditions) SetMatchETag(etag string) error { |
||||
if etag == "" { |
||||
return ErrInvalidArgument("ETag cannot be empty.") |
||||
} |
||||
c.conditions = append(c.conditions, copyCondition{ |
||||
key: "x-amz-copy-source-if-match", |
||||
value: etag, |
||||
}) |
||||
return nil |
||||
} |
||||
|
||||
// SetMatchETagExcept - set match etag except.
|
||||
func (c *CopyConditions) SetMatchETagExcept(etag string) error { |
||||
if etag == "" { |
||||
return ErrInvalidArgument("ETag cannot be empty.") |
||||
} |
||||
c.conditions = append(c.conditions, copyCondition{ |
||||
key: "x-amz-copy-source-if-none-match", |
||||
value: etag, |
||||
}) |
||||
return nil |
||||
} |
||||
|
||||
// SetUnmodified - set unmodified time since.
|
||||
func (c *CopyConditions) SetUnmodified(modTime time.Time) error { |
||||
if modTime.IsZero() { |
||||
return ErrInvalidArgument("Modified since cannot be empty.") |
||||
} |
||||
c.conditions = append(c.conditions, copyCondition{ |
||||
key: "x-amz-copy-source-if-unmodified-since", |
||||
value: modTime.Format(http.TimeFormat), |
||||
}) |
||||
return nil |
||||
} |
||||
|
||||
// SetModified - set modified time since.
|
||||
func (c *CopyConditions) SetModified(modTime time.Time) error { |
||||
if modTime.IsZero() { |
||||
return ErrInvalidArgument("Modified since cannot be empty.") |
||||
} |
||||
c.conditions = append(c.conditions, copyCondition{ |
||||
key: "x-amz-copy-source-if-modified-since", |
||||
value: modTime.Format(http.TimeFormat), |
||||
}) |
||||
return nil |
||||
} |
||||
@ -1,60 +0,0 @@ |
||||
/* |
||||
* Minio Go Library for Amazon S3 Compatible Cloud Storage (C) 2015 Minio, Inc. |
||||
* |
||||
* 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. |
||||
*/ |
||||
|
||||
package minio |
||||
|
||||
import ( |
||||
"io/ioutil" |
||||
"os" |
||||
"sync" |
||||
) |
||||
|
||||
// tempFile - temporary file container.
|
||||
type tempFile struct { |
||||
*os.File |
||||
mutex *sync.Mutex |
||||
} |
||||
|
||||
// newTempFile returns a new temporary file, once closed it automatically deletes itself.
|
||||
func newTempFile(prefix string) (*tempFile, error) { |
||||
// use platform specific temp directory.
|
||||
file, err := ioutil.TempFile(os.TempDir(), prefix) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
return &tempFile{ |
||||
File: file, |
||||
mutex: &sync.Mutex{}, |
||||
}, nil |
||||
} |
||||
|
||||
// Close - closer wrapper to close and remove temporary file.
|
||||
func (t *tempFile) Close() error { |
||||
t.mutex.Lock() |
||||
defer t.mutex.Unlock() |
||||
if t.File != nil { |
||||
// Close the file.
|
||||
if err := t.File.Close(); err != nil { |
||||
return err |
||||
} |
||||
// Remove file.
|
||||
if err := os.Remove(t.File.Name()); err != nil { |
||||
return err |
||||
} |
||||
t.File = nil |
||||
} |
||||
return nil |
||||
} |
||||
@ -0,0 +1,48 @@ |
||||
// +build go1.7 go1.8
|
||||
|
||||
/* |
||||
* Minio Go Library for Amazon S3 Compatible Cloud Storage |
||||
* (C) 2017 Minio, Inc. |
||||
* |
||||
* 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. |
||||
*/ |
||||
|
||||
package minio |
||||
|
||||
import ( |
||||
"net" |
||||
"net/http" |
||||
"time" |
||||
) |
||||
|
||||
// This default transport is similar to http.DefaultTransport
|
||||
// but with additional DisableCompression:
|
||||
var defaultMinioTransport http.RoundTripper = &http.Transport{ |
||||
Proxy: http.ProxyFromEnvironment, |
||||
DialContext: (&net.Dialer{ |
||||
Timeout: 30 * time.Second, |
||||
KeepAlive: 30 * time.Second, |
||||
DualStack: true, |
||||
}).DialContext, |
||||
MaxIdleConns: 100, |
||||
IdleConnTimeout: 90 * time.Second, |
||||
TLSHandshakeTimeout: 10 * time.Second, |
||||
ExpectContinueTimeout: 1 * time.Second, |
||||
// Set this value so that the underlying transport round-tripper
|
||||
// doesn't try to auto decode the body of objects with
|
||||
// content-encoding set to `gzip`.
|
||||
//
|
||||
// Refer:
|
||||
// https://golang.org/src/net/http/transport.go?h=roundTrip#L1843
|
||||
DisableCompression: true, |
||||
} |
||||
@ -0,0 +1,39 @@ |
||||
// +build go1.5,!go1.6,!go1.7,!go1.8
|
||||
|
||||
/* |
||||
* Minio Go Library for Amazon S3 Compatible Cloud Storage |
||||
* (C) 2017 Minio, Inc. |
||||
* |
||||
* 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. |
||||
*/ |
||||
|
||||
package minio |
||||
|
||||
import ( |
||||
"net/http" |
||||
"time" |
||||
) |
||||
|
||||
// This default transport is similar to http.DefaultTransport
|
||||
// but with additional DisableCompression:
|
||||
var defaultMinioTransport http.RoundTripper = &http.Transport{ |
||||
Proxy: http.ProxyFromEnvironment, |
||||
TLSHandshakeTimeout: 10 * time.Second, |
||||
// Set this value so that the underlying transport round-tripper
|
||||
// doesn't try to auto decode the body of objects with
|
||||
// content-encoding set to `gzip`.
|
||||
//
|
||||
// Refer:
|
||||
// https://golang.org/src/net/http/transport.go?h=roundTrip#L1843
|
||||
DisableCompression: true, |
||||
} |
||||
@ -0,0 +1,40 @@ |
||||
// +build go1.6,!go1.7,!go1.8
|
||||
|
||||
/* |
||||
* Minio Go Library for Amazon S3 Compatible Cloud Storage |
||||
* (C) 2017 Minio, Inc. |
||||
* |
||||
* 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. |
||||
*/ |
||||
|
||||
package minio |
||||
|
||||
import ( |
||||
"net/http" |
||||
"time" |
||||
) |
||||
|
||||
// This default transport is similar to http.DefaultTransport
|
||||
// but with additional DisableCompression:
|
||||
var defaultMinioTransport http.RoundTripper = &http.Transport{ |
||||
Proxy: http.ProxyFromEnvironment, |
||||
TLSHandshakeTimeout: 10 * time.Second, |
||||
ExpectContinueTimeout: 1 * time.Second, |
||||
// Set this value so that the underlying transport round-tripper
|
||||
// doesn't try to auto decode the body of objects with
|
||||
// content-encoding set to `gzip`.
|
||||
//
|
||||
// Refer:
|
||||
// https://golang.org/src/net/http/transport.go?h=roundTrip#L1843
|
||||
DisableCompression: true, |
||||
} |
||||
Loading…
Reference in new issue