mirror of https://github.com/grafana/grafana
ImgUploader: add support for non-amazon S3 (#20354)
* imguploader: add support for non-Amazon S3 endpoints and forcing of path-style S3 addressing fixes #11240 Co-authored-by: Arve Knudsen <arve.knudsen@gmail.com>pull/21333/head
Paul Emmerich
6 years ago
committed by
Arve Knudsen
parent
85072b78a4
commit
42032f6c03
@ -0,0 +1,443 @@ |
||||
// 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) |
||||
|
||||
DeleteBucketEncryption(*s3.DeleteBucketEncryptionInput) (*s3.DeleteBucketEncryptionOutput, error) |
||||
DeleteBucketEncryptionWithContext(aws.Context, *s3.DeleteBucketEncryptionInput, ...request.Option) (*s3.DeleteBucketEncryptionOutput, error) |
||||
DeleteBucketEncryptionRequest(*s3.DeleteBucketEncryptionInput) (*request.Request, *s3.DeleteBucketEncryptionOutput) |
||||
|
||||
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) |
||||
|
||||
DeletePublicAccessBlock(*s3.DeletePublicAccessBlockInput) (*s3.DeletePublicAccessBlockOutput, error) |
||||
DeletePublicAccessBlockWithContext(aws.Context, *s3.DeletePublicAccessBlockInput, ...request.Option) (*s3.DeletePublicAccessBlockOutput, error) |
||||
DeletePublicAccessBlockRequest(*s3.DeletePublicAccessBlockInput) (*request.Request, *s3.DeletePublicAccessBlockOutput) |
||||
|
||||
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) |
||||
|
||||
GetBucketEncryption(*s3.GetBucketEncryptionInput) (*s3.GetBucketEncryptionOutput, error) |
||||
GetBucketEncryptionWithContext(aws.Context, *s3.GetBucketEncryptionInput, ...request.Option) (*s3.GetBucketEncryptionOutput, error) |
||||
GetBucketEncryptionRequest(*s3.GetBucketEncryptionInput) (*request.Request, *s3.GetBucketEncryptionOutput) |
||||
|
||||
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) |
||||
|
||||
GetBucketPolicyStatus(*s3.GetBucketPolicyStatusInput) (*s3.GetBucketPolicyStatusOutput, error) |
||||
GetBucketPolicyStatusWithContext(aws.Context, *s3.GetBucketPolicyStatusInput, ...request.Option) (*s3.GetBucketPolicyStatusOutput, error) |
||||
GetBucketPolicyStatusRequest(*s3.GetBucketPolicyStatusInput) (*request.Request, *s3.GetBucketPolicyStatusOutput) |
||||
|
||||
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) |
||||
|
||||
GetObjectLegalHold(*s3.GetObjectLegalHoldInput) (*s3.GetObjectLegalHoldOutput, error) |
||||
GetObjectLegalHoldWithContext(aws.Context, *s3.GetObjectLegalHoldInput, ...request.Option) (*s3.GetObjectLegalHoldOutput, error) |
||||
GetObjectLegalHoldRequest(*s3.GetObjectLegalHoldInput) (*request.Request, *s3.GetObjectLegalHoldOutput) |
||||
|
||||
GetObjectLockConfiguration(*s3.GetObjectLockConfigurationInput) (*s3.GetObjectLockConfigurationOutput, error) |
||||
GetObjectLockConfigurationWithContext(aws.Context, *s3.GetObjectLockConfigurationInput, ...request.Option) (*s3.GetObjectLockConfigurationOutput, error) |
||||
GetObjectLockConfigurationRequest(*s3.GetObjectLockConfigurationInput) (*request.Request, *s3.GetObjectLockConfigurationOutput) |
||||
|
||||
GetObjectRetention(*s3.GetObjectRetentionInput) (*s3.GetObjectRetentionOutput, error) |
||||
GetObjectRetentionWithContext(aws.Context, *s3.GetObjectRetentionInput, ...request.Option) (*s3.GetObjectRetentionOutput, error) |
||||
GetObjectRetentionRequest(*s3.GetObjectRetentionInput) (*request.Request, *s3.GetObjectRetentionOutput) |
||||
|
||||
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) |
||||
|
||||
GetPublicAccessBlock(*s3.GetPublicAccessBlockInput) (*s3.GetPublicAccessBlockOutput, error) |
||||
GetPublicAccessBlockWithContext(aws.Context, *s3.GetPublicAccessBlockInput, ...request.Option) (*s3.GetPublicAccessBlockOutput, error) |
||||
GetPublicAccessBlockRequest(*s3.GetPublicAccessBlockInput) (*request.Request, *s3.GetPublicAccessBlockOutput) |
||||
|
||||
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) |
||||
|
||||
PutBucketEncryption(*s3.PutBucketEncryptionInput) (*s3.PutBucketEncryptionOutput, error) |
||||
PutBucketEncryptionWithContext(aws.Context, *s3.PutBucketEncryptionInput, ...request.Option) (*s3.PutBucketEncryptionOutput, error) |
||||
PutBucketEncryptionRequest(*s3.PutBucketEncryptionInput) (*request.Request, *s3.PutBucketEncryptionOutput) |
||||
|
||||
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) |
||||
|
||||
PutObjectLegalHold(*s3.PutObjectLegalHoldInput) (*s3.PutObjectLegalHoldOutput, error) |
||||
PutObjectLegalHoldWithContext(aws.Context, *s3.PutObjectLegalHoldInput, ...request.Option) (*s3.PutObjectLegalHoldOutput, error) |
||||
PutObjectLegalHoldRequest(*s3.PutObjectLegalHoldInput) (*request.Request, *s3.PutObjectLegalHoldOutput) |
||||
|
||||
PutObjectLockConfiguration(*s3.PutObjectLockConfigurationInput) (*s3.PutObjectLockConfigurationOutput, error) |
||||
PutObjectLockConfigurationWithContext(aws.Context, *s3.PutObjectLockConfigurationInput, ...request.Option) (*s3.PutObjectLockConfigurationOutput, error) |
||||
PutObjectLockConfigurationRequest(*s3.PutObjectLockConfigurationInput) (*request.Request, *s3.PutObjectLockConfigurationOutput) |
||||
|
||||
PutObjectRetention(*s3.PutObjectRetentionInput) (*s3.PutObjectRetentionOutput, error) |
||||
PutObjectRetentionWithContext(aws.Context, *s3.PutObjectRetentionInput, ...request.Option) (*s3.PutObjectRetentionOutput, error) |
||||
PutObjectRetentionRequest(*s3.PutObjectRetentionInput) (*request.Request, *s3.PutObjectRetentionOutput) |
||||
|
||||
PutObjectTagging(*s3.PutObjectTaggingInput) (*s3.PutObjectTaggingOutput, error) |
||||
PutObjectTaggingWithContext(aws.Context, *s3.PutObjectTaggingInput, ...request.Option) (*s3.PutObjectTaggingOutput, error) |
||||
PutObjectTaggingRequest(*s3.PutObjectTaggingInput) (*request.Request, *s3.PutObjectTaggingOutput) |
||||
|
||||
PutPublicAccessBlock(*s3.PutPublicAccessBlockInput) (*s3.PutPublicAccessBlockOutput, error) |
||||
PutPublicAccessBlockWithContext(aws.Context, *s3.PutPublicAccessBlockInput, ...request.Option) (*s3.PutPublicAccessBlockOutput, error) |
||||
PutPublicAccessBlockRequest(*s3.PutPublicAccessBlockInput) (*request.Request, *s3.PutPublicAccessBlockOutput) |
||||
|
||||
RestoreObject(*s3.RestoreObjectInput) (*s3.RestoreObjectOutput, error) |
||||
RestoreObjectWithContext(aws.Context, *s3.RestoreObjectInput, ...request.Option) (*s3.RestoreObjectOutput, error) |
||||
RestoreObjectRequest(*s3.RestoreObjectInput) (*request.Request, *s3.RestoreObjectOutput) |
||||
|
||||
SelectObjectContent(*s3.SelectObjectContentInput) (*s3.SelectObjectContentOutput, error) |
||||
SelectObjectContentWithContext(aws.Context, *s3.SelectObjectContentInput, ...request.Option) (*s3.SelectObjectContentOutput, error) |
||||
SelectObjectContentRequest(*s3.SelectObjectContentInput) (*request.Request, *s3.SelectObjectContentOutput) |
||||
|
||||
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) |
||||
@ -0,0 +1,529 @@ |
||||
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 { |
||||
origErr := "" |
||||
if err.OrigErr != nil { |
||||
origErr = ":\n" + err.OrigErr.Error() |
||||
} |
||||
return fmt.Sprintf("failed to perform batch operation on %q to %q%s", |
||||
aws.StringValue(err.Key), |
||||
aws.StringValue(err.Bucket), |
||||
origErr, |
||||
) |
||||
} |
||||
|
||||
// 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(aws.BackgroundContext(), &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(aws.BackgroundContext(), &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 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(ctx, 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, |
||||
}) |
||||
} |
||||
} |
||||
} |
||||
|
||||
// iter.Next() could return false (above) plus populate iter.Err()
|
||||
if iter.Err() != nil { |
||||
errs = append(errs, newError(iter.Err(), nil, nil)) |
||||
} |
||||
|
||||
if input != nil && len(input.Delete.Objects) > 0 { |
||||
if err := deleteBatch(ctx, 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{}, |
||||
} |
||||
} |
||||
|
||||
const ( |
||||
// ErrDeleteBatchFailCode represents an error code which will be returned
|
||||
// only when DeleteObjects.Errors has an error that does not contain a code.
|
||||
ErrDeleteBatchFailCode = "DeleteBatchError" |
||||
errDefaultDeleteBatchMessage = "failed to delete" |
||||
) |
||||
|
||||
// deleteBatch will delete a batch of items in the objects parameters.
|
||||
func deleteBatch(ctx aws.Context, d *BatchDelete, input *s3.DeleteObjectsInput, objects []BatchDeleteObject) []Error { |
||||
errs := []Error{} |
||||
|
||||
if result, err := d.Client.DeleteObjectsWithContext(ctx, 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++ { |
||||
code := ErrDeleteBatchFailCode |
||||
msg := errDefaultDeleteBatchMessage |
||||
if result.Errors[i].Message != nil { |
||||
msg = *result.Errors[i].Message |
||||
} |
||||
if result.Errors[i].Code != nil { |
||||
code = *result.Errors[i].Code |
||||
} |
||||
|
||||
errs = append(errs, newError(awserr.New(code, msg, 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 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 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 |
||||
} |
||||
@ -0,0 +1,88 @@ |
||||
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. If the regionHint parameter is an
|
||||
// empty string GetBucketRegion will fallback to the ConfigProvider's region
|
||||
// config. If the regionHint is empty, and the ConfigProvider does not have a
|
||||
// region value, an error will be returned..
|
||||
//
|
||||
// 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) { |
||||
var cfg aws.Config |
||||
if len(regionHint) != 0 { |
||||
cfg.Region = aws.String(regionHint) |
||||
} |
||||
svc := s3.New(c, &cfg) |
||||
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 |
||||
} |
||||
@ -0,0 +1,81 @@ |
||||
package s3manager |
||||
|
||||
import ( |
||||
"io" |
||||
|
||||
"github.com/aws/aws-sdk-go/internal/sdkio" |
||||
) |
||||
|
||||
// BufferedReadSeeker is buffered io.ReadSeeker
|
||||
type BufferedReadSeeker struct { |
||||
r io.ReadSeeker |
||||
buffer []byte |
||||
readIdx, writeIdx int |
||||
} |
||||
|
||||
// NewBufferedReadSeeker returns a new BufferedReadSeeker
|
||||
// if len(b) == 0 then the buffer will be initialized to 64 KiB.
|
||||
func NewBufferedReadSeeker(r io.ReadSeeker, b []byte) *BufferedReadSeeker { |
||||
if len(b) == 0 { |
||||
b = make([]byte, 64*1024) |
||||
} |
||||
return &BufferedReadSeeker{r: r, buffer: b} |
||||
} |
||||
|
||||
func (b *BufferedReadSeeker) reset(r io.ReadSeeker) { |
||||
b.r = r |
||||
b.readIdx, b.writeIdx = 0, 0 |
||||
} |
||||
|
||||
// Read will read up len(p) bytes into p and will return
|
||||
// the number of bytes read and any error that occurred.
|
||||
// If the len(p) > the buffer size then a single read request
|
||||
// will be issued to the underlying io.ReadSeeker for len(p) bytes.
|
||||
// A Read request will at most perform a single Read to the underlying
|
||||
// io.ReadSeeker, and may return < len(p) if serviced from the buffer.
|
||||
func (b *BufferedReadSeeker) Read(p []byte) (n int, err error) { |
||||
if len(p) == 0 { |
||||
return n, err |
||||
} |
||||
|
||||
if b.readIdx == b.writeIdx { |
||||
if len(p) >= len(b.buffer) { |
||||
n, err = b.r.Read(p) |
||||
return n, err |
||||
} |
||||
b.readIdx, b.writeIdx = 0, 0 |
||||
|
||||
n, err = b.r.Read(b.buffer) |
||||
if n == 0 { |
||||
return n, err |
||||
} |
||||
|
||||
b.writeIdx += n |
||||
} |
||||
|
||||
n = copy(p, b.buffer[b.readIdx:b.writeIdx]) |
||||
b.readIdx += n |
||||
|
||||
return n, err |
||||
} |
||||
|
||||
// Seek will position then underlying io.ReadSeeker to the given offset
|
||||
// and will clear the buffer.
|
||||
func (b *BufferedReadSeeker) Seek(offset int64, whence int) (int64, error) { |
||||
n, err := b.r.Seek(offset, whence) |
||||
|
||||
b.reset(b.r) |
||||
|
||||
return n, err |
||||
} |
||||
|
||||
// ReadAt will read up to len(p) bytes at the given file offset.
|
||||
// This will result in the buffer being cleared.
|
||||
func (b *BufferedReadSeeker) ReadAt(p []byte, off int64) (int, error) { |
||||
_, err := b.Seek(off, sdkio.SeekStart) |
||||
if err != nil { |
||||
return 0, err |
||||
} |
||||
|
||||
return b.Read(p) |
||||
} |
||||
@ -0,0 +1,7 @@ |
||||
// +build !windows
|
||||
|
||||
package s3manager |
||||
|
||||
func defaultUploadBufferProvider() ReadSeekerWriteToProvider { |
||||
return nil |
||||
} |
||||
@ -0,0 +1,5 @@ |
||||
package s3manager |
||||
|
||||
func defaultUploadBufferProvider() ReadSeekerWriteToProvider { |
||||
return NewBufferedReadSeekerWriteToPool(1024 * 1024) |
||||
} |
||||
7
vendor/github.com/aws/aws-sdk-go/service/s3/s3manager/default_writer_read_from.go
generated
vendored
7
vendor/github.com/aws/aws-sdk-go/service/s3/s3manager/default_writer_read_from.go
generated
vendored
@ -0,0 +1,7 @@ |
||||
// +build !windows
|
||||
|
||||
package s3manager |
||||
|
||||
func defaultDownloadBufferProvider() WriterReadFromProvider { |
||||
return nil |
||||
} |
||||
@ -0,0 +1,5 @@ |
||||
package s3manager |
||||
|
||||
func defaultDownloadBufferProvider() WriterReadFromProvider { |
||||
return NewPooledBufferedWriterReadFromProvider(1024 * 1024) |
||||
} |
||||
@ -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 |
||||
@ -0,0 +1,597 @@ |
||||
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 |
||||
|
||||
type errReadingBody struct { |
||||
err error |
||||
} |
||||
|
||||
func (e *errReadingBody) Error() string { |
||||
return fmt.Sprintf("failed to read part body: %v", e.err) |
||||
} |
||||
|
||||
func (e *errReadingBody) Unwrap() error { |
||||
return e.err |
||||
} |
||||
|
||||
// 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 size (in bytes) to request from S3 for each part.
|
||||
// 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 of 1 will download the parts sequentially.
|
||||
//
|
||||
// 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 |
||||
|
||||
// Defines the buffer strategy used when downloading a part.
|
||||
//
|
||||
// If a WriterReadFromProvider is given the Download manager
|
||||
// will pass the io.WriterAt of the Download request to the provider
|
||||
// and will use the returned WriterReadFrom from the provider as the
|
||||
// destination writer when copying from http response body.
|
||||
BufferProvider WriterReadFromProvider |
||||
} |
||||
|
||||
// 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 { |
||||
return newDownloader(s3.New(c), options...) |
||||
} |
||||
|
||||
func newDownloader(client s3iface.S3API, options ...func(*Downloader)) *Downloader { |
||||
d := &Downloader{ |
||||
S3: client, |
||||
PartSize: DefaultDownloadPartSize, |
||||
Concurrency: DefaultDownloadConcurrency, |
||||
BufferProvider: defaultDownloadBufferProvider(), |
||||
} |
||||
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 { |
||||
return newDownloader(svc, options...) |
||||
} |
||||
|
||||
type maxRetrier interface { |
||||
MaxRetries() int |
||||
} |
||||
|
||||
// Download downloads an object in S3 and writes the payload into w using
|
||||
// concurrent GET requests. The n int64 returned is the size of the object downloaded
|
||||
// in bytes.
|
||||
//
|
||||
// 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.
|
||||
//
|
||||
// Specifying a Downloader.Concurrency of 1 will cause the Downloader to
|
||||
// download the parts from S3 sequentially.
|
||||
//
|
||||
// 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. The n int64 returned is the size of the object downloaded
|
||||
// in bytes.
|
||||
//
|
||||
// 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, etc. 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.
|
||||
//
|
||||
// Specifying a Downloader.Concurrency of 1 will cause the Downloader to
|
||||
// download the parts from S3 sequentially.
|
||||
//
|
||||
// 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 {
|
||||
// {
|
||||
// Object: &s3.GetObjectInput {
|
||||
// Bucket: aws.String("bucket"),
|
||||
// Key: aws.String("foo"),
|
||||
// },
|
||||
// Writer: fooFile,
|
||||
// },
|
||||
// {
|
||||
// Object: &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++ { |
||||
n, err = d.tryDownloadChunk(in, &chunk) |
||||
if err == nil { |
||||
break |
||||
} |
||||
// Check if the returned error is an errReadingBody.
|
||||
// If err is errReadingBody this indicates that an error
|
||||
// occurred while copying the http response body.
|
||||
// If this occurs we unwrap the err to set the underlying error
|
||||
// and attempt any remaining retries.
|
||||
if bodyErr, ok := err.(*errReadingBody); ok { |
||||
err = bodyErr.Unwrap() |
||||
} else { |
||||
return err |
||||
} |
||||
|
||||
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 (d *downloader) tryDownloadChunk(in *s3.GetObjectInput, w io.Writer) (int64, error) { |
||||
cleanup := func() {} |
||||
if d.cfg.BufferProvider != nil { |
||||
w, cleanup = d.cfg.BufferProvider.GetReadFrom(w) |
||||
} |
||||
defer cleanup() |
||||
|
||||
resp, err := d.cfg.S3.GetObjectWithContext(d.ctx, in, d.cfg.RequestOptions...) |
||||
if err != nil { |
||||
return 0, err |
||||
} |
||||
d.setTotalBytes(resp) // Set total if not yet set.
|
||||
|
||||
n, err := io.Copy(w, resp.Body) |
||||
resp.Body.Close() |
||||
if err != nil { |
||||
return n, &errReadingBody{err: err} |
||||
} |
||||
|
||||
return n, nil |
||||
} |
||||
|
||||
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) |
||||
} |
||||
@ -0,0 +1,65 @@ |
||||
package s3manager |
||||
|
||||
import ( |
||||
"io" |
||||
"sync" |
||||
) |
||||
|
||||
// ReadSeekerWriteTo defines an interface implementing io.WriteTo and io.ReadSeeker
|
||||
type ReadSeekerWriteTo interface { |
||||
io.ReadSeeker |
||||
io.WriterTo |
||||
} |
||||
|
||||
// BufferedReadSeekerWriteTo wraps a BufferedReadSeeker with an io.WriteAt
|
||||
// implementation.
|
||||
type BufferedReadSeekerWriteTo struct { |
||||
*BufferedReadSeeker |
||||
} |
||||
|
||||
// WriteTo writes to the given io.Writer from BufferedReadSeeker until there's no more data to write or
|
||||
// an error occurs. Returns the number of bytes written and any error encountered during the write.
|
||||
func (b *BufferedReadSeekerWriteTo) WriteTo(writer io.Writer) (int64, error) { |
||||
return io.Copy(writer, b.BufferedReadSeeker) |
||||
} |
||||
|
||||
// ReadSeekerWriteToProvider provides an implementation of io.WriteTo for an io.ReadSeeker
|
||||
type ReadSeekerWriteToProvider interface { |
||||
GetWriteTo(seeker io.ReadSeeker) (r ReadSeekerWriteTo, cleanup func()) |
||||
} |
||||
|
||||
// BufferedReadSeekerWriteToPool uses a sync.Pool to create and reuse
|
||||
// []byte slices for buffering parts in memory
|
||||
type BufferedReadSeekerWriteToPool struct { |
||||
pool sync.Pool |
||||
} |
||||
|
||||
// NewBufferedReadSeekerWriteToPool will return a new BufferedReadSeekerWriteToPool that will create
|
||||
// a pool of reusable buffers . If size is less then < 64 KiB then the buffer
|
||||
// will default to 64 KiB. Reason: io.Copy from writers or readers that don't support io.WriteTo or io.ReadFrom
|
||||
// respectively will default to copying 32 KiB.
|
||||
func NewBufferedReadSeekerWriteToPool(size int) *BufferedReadSeekerWriteToPool { |
||||
if size < 65536 { |
||||
size = 65536 |
||||
} |
||||
|
||||
return &BufferedReadSeekerWriteToPool{ |
||||
pool: sync.Pool{New: func() interface{} { |
||||
return make([]byte, size) |
||||
}}, |
||||
} |
||||
} |
||||
|
||||
// GetWriteTo will wrap the provided io.ReadSeeker with a BufferedReadSeekerWriteTo.
|
||||
// The provided cleanup must be called after operations have been completed on the
|
||||
// returned io.ReadSeekerWriteTo in order to signal the return of resources to the pool.
|
||||
func (p *BufferedReadSeekerWriteToPool) GetWriteTo(seeker io.ReadSeeker) (r ReadSeekerWriteTo, cleanup func()) { |
||||
buffer := p.pool.Get().([]byte) |
||||
|
||||
r = &BufferedReadSeekerWriteTo{BufferedReadSeeker: NewBufferedReadSeeker(seeker, buffer)} |
||||
cleanup = func() { |
||||
p.pool.Put(buffer) |
||||
} |
||||
|
||||
return r, cleanup |
||||
} |
||||
@ -0,0 +1,774 @@ |
||||
package s3manager |
||||
|
||||
import ( |
||||
"bytes" |
||||
"fmt" |
||||
"io" |
||||
"sort" |
||||
"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/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" |
||||
) |
||||
|
||||
// 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 |
||||
} |
||||
|
||||
// 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 per call to Upload when
|
||||
// sending parts. If this is set to zero, the DefaultUploadConcurrency value
|
||||
// will be used.
|
||||
//
|
||||
// The concurrency pool is not shared between calls to Upload.
|
||||
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 must not be used to limit the total number of bytes uploaded.
|
||||
// Use a type like to io.LimitReader (https://golang.org/pkg/io/#LimitedReader)
|
||||
// instead. An io.LimitReader is helpful when uploading an unbounded reader
|
||||
// to S3, and you know its maximum size. Otherwise the reader's io.EOF returned
|
||||
// error must be used to signal end of stream.
|
||||
//
|
||||
// Defaults to package const's MaxUploadParts value.
|
||||
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 |
||||
|
||||
// Defines the buffer strategy used when uploading a part
|
||||
BufferProvider ReadSeekerWriteToProvider |
||||
|
||||
// partPool allows for the re-usage of streaming payload part buffers between upload calls
|
||||
partPool *partPool |
||||
} |
||||
|
||||
// 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 { |
||||
return newUploader(s3.New(c), options...) |
||||
} |
||||
|
||||
func newUploader(client s3iface.S3API, options ...func(*Uploader)) *Uploader { |
||||
u := &Uploader{ |
||||
S3: client, |
||||
PartSize: DefaultUploadPartSize, |
||||
Concurrency: DefaultUploadConcurrency, |
||||
LeavePartsOnError: false, |
||||
MaxUploadParts: MaxUploadParts, |
||||
BufferProvider: defaultUploadBufferProvider(), |
||||
} |
||||
|
||||
for _, option := range options { |
||||
option(u) |
||||
} |
||||
|
||||
u.partPool = newPartPool(u.PartSize) |
||||
|
||||
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 { |
||||
return newUploader(svc, options...) |
||||
} |
||||
|
||||
// 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, etc. 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 := &s3manager.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) { |
||||
if err := u.init(); err != nil { |
||||
return nil, awserr.New("ReadRequestBody", "unable to initialize upload", err) |
||||
} |
||||
|
||||
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, _, cleanup, err := u.nextReader() |
||||
if err == io.EOF { // single part
|
||||
return u.singlePart(reader, cleanup) |
||||
} else if err != nil { |
||||
cleanup() |
||||
return nil, awserr.New("ReadRequestBody", "read upload data failed", err) |
||||
} |
||||
|
||||
mu := multiuploader{uploader: u} |
||||
return mu.upload(reader, cleanup) |
||||
} |
||||
|
||||
// init will initialize all default options.
|
||||
func (u *uploader) init() error { |
||||
if u.cfg.Concurrency == 0 { |
||||
u.cfg.Concurrency = DefaultUploadConcurrency |
||||
} |
||||
if u.cfg.PartSize == 0 { |
||||
u.cfg.PartSize = DefaultUploadPartSize |
||||
} |
||||
if u.cfg.MaxUploadParts == 0 { |
||||
u.cfg.MaxUploadParts = MaxUploadParts |
||||
} |
||||
|
||||
// If PartSize was changed or partPool was never setup then we need to allocated a new pool
|
||||
// so that we return []byte slices of the correct size
|
||||
if u.cfg.partPool == nil || u.cfg.partPool.partSize != u.cfg.PartSize { |
||||
u.cfg.partPool = newPartPool(u.cfg.PartSize) |
||||
} |
||||
|
||||
// Try to get the total size for some optimizations
|
||||
return 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() error { |
||||
u.totalSize = -1 |
||||
|
||||
switch r := u.in.Body.(type) { |
||||
case io.Seeker: |
||||
n, err := aws.SeekerLen(r) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
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 |
||||
} |
||||
} |
||||
|
||||
return nil |
||||
} |
||||
|
||||
// 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, func(), 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 |
||||
} |
||||
} |
||||
|
||||
var ( |
||||
reader io.ReadSeeker |
||||
cleanup func() |
||||
) |
||||
|
||||
reader = io.NewSectionReader(r, u.readerPos, n) |
||||
if u.cfg.BufferProvider != nil { |
||||
reader, cleanup = u.cfg.BufferProvider.GetWriteTo(reader) |
||||
} else { |
||||
cleanup = func() {} |
||||
} |
||||
|
||||
u.readerPos += n |
||||
|
||||
return reader, int(n), cleanup, err |
||||
|
||||
default: |
||||
part := u.cfg.partPool.Get().([]byte) |
||||
n, err := readFillBuf(r, part) |
||||
u.readerPos += int64(n) |
||||
|
||||
cleanup := func() { |
||||
u.cfg.partPool.Put(part) |
||||
} |
||||
|
||||
return bytes.NewReader(part[0:n]), n, cleanup, 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(r io.ReadSeeker, cleanup func()) (*UploadOutput, error) { |
||||
defer cleanup() |
||||
|
||||
params := &s3.PutObjectInput{} |
||||
awsutil.Copy(params, u.in) |
||||
params.Body = r |
||||
|
||||
// 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 |
||||
cleanup func() |
||||
} |
||||
|
||||
// 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, cleanup func()) (*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, cleanup: cleanup} |
||||
|
||||
// Read and queue the rest of the parts
|
||||
for u.geterr() == nil && err == nil { |
||||
var ( |
||||
reader io.ReadSeeker |
||||
nextChunkLen int |
||||
ok bool |
||||
) |
||||
|
||||
reader, nextChunkLen, cleanup, err = u.nextReader() |
||||
ok, err = u.shouldContinue(num, nextChunkLen, err) |
||||
if !ok { |
||||
cleanup() |
||||
if err != nil { |
||||
u.seterr(err) |
||||
} |
||||
break |
||||
} |
||||
|
||||
num++ |
||||
|
||||
ch <- chunk{buf: reader, num: num, cleanup: cleanup} |
||||
} |
||||
|
||||
// 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, |
||||
} |
||||
} |
||||
|
||||
// Create a presigned URL of the S3 Get Object in order to have parity with
|
||||
// single part upload.
|
||||
getReq, _ := u.cfg.S3.GetObjectRequest(&s3.GetObjectInput{ |
||||
Bucket: u.in.Bucket, |
||||
Key: u.in.Key, |
||||
}) |
||||
getReq.Config.Credentials = credentials.AnonymousCredentials |
||||
uploadLocation, _, _ := getReq.PresignRequest(1) |
||||
|
||||
return &UploadOutput{ |
||||
Location: uploadLocation, |
||||
VersionID: complete.VersionId, |
||||
UploadID: u.uploadID, |
||||
}, nil |
||||
} |
||||
|
||||
func (u *multiuploader) shouldContinue(part int64, nextChunkLen int, err error) (bool, error) { |
||||
if err != nil && err != io.EOF { |
||||
return false, awserr.New("ReadRequestBody", "read multipart upload data failed", err) |
||||
} |
||||
|
||||
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.
|
||||
return false, nil |
||||
} |
||||
|
||||
part++ |
||||
// This upload exceeded maximum number of supported parts, error now.
|
||||
if part > int64(u.cfg.MaxUploadParts) || part > int64(MaxUploadParts) { |
||||
var msg string |
||||
if part > 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) |
||||
} |
||||
return false, awserr.New("TotalPartsExceeded", msg, nil) |
||||
} |
||||
|
||||
return true, err |
||||
} |
||||
|
||||
// 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...) |
||||
c.cleanup() |
||||
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 |
||||
} |
||||
|
||||
type partPool struct { |
||||
partSize int64 |
||||
sync.Pool |
||||
} |
||||
|
||||
func newPartPool(partSize int64) *partPool { |
||||
p := &partPool{partSize: partSize} |
||||
|
||||
p.New = func() interface{} { |
||||
return make([]byte, p.partSize) |
||||
} |
||||
|
||||
return p |
||||
} |
||||
@ -0,0 +1,75 @@ |
||||
package s3manager |
||||
|
||||
import ( |
||||
"bufio" |
||||
"io" |
||||
"sync" |
||||
|
||||
"github.com/aws/aws-sdk-go/internal/sdkio" |
||||
) |
||||
|
||||
// WriterReadFrom defines an interface implementing io.Writer and io.ReaderFrom
|
||||
type WriterReadFrom interface { |
||||
io.Writer |
||||
io.ReaderFrom |
||||
} |
||||
|
||||
// WriterReadFromProvider provides an implementation of io.ReadFrom for the given io.Writer
|
||||
type WriterReadFromProvider interface { |
||||
GetReadFrom(writer io.Writer) (w WriterReadFrom, cleanup func()) |
||||
} |
||||
|
||||
type bufferedWriter interface { |
||||
WriterReadFrom |
||||
Flush() error |
||||
Reset(io.Writer) |
||||
} |
||||
|
||||
type bufferedReadFrom struct { |
||||
bufferedWriter |
||||
} |
||||
|
||||
func (b *bufferedReadFrom) ReadFrom(r io.Reader) (int64, error) { |
||||
n, err := b.bufferedWriter.ReadFrom(r) |
||||
if flushErr := b.Flush(); flushErr != nil && err == nil { |
||||
err = flushErr |
||||
} |
||||
return n, err |
||||
} |
||||
|
||||
// PooledBufferedReadFromProvider is a WriterReadFromProvider that uses a sync.Pool
|
||||
// to manage allocation and reuse of *bufio.Writer structures.
|
||||
type PooledBufferedReadFromProvider struct { |
||||
pool sync.Pool |
||||
} |
||||
|
||||
// NewPooledBufferedWriterReadFromProvider returns a new PooledBufferedReadFromProvider
|
||||
// Size is used to control the size of the underlying *bufio.Writer created for
|
||||
// calls to GetReadFrom.
|
||||
func NewPooledBufferedWriterReadFromProvider(size int) *PooledBufferedReadFromProvider { |
||||
if size < int(32*sdkio.KibiByte) { |
||||
size = int(64 * sdkio.KibiByte) |
||||
} |
||||
|
||||
return &PooledBufferedReadFromProvider{ |
||||
pool: sync.Pool{ |
||||
New: func() interface{} { |
||||
return &bufferedReadFrom{bufferedWriter: bufio.NewWriterSize(nil, size)} |
||||
}, |
||||
}, |
||||
} |
||||
} |
||||
|
||||
// GetReadFrom takes an io.Writer and wraps it with a type which satisfies the WriterReadFrom
|
||||
// interface/ Additionally a cleanup function is provided which must be called after usage of the WriterReadFrom
|
||||
// has been completed in order to allow the reuse of the *bufio.Writer
|
||||
func (p *PooledBufferedReadFromProvider) GetReadFrom(writer io.Writer) (r WriterReadFrom, cleanup func()) { |
||||
buffer := p.pool.Get().(*bufferedReadFrom) |
||||
buffer.Reset(writer) |
||||
r = buffer |
||||
cleanup = func() { |
||||
buffer.Reset(nil) // Reset to nil writer to release reference
|
||||
p.pool.Put(buffer) |
||||
} |
||||
return r, cleanup |
||||
} |
||||
Loading…
Reference in new issue