Merge pull request #174 from skriss/obj-store-reader

Switch PutObject to take an io.Reader rather than an io.ReadSeeker

Gopkg.lock

@@ -33,7 +33,7 @@
 
 [[projects]]
   name = "github.com/aws/aws-sdk-go"
-  packages = ["aws","aws/awserr","aws/awsutil","aws/client","aws/client/metadata","aws/corehandlers","aws/credentials","aws/credentials/ec2rolecreds","aws/credentials/endpointcreds","aws/credentials/stscreds","aws/defaults","aws/ec2metadata","aws/endpoints","aws/request","aws/session","aws/signer/v4","internal/shareddefaults","private/protocol","private/protocol/ec2query","private/protocol/query","private/protocol/query/queryutil","private/protocol/rest","private/protocol/restxml","private/protocol/xml/xmlutil","service/ec2","service/s3","service/sts"]
+  packages = ["aws","aws/awserr","aws/awsutil","aws/client","aws/client/metadata","aws/corehandlers","aws/credentials","aws/credentials/ec2rolecreds","aws/credentials/endpointcreds","aws/credentials/stscreds","aws/defaults","aws/ec2metadata","aws/endpoints","aws/request","aws/session","aws/signer/v4","internal/shareddefaults","private/protocol","private/protocol/ec2query","private/protocol/query","private/protocol/query/queryutil","private/protocol/rest","private/protocol/restxml","private/protocol/xml/xmlutil","service/ec2","service/s3","service/s3/s3iface","service/s3/s3manager","service/sts"]
   revision = "1850f427c33c2558a2118dc55c1cf95a633d7432"
   version = "v1.10.27"
 

pkg/cloudprovider/aws/object_storage_adapter.go

@@ -23,6 +23,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/endpoints"
 	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/aws/aws-sdk-go/service/s3/s3manager"
 	"github.com/pkg/errors"
 
 	"github.com/heptio/ark/pkg/cloudprovider"
