Update vendor dependencies

2016-11-19 10:05:20 +00:00
parent 8118623680
commit 00af021abb
247 changed files with 45514 additions and 11689 deletions
@@ -8,6 +8,7 @@ import (
 	"strings"
 	"sync"

+	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/aws/aws-sdk-go/aws/awsutil"
 	"github.com/aws/aws-sdk-go/aws/client"
@@ -34,7 +35,7 @@ type Downloader struct {
 	PartSize int64

 	// The number of goroutines to spin up in parallel when sending parts.
-	// If this is set to zero, the DefaultConcurrency value will be used.
+	// If this is set to zero, the DefaultDownloadConcurrency value will be used.
 	Concurrency int

 	// An S3 client to use when performing downloads.
@@ -49,13 +50,13 @@ type Downloader struct {
 //
 // Example:
 //     // The session the S3 Downloader will use
-//     sess := session.New()
+//     sess, err := session.NewSession()
 //
 //     // Create a downloader with the session and default options
 //     downloader := s3manager.NewDownloader(sess)
 //
 //     // Create a downloader with the session and custom options
-//     downloader := s3manager.NewDownloader(sess, func(d *s3manager.Uploader) {
+//     downloader := s3manager.NewDownloader(sess, func(d *s3manager.Downloader) {
 //          d.PartSize = 64 * 1024 * 1024 // 64MB per part
 //     })
 func NewDownloader(c client.ConfigProvider, options ...func(*Downloader)) *Downloader {
@@ -77,14 +78,17 @@ func NewDownloader(c client.ConfigProvider, options ...func(*Downloader)) *Downl
 // to make S3 API calls.
 //
 // Example:
+//     // The session the S3 Downloader will use
+//     sess, err := session.NewSession()
+//
 //     // The S3 client the S3 Downloader will use
-//     s3Svc := s3.new(session.New())
+//     s3Svc := s3.new(sess)
 //
 //     // Create a downloader with the s3 client and default options
 //     downloader := s3manager.NewDownloaderWithClient(s3Svc)
 //
 //     // Create a downloader with the s3 client and custom options
-//     downloader := s3manager.NewDownloaderWithClient(s3Svc, func(d *s3manager.Uploader) {
+//     downloader := s3manager.NewDownloaderWithClient(s3Svc, func(d *s3manager.Downloader) {
 //          d.PartSize = 64 * 1024 * 1024 // 64MB per part
 //     })
 func NewDownloaderWithClient(svc s3iface.S3API, options ...func(*Downloader)) *Downloader {
@@ -100,12 +104,16 @@ func NewDownloaderWithClient(svc s3iface.S3API, options ...func(*Downloader)) *D
 	return d
 }

+type maxRetrier interface {
+	MaxRetries() int
+}
+
 // Download downloads an object in S3 and writes the payload into w using
 // concurrent GET requests.
 //
 // Additional functional options can be provided to configure the individual
-// upload. These options are copies of the Uploader instance Upload is called from.
-// Modifying the options will not impact the original Uploader instance.
+// download. These options are copies of the Downloader instance Download is called from.
+// Modifying the options will not impact the original Downloader instance.
 //
 // It is safe to call this method concurrently across goroutines.
 //
@@ -118,6 +126,19 @@ func (d Downloader) Download(w io.WriterAt, input *s3.GetObjectInput, options ..
 		option(&impl.ctx)
 	}

+	if s, ok := d.S3.(maxRetrier); ok {
+		impl.partBodyMaxRetries = s.MaxRetries()
+	}
+
+	impl.totalBytes = -1
+	if impl.ctx.Concurrency == 0 {
+		impl.ctx.Concurrency = DefaultDownloadConcurrency
+	}
+
+	if impl.ctx.PartSize == 0 {
+		impl.ctx.PartSize = DefaultDownloadPartSize
+	}
+
 	return impl.download()
 }

@@ -135,26 +156,13 @@ type downloader struct {
 	totalBytes int64
 	written    int64
 	err        error
-}

-// init initializes the downloader with default options.
-func (d *downloader) init() {
-	d.totalBytes = -1
-
-	if d.ctx.Concurrency == 0 {
-		d.ctx.Concurrency = DefaultDownloadConcurrency
-	}
-
-	if d.ctx.PartSize == 0 {
-		d.ctx.PartSize = DefaultDownloadPartSize
-	}
+	partBodyMaxRetries int
 }

 // download performs the implementation of the object download across ranged
 // GETs.
 func (d *downloader) download() (n int64, err error) {
-	d.init()
-
 	// Spin off first worker to check additional header information
 	d.getChunk()

@@ -170,7 +178,7 @@ func (d *downloader) download() (n int64, err error) {
 		// Assign work
 		for d.getErr() == nil {
 			if d.pos >= total {
-				break // We're finished queueing chunks
+				break // We're finished queuing chunks
 			}

 			// Queue the next range of bytes to read.
@@ -211,49 +219,82 @@ func (d *downloader) downloadPart(ch chan dlchunk) {
 	defer d.wg.Done()
 	for {
 		chunk, ok := <-ch
-		if !ok {
+		if !ok || d.getErr() != nil {
+			break
+		}
+
+		if err := d.downloadChunk(chunk); err != nil {
+			d.setErr(err)
 			break
 		}
-		d.downloadChunk(chunk)
 	}
 }

 // getChunk grabs a chunk of data from the body.
 // Not thread safe. Should only used when grabbing data on a single thread.
 func (d *downloader) getChunk() {
-	chunk := dlchunk{w: d.w, start: d.pos, size: d.ctx.PartSize}
-	d.pos += d.ctx.PartSize
-	d.downloadChunk(chunk)
-}
-
-// downloadChunk downloads the chunk froom s3
-func (d *downloader) downloadChunk(chunk dlchunk) {
 	if d.getErr() != nil {
 		return
 	}
-	// Get the next byte range of data
+
+	chunk := dlchunk{w: d.w, start: d.pos, size: d.ctx.PartSize}
+	d.pos += d.ctx.PartSize
+
+	if err := d.downloadChunk(chunk); err != nil {
+		d.setErr(err)
+	}
+}
+
+// downloadChunk downloads the chunk froom s3
+func (d *downloader) downloadChunk(chunk dlchunk) error {
 	in := &s3.GetObjectInput{}
 	awsutil.Copy(in, d.in)
-	rng := fmt.Sprintf("bytes=%d-%d",
-		chunk.start, chunk.start+chunk.size-1)
+
+	// Get the next byte range of data
+	rng := fmt.Sprintf("bytes=%d-%d", chunk.start, chunk.start+chunk.size-1)
 	in.Range = &rng

-	req, resp := d.ctx.S3.GetObjectRequest(in)
-	req.Handlers.Build.PushBack(request.MakeAddToUserAgentFreeFormHandler("S3Manager"))
-	err := req.Send()
+	var n int64
+	var err error
+	for retry := 0; retry <= d.partBodyMaxRetries; retry++ {
+		req, resp := d.ctx.S3.GetObjectRequest(in)
+		req.Handlers.Build.PushBack(request.MakeAddToUserAgentFreeFormHandler("S3Manager"))

-	if err != nil {
-		d.setErr(err)
-	} else {
+		err = req.Send()
+		if err != nil {
+			return err
+		}
 		d.setTotalBytes(resp) // Set total if not yet set.

-		n, err := io.Copy(&chunk, resp.Body)
+		n, err = io.Copy(&chunk, resp.Body)
 		resp.Body.Close()
-
-		if err != nil {
-			d.setErr(err)
+		if err == nil {
+			break
 		}
-		d.incrWritten(n)
+
+		chunk.cur = 0
+		logMessage(d.ctx.S3, aws.LogDebugWithRequestRetries,
+			fmt.Sprintf("DEBUG: object part body download interrupted %s, err, %v, retrying attempt %d",
+				aws.StringValue(in.Key), err, retry))
+	}
+
+	d.incrWritten(n)
+
+	return err
+}
+
+func logMessage(svc s3iface.S3API, level aws.LogLevelType, msg string) {
+	s, ok := svc.(*s3.S3)
+	if !ok {
+		return
+	}
+
+	if s.Config.Logger == nil {
+		return
+	}
+
+	if s.Config.LogLevel.Matches(level) {
+		s.Config.Logger.Log(msg)
 	}
 }

@@ -8,6 +8,7 @@ import (
 	"sync"
 	"time"

+	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/aws/aws-sdk-go/aws/awsutil"
 	"github.com/aws/aws-sdk-go/aws/client"
@@ -42,7 +43,7 @@ const DefaultUploadConcurrency = 5
 //     u := s3manager.NewUploader(opts)
 //     output, err := u.upload(input)
 //     if err != nil {
-//         if multierr, ok := err.(MultiUploadFailure); ok {
+//         if multierr, ok := err.(s3manager.MultiUploadFailure); ok {
 //             // Process error and its associated uploadID
 //             fmt.Println("Error:", multierr.Code(), multierr.Message(), multierr.UploadID())
 //         } else {
@@ -165,7 +166,7 @@ type UploadInput struct {
 	// requests for an object protected by AWS KMS will fail if not made via SSL
 	// or using SigV4. Documentation on configuring any of the officially supported
 	// AWS SDKs and CLI can be found at http://docs.aws.amazon.com/AmazonS3/latest/dev/UsingAWSSDK.html#specify-signature-version
-	SSEKMSKeyID *string `location:"header" locationName:"x-amz-server-side-encryption-aws-kms-key-id" type:"string"`
+	SSEKMSKeyId *string `location:"header" locationName:"x-amz-server-side-encryption-aws-kms-key-id" type:"string"`

 	// The Server-side encryption algorithm used when storing this object in S3
 	// (e.g., AES256, aws:kms).
@@ -208,7 +209,7 @@ type Uploader struct {
 	PartSize int64

 	// The number of goroutines to spin up in parallel when sending parts.
-	// If this is set to zero, the DefaultConcurrency value will be used.
+	// If this is set to zero, the DefaultUploadConcurrency value will be used.
 	Concurrency int

 	// Setting this value to true will cause the SDK to avoid calling
@@ -237,7 +238,7 @@ type Uploader struct {
 //
 // Example:
 //     // The session the S3 Uploader will use
-//     sess := session.New()
+//     sess, err := session.NewSession()
 //
 //     // Create an uploader with the session and default options
 //     uploader := s3manager.NewUploader(sess)
@@ -267,8 +268,11 @@ func NewUploader(c client.ConfigProvider, options ...func(*Uploader)) *Uploader
 // a S3 service client to make S3 API calls.
 //
 // Example:
+//     // The session the S3 Uploader will use
+//     sess, err := session.NewSession()
+//
 //     // S3 service client the Upload manager will use.
-//     s3Svc := s3.New(session.New())
+//     s3Svc := s3.New(sess)
 //
 //     // Create an uploader with S3 client and default options
 //     uploader := s3manager.NewUploaderWithClient(s3Svc)
@@ -317,7 +321,7 @@ func NewUploaderWithClient(svc s3iface.S3API, options ...func(*Uploader)) *Uploa
 //     // Perform upload with options different than the those in the Uploader.
 //     result, err := uploader.Upload(upParams, func(u *s3manager.Uploader) {
 //          u.PartSize = 10 * 1024 * 1024 // 10MB part size
-//          u.LeavePartsOnError = true    // Dont delete the parts if the upload fails.
+//          u.LeavePartsOnError = true    // Don't delete the parts if the upload fails.
 //     })
 func (u Uploader) Upload(input *UploadInput, options ...func(*Uploader)) (*UploadOutput, error) {
 	i := uploader{in: input, ctx: u}
@@ -350,15 +354,15 @@ func (u *uploader) upload() (*UploadOutput, error) {
 	}

 	// Do one read to determine if we have more than one part
-	buf, err := u.nextReader()
-	if err == io.EOF || err == io.ErrUnexpectedEOF { // single part
-		return u.singlePart(buf)
+	reader, _, err := u.nextReader()
+	if err == io.EOF { // single part
+		return u.singlePart(reader)
 	} else if err != nil {
 		return nil, awserr.New("ReadRequestBody", "read upload data failed", err)
 	}

 	mu := multiuploader{uploader: u}
-	return mu.upload(buf)
+	return mu.upload(reader)
 }

 // init will initialize all default options.
@@ -404,38 +408,49 @@ func (u *uploader) initSize() {
 // This operation increases the shared u.readerPos counter, but note that it
 // does not need to be wrapped in a mutex because nextReader is only called
 // from the main thread.
-func (u *uploader) nextReader() (io.ReadSeeker, error) {
+func (u *uploader) nextReader() (io.ReadSeeker, int, error) {
+	type readerAtSeeker interface {
+		io.ReaderAt
+		io.ReadSeeker
+	}
 	switch r := u.in.Body.(type) {
-	case io.ReaderAt:
+	case readerAtSeeker:
 		var err error

 		n := u.ctx.PartSize
 		if u.totalSize >= 0 {
 			bytesLeft := u.totalSize - u.readerPos

-			if bytesLeft == 0 {
+			if bytesLeft <= u.ctx.PartSize {
 				err = io.EOF
 				n = bytesLeft
-			} else if bytesLeft <= u.ctx.PartSize {
-				err = io.ErrUnexpectedEOF
-				n = bytesLeft
 			}
 		}

-		buf := io.NewSectionReader(r, u.readerPos, n)
+		reader := io.NewSectionReader(r, u.readerPos, n)
 		u.readerPos += n

-		return buf, err
+		return reader, int(n), err

 	default:
-		packet := make([]byte, u.ctx.PartSize)
-		n, err := io.ReadFull(u.in.Body, packet)
+		part := make([]byte, u.ctx.PartSize)
+		n, err := readFillBuf(r, part)
 		u.readerPos += int64(n)

-		return bytes.NewReader(packet[0:n]), err
+		return bytes.NewReader(part[0:n]), n, err
 	}
 }

+func readFillBuf(r io.Reader, b []byte) (offset int, err error) {
+	for offset < len(b) && err == nil {
+		var n int
+		n, err = r.Read(b[offset:])
+		offset += n
+	}
+
+	return offset, err
+}
+
 // singlePart contains upload logic for uploading a single chunk via
 // a regular PutObject request. Multipart requests require at least two
 // parts, or at least 5MB of data.
@@ -507,7 +522,9 @@ func (u *multiuploader) upload(firstBuf io.ReadSeeker) (*UploadOutput, error) {
 	ch <- chunk{buf: firstBuf, num: num}

 	// Read and queue the rest of the parts
-	for u.geterr() == nil {
+	var err error
+	for u.geterr() == nil && err == nil {
+		num++
 		// This upload exceeded maximum number of supported parts, error now.
 		if num > int64(u.ctx.MaxUploadParts) || num > int64(MaxUploadParts) {
 			var msg string
@@ -521,22 +538,27 @@ func (u *multiuploader) upload(firstBuf io.ReadSeeker) (*UploadOutput, error) {
 			u.seterr(awserr.New("TotalPartsExceeded", msg, nil))
 			break
 		}
-		num++

-		buf, err := u.nextReader()
-		if err == io.EOF {
-			break
-		}
+		var reader io.ReadSeeker
+		var nextChunkLen int
+		reader, nextChunkLen, err = u.nextReader()

-		ch <- chunk{buf: buf, num: num}
-
-		if err != nil && err != io.ErrUnexpectedEOF {
+		if err != nil && err != io.EOF {
 			u.seterr(awserr.New(
 				"ReadRequestBody",
 				"read multipart upload data failed",
 				err))
 			break
 		}
+
+		if nextChunkLen == 0 {
+			// No need to upload empty part, if file was empty to start
+			// with empty single part would of been created and never
+			// started multipart upload.
+			break
+		}
+
+		ch <- chunk{buf: reader, num: num}
 	}

 	// Close the channel, wait for workers, and complete upload
@@ -554,7 +576,7 @@ func (u *multiuploader) upload(firstBuf io.ReadSeeker) (*UploadOutput, error) {
 		}
 	}
 	return &UploadOutput{
-		Location:  *complete.Location,
+		Location:  aws.StringValue(complete.Location),
 		VersionID: complete.VersionId,
 		UploadID:  u.uploadID,
 	}, nil