2
2
mirror of https://github.com/octoleo/restic.git synced 2024-12-11 13:52:31 +00:00
restic/vendor/github.com/minio/minio-go/v6/api-put-object-streaming.go

418 lines
14 KiB
Go
Raw Normal View History

2017-07-23 12:24:45 +00:00
/*
2019-11-22 13:57:56 +00:00
* MinIO Go Library for Amazon S3 Compatible Cloud Storage
* Copyright 2017 MinIO, Inc.
2017-07-23 12:24:45 +00:00
*
* 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 (
2017-12-08 19:45:59 +00:00
"context"
2017-07-23 12:24:45 +00:00
"fmt"
"io"
"net/http"
"sort"
"strings"
2019-11-22 13:57:56 +00:00
"github.com/minio/minio-go/v6/pkg/s3utils"
2017-07-23 12:24:45 +00:00
)
// 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.
//
// - *minio.Object
// - Any reader which has a method 'ReadAt()'
//
2017-12-08 19:45:59 +00:00
func (c Client) putObjectMultipartStream(ctx context.Context, bucketName, objectName string,
reader io.Reader, size int64, opts PutObjectOptions) (n int64, err error) {
if !isObject(reader) && isReadAt(reader) {
// Verify if the reader implements ReadAt and it is not a *minio.Object then we will use parallel uploader.
n, err = c.putObjectMultipartStreamFromReadAt(ctx, bucketName, objectName, reader.(io.ReaderAt), size, opts)
2017-07-23 12:24:45 +00:00
} else {
2017-12-08 19:45:59 +00:00
n, err = c.putObjectMultipartStreamNoChecksum(ctx, bucketName, objectName, reader, size, opts)
2017-07-23 12:24:45 +00:00
}
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.
2017-12-08 19:45:59 +00:00
return c.putObjectNoChecksum(ctx, bucketName, objectName, reader, size, opts)
2017-07-23 12:24:45 +00:00
}
}
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.
}
2019-11-22 13:57:56 +00:00
// putObjectMultipartFromReadAt - Uploads files bigger than 128MiB.
2017-07-23 12:24:45 +00:00
// 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.
2017-12-08 19:45:59 +00:00
func (c Client) putObjectMultipartStreamFromReadAt(ctx context.Context, bucketName, objectName string,
reader io.ReaderAt, size int64, opts PutObjectOptions) (n int64, err error) {
2017-07-23 12:24:45 +00:00
// 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.
2019-11-22 13:57:56 +00:00
totalPartsCount, partSize, lastPartSize, err := optimalPartInfo(size, opts.PartSize)
2017-07-23 12:24:45 +00:00
if err != nil {
return 0, err
}
// Initiate a new multipart upload.
2017-12-08 19:45:59 +00:00
uploadID, err := c.newUploadID(ctx, bucketName, objectName, opts)
2017-07-23 12:24:45 +00:00
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 {
2017-12-08 19:45:59 +00:00
c.abortMultipartUpload(ctx, bucketName, objectName, uploadID)
2017-07-23 12:24:45 +00:00
}
}()
// 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.
2017-12-08 19:45:59 +00:00
for w := 1; w <= opts.getNumThreads(); w++ {
2017-09-13 12:09:48 +00:00
go func(partSize int64) {
2017-07-23 12:24:45 +00:00
// 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.
2017-12-08 19:45:59 +00:00
sectionReader := newHook(io.NewSectionReader(reader, readOffset, partSize), opts.Progress)
2017-07-23 12:24:45 +00:00
// Proceed to upload the part.
var objPart ObjectPart
2017-12-08 19:45:59 +00:00
objPart, err = c.uploadPart(ctx, bucketName, objectName, uploadID,
2017-07-23 12:24:45 +00:00
sectionReader, uploadReq.PartNum,
"", "", partSize, opts.ServerSideEncryption)
2017-07-23 12:24:45 +00:00
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,
}
}
2017-09-13 12:09:48 +00:00
}(partSize)
2017-07-23 12:24:45 +00:00
}
// 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))
2017-12-08 19:45:59 +00:00
_, err = c.completeMultipartUpload(ctx, bucketName, objectName, uploadID, complMultipartUpload)
2017-07-23 12:24:45 +00:00
if err != nil {
return totalUploadedSize, err
}
// Return final size.
return totalUploadedSize, nil
}
2017-12-08 19:45:59 +00:00
func (c Client) putObjectMultipartStreamNoChecksum(ctx context.Context, bucketName, objectName string,
reader io.Reader, size int64, opts PutObjectOptions) (n int64, err error) {
2017-07-23 12:24:45 +00:00
// 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.
2019-11-22 13:57:56 +00:00
totalPartsCount, partSize, lastPartSize, err := optimalPartInfo(size, opts.PartSize)
2017-07-23 12:24:45 +00:00
if err != nil {
return 0, err
}
// Initiates a new multipart request
2017-12-08 19:45:59 +00:00
uploadID, err := c.newUploadID(ctx, bucketName, objectName, opts)
2017-07-23 12:24:45 +00:00
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 {
2017-12-08 19:45:59 +00:00
c.abortMultipartUpload(ctx, bucketName, objectName, uploadID)
2017-07-23 12:24:45 +00:00
}
}()
// 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.
2017-12-08 19:45:59 +00:00
hookReader := newHook(reader, opts.Progress)
2017-07-23 12:24:45 +00:00
// Proceed to upload the part.
if partNumber == totalPartsCount {
partSize = lastPartSize
}
var objPart ObjectPart
2017-12-08 19:45:59 +00:00
objPart, err = c.uploadPart(ctx, bucketName, objectName, uploadID,
2017-07-23 12:24:45 +00:00
io.LimitReader(hookReader, partSize),
partNumber, "", "", partSize, opts.ServerSideEncryption)
2017-07-23 12:24:45 +00:00
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))
2017-12-08 19:45:59 +00:00
_, err = c.completeMultipartUpload(ctx, bucketName, objectName, uploadID, complMultipartUpload)
2017-07-23 12:24:45 +00:00
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.
2017-12-08 19:45:59 +00:00
func (c Client) putObjectNoChecksum(ctx context.Context, bucketName, objectName string, reader io.Reader, size int64, opts PutObjectOptions) (n int64, err error) {
2017-07-23 12:24:45 +00:00
// 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) {
2017-07-23 12:24:45 +00:00
return 0, ErrEntityTooSmall(size, bucketName, objectName)
}
if size > 0 {
if isReadAt(reader) && !isObject(reader) {
2017-12-08 19:45:59 +00:00
seeker, _ := reader.(io.Seeker)
offset, err := seeker.Seek(0, io.SeekCurrent)
if err != nil {
return 0, ErrInvalidArgument(err.Error())
}
reader = io.NewSectionReader(reader.(io.ReaderAt), offset, size)
2017-07-23 12:24:45 +00:00
}
}
// Update progress reader appropriately to the latest offset as we
// read from the source.
2017-12-08 19:45:59 +00:00
readSeeker := newHook(reader, opts.Progress)
2017-07-23 12:24:45 +00:00
// This function does not calculate sha256 and md5sum for payload.
// Execute put object.
2017-12-08 19:45:59 +00:00
st, err := c.putObjectDo(ctx, bucketName, objectName, readSeeker, "", "", size, opts)
2017-07-23 12:24:45 +00:00
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.
2017-12-08 19:45:59 +00:00
func (c Client) putObjectDo(ctx context.Context, bucketName, objectName string, reader io.Reader, md5Base64, sha256Hex string, size int64, opts PutObjectOptions) (ObjectInfo, error) {
2017-07-23 12:24:45 +00:00
// Input validation.
if err := s3utils.CheckValidBucketName(bucketName); err != nil {
return ObjectInfo{}, err
}
if err := s3utils.CheckValidObjectName(objectName); err != nil {
return ObjectInfo{}, err
}
// Set headers.
2017-12-08 19:45:59 +00:00
customHeader := opts.Header()
2017-07-23 12:24:45 +00:00
// Populate request metadata.
reqMetadata := requestMetadata{
2017-12-08 19:45:59 +00:00
bucketName: bucketName,
objectName: objectName,
customHeader: customHeader,
contentBody: reader,
contentLength: size,
contentMD5Base64: md5Base64,
contentSHA256Hex: sha256Hex,
2017-07-23 12:24:45 +00:00
}
// Execute PUT an objectName.
2017-12-08 19:45:59 +00:00
resp, err := c.executeMethod(ctx, "PUT", reqMetadata)
2017-07-23 12:24:45 +00:00
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
}