@@ -31,8 +32,9 @@ import (
 var _ cloudprovider.ObjectStorageAdapter = &objectStorageAdapter{}
 
 type objectStorageAdapter struct {
-	s3       *s3.S3
-	kmsKeyID string
+	s3         *s3.S3
+	s3Uploader *s3manager.Uploader
+	kmsKeyID   string
 }
 
 func NewObjectStorageAdapter(region, s3URL, kmsKeyID string, s3ForcePathStyle bool) (cloudprovider.ObjectStorageAdapter, error) {
@@ -64,13 +66,14 @@ func NewObjectStorageAdapter(region, s3URL, kmsKeyID string, s3ForcePathStyle bo
 	}
 
 	return &objectStorageAdapter{
-		s3:       s3.New(sess),
-		kmsKeyID: kmsKeyID,
+		s3:         s3.New(sess),
+		s3Uploader: s3manager.NewUploader(sess),
+		kmsKeyID:   kmsKeyID,
 	}, nil
 }
 
-func (op *objectStorageAdapter) PutObject(bucket string, key string, body io.ReadSeeker) error {
-	req := &s3.PutObjectInput{
+func (op *objectStorageAdapter) PutObject(bucket string, key string, body io.Reader) error {
+	req := &s3manager.UploadInput{
 		Bucket: &bucket,
 		Key:    &key,
 		Body:   body,
@@ -82,7 +85,7 @@ func (op *objectStorageAdapter) PutObject(bucket string, key string, body io.Rea
 		req.SSEKMSKeyId = &op.kmsKeyID
 	}
 
-	_, err := op.s3.PutObject(req)
+	_, err := op.s3Uploader.Upload(req)
 
 	return errors.Wrapf(err, "error putting object %s", key)
 }

pkg/cloudprovider/azure/object_storage_adapter.go

@@ -50,7 +50,7 @@ func NewObjectStorageAdapter() (cloudprovider.ObjectStorageAdapter, error) {
 	}, nil
 }
 
-func (op *objectStorageAdapter) PutObject(bucket string, key string, body io.ReadSeeker) error {
+func (op *objectStorageAdapter) PutObject(bucket string, key string, body io.Reader) error {
 	container, err := getContainerReference(op.blobClient, bucket)
 	if err != nil {
 		return err
@@ -61,19 +61,6 @@ func (op *objectStorageAdapter) PutObject(bucket string, key string, body io.Rea
 		return err
 	}
 
-	// TODO having to seek to end/back to beginning to get
-	// length here is ugly. refactor to make this better.
-	len, err := body.Seek(0, io.SeekEnd)
-	if err != nil {
-		return errors.WithStack(err)
-	}
-
-	blob.Properties.ContentLength = len
-
-	if _, err := body.Seek(0, 0); err != nil {
-		return errors.WithStack(err)
-	}
-
 	return errors.WithStack(blob.CreateBlockBlobFromReader(body, nil))
 }
 

pkg/cloudprovider/backup_service.go

@@ -40,7 +40,7 @@ type BackupService interface {
 	// UploadBackup uploads the specified Ark backup of a set of Kubernetes API objects, whose manifests are
 	// stored in the specified file, into object storage in an Ark bucket, tagged with Ark metadata. Returns
 	// an error if a problem is encountered accessing the file or performing the upload via the cloud API.
-	UploadBackup(bucket, name string, metadata, backup, log io.ReadSeeker) error
+	UploadBackup(bucket, name string, metadata, backup, log io.Reader) error
 
 	// DownloadBackup downloads an Ark backup with the specified object key from object storage via the cloud API.
 	// It returns the snapshot metadata and data (separately), or an error if a problem is encountered
@@ -58,7 +58,7 @@ type BackupService interface {
 	CreateSignedURL(target api.DownloadTarget, bucket string, ttl time.Duration) (string, error)
 
 	// UploadRestoreLog uploads the restore's log file to object storage.
-	UploadRestoreLog(bucket, backup, restore string, log io.ReadSeeker) error
+	UploadRestoreLog(bucket, backup, restore string, log io.Reader) error
 }
 
 // BackupGetter knows how to list backups in object storage.
@@ -108,7 +108,7 @@ func NewBackupService(objectStorage ObjectStorageAdapter, logger *logrus.Logger)
 	}
 }
 
-func (br *backupService) UploadBackup(bucket, backupName string, metadata, backup, log io.ReadSeeker) error {
+func (br *backupService) UploadBackup(bucket, backupName string, metadata, backup, log io.Reader) error {
 	// upload metadata file
 	metadataKey := getMetadataKey(backupName)
 	if err := br.objectStorage.PutObject(bucket, metadataKey, metadata); err != nil {
@@ -231,7 +231,7 @@ func (br *backupService) CreateSignedURL(target api.DownloadTarget, bucket strin
 	}
 }
 
-func (br *backupService) UploadRestoreLog(bucket, backup, restore string, log io.ReadSeeker) error {
+func (br *backupService) UploadRestoreLog(bucket, backup, restore string, log io.Reader) error {
 	key := getRestoreLogKey(backup, restore)
 	return br.objectStorage.PutObject(bucket, key, log)
 }

pkg/cloudprovider/gcp/object_storage_adapter.go

@@ -57,7 +57,7 @@ func NewObjectStorageAdapter(googleAccessID string, privateKey []byte) (cloudpro
 	}, nil
 }
 
-func (op *objectStorageAdapter) PutObject(bucket string, key string, body io.ReadSeeker) error {
+func (op *objectStorageAdapter) PutObject(bucket string, key string, body io.Reader) error {
 	obj := &storage.Object{
 		Name: key,
 	}

pkg/cloudprovider/storage_interfaces.go

@@ -26,7 +26,7 @@ import (
 type ObjectStorageAdapter interface {
 	// PutObject creates a new object using the data in body within the specified
 	// object storage bucket with the given key.
-	PutObject(bucket string, key string, body io.ReadSeeker) error
+	PutObject(bucket string, key string, body io.Reader) error
 
 	// GetObject retrieves the object with the given key from the specified
 	// bucket in object storage.

pkg/controller/backup_controller.go

@@ -338,5 +338,5 @@ func (controller *backupController) runBackup(backup *api.Backup, bucket string)
 		return errors.Wrap(err, "error resetting Backup log file offset")
 	}
 
-	return controller.backupService.UploadBackup(bucket, backup.Name, bytes.NewReader(buf.Bytes()), backupFile, logFile)
+	return controller.backupService.UploadBackup(bucket, backup.Name, buf, backupFile, logFile)
 }

pkg/util/test/backup_service.go

@@ -132,11 +132,11 @@ func (_m *BackupService) GetBackup(bucket string, name string) (*v1.Backup, erro
 }
 
 // UploadBackup provides a mock function with given fields: bucket, name, metadata, backup, log
-func (_m *BackupService) UploadBackup(bucket string, name string, metadata io.ReadSeeker, backup io.ReadSeeker, log io.ReadSeeker) error {
+func (_m *BackupService) UploadBackup(bucket string, name string, metadata, backup, log io.Reader) error {
 	ret := _m.Called(bucket, name, metadata, backup, log)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(string, string, io.ReadSeeker, io.ReadSeeker, io.ReadSeeker) error); ok {
+	if rf, ok := ret.Get(0).(func(string, string, io.Reader, io.Reader, io.Reader) error); ok {
 		r0 = rf(bucket, name, metadata, backup, log)
 	} else {
 		r0 = ret.Error(0)
@@ -146,11 +146,11 @@ func (_m *BackupService) UploadBackup(bucket string, name string, metadata io.Re
 }
 
 // UploadRestoreLog provides a mock function with given fields: bucket, backup, restore, log
-func (_m *BackupService) UploadRestoreLog(bucket string, backup string, restore string, log io.ReadSeeker) error {
+func (_m *BackupService) UploadRestoreLog(bucket string, backup string, restore string, log io.Reader) error {
 	ret := _m.Called(bucket, backup, restore, log)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(string, string, string, io.ReadSeeker) error); ok {
+	if rf, ok := ret.Get(0).(func(string, string, string, io.Reader) error); ok {
 		r0 = rf(bucket, backup, restore, log)
 	} else {
 		r0 = ret.Error(0)

pkg/util/test/fake_backup_service.go

@@ -41,7 +41,7 @@ func (f *FakeBackupService) GetAllBackups(bucket string) ([]*v1.Backup, error) {
 	return backups, args.Error(1)
 }
 
-func (f *FakeBackupService) UploadBackup(bucket, name string, metadata, backup io.ReadSeeker) error {
+func (f *FakeBackupService) UploadBackup(bucket, name string, metadata, backup io.Reader) error {
 	args := f.Called(bucket, name, metadata, backup)
 	return args.Error(0)
 }

pkg/util/test/object_storage_adapter.go

@@ -131,11 +131,11 @@ func (_m *ObjectStorageAdapter) ListObjects(bucket string, prefix string) ([]str
 }
 
 // PutObject provides a mock function with given fields: bucket, key, body
-func (_m *ObjectStorageAdapter) PutObject(bucket string, key string, body io.ReadSeeker) error {
+func (_m *ObjectStorageAdapter) PutObject(bucket string, key string, body io.Reader) error {
 	ret := _m.Called(bucket, key, body)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(string, string, io.ReadSeeker) error); ok {
+	if rf, ok := ret.Get(0).(func(string, string, io.Reader) error); ok {
 		r0 = rf(bucket, key, body)
 	} else {
 		r0 = ret.Error(0)

vendor/github.com/aws/aws-sdk-go/service/s3/s3iface/interface.go

@@ -0,0 +1,387 @@
+// Code generated by private/model/cli/gen-api/main.go. DO NOT EDIT.
+
+// Package s3iface provides an interface to enable mocking the Amazon Simple Storage Service service client
+// for testing your code.
+//
+// It is important to note that this interface will have breaking changes
+// when the service model is updated and adds new API operations, paginators,
+// and waiters.
+package s3iface
+
+import (
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/request"
+	"github.com/aws/aws-sdk-go/service/s3"
+)
+
+// S3API provides an interface to enable mocking the
+// s3.S3 service client's API operation,
+// paginators, and waiters. This make unit testing your code that calls out
+// to the SDK's service client's calls easier.
+//
+// The best way to use this interface is so the SDK's service client's calls
+// can be stubbed out for unit testing your code with the SDK without needing
+// to inject custom request handlers into the SDK's request pipeline.
+//
+//    // myFunc uses an SDK service client to make a request to
+//    // Amazon Simple Storage Service.
+//    func myFunc(svc s3iface.S3API) bool {
+//        // Make svc.AbortMultipartUpload request
+//    }
+//
+//    func main() {
+//        sess := session.New()
+//        svc := s3.New(sess)
+//
+//        myFunc(svc)
+//    }
+//
+// In your _test.go file:
+//
+//    // Define a mock struct to be used in your unit tests of myFunc.
+//    type mockS3Client struct {
+//        s3iface.S3API
+//    }
+//    func (m *mockS3Client) AbortMultipartUpload(input *s3.AbortMultipartUploadInput) (*s3.AbortMultipartUploadOutput, error) {
+//        // mock response/functionality
+//    }
+//
+//    func TestMyFunc(t *testing.T) {
+//        // Setup Test
+//        mockSvc := &mockS3Client{}
+//
+//        myfunc(mockSvc)
+//
+//        // Verify myFunc's functionality
+//    }
+//
+// It is important to note that this interface will have breaking changes
+// when the service model is updated and adds new API operations, paginators,
+// and waiters. Its suggested to use the pattern above for testing, or using
+// tooling to generate mocks to satisfy the interfaces.
+type S3API interface {
+	AbortMultipartUpload(*s3.AbortMultipartUploadInput) (*s3.AbortMultipartUploadOutput, error)
+	AbortMultipartUploadWithContext(aws.Context, *s3.AbortMultipartUploadInput, ...request.Option) (*s3.AbortMultipartUploadOutput, error)
+	AbortMultipartUploadRequest(*s3.AbortMultipartUploadInput) (*request.Request, *s3.AbortMultipartUploadOutput)
+
+	CompleteMultipartUpload(*s3.CompleteMultipartUploadInput) (*s3.CompleteMultipartUploadOutput, error)
+	CompleteMultipartUploadWithContext(aws.Context, *s3.CompleteMultipartUploadInput, ...request.Option) (*s3.CompleteMultipartUploadOutput, error)
+	CompleteMultipartUploadRequest(*s3.CompleteMultipartUploadInput) (*request.Request, *s3.CompleteMultipartUploadOutput)
+
+	CopyObject(*s3.CopyObjectInput) (*s3.CopyObjectOutput, error)
+	CopyObjectWithContext(aws.Context, *s3.CopyObjectInput, ...request.Option) (*s3.CopyObjectOutput, error)
+	CopyObjectRequest(*s3.CopyObjectInput) (*request.Request, *s3.CopyObjectOutput)
+
+	CreateBucket(*s3.CreateBucketInput) (*s3.CreateBucketOutput, error)
+	CreateBucketWithContext(aws.Context, *s3.CreateBucketInput, ...request.Option) (*s3.CreateBucketOutput, error)
+	CreateBucketRequest(*s3.CreateBucketInput) (*request.Request, *s3.CreateBucketOutput)
+
+	CreateMultipartUpload(*s3.CreateMultipartUploadInput) (*s3.CreateMultipartUploadOutput, error)
+	CreateMultipartUploadWithContext(aws.Context, *s3.CreateMultipartUploadInput, ...request.Option) (*s3.CreateMultipartUploadOutput, error)
+	CreateMultipartUploadRequest(*s3.CreateMultipartUploadInput) (*request.Request, *s3.CreateMultipartUploadOutput)
+
+	DeleteBucket(*s3.DeleteBucketInput) (*s3.DeleteBucketOutput, error)
+	DeleteBucketWithContext(aws.Context, *s3.DeleteBucketInput, ...request.Option) (*s3.DeleteBucketOutput, error)
+	DeleteBucketRequest(*s3.DeleteBucketInput) (*request.Request, *s3.DeleteBucketOutput)
+
+	DeleteBucketAnalyticsConfiguration(*s3.DeleteBucketAnalyticsConfigurationInput) (*s3.DeleteBucketAnalyticsConfigurationOutput, error)
+	DeleteBucketAnalyticsConfigurationWithContext(aws.Context, *s3.DeleteBucketAnalyticsConfigurationInput, ...request.Option) (*s3.DeleteBucketAnalyticsConfigurationOutput, error)
+	DeleteBucketAnalyticsConfigurationRequest(*s3.DeleteBucketAnalyticsConfigurationInput) (*request.Request, *s3.DeleteBucketAnalyticsConfigurationOutput)
+
+	DeleteBucketCors(*s3.DeleteBucketCorsInput) (*s3.DeleteBucketCorsOutput, error)
+	DeleteBucketCorsWithContext(aws.Context, *s3.DeleteBucketCorsInput, ...request.Option) (*s3.DeleteBucketCorsOutput, error)
+	DeleteBucketCorsRequest(*s3.DeleteBucketCorsInput) (*request.Request, *s3.DeleteBucketCorsOutput)
+
+	DeleteBucketInventoryConfiguration(*s3.DeleteBucketInventoryConfigurationInput) (*s3.DeleteBucketInventoryConfigurationOutput, error)
+	DeleteBucketInventoryConfigurationWithContext(aws.Context, *s3.DeleteBucketInventoryConfigurationInput, ...request.Option) (*s3.DeleteBucketInventoryConfigurationOutput, error)
+	DeleteBucketInventoryConfigurationRequest(*s3.DeleteBucketInventoryConfigurationInput) (*request.Request, *s3.DeleteBucketInventoryConfigurationOutput)
+
+	DeleteBucketLifecycle(*s3.DeleteBucketLifecycleInput) (*s3.DeleteBucketLifecycleOutput, error)
+	DeleteBucketLifecycleWithContext(aws.Context, *s3.DeleteBucketLifecycleInput, ...request.Option) (*s3.DeleteBucketLifecycleOutput, error)
+	DeleteBucketLifecycleRequest(*s3.DeleteBucketLifecycleInput) (*request.Request, *s3.DeleteBucketLifecycleOutput)
+
+	DeleteBucketMetricsConfiguration(*s3.DeleteBucketMetricsConfigurationInput) (*s3.DeleteBucketMetricsConfigurationOutput, error)
+	DeleteBucketMetricsConfigurationWithContext(aws.Context, *s3.DeleteBucketMetricsConfigurationInput, ...request.Option) (*s3.DeleteBucketMetricsConfigurationOutput, error)
+	DeleteBucketMetricsConfigurationRequest(*s3.DeleteBucketMetricsConfigurationInput) (*request.Request, *s3.DeleteBucketMetricsConfigurationOutput)
+
+	DeleteBucketPolicy(*s3.DeleteBucketPolicyInput) (*s3.DeleteBucketPolicyOutput, error)
+	DeleteBucketPolicyWithContext(aws.Context, *s3.DeleteBucketPolicyInput, ...request.Option) (*s3.DeleteBucketPolicyOutput, error)
+	DeleteBucketPolicyRequest(*s3.DeleteBucketPolicyInput) (*request.Request, *s3.DeleteBucketPolicyOutput)
+
+	DeleteBucketReplication(*s3.DeleteBucketReplicationInput) (*s3.DeleteBucketReplicationOutput, error)
+	DeleteBucketReplicationWithContext(aws.Context, *s3.DeleteBucketReplicationInput, ...request.Option) (*s3.DeleteBucketReplicationOutput, error)
+	DeleteBucketReplicationRequest(*s3.DeleteBucketReplicationInput) (*request.Request, *s3.DeleteBucketReplicationOutput)
+
+	DeleteBucketTagging(*s3.DeleteBucketTaggingInput) (*s3.DeleteBucketTaggingOutput, error)
+	DeleteBucketTaggingWithContext(aws.Context, *s3.DeleteBucketTaggingInput, ...request.Option) (*s3.DeleteBucketTaggingOutput, error)
+	DeleteBucketTaggingRequest(*s3.DeleteBucketTaggingInput) (*request.Request, *s3.DeleteBucketTaggingOutput)
+
+	DeleteBucketWebsite(*s3.DeleteBucketWebsiteInput) (*s3.DeleteBucketWebsiteOutput, error)
+	DeleteBucketWebsiteWithContext(aws.Context, *s3.DeleteBucketWebsiteInput, ...request.Option) (*s3.DeleteBucketWebsiteOutput, error)
+	DeleteBucketWebsiteRequest(*s3.DeleteBucketWebsiteInput) (*request.Request, *s3.DeleteBucketWebsiteOutput)
+
+	DeleteObject(*s3.DeleteObjectInput) (*s3.DeleteObjectOutput, error)
+	DeleteObjectWithContext(aws.Context, *s3.DeleteObjectInput, ...request.Option) (*s3.DeleteObjectOutput, error)
+	DeleteObjectRequest(*s3.DeleteObjectInput) (*request.Request, *s3.DeleteObjectOutput)
+
+	DeleteObjectTagging(*s3.DeleteObjectTaggingInput) (*s3.DeleteObjectTaggingOutput, error)
+	DeleteObjectTaggingWithContext(aws.Context, *s3.DeleteObjectTaggingInput, ...request.Option) (*s3.DeleteObjectTaggingOutput, error)
+	DeleteObjectTaggingRequest(*s3.DeleteObjectTaggingInput) (*request.Request, *s3.DeleteObjectTaggingOutput)
+
+	DeleteObjects(*s3.DeleteObjectsInput) (*s3.DeleteObjectsOutput, error)
+	DeleteObjectsWithContext(aws.Context, *s3.DeleteObjectsInput, ...request.Option) (*s3.DeleteObjectsOutput, error)
+	DeleteObjectsRequest(*s3.DeleteObjectsInput) (*request.Request, *s3.DeleteObjectsOutput)
+
+	GetBucketAccelerateConfiguration(*s3.GetBucketAccelerateConfigurationInput) (*s3.GetBucketAccelerateConfigurationOutput, error)
+	GetBucketAccelerateConfigurationWithContext(aws.Context, *s3.GetBucketAccelerateConfigurationInput, ...request.Option) (*s3.GetBucketAccelerateConfigurationOutput, error)
+	GetBucketAccelerateConfigurationRequest(*s3.GetBucketAccelerateConfigurationInput) (*request.Request, *s3.GetBucketAccelerateConfigurationOutput)
+
+	GetBucketAcl(*s3.GetBucketAclInput) (*s3.GetBucketAclOutput, error)
+	GetBucketAclWithContext(aws.Context, *s3.GetBucketAclInput, ...request.Option) (*s3.GetBucketAclOutput, error)
+	GetBucketAclRequest(*s3.GetBucketAclInput) (*request.Request, *s3.GetBucketAclOutput)
+
+	GetBucketAnalyticsConfiguration(*s3.GetBucketAnalyticsConfigurationInput) (*s3.GetBucketAnalyticsConfigurationOutput, error)
+	GetBucketAnalyticsConfigurationWithContext(aws.Context, *s3.GetBucketAnalyticsConfigurationInput, ...request.Option) (*s3.GetBucketAnalyticsConfigurationOutput, error)
+	GetBucketAnalyticsConfigurationRequest(*s3.GetBucketAnalyticsConfigurationInput) (*request.Request, *s3.GetBucketAnalyticsConfigurationOutput)
+
+	GetBucketCors(*s3.GetBucketCorsInput) (*s3.GetBucketCorsOutput, error)
+	GetBucketCorsWithContext(aws.Context, *s3.GetBucketCorsInput, ...request.Option) (*s3.GetBucketCorsOutput, error)
+	GetBucketCorsRequest(*s3.GetBucketCorsInput) (*request.Request, *s3.GetBucketCorsOutput)
+
+	GetBucketInventoryConfiguration(*s3.GetBucketInventoryConfigurationInput) (*s3.GetBucketInventoryConfigurationOutput, error)
+	GetBucketInventoryConfigurationWithContext(aws.Context, *s3.GetBucketInventoryConfigurationInput, ...request.Option) (*s3.GetBucketInventoryConfigurationOutput, error)
+	GetBucketInventoryConfigurationRequest(*s3.GetBucketInventoryConfigurationInput) (*request.Request, *s3.GetBucketInventoryConfigurationOutput)
+
+	GetBucketLifecycle(*s3.GetBucketLifecycleInput) (*s3.GetBucketLifecycleOutput, error)
+	GetBucketLifecycleWithContext(aws.Context, *s3.GetBucketLifecycleInput, ...request.Option) (*s3.GetBucketLifecycleOutput, error)
+	GetBucketLifecycleRequest(*s3.GetBucketLifecycleInput) (*request.Request, *s3.GetBucketLifecycleOutput)
+
+	GetBucketLifecycleConfiguration(*s3.GetBucketLifecycleConfigurationInput) (*s3.GetBucketLifecycleConfigurationOutput, error)
+	GetBucketLifecycleConfigurationWithContext(aws.Context, *s3.GetBucketLifecycleConfigurationInput, ...request.Option) (*s3.GetBucketLifecycleConfigurationOutput, error)
+	GetBucketLifecycleConfigurationRequest(*s3.GetBucketLifecycleConfigurationInput) (*request.Request, *s3.GetBucketLifecycleConfigurationOutput)
+
+	GetBucketLocation(*s3.GetBucketLocationInput) (*s3.GetBucketLocationOutput, error)
+	GetBucketLocationWithContext(aws.Context, *s3.GetBucketLocationInput, ...request.Option) (*s3.GetBucketLocationOutput, error)
+	GetBucketLocationRequest(*s3.GetBucketLocationInput) (*request.Request, *s3.GetBucketLocationOutput)
+
+	GetBucketLogging(*s3.GetBucketLoggingInput) (*s3.GetBucketLoggingOutput, error)
+	GetBucketLoggingWithContext(aws.Context, *s3.GetBucketLoggingInput, ...request.Option) (*s3.GetBucketLoggingOutput, error)
+	GetBucketLoggingRequest(*s3.GetBucketLoggingInput) (*request.Request, *s3.GetBucketLoggingOutput)
+
+	GetBucketMetricsConfiguration(*s3.GetBucketMetricsConfigurationInput) (*s3.GetBucketMetricsConfigurationOutput, error)
+	GetBucketMetricsConfigurationWithContext(aws.Context, *s3.GetBucketMetricsConfigurationInput, ...request.Option) (*s3.GetBucketMetricsConfigurationOutput, error)
+	GetBucketMetricsConfigurationRequest(*s3.GetBucketMetricsConfigurationInput) (*request.Request, *s3.GetBucketMetricsConfigurationOutput)
+
+	GetBucketNotification(*s3.GetBucketNotificationConfigurationRequest) (*s3.NotificationConfigurationDeprecated, error)
+	GetBucketNotificationWithContext(aws.Context, *s3.GetBucketNotificationConfigurationRequest, ...request.Option) (*s3.NotificationConfigurationDeprecated, error)
+	GetBucketNotificationRequest(*s3.GetBucketNotificationConfigurationRequest) (*request.Request, *s3.NotificationConfigurationDeprecated)
+
+	GetBucketNotificationConfiguration(*s3.GetBucketNotificationConfigurationRequest) (*s3.NotificationConfiguration, error)
+	GetBucketNotificationConfigurationWithContext(aws.Context, *s3.GetBucketNotificationConfigurationRequest, ...request.Option) (*s3.NotificationConfiguration, error)
+	GetBucketNotificationConfigurationRequest(*s3.GetBucketNotificationConfigurationRequest) (*request.Request, *s3.NotificationConfiguration)
+
+	GetBucketPolicy(*s3.GetBucketPolicyInput) (*s3.GetBucketPolicyOutput, error)
+	GetBucketPolicyWithContext(aws.Context, *s3.GetBucketPolicyInput, ...request.Option) (*s3.GetBucketPolicyOutput, error)
+	GetBucketPolicyRequest(*s3.GetBucketPolicyInput) (*request.Request, *s3.GetBucketPolicyOutput)
+
+	GetBucketReplication(*s3.GetBucketReplicationInput) (*s3.GetBucketReplicationOutput, error)
+	GetBucketReplicationWithContext(aws.Context, *s3.GetBucketReplicationInput, ...request.Option) (*s3.GetBucketReplicationOutput, error)
+	GetBucketReplicationRequest(*s3.GetBucketReplicationInput) (*request.Request, *s3.GetBucketReplicationOutput)
+
+	GetBucketRequestPayment(*s3.GetBucketRequestPaymentInput) (*s3.GetBucketRequestPaymentOutput, error)
+	GetBucketRequestPaymentWithContext(aws.Context, *s3.GetBucketRequestPaymentInput, ...request.Option) (*s3.GetBucketRequestPaymentOutput, error)
+	GetBucketRequestPaymentRequest(*s3.GetBucketRequestPaymentInput) (*request.Request, *s3.GetBucketRequestPaymentOutput)
+
+	GetBucketTagging(*s3.GetBucketTaggingInput) (*s3.GetBucketTaggingOutput, error)
+	GetBucketTaggingWithContext(aws.Context, *s3.GetBucketTaggingInput, ...request.Option) (*s3.GetBucketTaggingOutput, error)
+	GetBucketTaggingRequest(*s3.GetBucketTaggingInput) (*request.Request, *s3.GetBucketTaggingOutput)
+
+	GetBucketVersioning(*s3.GetBucketVersioningInput) (*s3.GetBucketVersioningOutput, error)
+	GetBucketVersioningWithContext(aws.Context, *s3.GetBucketVersioningInput, ...request.Option) (*s3.GetBucketVersioningOutput, error)
+	GetBucketVersioningRequest(*s3.GetBucketVersioningInput) (*request.Request, *s3.GetBucketVersioningOutput)
+
+	GetBucketWebsite(*s3.GetBucketWebsiteInput) (*s3.GetBucketWebsiteOutput, error)
+	GetBucketWebsiteWithContext(aws.Context, *s3.GetBucketWebsiteInput, ...request.Option) (*s3.GetBucketWebsiteOutput, error)
+	GetBucketWebsiteRequest(*s3.GetBucketWebsiteInput) (*request.Request, *s3.GetBucketWebsiteOutput)
+
+	GetObject(*s3.GetObjectInput) (*s3.GetObjectOutput, error)
+	GetObjectWithContext(aws.Context, *s3.GetObjectInput, ...request.Option) (*s3.GetObjectOutput, error)
+	GetObjectRequest(*s3.GetObjectInput) (*request.Request, *s3.GetObjectOutput)
+
+	GetObjectAcl(*s3.GetObjectAclInput) (*s3.GetObjectAclOutput, error)
+	GetObjectAclWithContext(aws.Context, *s3.GetObjectAclInput, ...request.Option) (*s3.GetObjectAclOutput, error)
+	GetObjectAclRequest(*s3.GetObjectAclInput) (*request.Request, *s3.GetObjectAclOutput)
+
+	GetObjectTagging(*s3.GetObjectTaggingInput) (*s3.GetObjectTaggingOutput, error)
+	GetObjectTaggingWithContext(aws.Context, *s3.GetObjectTaggingInput, ...request.Option) (*s3.GetObjectTaggingOutput, error)
+	GetObjectTaggingRequest(*s3.GetObjectTaggingInput) (*request.Request, *s3.GetObjectTaggingOutput)
+
+	GetObjectTorrent(*s3.GetObjectTorrentInput) (*s3.GetObjectTorrentOutput, error)
+	GetObjectTorrentWithContext(aws.Context, *s3.GetObjectTorrentInput, ...request.Option) (*s3.GetObjectTorrentOutput, error)
+	GetObjectTorrentRequest(*s3.GetObjectTorrentInput) (*request.Request, *s3.GetObjectTorrentOutput)
+
+	HeadBucket(*s3.HeadBucketInput) (*s3.HeadBucketOutput, error)
+	HeadBucketWithContext(aws.Context, *s3.HeadBucketInput, ...request.Option) (*s3.HeadBucketOutput, error)
+	HeadBucketRequest(*s3.HeadBucketInput) (*request.Request, *s3.HeadBucketOutput)
+
+	HeadObject(*s3.HeadObjectInput) (*s3.HeadObjectOutput, error)
+	HeadObjectWithContext(aws.Context, *s3.HeadObjectInput, ...request.Option) (*s3.HeadObjectOutput, error)
+	HeadObjectRequest(*s3.HeadObjectInput) (*request.Request, *s3.HeadObjectOutput)
+
+	ListBucketAnalyticsConfigurations(*s3.ListBucketAnalyticsConfigurationsInput) (*s3.ListBucketAnalyticsConfigurationsOutput, error)
+	ListBucketAnalyticsConfigurationsWithContext(aws.Context, *s3.ListBucketAnalyticsConfigurationsInput, ...request.Option) (*s3.ListBucketAnalyticsConfigurationsOutput, error)
+	ListBucketAnalyticsConfigurationsRequest(*s3.ListBucketAnalyticsConfigurationsInput) (*request.Request, *s3.ListBucketAnalyticsConfigurationsOutput)
+
+	ListBucketInventoryConfigurations(*s3.ListBucketInventoryConfigurationsInput) (*s3.ListBucketInventoryConfigurationsOutput, error)
+	ListBucketInventoryConfigurationsWithContext(aws.Context, *s3.ListBucketInventoryConfigurationsInput, ...request.Option) (*s3.ListBucketInventoryConfigurationsOutput, error)
+	ListBucketInventoryConfigurationsRequest(*s3.ListBucketInventoryConfigurationsInput) (*request.Request, *s3.ListBucketInventoryConfigurationsOutput)
+
+	ListBucketMetricsConfigurations(*s3.ListBucketMetricsConfigurationsInput) (*s3.ListBucketMetricsConfigurationsOutput, error)
+	ListBucketMetricsConfigurationsWithContext(aws.Context, *s3.ListBucketMetricsConfigurationsInput, ...request.Option) (*s3.ListBucketMetricsConfigurationsOutput, error)
+	ListBucketMetricsConfigurationsRequest(*s3.ListBucketMetricsConfigurationsInput) (*request.Request, *s3.ListBucketMetricsConfigurationsOutput)
+
+	ListBuckets(*s3.ListBucketsInput) (*s3.ListBucketsOutput, error)
+	ListBucketsWithContext(aws.Context, *s3.ListBucketsInput, ...request.Option) (*s3.ListBucketsOutput, error)
+	ListBucketsRequest(*s3.ListBucketsInput) (*request.Request, *s3.ListBucketsOutput)
+
+	ListMultipartUploads(*s3.ListMultipartUploadsInput) (*s3.ListMultipartUploadsOutput, error)
+	ListMultipartUploadsWithContext(aws.Context, *s3.ListMultipartUploadsInput, ...request.Option) (*s3.ListMultipartUploadsOutput, error)
+	ListMultipartUploadsRequest(*s3.ListMultipartUploadsInput) (*request.Request, *s3.ListMultipartUploadsOutput)
+
+	ListMultipartUploadsPages(*s3.ListMultipartUploadsInput, func(*s3.ListMultipartUploadsOutput, bool) bool) error
+	ListMultipartUploadsPagesWithContext(aws.Context, *s3.ListMultipartUploadsInput, func(*s3.ListMultipartUploadsOutput, bool) bool, ...request.Option) error
+
+	ListObjectVersions(*s3.ListObjectVersionsInput) (*s3.ListObjectVersionsOutput, error)
+	ListObjectVersionsWithContext(aws.Context, *s3.ListObjectVersionsInput, ...request.Option) (*s3.ListObjectVersionsOutput, error)
+	ListObjectVersionsRequest(*s3.ListObjectVersionsInput) (*request.Request, *s3.ListObjectVersionsOutput)
+
+	ListObjectVersionsPages(*s3.ListObjectVersionsInput, func(*s3.ListObjectVersionsOutput, bool) bool) error
+	ListObjectVersionsPagesWithContext(aws.Context, *s3.ListObjectVersionsInput, func(*s3.ListObjectVersionsOutput, bool) bool, ...request.Option) error
+
+	ListObjects(*s3.ListObjectsInput) (*s3.ListObjectsOutput, error)
+	ListObjectsWithContext(aws.Context, *s3.ListObjectsInput, ...request.Option) (*s3.ListObjectsOutput, error)
+	ListObjectsRequest(*s3.ListObjectsInput) (*request.Request, *s3.ListObjectsOutput)
+
+	ListObjectsPages(*s3.ListObjectsInput, func(*s3.ListObjectsOutput, bool) bool) error
+	ListObjectsPagesWithContext(aws.Context, *s3.ListObjectsInput, func(*s3.ListObjectsOutput, bool) bool, ...request.Option) error
+
+	ListObjectsV2(*s3.ListObjectsV2Input) (*s3.ListObjectsV2Output, error)
+	ListObjectsV2WithContext(aws.Context, *s3.ListObjectsV2Input, ...request.Option) (*s3.ListObjectsV2Output, error)
+	ListObjectsV2Request(*s3.ListObjectsV2Input) (*request.Request, *s3.ListObjectsV2Output)
+
+	ListObjectsV2Pages(*s3.ListObjectsV2Input, func(*s3.ListObjectsV2Output, bool) bool) error
+	ListObjectsV2PagesWithContext(aws.Context, *s3.ListObjectsV2Input, func(*s3.ListObjectsV2Output, bool) bool, ...request.Option) error
+
+	ListParts(*s3.ListPartsInput) (*s3.ListPartsOutput, error)
+	ListPartsWithContext(aws.Context, *s3.ListPartsInput, ...request.Option) (*s3.ListPartsOutput, error)
+	ListPartsRequest(*s3.ListPartsInput) (*request.Request, *s3.ListPartsOutput)
+
+	ListPartsPages(*s3.ListPartsInput, func(*s3.ListPartsOutput, bool) bool) error
+	ListPartsPagesWithContext(aws.Context, *s3.ListPartsInput, func(*s3.ListPartsOutput, bool) bool, ...request.Option) error
+
+	PutBucketAccelerateConfiguration(*s3.PutBucketAccelerateConfigurationInput) (*s3.PutBucketAccelerateConfigurationOutput, error)
+	PutBucketAccelerateConfigurationWithContext(aws.Context, *s3.PutBucketAccelerateConfigurationInput, ...request.Option) (*s3.PutBucketAccelerateConfigurationOutput, error)
+	PutBucketAccelerateConfigurationRequest(*s3.PutBucketAccelerateConfigurationInput) (*request.Request, *s3.PutBucketAccelerateConfigurationOutput)
+
+	PutBucketAcl(*s3.PutBucketAclInput) (*s3.PutBucketAclOutput, error)
+	PutBucketAclWithContext(aws.Context, *s3.PutBucketAclInput, ...request.Option) (*s3.PutBucketAclOutput, error)
+	PutBucketAclRequest(*s3.PutBucketAclInput) (*request.Request, *s3.PutBucketAclOutput)
+
+	PutBucketAnalyticsConfiguration(*s3.PutBucketAnalyticsConfigurationInput) (*s3.PutBucketAnalyticsConfigurationOutput, error)
+	PutBucketAnalyticsConfigurationWithContext(aws.Context, *s3.PutBucketAnalyticsConfigurationInput, ...request.Option) (*s3.PutBucketAnalyticsConfigurationOutput, error)
+	PutBucketAnalyticsConfigurationRequest(*s3.PutBucketAnalyticsConfigurationInput) (*request.Request, *s3.PutBucketAnalyticsConfigurationOutput)
+
+	PutBucketCors(*s3.PutBucketCorsInput) (*s3.PutBucketCorsOutput, error)
+	PutBucketCorsWithContext(aws.Context, *s3.PutBucketCorsInput, ...request.Option) (*s3.PutBucketCorsOutput, error)
+	PutBucketCorsRequest(*s3.PutBucketCorsInput) (*request.Request, *s3.PutBucketCorsOutput)
+
+	PutBucketInventoryConfiguration(*s3.PutBucketInventoryConfigurationInput) (*s3.PutBucketInventoryConfigurationOutput, error)
+	PutBucketInventoryConfigurationWithContext(aws.Context, *s3.PutBucketInventoryConfigurationInput, ...request.Option) (*s3.PutBucketInventoryConfigurationOutput, error)
+	PutBucketInventoryConfigurationRequest(*s3.PutBucketInventoryConfigurationInput) (*request.Request, *s3.PutBucketInventoryConfigurationOutput)
+
+	PutBucketLifecycle(*s3.PutBucketLifecycleInput) (*s3.PutBucketLifecycleOutput, error)
+	PutBucketLifecycleWithContext(aws.Context, *s3.PutBucketLifecycleInput, ...request.Option) (*s3.PutBucketLifecycleOutput, error)
+	PutBucketLifecycleRequest(*s3.PutBucketLifecycleInput) (*request.Request, *s3.PutBucketLifecycleOutput)
+
+	PutBucketLifecycleConfiguration(*s3.PutBucketLifecycleConfigurationInput) (*s3.PutBucketLifecycleConfigurationOutput, error)
+	PutBucketLifecycleConfigurationWithContext(aws.Context, *s3.PutBucketLifecycleConfigurationInput, ...request.Option) (*s3.PutBucketLifecycleConfigurationOutput, error)
+	PutBucketLifecycleConfigurationRequest(*s3.PutBucketLifecycleConfigurationInput) (*request.Request, *s3.PutBucketLifecycleConfigurationOutput)
+
+	PutBucketLogging(*s3.PutBucketLoggingInput) (*s3.PutBucketLoggingOutput, error)
+	PutBucketLoggingWithContext(aws.Context, *s3.PutBucketLoggingInput, ...request.Option) (*s3.PutBucketLoggingOutput, error)
+	PutBucketLoggingRequest(*s3.PutBucketLoggingInput) (*request.Request, *s3.PutBucketLoggingOutput)
+
+	PutBucketMetricsConfiguration(*s3.PutBucketMetricsConfigurationInput) (*s3.PutBucketMetricsConfigurationOutput, error)
+	PutBucketMetricsConfigurationWithContext(aws.Context, *s3.PutBucketMetricsConfigurationInput, ...request.Option) (*s3.PutBucketMetricsConfigurationOutput, error)
+	PutBucketMetricsConfigurationRequest(*s3.PutBucketMetricsConfigurationInput) (*request.Request, *s3.PutBucketMetricsConfigurationOutput)
+
+	PutBucketNotification(*s3.PutBucketNotificationInput) (*s3.PutBucketNotificationOutput, error)
+	PutBucketNotificationWithContext(aws.Context, *s3.PutBucketNotificationInput, ...request.Option) (*s3.PutBucketNotificationOutput, error)
+	PutBucketNotificationRequest(*s3.PutBucketNotificationInput) (*request.Request, *s3.PutBucketNotificationOutput)
+
+	PutBucketNotificationConfiguration(*s3.PutBucketNotificationConfigurationInput) (*s3.PutBucketNotificationConfigurationOutput, error)
+	PutBucketNotificationConfigurationWithContext(aws.Context, *s3.PutBucketNotificationConfigurationInput, ...request.Option) (*s3.PutBucketNotificationConfigurationOutput, error)
+	PutBucketNotificationConfigurationRequest(*s3.PutBucketNotificationConfigurationInput) (*request.Request, *s3.PutBucketNotificationConfigurationOutput)
+
+	PutBucketPolicy(*s3.PutBucketPolicyInput) (*s3.PutBucketPolicyOutput, error)
+	PutBucketPolicyWithContext(aws.Context, *s3.PutBucketPolicyInput, ...request.Option) (*s3.PutBucketPolicyOutput, error)
+	PutBucketPolicyRequest(*s3.PutBucketPolicyInput) (*request.Request, *s3.PutBucketPolicyOutput)
+
+	PutBucketReplication(*s3.PutBucketReplicationInput) (*s3.PutBucketReplicationOutput, error)
+	PutBucketReplicationWithContext(aws.Context, *s3.PutBucketReplicationInput, ...request.Option) (*s3.PutBucketReplicationOutput, error)
+	PutBucketReplicationRequest(*s3.PutBucketReplicationInput) (*request.Request, *s3.PutBucketReplicationOutput)
+
+	PutBucketRequestPayment(*s3.PutBucketRequestPaymentInput) (*s3.PutBucketRequestPaymentOutput, error)
+	PutBucketRequestPaymentWithContext(aws.Context, *s3.PutBucketRequestPaymentInput, ...request.Option) (*s3.PutBucketRequestPaymentOutput, error)
+	PutBucketRequestPaymentRequest(*s3.PutBucketRequestPaymentInput) (*request.Request, *s3.PutBucketRequestPaymentOutput)
+
+	PutBucketTagging(*s3.PutBucketTaggingInput) (*s3.PutBucketTaggingOutput, error)
+	PutBucketTaggingWithContext(aws.Context, *s3.PutBucketTaggingInput, ...request.Option) (*s3.PutBucketTaggingOutput, error)
+	PutBucketTaggingRequest(*s3.PutBucketTaggingInput) (*request.Request, *s3.PutBucketTaggingOutput)
+
+	PutBucketVersioning(*s3.PutBucketVersioningInput) (*s3.PutBucketVersioningOutput, error)
+	PutBucketVersioningWithContext(aws.Context, *s3.PutBucketVersioningInput, ...request.Option) (*s3.PutBucketVersioningOutput, error)
+	PutBucketVersioningRequest(*s3.PutBucketVersioningInput) (*request.Request, *s3.PutBucketVersioningOutput)
+
+	PutBucketWebsite(*s3.PutBucketWebsiteInput) (*s3.PutBucketWebsiteOutput, error)
+	PutBucketWebsiteWithContext(aws.Context, *s3.PutBucketWebsiteInput, ...request.Option) (*s3.PutBucketWebsiteOutput, error)
+	PutBucketWebsiteRequest(*s3.PutBucketWebsiteInput) (*request.Request, *s3.PutBucketWebsiteOutput)
+
+	PutObject(*s3.PutObjectInput) (*s3.PutObjectOutput, error)
+	PutObjectWithContext(aws.Context, *s3.PutObjectInput, ...request.Option) (*s3.PutObjectOutput, error)
+	PutObjectRequest(*s3.PutObjectInput) (*request.Request, *s3.PutObjectOutput)
+
+	PutObjectAcl(*s3.PutObjectAclInput) (*s3.PutObjectAclOutput, error)
+	PutObjectAclWithContext(aws.Context, *s3.PutObjectAclInput, ...request.Option) (*s3.PutObjectAclOutput, error)
+	PutObjectAclRequest(*s3.PutObjectAclInput) (*request.Request, *s3.PutObjectAclOutput)
+
+	PutObjectTagging(*s3.PutObjectTaggingInput) (*s3.PutObjectTaggingOutput, error)
+	PutObjectTaggingWithContext(aws.Context, *s3.PutObjectTaggingInput, ...request.Option) (*s3.PutObjectTaggingOutput, error)
+	PutObjectTaggingRequest(*s3.PutObjectTaggingInput) (*request.Request, *s3.PutObjectTaggingOutput)
+
+	RestoreObject(*s3.RestoreObjectInput) (*s3.RestoreObjectOutput, error)
+	RestoreObjectWithContext(aws.Context, *s3.RestoreObjectInput, ...request.Option) (*s3.RestoreObjectOutput, error)
+	RestoreObjectRequest(*s3.RestoreObjectInput) (*request.Request, *s3.RestoreObjectOutput)
+
+	UploadPart(*s3.UploadPartInput) (*s3.UploadPartOutput, error)
+	UploadPartWithContext(aws.Context, *s3.UploadPartInput, ...request.Option) (*s3.UploadPartOutput, error)
+	UploadPartRequest(*s3.UploadPartInput) (*request.Request, *s3.UploadPartOutput)
+
+	UploadPartCopy(*s3.UploadPartCopyInput) (*s3.UploadPartCopyOutput, error)
+	UploadPartCopyWithContext(aws.Context, *s3.UploadPartCopyInput, ...request.Option) (*s3.UploadPartCopyOutput, error)
+	UploadPartCopyRequest(*s3.UploadPartCopyInput) (*request.Request, *s3.UploadPartCopyOutput)
+
+	WaitUntilBucketExists(*s3.HeadBucketInput) error
+	WaitUntilBucketExistsWithContext(aws.Context, *s3.HeadBucketInput, ...request.WaiterOption) error
+
+	WaitUntilBucketNotExists(*s3.HeadBucketInput) error
+	WaitUntilBucketNotExistsWithContext(aws.Context, *s3.HeadBucketInput, ...request.WaiterOption) error
+
+	WaitUntilObjectExists(*s3.HeadObjectInput) error
+	WaitUntilObjectExistsWithContext(aws.Context, *s3.HeadObjectInput, ...request.WaiterOption) error
+
+	WaitUntilObjectNotExists(*s3.HeadObjectInput) error
+	WaitUntilObjectNotExistsWithContext(aws.Context, *s3.HeadObjectInput, ...request.WaiterOption) error
+}
+
+var _ S3API = (*s3.S3)(nil)

vendor/github.com/aws/aws-sdk-go/service/s3/s3manager/batch.go

@@ -0,0 +1,500 @@
+package s3manager
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/awserr"
+	"github.com/aws/aws-sdk-go/aws/client"
+	"github.com/aws/aws-sdk-go/aws/request"
+	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/aws/aws-sdk-go/service/s3/s3iface"
+)
+
+const (
+	// DefaultBatchSize is the batch size we initialize when constructing a batch delete client.
+	// This value is used when calling DeleteObjects. This represents how many objects to delete
+	// per DeleteObjects call.
+	DefaultBatchSize = 100
+)
+
+// BatchError will contain the key and bucket of the object that failed to
+// either upload or download.
+type BatchError struct {
+	Errors  Errors
+	code    string
+	message string
+}
+
+// Errors is a typed alias for a slice of errors to satisfy the error
+// interface.
+type Errors []Error
+
+func (errs Errors) Error() string {
+	buf := bytes.NewBuffer(nil)
+	for i, err := range errs {
+		buf.WriteString(err.Error())
+		if i+1 < len(errs) {
+			buf.WriteString("\n")
+		}
+	}
+	return buf.String()
+}
+
+// Error will contain the original error, bucket, and key of the operation that failed
+// during batch operations.
+type Error struct {
+	OrigErr error
+	Bucket  *string
+	Key     *string
+}
+
+func newError(err error, bucket, key *string) Error {
+	return Error{
+		err,
+		bucket,
+		key,
+	}
+}
+
+func (err *Error) Error() string {
+	return fmt.Sprintf("failed to upload %q to %q:\n%s", err.Key, err.Bucket, err.OrigErr.Error())
+}
+
+// NewBatchError will return a BatchError that satisfies the awserr.Error interface.
+func NewBatchError(code, message string, err []Error) awserr.Error {
+	return &BatchError{
+		Errors:  err,
+		code:    code,
+		message: message,
+	}
+}
+
+// Code will return the code associated with the batch error.
+func (err *BatchError) Code() string {
+	return err.code
+}
+
+// Message will return the message associated with the batch error.
+func (err *BatchError) Message() string {
+	return err.message
+}
+
+func (err *BatchError) Error() string {
+	return awserr.SprintError(err.Code(), err.Message(), "", err.Errors)
+}
+
+// OrigErr will return the original error. Which, in this case, will always be nil
+// for batched operations.
+func (err *BatchError) OrigErr() error {
+	return err.Errors
+}
+
+// BatchDeleteIterator is an interface that uses the scanner pattern to
+// iterate through what needs to be deleted.
+type BatchDeleteIterator interface {
+	Next() bool
+	Err() error
+	DeleteObject() BatchDeleteObject
+}
+
+// DeleteListIterator is an alternative iterator for the BatchDelete client. This will
+// iterate through a list of objects and delete the objects.
+//
+// Example:
+//	iter := &s3manager.DeleteListIterator{
+//		Client: svc,
+//		Input: &s3.ListObjectsInput{
+//			Bucket:  aws.String("bucket"),
+//			MaxKeys: aws.Int64(5),
+//		},
+//		Paginator: request.Pagination{
+//			NewRequest: func() (*request.Request, error) {
+//				var inCpy *ListObjectsInput
+//				if input != nil {
+//					tmp := *input
+//					inCpy = &tmp
+//				}
+//				req, _ := c.ListObjectsRequest(inCpy)
+//				return req, nil
+//			},
+//		},
+//	}
+//
+//	batcher := s3manager.NewBatchDeleteWithClient(svc)
+//	if err := batcher.Delete(aws.BackgroundContext(), iter); err != nil {
+//		return err
+//	}
+type DeleteListIterator struct {
+	Bucket    *string
+	Paginator request.Pagination
+	objects   []*s3.Object
+}
+
+// NewDeleteListIterator will return a new DeleteListIterator.
+func NewDeleteListIterator(svc s3iface.S3API, input *s3.ListObjectsInput, opts ...func(*DeleteListIterator)) BatchDeleteIterator {
+	iter := &DeleteListIterator{
+		Bucket: input.Bucket,
+		Paginator: request.Pagination{
+			NewRequest: func() (*request.Request, error) {
+				var inCpy *s3.ListObjectsInput
+				if input != nil {
+					tmp := *input
+					inCpy = &tmp
+				}
+				req, _ := svc.ListObjectsRequest(inCpy)
+				return req, nil
+			},
+		},
+	}
+
+	for _, opt := range opts {
+		opt(iter)
+	}
+	return iter
+}
+
+// Next will use the S3API client to iterate through a list of objects.
+func (iter *DeleteListIterator) Next() bool {
+	if len(iter.objects) > 0 {
+		iter.objects = iter.objects[1:]
+	}
+
+	if len(iter.objects) == 0 && iter.Paginator.Next() {
+		iter.objects = iter.Paginator.Page().(*s3.ListObjectsOutput).Contents
+	}
+
+	return len(iter.objects) > 0
+}
+
+// Err will return the last known error from Next.
+func (iter *DeleteListIterator) Err() error {
+	return iter.Paginator.Err()
+}
+
+// DeleteObject will return the current object to be deleted.
+func (iter *DeleteListIterator) DeleteObject() BatchDeleteObject {
+	return BatchDeleteObject{
+		Object: &s3.DeleteObjectInput{
+			Bucket: iter.Bucket,
+			Key:    iter.objects[0].Key,
+		},
+	}
+}
+
+// BatchDelete will use the s3 package's service client to perform a batch
+// delete.
+type BatchDelete struct {
+	Client    s3iface.S3API
+	BatchSize int
+}
+
+// NewBatchDeleteWithClient will return a new delete client that can delete a batched amount of
+// objects.
+//
+// Example:
+//	batcher := s3manager.NewBatchDeleteWithClient(client, size)
+//
+//	objects := []BatchDeleteObject{
+//		{
+//			Object:	&s3.DeleteObjectInput {
+//				Key: aws.String("key"),
+//				Bucket: aws.String("bucket"),
+//			},
+//		},
+//	}
+//
+//	if err := batcher.Delete(&s3manager.DeleteObjectsIterator{
+//		Objects: objects,
+//	}); err != nil {
+//		return err
+//	}
+func NewBatchDeleteWithClient(client s3iface.S3API, options ...func(*BatchDelete)) *BatchDelete {
+	svc := &BatchDelete{
+		Client:    client,
+		BatchSize: DefaultBatchSize,
+	}
+
+	for _, opt := range options {
+		opt(svc)
+	}
+
+	return svc
+}
+
+// NewBatchDelete will return a new delete client that can delete a batched amount of
+// objects.
+//
+// Example:
+//	batcher := s3manager.NewBatchDelete(sess, size)
+//
+//	objects := []BatchDeleteObject{
+//		{
+//			Object:	&s3.DeleteObjectInput {
+//				Key: aws.String("key"),
+//				Bucket: aws.String("bucket"),
+//			},
+//		},
+//	}
+//
+//	if err := batcher.Delete(&s3manager.DeleteObjectsIterator{
+//		Objects: objects,
+//	}); err != nil {
+//		return err
+//	}
+func NewBatchDelete(c client.ConfigProvider, options ...func(*BatchDelete)) *BatchDelete {
+	client := s3.New(c)
+	return NewBatchDeleteWithClient(client, options...)
+}
+
+// BatchDeleteObject is a wrapper object for calling the batch delete operation.
+type BatchDeleteObject struct {
+	Object *s3.DeleteObjectInput
+	// After will run after each iteration during the batch process. This function will
+	// be executed whether or not the request was successful.
+	After func() error
+}
+
+// DeleteObjectsIterator is an interface that uses the scanner pattern to iterate
+// through a series of objects to be deleted.
+type DeleteObjectsIterator struct {
+	Objects []BatchDeleteObject
+	index   int
+	inc     bool
+}
+
+// Next will increment the default iterator's index and and ensure that there
+// is another object to iterator to.
+func (iter *DeleteObjectsIterator) Next() bool {
+	if iter.inc {
+		iter.index++
+	} else {
+		iter.inc = true
+	}
+	return iter.index < len(iter.Objects)
+}
+
+// Err will return an error. Since this is just used to satisfy the BatchDeleteIterator interface
+// this will only return nil.
+func (iter *DeleteObjectsIterator) Err() error {
+	return nil
+}
+
+// DeleteObject will return the BatchDeleteObject at the current batched index.
+func (iter *DeleteObjectsIterator) DeleteObject() BatchDeleteObject {
+	object := iter.Objects[iter.index]
+	return object
+}
+
+// Delete will use the iterator to queue up objects that need to be deleted.
+// Once the batch size is met, this will call the deleteBatch function.
+func (d *BatchDelete) Delete(ctx aws.Context, iter BatchDeleteIterator) error {
+	var errs []Error
+	objects := []BatchDeleteObject{}
+	var input *s3.DeleteObjectsInput
+
+	for iter.Next() {
+		o := iter.DeleteObject()
+
+		if input == nil {
+			input = initDeleteObjectsInput(o.Object)
+		}
+
+		parity := hasParity(input, o)
+		if parity {
+			input.Delete.Objects = append(input.Delete.Objects, &s3.ObjectIdentifier{
+				Key:       o.Object.Key,
+				VersionId: o.Object.VersionId,
+			})
+			objects = append(objects, o)
+		}
+
+		if len(input.Delete.Objects) == d.BatchSize || !parity {
+			if err := deleteBatch(d, input, objects); err != nil {
+				errs = append(errs, err...)
+			}
+
+			objects = objects[:0]
+			input = nil
+
+			if !parity {
+				objects = append(objects, o)
+				input = initDeleteObjectsInput(o.Object)
+				input.Delete.Objects = append(input.Delete.Objects, &s3.ObjectIdentifier{
+					Key:       o.Object.Key,
+					VersionId: o.Object.VersionId,
+				})
+			}
+		}
+	}
+
+	if input != nil && len(input.Delete.Objects) > 0 {
+		if err := deleteBatch(d, input, objects); err != nil {
+			errs = append(errs, err...)
+		}
+	}
+
+	if len(errs) > 0 {
+		return NewBatchError("BatchedDeleteIncomplete", "some objects have failed to be deleted.", errs)
+	}
+	return nil
+}
+
+func initDeleteObjectsInput(o *s3.DeleteObjectInput) *s3.DeleteObjectsInput {
+	return &s3.DeleteObjectsInput{
+		Bucket:       o.Bucket,
+		MFA:          o.MFA,
+		RequestPayer: o.RequestPayer,
+		Delete:       &s3.Delete{},
+	}
+}
+
+// deleteBatch will delete a batch of items in the objects parameters.
+func deleteBatch(d *BatchDelete, input *s3.DeleteObjectsInput, objects []BatchDeleteObject) []Error {
+	errs := []Error{}
+
+	if result, err := d.Client.DeleteObjects(input); err != nil {
+		for i := 0; i < len(input.Delete.Objects); i++ {
+			errs = append(errs, newError(err, input.Bucket, input.Delete.Objects[i].Key))
+		}
+	} else if len(result.Errors) > 0 {
+		for i := 0; i < len(result.Errors); i++ {
+			errs = append(errs, newError(err, input.Bucket, result.Errors[i].Key))
+		}
+	}
+	for _, object := range objects {
+		if object.After == nil {
+			continue
+		}
+		if err := object.After(); err != nil {
+			errs = append(errs, newError(err, object.Object.Bucket, object.Object.Key))
+		}
+	}
+
+	return errs
+}
+
+func hasParity(o1 *s3.DeleteObjectsInput, o2 BatchDeleteObject) bool {
+	if o1.Bucket != nil && o2.Object.Bucket != nil {
+		if *o1.Bucket != *o2.Object.Bucket {
+			return false
+		}
+	} else if o1.Bucket != o2.Object.Bucket {
+		return false
+	}
+
+	if o1.MFA != nil && o2.Object.MFA != nil {
+		if *o1.MFA != *o2.Object.MFA {
+			return false
+		}
+	} else if o1.MFA != o2.Object.MFA {
+		return false
+	}
+
+	if o1.RequestPayer != nil && o2.Object.RequestPayer != nil {
+		if *o1.RequestPayer != *o2.Object.RequestPayer {
+			return false
+		}
+	} else if o1.RequestPayer != o2.Object.RequestPayer {
+		return false
+	}
+
+	return true
+}
+
+// BatchDownloadIterator is an interface that uses the scanner pattern to iterate
+// through a series of objects to be downloaded.
+type BatchDownloadIterator interface {
+	Next() bool
+	Err() error
+	DownloadObject() BatchDownloadObject
+}
+
+// BatchDownloadObject contains all necessary information to run a batch operation once.
+type BatchDownloadObject struct {
+	Object *s3.GetObjectInput
+	Writer io.WriterAt
+	// After will run after each iteration during the batch process. This function will
+	// be executed whether or not the request was successful.
+	After func() error
+}
+
+// DownloadObjectsIterator implements the BatchDownloadIterator interface and allows for batched
+// download of objects.
+type DownloadObjectsIterator struct {
+	Objects []BatchDownloadObject
+	index   int
+	inc     bool
+}
+
+// Next will increment the default iterator's index and and ensure that there
+// is another object to iterator to.
+func (batcher *DownloadObjectsIterator) Next() bool {
+	if batcher.inc {
+		batcher.index++
+	} else {
+		batcher.inc = true
+	}
+	return batcher.index < len(batcher.Objects)
+}
+
+// DownloadObject will return the BatchDownloadObject at the current batched index.
+func (batcher *DownloadObjectsIterator) DownloadObject() BatchDownloadObject {
+	object := batcher.Objects[batcher.index]
+	return object
+}
+
+// Err will return an error. Since this is just used to satisfy the BatchDeleteIterator interface
+// this will only return nil.
+func (batcher *DownloadObjectsIterator) Err() error {
+	return nil
+}
+
+// BatchUploadIterator is an interface that uses the scanner pattern to
+// iterate through what needs to be uploaded.
+type BatchUploadIterator interface {
+	Next() bool
+	Err() error
+	UploadObject() BatchUploadObject
+}
+
+// UploadObjectsIterator implements the BatchUploadIterator interface and allows for batched
+// upload of objects.
+type UploadObjectsIterator struct {
+	Objects []BatchUploadObject
+	index   int
+	inc     bool
+}
+
+// Next will increment the default iterator's index and and ensure that there
+// is another object to iterator to.
+func (batcher *UploadObjectsIterator) Next() bool {
+	if batcher.inc {
+		batcher.index++
+	} else {
+		batcher.inc = true
+	}
+	return batcher.index < len(batcher.Objects)
+}
+
+// Err will return an error. Since this is just used to satisfy the BatchUploadIterator interface
+// this will only return nil.
+func (batcher *UploadObjectsIterator) Err() error {
+	return nil
+}
+
+// UploadObject will return the BatchUploadObject at the current batched index.
+func (batcher *UploadObjectsIterator) UploadObject() BatchUploadObject {
+	object := batcher.Objects[batcher.index]
+	return object
+}
+
+// BatchUploadObject contains all necessary information to run a batch operation once.
+type BatchUploadObject struct {
+	Object *UploadInput
+	// After will run after each iteration during the batch process. This function will
+	// be executed whether or not the request was successful.
+	After func() error
+}

vendor/github.com/aws/aws-sdk-go/service/s3/s3manager/bucket_region.go

@@ -0,0 +1,83 @@
+package s3manager
+
+import (
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/client"
+	"github.com/aws/aws-sdk-go/aws/credentials"
+	"github.com/aws/aws-sdk-go/aws/request"
+	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/aws/aws-sdk-go/service/s3/s3iface"
+)
+
+// GetBucketRegion will attempt to get the region for a bucket using the
+// regionHint to determine which AWS partition to perform the query on.
+//
+// The request will not be signed, and will not use your AWS credentials.
+//
+// A "NotFound" error code will be returned if the bucket does not exist in
+// the AWS partition the regionHint belongs to.
+//
+// For example to get the region of a bucket which exists in "eu-central-1"
+// you could provide a region hint of "us-west-2".
+//
+//    sess := session.Must(session.NewSession())
+//
+//    bucket := "my-bucket"
+//    region, err := s3manager.GetBucketRegion(ctx, sess, bucket, "us-west-2")
+//    if err != nil {
+//        if aerr, ok := err.(awserr.Error); ok && aerr.Code() == "NotFound" {
+//             fmt.Fprintf(os.Stderr, "unable to find bucket %s's region not found\n", bucket)
+//        }
+//        return err
+//    }
+//    fmt.Printf("Bucket %s is in %s region\n", bucket, region)
+//
+func GetBucketRegion(ctx aws.Context, c client.ConfigProvider, bucket, regionHint string, opts ...request.Option) (string, error) {
+	svc := s3.New(c, &aws.Config{
+		Region: aws.String(regionHint),
+	})
+	return GetBucketRegionWithClient(ctx, svc, bucket, opts...)
+}
+
+const bucketRegionHeader = "X-Amz-Bucket-Region"
+
+// GetBucketRegionWithClient is the same as GetBucketRegion with the exception
+// that it takes a S3 service client instead of a Session. The regionHint is
+// derived from the region the S3 service client was created in.
+//
+// See GetBucketRegion for more information.
+func GetBucketRegionWithClient(ctx aws.Context, svc s3iface.S3API, bucket string, opts ...request.Option) (string, error) {
+	req, _ := svc.HeadBucketRequest(&s3.HeadBucketInput{
+		Bucket: aws.String(bucket),
+	})
+	req.Config.S3ForcePathStyle = aws.Bool(true)
+	req.Config.Credentials = credentials.AnonymousCredentials
+	req.SetContext(ctx)
+
+	// Disable HTTP redirects to prevent an invalid 301 from eating the response
+	// because Go's HTTP client will fail, and drop the response if an 301 is
+	// received without a location header. S3 will return a 301 without the
+	// location header for HeadObject API calls.
+	req.DisableFollowRedirects = true
+
+	var bucketRegion string
+	req.Handlers.Send.PushBack(func(r *request.Request) {
+		bucketRegion = r.HTTPResponse.Header.Get(bucketRegionHeader)
+		if len(bucketRegion) == 0 {
+			return
+		}
+		r.HTTPResponse.StatusCode = 200
+		r.HTTPResponse.Status = "OK"
+		r.Error = nil
+	})
+
+	req.ApplyOptions(opts...)
+
+	if err := req.Send(); err != nil {
+		return "", err
+	}
+
+	bucketRegion = s3.NormalizeBucketLocation(bucketRegion)
+
+	return bucketRegion, nil
+}

vendor/github.com/aws/aws-sdk-go/service/s3/s3manager/doc.go

@@ -0,0 +1,3 @@
+// Package s3manager provides utilities to upload and download objects from
+// S3 concurrently. Helpful for when working with large objects.
+package s3manager

vendor/github.com/aws/aws-sdk-go/service/s3/s3manager/download.go

@@ -0,0 +1,547 @@
+package s3manager
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"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"
+	"github.com/aws/aws-sdk-go/aws/request"
+	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/aws/aws-sdk-go/service/s3/s3iface"
+)
+
+// DefaultDownloadPartSize is the default range of bytes to get at a time when
+// using Download().
+const DefaultDownloadPartSize = 1024 * 1024 * 5
+
+// DefaultDownloadConcurrency is the default number of goroutines to spin up
+// when using Download().
+const DefaultDownloadConcurrency = 5
+
+// The Downloader structure that calls Download(). It is safe to call Download()
+// on this structure for multiple objects and across concurrent goroutines.
+// Mutating the Downloader's properties is not safe to be done concurrently.
+type Downloader struct {
+	// The buffer size (in bytes) to use when buffering data into chunks and
+	// sending them as parts to S3. The minimum allowed part size is 5MB, and
+	// if this value is set to zero, the DefaultDownloadPartSize value will be used.
+	//
+	// PartSize is ignored if the Range input parameter is provided.
+	PartSize int64
+
+	// The number of goroutines to spin up in parallel when sending parts.
+	// If this is set to zero, the DefaultDownloadConcurrency value will be used.
+	//
+	// Concurrency is ignored if the Range input parameter is provided.
+	Concurrency int
+
+	// An S3 client to use when performing downloads.
+	S3 s3iface.S3API
+
+	// List of request options that will be passed down to individual API
+	// operation requests made by the downloader.
+	RequestOptions []request.Option
+}
+
+// WithDownloaderRequestOptions appends to the Downloader's API request options.
+func WithDownloaderRequestOptions(opts ...request.Option) func(*Downloader) {
+	return func(d *Downloader) {
+		d.RequestOptions = append(d.RequestOptions, opts...)
+	}
+}
+
+// NewDownloader creates a new Downloader instance to downloads objects from
+// S3 in concurrent chunks. Pass in additional functional options  to customize
+// the downloader behavior. Requires a client.ConfigProvider in order to create
+// a S3 service client. The session.Session satisfies the client.ConfigProvider
+// interface.
+//
+// Example:
+//     // The session the S3 Downloader will use
+//     sess := session.Must(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.Downloader) {
+//          d.PartSize = 64 * 1024 * 1024 // 64MB per part
+//     })
+func NewDownloader(c client.ConfigProvider, options ...func(*Downloader)) *Downloader {
+	d := &Downloader{
+		S3:          s3.New(c),
+		PartSize:    DefaultDownloadPartSize,
+		Concurrency: DefaultDownloadConcurrency,
+	}
+	for _, option := range options {
+		option(d)
+	}
+
+	return d
+}
+
+// NewDownloaderWithClient creates a new Downloader instance to downloads
+// objects from S3 in concurrent chunks. Pass in additional functional
+// options to customize the downloader behavior. Requires a S3 service client
+// to make S3 API calls.
+//
+// Example:
+//     // The session the S3 Downloader will use
+//     sess := session.Must(session.NewSession())
+//
+//     // The S3 client the S3 Downloader will use
+//     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.Downloader) {
+//          d.PartSize = 64 * 1024 * 1024 // 64MB per part
+//     })
+func NewDownloaderWithClient(svc s3iface.S3API, options ...func(*Downloader)) *Downloader {
+	d := &Downloader{
+		S3:          svc,
+		PartSize:    DefaultDownloadPartSize,
+		Concurrency: DefaultDownloadConcurrency,
+	}
+	for _, option := range options {
+		option(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
+// 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.
+//
+// The w io.WriterAt can be satisfied by an os.File to do multipart concurrent
+// downloads, or in memory []byte wrapper using aws.WriteAtBuffer.
+//
+// If the GetObjectInput's Range value is provided that will cause the downloader
+// to perform a single GetObjectInput request for that object's range. This will
+// caused the part size, and concurrency configurations to be ignored.
+func (d Downloader) Download(w io.WriterAt, input *s3.GetObjectInput, options ...func(*Downloader)) (n int64, err error) {
+	return d.DownloadWithContext(aws.BackgroundContext(), w, input, options...)
+}
+
+// DownloadWithContext downloads an object in S3 and writes the payload into w
+// using concurrent GET requests.
+//
+// DownloadWithContext is the same as Download with the additional support for
+// Context input parameters. The Context must not be nil. A nil Context will
+// cause a panic. Use the Context to add deadlining, timeouts, ect. The
+// DownloadWithContext may create sub-contexts for individual underlying
+// requests.
+//
+// Additional functional options can be provided to configure the individual
+// download. These options are copies of the Downloader instance Download is
+// called from. Modifying the options will not impact the original Downloader
+// instance. Use the WithDownloaderRequestOptions helper function to pass in request
+// options that will be applied to all API operations made with this downloader.
+//
+// The w io.WriterAt can be satisfied by an os.File to do multipart concurrent
+// downloads, or in memory []byte wrapper using aws.WriteAtBuffer.
+//
+// It is safe to call this method concurrently across goroutines.
+//
+// If the GetObjectInput's Range value is provided that will cause the downloader
+// to perform a single GetObjectInput request for that object's range. This will
+// caused the part size, and concurrency configurations to be ignored.
+func (d Downloader) DownloadWithContext(ctx aws.Context, w io.WriterAt, input *s3.GetObjectInput, options ...func(*Downloader)) (n int64, err error) {
+	impl := downloader{w: w, in: input, cfg: d, ctx: ctx}
+
+	for _, option := range options {
+		option(&impl.cfg)
+	}
+	impl.cfg.RequestOptions = append(impl.cfg.RequestOptions, request.WithAppendUserAgent("S3Manager"))
+
+	if s, ok := d.S3.(maxRetrier); ok {
+		impl.partBodyMaxRetries = s.MaxRetries()
+	}
+
+	impl.totalBytes = -1
+	if impl.cfg.Concurrency == 0 {
+		impl.cfg.Concurrency = DefaultDownloadConcurrency
+	}
+
+	if impl.cfg.PartSize == 0 {
+		impl.cfg.PartSize = DefaultDownloadPartSize
+	}
+
+	return impl.download()
+}
+
+// DownloadWithIterator will download a batched amount of objects in S3 and writes them
+// to the io.WriterAt specificed in the iterator.
+//
+// Example:
+//	svc := s3manager.NewDownloader(session)
+//
+//	fooFile, err := os.Open("/tmp/foo.file")
+//	if err != nil {
+//		return err
+//	}
+//
+//	barFile, err := os.Open("/tmp/bar.file")
+//	if err != nil {
+//		return err
+//	}
+//
+//	objects := []s3manager.BatchDownloadObject {
+//		{
+//			Input: &s3.GetObjectInput {
+//				Bucket: aws.String("bucket"),
+//				Key: aws.String("foo"),
+//			},
+//			Writer: fooFile,
+//		},
+//		{
+//			Input: &s3.GetObjectInput {
+//				Bucket: aws.String("bucket"),
+//				Key: aws.String("bar"),
+//			},
+//			Writer: barFile,
+//		},
+//	}
+//
+//	iter := &s3manager.DownloadObjectsIterator{Objects: objects}
+//	if err := svc.DownloadWithIterator(aws.BackgroundContext(), iter); err != nil {
+//		return err
+//	}
+func (d Downloader) DownloadWithIterator(ctx aws.Context, iter BatchDownloadIterator, opts ...func(*Downloader)) error {
+	var errs []Error
+	for iter.Next() {
+		object := iter.DownloadObject()
+		if _, err := d.DownloadWithContext(ctx, object.Writer, object.Object, opts...); err != nil {
+			errs = append(errs, newError(err, object.Object.Bucket, object.Object.Key))
+		}
+
+		if object.After == nil {
+			continue
+		}
+
+		if err := object.After(); err != nil {
+			errs = append(errs, newError(err, object.Object.Bucket, object.Object.Key))
+		}
+	}
+
+	if len(errs) > 0 {
+		return NewBatchError("BatchedDownloadIncomplete", "some objects have failed to download.", errs)
+	}
+	return nil
+}
+
+// downloader is the implementation structure used internally by Downloader.
+type downloader struct {
+	ctx aws.Context
+	cfg Downloader
+
+	in *s3.GetObjectInput
+	w  io.WriterAt
+
+	wg sync.WaitGroup
+	m  sync.Mutex
+
+	pos        int64
+	totalBytes int64
+	written    int64
+	err        error
+
+	partBodyMaxRetries int
+}
+
+// download performs the implementation of the object download across ranged
+// GETs.
+func (d *downloader) download() (n int64, err error) {
+	// If range is specified fall back to single download of that range
+	// this enables the functionality of ranged gets with the downloader but
+	// at the cost of no multipart downloads.
+	if rng := aws.StringValue(d.in.Range); len(rng) > 0 {
+		d.downloadRange(rng)
+		return d.written, d.err
+	}
+
+	// Spin off first worker to check additional header information
+	d.getChunk()
+
+	if total := d.getTotalBytes(); total >= 0 {
+		// Spin up workers
+		ch := make(chan dlchunk, d.cfg.Concurrency)
+
+		for i := 0; i < d.cfg.Concurrency; i++ {
+			d.wg.Add(1)
+			go d.downloadPart(ch)
+		}
+
+		// Assign work
+		for d.getErr() == nil {
+			if d.pos >= total {
+				break // We're finished queuing chunks
+			}
+
+			// Queue the next range of bytes to read.
+			ch <- dlchunk{w: d.w, start: d.pos, size: d.cfg.PartSize}
+			d.pos += d.cfg.PartSize
+		}
+
+		// Wait for completion
+		close(ch)
+		d.wg.Wait()
+	} else {
+		// Checking if we read anything new
+		for d.err == nil {
+			d.getChunk()
+		}
+
+		// We expect a 416 error letting us know we are done downloading the
+		// total bytes. Since we do not know the content's length, this will
+		// keep grabbing chunks of data until the range of bytes specified in
+		// the request is out of range of the content. Once, this happens, a
+		// 416 should occur.
+		e, ok := d.err.(awserr.RequestFailure)
+		if ok && e.StatusCode() == http.StatusRequestedRangeNotSatisfiable {
+			d.err = nil
+		}
+	}
+
+	// Return error
+	return d.written, d.err
+}
+
+// downloadPart is an individual goroutine worker reading from the ch channel
+// and performing a GetObject request on the data with a given byte range.
+//
+// If this is the first worker, this operation also resolves the total number
+// of bytes to be read so that the worker manager knows when it is finished.
+func (d *downloader) downloadPart(ch chan dlchunk) {
+	defer d.wg.Done()
+	for {
+		chunk, ok := <-ch
+		if !ok {
+			break
+		}
+		if d.getErr() != nil {
+			// Drain the channel if there is an error, to prevent deadlocking
+			// of download producer.
+			continue
+		}
+
+		if err := d.downloadChunk(chunk); err != nil {
+			d.setErr(err)
+		}
+	}
+}
+
+// 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() {
+	if d.getErr() != nil {
+		return
+	}
+
+	chunk := dlchunk{w: d.w, start: d.pos, size: d.cfg.PartSize}
+	d.pos += d.cfg.PartSize
+
+	if err := d.downloadChunk(chunk); err != nil {
+		d.setErr(err)
+	}
+}
+
+// downloadRange downloads an Object given the passed in Byte-Range value.
+// The chunk used down download the range will be configured for that range.
+func (d *downloader) downloadRange(rng string) {
+	if d.getErr() != nil {
+		return
+	}
+
+	chunk := dlchunk{w: d.w, start: d.pos}
+	// Ranges specified will short circuit the multipart download
+	chunk.withRange = rng
+
+	if err := d.downloadChunk(chunk); err != nil {
+		d.setErr(err)
+	}
+
+	// Update the position based on the amount of data received.
+	d.pos = d.written
+}
+
+// downloadChunk downloads the chunk from s3
+func (d *downloader) downloadChunk(chunk dlchunk) error {
+	in := &s3.GetObjectInput{}
+	awsutil.Copy(in, d.in)
+
+	// Get the next byte range of data
+	in.Range = aws.String(chunk.ByteRange())
+
+	var n int64
+	var err error
+	for retry := 0; retry <= d.partBodyMaxRetries; retry++ {
+		var resp *s3.GetObjectOutput
+		resp, err = d.cfg.S3.GetObjectWithContext(d.ctx, in, d.cfg.RequestOptions...)
+		if err != nil {
+			return err
+		}
+		d.setTotalBytes(resp) // Set total if not yet set.
+
+		n, err = io.Copy(&chunk, resp.Body)
+		resp.Body.Close()
+		if err == nil {
+			break
+		}
+
+		chunk.cur = 0
+		logMessage(d.cfg.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)
+	}
+}
+
+// getTotalBytes is a thread-safe getter for retrieving the total byte status.
+func (d *downloader) getTotalBytes() int64 {
+	d.m.Lock()
+	defer d.m.Unlock()
+
+	return d.totalBytes
+}
+
+// setTotalBytes is a thread-safe setter for setting the total byte status.
+// Will extract the object's total bytes from the Content-Range if the file
+// will be chunked, or Content-Length. Content-Length is used when the response
+// does not include a Content-Range. Meaning the object was not chunked. This
+// occurs when the full file fits within the PartSize directive.
+func (d *downloader) setTotalBytes(resp *s3.GetObjectOutput) {
+	d.m.Lock()
+	defer d.m.Unlock()
+
+	if d.totalBytes >= 0 {
+		return
+	}
+
+	if resp.ContentRange == nil {
+		// ContentRange is nil when the full file contents is provided, and
+		// is not chunked. Use ContentLength instead.
+		if resp.ContentLength != nil {
+			d.totalBytes = *resp.ContentLength
+			return
+		}
+	} else {
+		parts := strings.Split(*resp.ContentRange, "/")
+
+		total := int64(-1)
+		var err error
+		// Checking for whether or not a numbered total exists
+		// If one does not exist, we will assume the total to be -1, undefined,
+		// and sequentially download each chunk until hitting a 416 error
+		totalStr := parts[len(parts)-1]
+		if totalStr != "*" {
+			total, err = strconv.ParseInt(totalStr, 10, 64)
+			if err != nil {
+				d.err = err
+				return
+			}
+		}
+
+		d.totalBytes = total
+	}
+}
+
+func (d *downloader) incrWritten(n int64) {
+	d.m.Lock()
+	defer d.m.Unlock()
+
+	d.written += n
+}
+
+// getErr is a thread-safe getter for the error object
+func (d *downloader) getErr() error {
+	d.m.Lock()
+	defer d.m.Unlock()
+
+	return d.err
+}
+
+// setErr is a thread-safe setter for the error object
+func (d *downloader) setErr(e error) {
+	d.m.Lock()
+	defer d.m.Unlock()
+
+	d.err = e
+}
+
+// dlchunk represents a single chunk of data to write by the worker routine.
+// This structure also implements an io.SectionReader style interface for
+// io.WriterAt, effectively making it an io.SectionWriter (which does not
+// exist).
+type dlchunk struct {
+	w     io.WriterAt
+	start int64
+	size  int64
+	cur   int64
+
+	// specifies the byte range the chunk should be downloaded with.
+	withRange string
+}
+
+// Write wraps io.WriterAt for the dlchunk, writing from the dlchunk's start
+// position to its end (or EOF).
+//
+// If a range is specified on the dlchunk the size will be ignored when writing.
+// as the total size may not of be known ahead of time.
+func (c *dlchunk) Write(p []byte) (n int, err error) {
+	if c.cur >= c.size && len(c.withRange) == 0 {
+		return 0, io.EOF
+	}
+
+	n, err = c.w.WriteAt(p, c.start+c.cur)
+	c.cur += int64(n)
+
+	return
+}
+
+// ByteRange returns a HTTP Byte-Range header value that should be used by the
+// client to request the chunk's range.
+func (c *dlchunk) ByteRange() string {
+	if len(c.withRange) != 0 {
+		return c.withRange
+	}
+
+	return fmt.Sprintf("bytes=%d-%d", c.start, c.start+c.size-1)
+}

vendor/github.com/aws/aws-sdk-go/service/s3/s3manager/upload.go

@@ -0,0 +1,784 @@
+package s3manager
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"sort"
+	"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"
+	"github.com/aws/aws-sdk-go/aws/request"
+	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/aws/aws-sdk-go/service/s3/s3iface"
+)
+
+// MaxUploadParts is the maximum allowed number of parts in a multi-part upload
+// on Amazon S3.
+const MaxUploadParts = 10000
+
+// MinUploadPartSize is the minimum allowed part size when uploading a part to
+// Amazon S3.
+const MinUploadPartSize int64 = 1024 * 1024 * 5
+
+// DefaultUploadPartSize is the default part size to buffer chunks of a
+// payload into.
+const DefaultUploadPartSize = MinUploadPartSize
+
+// DefaultUploadConcurrency is the default number of goroutines to spin up when
+// using Upload().
+const DefaultUploadConcurrency = 5
+
+// A MultiUploadFailure wraps a failed S3 multipart upload. An error returned
+// will satisfy this interface when a multi part upload failed to upload all
+// chucks to S3. In the case of a failure the UploadID is needed to operate on
+// the chunks, if any, which were uploaded.
+//
+// Example:
+//
+//     u := s3manager.NewUploader(opts)
+//     output, err := u.upload(input)
+//     if err != nil {
+//         if multierr, ok := err.(s3manager.MultiUploadFailure); ok {
+//             // Process error and its associated uploadID
+//             fmt.Println("Error:", multierr.Code(), multierr.Message(), multierr.UploadID())
+//         } else {
+//             // Process error generically
+//             fmt.Println("Error:", err.Error())
+//         }
+//     }
+//
+type MultiUploadFailure interface {
+	awserr.Error
+
+	// Returns the upload id for the S3 multipart upload that failed.
+	UploadID() string
+}
+
+// So that the Error interface type can be included as an anonymous field
+// in the multiUploadError struct and not conflict with the error.Error() method.
+type awsError awserr.Error
+
+// A multiUploadError wraps the upload ID of a failed s3 multipart upload.
+// Composed of BaseError for code, message, and original error
+//
+// Should be used for an error that occurred failing a S3 multipart upload,
+// and a upload ID is available. If an uploadID is not available a more relevant
+type multiUploadError struct {
+	awsError
+
+	// ID for multipart upload which failed.
+	uploadID string
+}
+
+// Error returns the string representation of the error.
+//
+// See apierr.BaseError ErrorWithExtra for output format
+//
+// Satisfies the error interface.
+func (m multiUploadError) Error() string {
+	extra := fmt.Sprintf("upload id: %s", m.uploadID)
+	return awserr.SprintError(m.Code(), m.Message(), extra, m.OrigErr())
+}
+
+// String returns the string representation of the error.
+// Alias for Error to satisfy the stringer interface.
+func (m multiUploadError) String() string {
+	return m.Error()
+}
+
+// UploadID returns the id of the S3 upload which failed.
+func (m multiUploadError) UploadID() string {
+	return m.uploadID
+}
+
+// UploadInput contains all input for upload requests to Amazon S3.
+type UploadInput struct {
+	// The canned ACL to apply to the object.
+	ACL *string `location:"header" locationName:"x-amz-acl" type:"string"`
+
+	Bucket *string `location:"uri" locationName:"Bucket" type:"string" required:"true"`
+
+	// Specifies caching behavior along the request/reply chain.
+	CacheControl *string `location:"header" locationName:"Cache-Control" type:"string"`
+
+	// Specifies presentational information for the object.
+	ContentDisposition *string `location:"header" locationName:"Content-Disposition" type:"string"`
+
+	// Specifies what content encodings have been applied to the object and thus
+	// what decoding mechanisms must be applied to obtain the media-type referenced
+	// by the Content-Type header field.
+	ContentEncoding *string `location:"header" locationName:"Content-Encoding" type:"string"`
+
+	// The language the content is in.
+	ContentLanguage *string `location:"header" locationName:"Content-Language" type:"string"`
+
+	// A standard MIME type describing the format of the object data.
+	ContentType *string `location:"header" locationName:"Content-Type" type:"string"`
+
+	// The date and time at which the object is no longer cacheable.
+	Expires *time.Time `location:"header" locationName:"Expires" type:"timestamp" timestampFormat:"rfc822"`
+
+	// Gives the grantee READ, READ_ACP, and WRITE_ACP permissions on the object.
+	GrantFullControl *string `location:"header" locationName:"x-amz-grant-full-control" type:"string"`
+
+	// Allows grantee to read the object data and its metadata.
+	GrantRead *string `location:"header" locationName:"x-amz-grant-read" type:"string"`
+
+	// Allows grantee to read the object ACL.
+	GrantReadACP *string `location:"header" locationName:"x-amz-grant-read-acp" type:"string"`
+
+	// Allows grantee to write the ACL for the applicable object.
+	GrantWriteACP *string `location:"header" locationName:"x-amz-grant-write-acp" type:"string"`
+
+	Key *string `location:"uri" locationName:"Key" type:"string" required:"true"`
+
+	// A map of metadata to store with the object in S3.
+	Metadata map[string]*string `location:"headers" locationName:"x-amz-meta-" type:"map"`
+
+	// Confirms that the requester knows that she or he will be charged for the
+	// request. Bucket owners need not specify this parameter in their requests.
+	// Documentation on downloading objects from requester pays buckets can be found
+	// at http://docs.aws.amazon.com/AmazonS3/latest/dev/ObjectsinRequesterPaysBuckets.html
+	RequestPayer *string `location:"header" locationName:"x-amz-request-payer" type:"string"`
+
+	// Specifies the algorithm to use to when encrypting the object (e.g., AES256,
+	// aws:kms).
+	SSECustomerAlgorithm *string `location:"header" locationName:"x-amz-server-side-encryption-customer-algorithm" type:"string"`
+
+	// Specifies the customer-provided encryption key for Amazon S3 to use in encrypting
+	// data. This value is used to store the object and then it is discarded; Amazon
+	// does not store the encryption key. The key must be appropriate for use with
+	// the algorithm specified in the x-amz-server-side​-encryption​-customer-algorithm
+	// header.
+	SSECustomerKey *string `location:"header" locationName:"x-amz-server-side-encryption-customer-key" type:"string"`
+
+	// Specifies the 128-bit MD5 digest of the encryption key according to RFC 1321.
+	// Amazon S3 uses this header for a message integrity check to ensure the encryption
+	// key was transmitted without error.
+	SSECustomerKeyMD5 *string `location:"header" locationName:"x-amz-server-side-encryption-customer-key-MD5" type:"string"`
+
+	// Specifies the AWS KMS key ID to use for object encryption. All GET and PUT
+	// 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"`
+
+	// The Server-side encryption algorithm used when storing this object in S3
+	// (e.g., AES256, aws:kms).
+	ServerSideEncryption *string `location:"header" locationName:"x-amz-server-side-encryption" type:"string"`
+
+	// The type of storage to use for the object. Defaults to 'STANDARD'.
+	StorageClass *string `location:"header" locationName:"x-amz-storage-class" type:"string"`
+
+	// The tag-set for the object. The tag-set must be encoded as URL Query parameters
+	Tagging *string `location:"header" locationName:"x-amz-tagging" type:"string"`
+
+	// If the bucket is configured as a website, redirects requests for this object
+	// to another object in the same bucket or to an external URL. Amazon S3 stores
+	// the value of this header in the object metadata.
+	WebsiteRedirectLocation *string `location:"header" locationName:"x-amz-website-redirect-location" type:"string"`
+
+	// The readable body payload to send to S3.
+	Body io.Reader
+}
+
+// UploadOutput represents a response from the Upload() call.
+type UploadOutput struct {
+	// The URL where the object was uploaded to.
+	Location string
+
+	// The version of the object that was uploaded. Will only be populated if
+	// the S3 Bucket is versioned. If the bucket is not versioned this field
+	// will not be set.
+	VersionID *string
+
+	// The ID for a multipart upload to S3. In the case of an error the error
+	// can be cast to the MultiUploadFailure interface to extract the upload ID.
+	UploadID string
+}
+
+// WithUploaderRequestOptions appends to the Uploader's API request options.
+func WithUploaderRequestOptions(opts ...request.Option) func(*Uploader) {
+	return func(u *Uploader) {
+		u.RequestOptions = append(u.RequestOptions, opts...)
+	}
+}
+
+// The Uploader structure that calls Upload(). It is safe to call Upload()
+// on this structure for multiple objects and across concurrent goroutines.
+// Mutating the Uploader's properties is not safe to be done concurrently.
+type Uploader struct {
+	// The buffer size (in bytes) to use when buffering data into chunks and
+	// sending them as parts to S3. The minimum allowed part size is 5MB, and
+	// if this value is set to zero, the DefaultUploadPartSize value will be used.
+	PartSize int64
+
+	// The number of goroutines to spin up in parallel when sending parts.
+	// 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
+	// AbortMultipartUpload on a failure, leaving all successfully uploaded
+	// parts on S3 for manual recovery.
+	//
+	// Note that storing parts of an incomplete multipart upload counts towards
+	// space usage on S3 and will add additional costs if not cleaned up.
+	LeavePartsOnError bool
+
+	// MaxUploadParts is the max number of parts which will be uploaded to S3.
+	// Will be used to calculate the partsize of the object to be uploaded.
+	// E.g: 5GB file, with MaxUploadParts set to 100, will upload the file
+	// as 100, 50MB parts.
+	// With a limited of s3.MaxUploadParts (10,000 parts).
+	MaxUploadParts int
+
+	// The client to use when uploading to S3.
+	S3 s3iface.S3API
+
+	// List of request options that will be passed down to individual API
+	// operation requests made by the uploader.
+	RequestOptions []request.Option
+}
+
+// NewUploader creates a new Uploader instance to upload objects to S3. Pass In
+// additional functional options to customize the uploader's behavior. Requires a
+// client.ConfigProvider in order to create a S3 service client. The session.Session
+// satisfies the client.ConfigProvider interface.
+//
+// Example:
+//     // The session the S3 Uploader will use
+//     sess := session.Must(session.NewSession())
+//
+//     // Create an uploader with the session and default options
+//     uploader := s3manager.NewUploader(sess)
+//
+//     // Create an uploader with the session and custom options
+//     uploader := s3manager.NewUploader(session, func(u *s3manager.Uploader) {
+//          u.PartSize = 64 * 1024 * 1024 // 64MB per part
+//     })
+func NewUploader(c client.ConfigProvider, options ...func(*Uploader)) *Uploader {
+	u := &Uploader{
+		S3:                s3.New(c),
+		PartSize:          DefaultUploadPartSize,
+		Concurrency:       DefaultUploadConcurrency,
+		LeavePartsOnError: false,
+		MaxUploadParts:    MaxUploadParts,
+	}
+
+	for _, option := range options {
+		option(u)
+	}
+
+	return u
+}
+
+// NewUploaderWithClient creates a new Uploader instance to upload objects to S3. Pass in
+// additional functional options to customize the uploader's behavior. Requires
+// a S3 service client to make S3 API calls.
+//
+// Example:
+//     // The session the S3 Uploader will use
+//     sess := session.Must(session.NewSession())
+//
+//     // S3 service client the Upload manager will use.
+//     s3Svc := s3.New(sess)
+//
+//     // Create an uploader with S3 client and default options
+//     uploader := s3manager.NewUploaderWithClient(s3Svc)
+//
+//     // Create an uploader with S3 client and custom options
+//     uploader := s3manager.NewUploaderWithClient(s3Svc, func(u *s3manager.Uploader) {
+//          u.PartSize = 64 * 1024 * 1024 // 64MB per part
+//     })
+func NewUploaderWithClient(svc s3iface.S3API, options ...func(*Uploader)) *Uploader {
+	u := &Uploader{
+		S3:                svc,
+		PartSize:          DefaultUploadPartSize,
+		Concurrency:       DefaultUploadConcurrency,
+		LeavePartsOnError: false,
+		MaxUploadParts:    MaxUploadParts,
+	}
+
+	for _, option := range options {
+		option(u)
+	}
+
+	return u
+}
+
+// Upload uploads an object to S3, intelligently buffering large files into
+// smaller chunks and sending them in parallel across multiple goroutines. You
+// can configure the buffer size and concurrency through the Uploader's parameters.
+//
+// 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.
+//
+// Use the WithUploaderRequestOptions helper function to pass in request
+// options that will be applied to all API operations made with this uploader.
+//
+// It is safe to call this method concurrently across goroutines.
+//
+// Example:
+//     // Upload input parameters
+//     upParams := &s3manager.UploadInput{
+//         Bucket: &bucketName,
+//         Key:    &keyName,
+//         Body:   file,
+//     }
+//
+//     // Perform an upload.
+//     result, err := uploader.Upload(upParams)
+//
+//     // 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    // Don't delete the parts if the upload fails.
+//     })
+func (u Uploader) Upload(input *UploadInput, options ...func(*Uploader)) (*UploadOutput, error) {
+	return u.UploadWithContext(aws.BackgroundContext(), input, options...)
+}
+
+// UploadWithContext uploads an object to S3, intelligently buffering large
+// files into smaller chunks and sending them in parallel across multiple
+// goroutines. You can configure the buffer size and concurrency through the
+// Uploader's parameters.
+//
+// UploadWithContext is the same as Upload with the additional support for
+// Context input parameters. The Context must not be nil. A nil Context will
+// cause a panic. Use the context to add deadlining, timeouts, ect. The
+// UploadWithContext may create sub-contexts for individual underlying 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.
+//
+// Use the WithUploaderRequestOptions helper function to pass in request
+// options that will be applied to all API operations made with this uploader.
+//
+// It is safe to call this method concurrently across goroutines.
+func (u Uploader) UploadWithContext(ctx aws.Context, input *UploadInput, opts ...func(*Uploader)) (*UploadOutput, error) {
+	i := uploader{in: input, cfg: u, ctx: ctx}
+
+	for _, opt := range opts {
+		opt(&i.cfg)
+	}
+	i.cfg.RequestOptions = append(i.cfg.RequestOptions, request.WithAppendUserAgent("S3Manager"))
+
+	return i.upload()
+}
+
+// UploadWithIterator will upload a batched amount of objects to S3. This operation uses
+// the iterator pattern to know which object to upload next. Since this is an interface this
+// allows for custom defined functionality.
+//
+// Example:
+//	svc:= s3manager.NewUploader(sess)
+//
+//	objects := []BatchUploadObject{
+//		{
+//			Object:	&s3manager.UploadInput {
+//				Key: aws.String("key"),
+//				Bucket: aws.String("bucket"),
+//			},
+//		},
+//	}
+//
+//	iter := &s3managee.UploadObjectsIterator{Objects: objects}
+//	if err := svc.UploadWithIterator(aws.BackgroundContext(), iter); err != nil {
+//		return err
+//	}
+func (u Uploader) UploadWithIterator(ctx aws.Context, iter BatchUploadIterator, opts ...func(*Uploader)) error {
+	var errs []Error
+	for iter.Next() {
+		object := iter.UploadObject()
+		if _, err := u.UploadWithContext(ctx, object.Object, opts...); err != nil {
+			s3Err := Error{
+				OrigErr: err,
+				Bucket:  object.Object.Bucket,
+				Key:     object.Object.Key,
+			}
+
+			errs = append(errs, s3Err)
+		}
+
+		if object.After == nil {
+			continue
+		}
+
+		if err := object.After(); err != nil {
+			s3Err := Error{
+				OrigErr: err,
+				Bucket:  object.Object.Bucket,
+				Key:     object.Object.Key,
+			}
+
+			errs = append(errs, s3Err)
+		}
+	}
+
+	if len(errs) > 0 {
+		return NewBatchError("BatchedUploadIncomplete", "some objects have failed to upload.", errs)
+	}
+	return nil
+}
+
+// internal structure to manage an upload to S3.
+type uploader struct {
+	ctx aws.Context
+	cfg Uploader
+
+	in *UploadInput
+
+	readerPos int64 // current reader position
+	totalSize int64 // set to -1 if the size is not known
+}
+
+// internal logic for deciding whether to upload a single part or use a
+// multipart upload.
+func (u *uploader) upload() (*UploadOutput, error) {
+	u.init()
+
+	if u.cfg.PartSize < MinUploadPartSize {
+		msg := fmt.Sprintf("part size must be at least %d bytes", MinUploadPartSize)
+		return nil, awserr.New("ConfigError", msg, nil)
+	}
+
+	// Do one read to determine if we have more than one part
+	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(reader)
+}
+
+// init will initialize all default options.
+func (u *uploader) init() {
+	if u.cfg.Concurrency == 0 {
+		u.cfg.Concurrency = DefaultUploadConcurrency
+	}
+	if u.cfg.PartSize == 0 {
+		u.cfg.PartSize = DefaultUploadPartSize
+	}
+
+	// Try to get the total size for some optimizations
+	u.initSize()
+}
+
+// initSize tries to detect the total stream size, setting u.totalSize. If
+// the size is not known, totalSize is set to -1.
+func (u *uploader) initSize() {
+	u.totalSize = -1
+
+	switch r := u.in.Body.(type) {
+	case io.Seeker:
+		pos, _ := r.Seek(0, 1)
+		defer r.Seek(pos, 0)
+
+		n, err := r.Seek(0, 2)
+		if err != nil {
+			return
+		}
+		u.totalSize = n
+
+		// Try to adjust partSize if it is too small and account for
+		// integer division truncation.
+		if u.totalSize/u.cfg.PartSize >= int64(u.cfg.MaxUploadParts) {
+			// Add one to the part size to account for remainders
+			// during the size calculation. e.g odd number of bytes.
+			u.cfg.PartSize = (u.totalSize / int64(u.cfg.MaxUploadParts)) + 1
+		}
+	}
+}
+
+// nextReader returns a seekable reader representing the next packet of data.
+// 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, int, error) {
+	type readerAtSeeker interface {
+		io.ReaderAt
+		io.ReadSeeker
+	}
+	switch r := u.in.Body.(type) {
+	case readerAtSeeker:
+		var err error
+
+		n := u.cfg.PartSize
+		if u.totalSize >= 0 {
+			bytesLeft := u.totalSize - u.readerPos
+
+			if bytesLeft <= u.cfg.PartSize {
+				err = io.EOF
+				n = bytesLeft
+			}
+		}
+
+		reader := io.NewSectionReader(r, u.readerPos, n)
+		u.readerPos += n
+
+		return reader, int(n), err
+
+	default:
+		part := make([]byte, u.cfg.PartSize)
+		n, err := readFillBuf(r, part)
+		u.readerPos += int64(n)
+
+		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.
+func (u *uploader) singlePart(buf io.ReadSeeker) (*UploadOutput, error) {
+	params := &s3.PutObjectInput{}
+	awsutil.Copy(params, u.in)
+	params.Body = buf
+
+	// Need to use request form because URL generated in request is
+	// used in return.
+	req, out := u.cfg.S3.PutObjectRequest(params)
+	req.SetContext(u.ctx)
+	req.ApplyOptions(u.cfg.RequestOptions...)
+	if err := req.Send(); err != nil {
+		return nil, err
+	}
+
+	url := req.HTTPRequest.URL.String()
+	return &UploadOutput{
+		Location:  url,
+		VersionID: out.VersionId,
+	}, nil
+}
+
+// internal structure to manage a specific multipart upload to S3.
+type multiuploader struct {
+	*uploader
+	wg       sync.WaitGroup
+	m        sync.Mutex
+	err      error
+	uploadID string
+	parts    completedParts
+}
+
+// keeps track of a single chunk of data being sent to S3.
+type chunk struct {
+	buf io.ReadSeeker
+	num int64
+}
+
+// completedParts is a wrapper to make parts sortable by their part number,
+// since S3 required this list to be sent in sorted order.
+type completedParts []*s3.CompletedPart
+
+func (a completedParts) Len() int           { return len(a) }
+func (a completedParts) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
+func (a completedParts) Less(i, j int) bool { return *a[i].PartNumber < *a[j].PartNumber }
+
+// upload will perform a multipart upload using the firstBuf buffer containing
+// the first chunk of data.
+func (u *multiuploader) upload(firstBuf io.ReadSeeker) (*UploadOutput, error) {
+	params := &s3.CreateMultipartUploadInput{}
+	awsutil.Copy(params, u.in)
+
+	// Create the multipart
+	resp, err := u.cfg.S3.CreateMultipartUploadWithContext(u.ctx, params, u.cfg.RequestOptions...)
+	if err != nil {
+		return nil, err
+	}
+	u.uploadID = *resp.UploadId
+
+	// Create the workers
+	ch := make(chan chunk, u.cfg.Concurrency)
+	for i := 0; i < u.cfg.Concurrency; i++ {
+		u.wg.Add(1)
+		go u.readChunk(ch)
+	}
+
+	// Send part 1 to the workers
+	var num int64 = 1
+	ch <- chunk{buf: firstBuf, num: num}
+
+	// Read and queue the rest of the parts
+	for u.geterr() == nil && err == nil {
+		num++
+		// This upload exceeded maximum number of supported parts, error now.
+		if num > int64(u.cfg.MaxUploadParts) || num > int64(MaxUploadParts) {
+			var msg string
+			if num > int64(u.cfg.MaxUploadParts) {
+				msg = fmt.Sprintf("exceeded total allowed configured MaxUploadParts (%d). Adjust PartSize to fit in this limit",
+					u.cfg.MaxUploadParts)
+			} else {
+				msg = fmt.Sprintf("exceeded total allowed S3 limit MaxUploadParts (%d). Adjust PartSize to fit in this limit",
+					MaxUploadParts)
+			}
+			u.seterr(awserr.New("TotalPartsExceeded", msg, nil))
+			break
+		}
+
+		var reader io.ReadSeeker
+		var nextChunkLen int
+		reader, nextChunkLen, err = u.nextReader()
+
+		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
+	close(ch)
+	u.wg.Wait()
+	complete := u.complete()
+
+	if err := u.geterr(); err != nil {
+		return nil, &multiUploadError{
+			awsError: awserr.New(
+				"MultipartUpload",
+				"upload multipart failed",
+				err),
+			uploadID: u.uploadID,
+		}
+	}
+	return &UploadOutput{
+		Location:  aws.StringValue(complete.Location),
+		VersionID: complete.VersionId,
+		UploadID:  u.uploadID,
+	}, nil
+}
+
+// readChunk runs in worker goroutines to pull chunks off of the ch channel
+// and send() them as UploadPart requests.
+func (u *multiuploader) readChunk(ch chan chunk) {
+	defer u.wg.Done()
+	for {
+		data, ok := <-ch
+
+		if !ok {
+			break
+		}
+
+		if u.geterr() == nil {
+			if err := u.send(data); err != nil {
+				u.seterr(err)
+			}
+		}
+	}
+}
+
+// send performs an UploadPart request and keeps track of the completed
+// part information.
+func (u *multiuploader) send(c chunk) error {
+	params := &s3.UploadPartInput{
+		Bucket:               u.in.Bucket,
+		Key:                  u.in.Key,
+		Body:                 c.buf,
+		UploadId:             &u.uploadID,
+		SSECustomerAlgorithm: u.in.SSECustomerAlgorithm,
+		SSECustomerKey:       u.in.SSECustomerKey,
+		PartNumber:           &c.num,
+	}
+	resp, err := u.cfg.S3.UploadPartWithContext(u.ctx, params, u.cfg.RequestOptions...)
+	if err != nil {
+		return err
+	}
+
+	n := c.num
+	completed := &s3.CompletedPart{ETag: resp.ETag, PartNumber: &n}
+
+	u.m.Lock()
+	u.parts = append(u.parts, completed)
+	u.m.Unlock()
+
+	return nil
+}
+
+// geterr is a thread-safe getter for the error object
+func (u *multiuploader) geterr() error {
+	u.m.Lock()
+	defer u.m.Unlock()
+
+	return u.err
+}
+
+// seterr is a thread-safe setter for the error object
+func (u *multiuploader) seterr(e error) {
+	u.m.Lock()
+	defer u.m.Unlock()
+
+	u.err = e
+}
+
+// fail will abort the multipart unless LeavePartsOnError is set to true.
+func (u *multiuploader) fail() {
+	if u.cfg.LeavePartsOnError {
+		return
+	}
+
+	params := &s3.AbortMultipartUploadInput{
+		Bucket:   u.in.Bucket,
+		Key:      u.in.Key,
+		UploadId: &u.uploadID,
+	}
+	_, err := u.cfg.S3.AbortMultipartUploadWithContext(u.ctx, params, u.cfg.RequestOptions...)
+	if err != nil {
+		logMessage(u.cfg.S3, aws.LogDebug, fmt.Sprintf("failed to abort multipart upload, %v", err))
+	}
+}
+
+// complete successfully completes a multipart upload and returns the response.
+func (u *multiuploader) complete() *s3.CompleteMultipartUploadOutput {
+	if u.geterr() != nil {
+		u.fail()
+		return nil
+	}
+
+	// Parts must be sorted in PartNumber order.
+	sort.Sort(u.parts)
+
+	params := &s3.CompleteMultipartUploadInput{
+		Bucket:          u.in.Bucket,
+		Key:             u.in.Key,
+		UploadId:        &u.uploadID,
+		MultipartUpload: &s3.CompletedMultipartUpload{Parts: u.parts},
+	}
+	resp, err := u.cfg.S3.CompleteMultipartUploadWithContext(u.ctx, params, u.cfg.RequestOptions...)
+	if err != nil {
+		u.seterr(err)
+		u.fail()
+	}
+
+	return resp
+}