2014-12-22 23:24:45 +01:00
// Package s3 provides a storagedriver.StorageDriver implementation to
// store blobs in Amazon S3 cloud storage.
//
2016-01-22 03:17:53 +01:00
// This package leverages the official aws client library for interfacing with
// S3.
2014-12-22 23:24:45 +01:00
//
2016-01-22 03:17:53 +01:00
// Because S3 is a key, value store the Stat call does not support last modification
2014-12-22 23:24:45 +01:00
// time for directories (directories are an abstraction for key, value stores)
//
2016-01-22 03:17:53 +01:00
// Keep in mind that S3 guarantees only read-after-write consistency for new
// objects, but no read-after-update or list-after-write consistency.
2014-10-25 01:37:25 +02:00
package s3
import (
"bytes"
2014-10-29 02:15:40 +01:00
"fmt"
2014-10-25 01:37:25 +02:00
"io"
2014-12-19 18:16:51 +01:00
"io/ioutil"
2014-10-25 01:37:25 +02:00
"net/http"
2015-04-22 23:31:34 +02:00
"reflect"
2016-04-23 12:13:15 +02:00
"sort"
2014-10-25 01:37:25 +02:00
"strconv"
2014-12-19 18:16:51 +01:00
"strings"
"time"
2014-10-25 01:37:25 +02:00
2016-01-22 03:17:53 +01:00
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/awserr"
"github.com/aws/aws-sdk-go/aws/credentials"
"github.com/aws/aws-sdk-go/aws/credentials/ec2rolecreds"
"github.com/aws/aws-sdk-go/aws/ec2metadata"
"github.com/aws/aws-sdk-go/aws/request"
"github.com/aws/aws-sdk-go/aws/session"
"github.com/aws/aws-sdk-go/service/s3"
2015-04-28 00:58:58 +02:00
"github.com/docker/distribution/context"
2016-01-21 01:40:58 +01:00
"github.com/docker/distribution/registry/client/transport"
2015-02-11 03:14:23 +01:00
storagedriver "github.com/docker/distribution/registry/storage/driver"
"github.com/docker/distribution/registry/storage/driver/base"
"github.com/docker/distribution/registry/storage/driver/factory"
2014-10-25 01:37:25 +02:00
)
2016-01-22 03:17:53 +01:00
const driverName = "s3aws"
2014-10-29 02:15:40 +01:00
2014-10-29 20:14:19 +01:00
// minChunkSize defines the minimum multipart upload chunk size
// S3 API requires multipart upload chunks to be at least 5MB
2015-01-24 01:46:43 +01:00
const minChunkSize = 5 << 20
const defaultChunkSize = 2 * minChunkSize
2014-10-25 01:37:25 +02:00
2014-12-19 18:16:51 +01:00
// listMax is the largest amount of objects you can request from S3 in a list call
const listMax = 1000
2014-10-25 01:37:25 +02:00
2016-01-22 03:17:53 +01:00
// validRegions maps known s3 region identifiers to region descriptors
var validRegions = map [ string ] struct { } { }
2015-01-07 11:18:42 +01:00
//DriverParameters A struct that encapsulates all of the driver parameters after all values have been set
type DriverParameters struct {
2016-03-05 19:46:44 +01:00
AccessKey string
SecretKey string
Bucket string
Region string
RegionEndpoint string
Encrypt bool
2016-03-10 01:52:59 +01:00
KeyID string
2016-03-05 19:46:44 +01:00
Secure bool
ChunkSize int64
RootDirectory string
StorageClass string
UserAgent string
2015-01-07 11:18:42 +01:00
}
2014-10-29 02:15:40 +01:00
func init ( ) {
2016-01-22 03:17:53 +01:00
for _ , region := range [ ] string {
"us-east-1" ,
"us-west-1" ,
"us-west-2" ,
"eu-west-1" ,
"eu-central-1" ,
"ap-southeast-1" ,
"ap-southeast-2" ,
"ap-northeast-1" ,
"ap-northeast-2" ,
"sa-east-1" ,
2016-04-21 05:51:34 +02:00
"cn-north-1" ,
2016-01-22 03:17:53 +01:00
} {
validRegions [ region ] = struct { } { }
}
// Register this as the default s3 driver in addition to s3aws
factory . Register ( "s3" , & s3DriverFactory { } )
2014-11-18 00:44:07 +01:00
factory . Register ( driverName , & s3DriverFactory { } )
2014-10-29 02:15:40 +01:00
}
2014-10-29 20:14:19 +01:00
// s3DriverFactory implements the factory.StorageDriverFactory interface
2014-10-29 02:15:40 +01:00
type s3DriverFactory struct { }
2014-12-19 18:16:51 +01:00
func ( factory * s3DriverFactory ) Create ( parameters map [ string ] interface { } ) ( storagedriver . StorageDriver , error ) {
2014-10-29 02:15:40 +01:00
return FromParameters ( parameters )
}
2015-02-04 01:54:52 +01:00
type driver struct {
2014-12-19 18:16:51 +01:00
S3 * s3 . S3
2016-01-22 03:17:53 +01:00
Bucket string
2015-01-24 01:46:43 +01:00
ChunkSize int64
2014-12-19 18:16:51 +01:00
Encrypt bool
2016-03-10 01:52:59 +01:00
KeyID string
2015-01-24 01:46:43 +01:00
RootDirectory string
2016-01-22 03:17:53 +01:00
StorageClass string
2014-10-25 01:37:25 +02:00
}
2015-02-04 01:54:52 +01:00
type baseEmbed struct {
base . Base
}
// Driver is a storagedriver.StorageDriver implementation backed by Amazon S3
// Objects are stored at absolute keys in the provided bucket.
type Driver struct {
baseEmbed
}
2014-11-18 00:44:07 +01:00
// FromParameters constructs a new Driver with a given parameters map
2014-10-29 02:15:40 +01:00
// Required parameters:
// - accesskey
// - secretkey
// - region
// - bucket
// - encrypt
2014-12-19 18:16:51 +01:00
func FromParameters ( parameters map [ string ] interface { } ) ( * Driver , error ) {
2014-12-22 23:24:45 +01:00
// Providing no values for these is valid in case the user is authenticating
// with an IAM on an ec2 instance (in which case the instance credentials will
// be summoned when GetAuth is called)
2016-03-07 20:50:46 +01:00
accessKey := parameters [ "accesskey" ]
if accessKey == nil {
2015-01-24 00:50:55 +01:00
accessKey = ""
}
2016-03-07 20:50:46 +01:00
secretKey := parameters [ "secretkey" ]
if secretKey == nil {
2015-01-24 00:50:55 +01:00
secretKey = ""
}
2014-10-29 02:15:40 +01:00
2016-04-06 13:34:14 +02:00
regionEndpoint := parameters [ "regionendpoint" ]
if regionEndpoint == nil {
regionEndpoint = ""
}
2016-01-22 03:17:53 +01:00
regionName , ok := parameters [ "region" ]
2016-03-07 20:50:46 +01:00
if regionName == nil || fmt . Sprint ( regionName ) == "" {
2014-10-29 02:15:40 +01:00
return nil , fmt . Errorf ( "No region parameter provided" )
}
2016-01-22 03:17:53 +01:00
region := fmt . Sprint ( regionName )
2016-04-06 13:34:14 +02:00
// Don't check the region value if a custom endpoint is provided.
if regionEndpoint == "" {
if _ , ok = validRegions [ region ] ; ! ok {
return nil , fmt . Errorf ( "Invalid region provided: %v" , region )
}
2014-10-29 02:15:40 +01:00
}
2016-03-07 20:50:46 +01:00
bucket := parameters [ "bucket" ]
if bucket == nil || fmt . Sprint ( bucket ) == "" {
2014-10-29 02:15:40 +01:00
return nil , fmt . Errorf ( "No bucket parameter provided" )
}
2015-01-07 10:51:29 +01:00
encryptBool := false
2016-03-07 20:50:46 +01:00
encrypt := parameters [ "encrypt" ]
switch encrypt := encrypt . ( type ) {
case string :
b , err := strconv . ParseBool ( encrypt )
if err != nil {
2014-12-30 18:31:12 +01:00
return nil , fmt . Errorf ( "The encrypt parameter should be a boolean" )
}
2016-03-07 20:50:46 +01:00
encryptBool = b
case bool :
encryptBool = encrypt
case nil :
// do nothing
default :
return nil , fmt . Errorf ( "The encrypt parameter should be a boolean" )
2014-10-29 02:15:40 +01:00
}
2014-12-19 18:16:51 +01:00
2015-01-07 10:45:31 +01:00
secureBool := true
2016-03-07 20:50:46 +01:00
secure := parameters [ "secure" ]
switch secure := secure . ( type ) {
case string :
b , err := strconv . ParseBool ( secure )
if err != nil {
2014-12-29 21:29:54 +01:00
return nil , fmt . Errorf ( "The secure parameter should be a boolean" )
}
2016-03-07 20:50:46 +01:00
secureBool = b
case bool :
secureBool = secure
case nil :
// do nothing
default :
return nil , fmt . Errorf ( "The secure parameter should be a boolean" )
2015-01-07 10:45:31 +01:00
}
2016-03-10 01:52:59 +01:00
keyID := parameters [ "keyid" ]
if keyID == nil {
keyID = ""
}
2015-01-24 01:46:43 +01:00
chunkSize := int64 ( defaultChunkSize )
2016-03-07 20:50:46 +01:00
chunkSizeParam := parameters [ "chunksize" ]
switch v := chunkSizeParam . ( type ) {
case string :
vv , err := strconv . ParseInt ( v , 0 , 64 )
if err != nil {
return nil , fmt . Errorf ( "chunksize parameter must be an integer, %v invalid" , chunkSizeParam )
2015-04-22 23:31:34 +02:00
}
2016-03-07 20:50:46 +01:00
chunkSize = vv
case int64 :
chunkSize = v
case int , uint , int32 , uint32 , uint64 :
chunkSize = reflect . ValueOf ( v ) . Convert ( reflect . TypeOf ( chunkSize ) ) . Int ( )
case nil :
// do nothing
default :
return nil , fmt . Errorf ( "invalid value for chunksize: %#v" , chunkSizeParam )
}
2015-04-22 23:31:34 +02:00
2016-03-07 20:50:46 +01:00
if chunkSize < minChunkSize {
return nil , fmt . Errorf ( "The chunksize %#v parameter should be a number that is larger than or equal to %d" , chunkSize , minChunkSize )
2015-01-24 01:46:43 +01:00
}
2016-03-07 20:50:46 +01:00
rootDirectory := parameters [ "rootdirectory" ]
if rootDirectory == nil {
2014-12-30 18:31:12 +01:00
rootDirectory = ""
2014-12-19 18:16:51 +01:00
}
2016-01-22 03:17:53 +01:00
storageClass := s3 . StorageClassStandard
2016-03-07 20:50:46 +01:00
storageClassParam := parameters [ "storageclass" ]
if storageClassParam != nil {
2016-01-29 00:48:49 +01:00
storageClassString , ok := storageClassParam . ( string )
if ! ok {
2016-01-22 03:17:53 +01:00
return nil , fmt . Errorf ( "The storageclass parameter must be one of %v, %v invalid" , [ ] string { s3 . StorageClassStandard , s3 . StorageClassReducedRedundancy } , storageClassParam )
2016-01-29 00:48:49 +01:00
}
// All valid storage class parameters are UPPERCASE, so be a bit more flexible here
2016-01-22 03:17:53 +01:00
storageClassString = strings . ToUpper ( storageClassString )
if storageClassString != s3 . StorageClassStandard && storageClassString != s3 . StorageClassReducedRedundancy {
return nil , fmt . Errorf ( "The storageclass parameter must be one of %v, %v invalid" , [ ] string { s3 . StorageClassStandard , s3 . StorageClassReducedRedundancy } , storageClassParam )
2016-01-29 00:48:49 +01:00
}
2016-01-22 03:17:53 +01:00
storageClass = storageClassString
2016-01-29 00:48:49 +01:00
}
2016-03-07 20:50:46 +01:00
userAgent := parameters [ "useragent" ]
if userAgent == nil {
2016-01-21 01:40:58 +01:00
userAgent = ""
}
2015-01-07 11:18:42 +01:00
params := DriverParameters {
fmt . Sprint ( accessKey ) ,
fmt . Sprint ( secretKey ) ,
fmt . Sprint ( bucket ) ,
region ,
2016-03-05 19:46:44 +01:00
fmt . Sprint ( regionEndpoint ) ,
2015-01-07 11:18:42 +01:00
encryptBool ,
2016-03-10 01:52:59 +01:00
fmt . Sprint ( keyID ) ,
2015-01-07 11:18:42 +01:00
secureBool ,
2015-01-24 01:46:43 +01:00
chunkSize ,
2015-01-07 11:18:42 +01:00
fmt . Sprint ( rootDirectory ) ,
2016-01-29 00:48:49 +01:00
storageClass ,
2016-01-21 01:40:58 +01:00
fmt . Sprint ( userAgent ) ,
2015-01-07 11:18:42 +01:00
}
return New ( params )
2014-10-29 02:15:40 +01:00
}
2014-11-18 00:44:07 +01:00
// New constructs a new Driver with the given AWS credentials, region, encryption flag, and
2014-10-29 20:14:19 +01:00
// bucketName
2015-01-07 11:18:42 +01:00
func New ( params DriverParameters ) ( * Driver , error ) {
2016-01-22 03:17:53 +01:00
awsConfig := aws . NewConfig ( )
2016-03-05 19:46:44 +01:00
var creds * credentials . Credentials
if params . RegionEndpoint == "" {
creds = credentials . NewChainCredentials ( [ ] credentials . Provider {
& credentials . StaticProvider {
Value : credentials . Value {
AccessKeyID : params . AccessKey ,
SecretAccessKey : params . SecretKey ,
} ,
} ,
& credentials . EnvProvider { } ,
& credentials . SharedCredentialsProvider { } ,
& ec2rolecreds . EC2RoleProvider { Client : ec2metadata . New ( session . New ( ) ) } ,
} )
} else {
creds = credentials . NewChainCredentials ( [ ] credentials . Provider {
& credentials . StaticProvider {
Value : credentials . Value {
AccessKeyID : params . AccessKey ,
SecretAccessKey : params . SecretKey ,
} ,
2016-01-22 03:17:53 +01:00
} ,
2016-03-05 19:46:44 +01:00
& credentials . EnvProvider { } ,
} )
awsConfig . WithS3ForcePathStyle ( true )
awsConfig . WithEndpoint ( params . RegionEndpoint )
}
2014-12-29 21:29:54 +01:00
2016-01-22 03:17:53 +01:00
awsConfig . WithCredentials ( creds )
awsConfig . WithRegion ( params . Region )
awsConfig . WithDisableSSL ( ! params . Secure )
2016-01-21 01:40:58 +01:00
if params . UserAgent != "" {
2016-01-22 03:17:53 +01:00
awsConfig . WithHTTPClient ( & http . Client {
Transport : transport . NewTransport ( http . DefaultTransport , transport . NewHeaderRequestModifier ( http . Header { http . CanonicalHeaderKey ( "User-Agent" ) : [ ] string { params . UserAgent } } ) ) ,
} )
2016-01-21 01:40:58 +01:00
}
2014-10-25 01:37:25 +02:00
2016-01-22 03:17:53 +01:00
s3obj := s3 . New ( session . New ( awsConfig ) )
2016-01-21 01:40:58 +01:00
2014-12-21 07:48:42 +01:00
// TODO Currently multipart uploads have no timestamps, so this would be unwise
// if you initiated a new s3driver while another one is running on the same bucket.
// multis, _, err := bucket.ListMulti("", "")
// if err != nil {
// return nil, err
// }
// for _, multi := range multis {
// err := multi.Abort()
// //TODO appropriate to do this error checking?
// if err != nil {
// return nil, err
// }
// }
2014-12-19 18:16:51 +01:00
2015-02-04 01:54:52 +01:00
d := & driver {
2015-01-24 01:46:43 +01:00
S3 : s3obj ,
2016-01-22 03:17:53 +01:00
Bucket : params . Bucket ,
2015-01-24 01:46:43 +01:00
ChunkSize : params . ChunkSize ,
Encrypt : params . Encrypt ,
2016-03-10 01:52:59 +01:00
KeyID : params . KeyID ,
2015-02-04 01:54:52 +01:00
RootDirectory : params . RootDirectory ,
2016-01-29 00:48:49 +01:00
StorageClass : params . StorageClass ,
2015-02-04 01:54:52 +01:00
}
return & Driver {
baseEmbed : baseEmbed {
Base : base . Base {
StorageDriver : d ,
} ,
} ,
} , nil
2014-10-25 01:37:25 +02:00
}
2014-10-29 20:14:19 +01:00
// Implement the storagedriver.StorageDriver interface
2015-04-23 02:30:01 +02:00
func ( d * driver ) Name ( ) string {
return driverName
}
2014-11-18 00:44:07 +01:00
// GetContent retrieves the content stored at "path" as a []byte.
2015-04-28 00:58:58 +02:00
func ( d * driver ) GetContent ( ctx context . Context , path string ) ( [ ] byte , error ) {
2016-02-08 23:29:21 +01:00
reader , err := d . Reader ( ctx , path , 0 )
2014-11-19 02:41:48 +01:00
if err != nil {
2016-01-22 03:17:53 +01:00
return nil , err
2014-11-19 02:41:48 +01:00
}
2016-01-22 03:17:53 +01:00
return ioutil . ReadAll ( reader )
2014-10-25 01:37:25 +02:00
}
2014-11-18 00:44:07 +01:00
// PutContent stores the []byte content at a location designated by "path".
2015-04-28 00:58:58 +02:00
func ( d * driver ) PutContent ( ctx context . Context , path string , contents [ ] byte ) error {
2016-01-22 03:17:53 +01:00
_ , err := d . S3 . PutObject ( & s3 . PutObjectInput {
Bucket : aws . String ( d . Bucket ) ,
Key : aws . String ( d . s3Path ( path ) ) ,
ContentType : d . getContentType ( ) ,
ACL : d . getACL ( ) ,
ServerSideEncryption : d . getEncryptionMode ( ) ,
2016-03-10 01:52:59 +01:00
SSEKMSKeyId : d . getSSEKMSKeyID ( ) ,
2016-01-22 03:17:53 +01:00
StorageClass : d . getStorageClass ( ) ,
Body : bytes . NewReader ( contents ) ,
} )
return parseError ( path , err )
2014-10-25 01:37:25 +02:00
}
2016-02-08 23:29:21 +01:00
// Reader retrieves an io.ReadCloser for the content stored at "path" with a
2014-11-18 00:44:07 +01:00
// given byte offset.
2016-02-08 23:29:21 +01:00
func ( d * driver ) Reader ( ctx context . Context , path string , offset int64 ) ( io . ReadCloser , error ) {
2016-01-22 03:17:53 +01:00
resp , err := d . S3 . GetObject ( & s3 . GetObjectInput {
Bucket : aws . String ( d . Bucket ) ,
Key : aws . String ( d . s3Path ( path ) ) ,
Range : aws . String ( "bytes=" + strconv . FormatInt ( offset , 10 ) + "-" ) ,
} )
2014-10-25 01:37:25 +02:00
2014-11-19 02:41:48 +01:00
if err != nil {
2016-01-22 03:17:53 +01:00
if s3Err , ok := err . ( awserr . Error ) ; ok && s3Err . Code ( ) == "InvalidRange" {
2014-12-19 18:16:51 +01:00
return ioutil . NopCloser ( bytes . NewReader ( nil ) ) , nil
}
return nil , parseError ( path , err )
2014-10-25 01:37:25 +02:00
}
2014-11-19 02:41:48 +01:00
return resp . Body , nil
2014-10-25 01:37:25 +02:00
}
2016-02-08 23:29:21 +01:00
// Writer returns a FileWriter which will store the content written to it
// at the location designated by "path" after the call to Commit.
func ( d * driver ) Writer ( ctx context . Context , path string , append bool ) ( storagedriver . FileWriter , error ) {
key := d . s3Path ( path )
if ! append {
// TODO (brianbland): cancel other uploads at this path
resp , err := d . S3 . CreateMultipartUpload ( & s3 . CreateMultipartUploadInput {
Bucket : aws . String ( d . Bucket ) ,
Key : aws . String ( key ) ,
ContentType : d . getContentType ( ) ,
ACL : d . getACL ( ) ,
ServerSideEncryption : d . getEncryptionMode ( ) ,
2016-03-10 01:52:59 +01:00
SSEKMSKeyId : d . getSSEKMSKeyID ( ) ,
2016-02-08 23:29:21 +01:00
StorageClass : d . getStorageClass ( ) ,
} )
2014-12-19 18:16:51 +01:00
if err != nil {
2016-02-08 23:29:21 +01:00
return nil , err
2014-12-19 18:20:07 +01:00
}
2016-02-08 23:29:21 +01:00
return d . newWriter ( key , * resp . UploadId , nil ) , nil
2014-12-19 18:20:07 +01:00
}
2016-02-08 23:29:21 +01:00
resp , err := d . S3 . ListMultipartUploads ( & s3 . ListMultipartUploadsInput {
Bucket : aws . String ( d . Bucket ) ,
Prefix : aws . String ( key ) ,
} )
if err != nil {
return nil , parseError ( path , err )
2014-12-19 18:20:07 +01:00
}
2016-02-08 23:29:21 +01:00
for _ , multi := range resp . Uploads {
if key != * multi . Key {
continue
}
resp , err := d . S3 . ListParts ( & s3 . ListPartsInput {
Bucket : aws . String ( d . Bucket ) ,
Key : aws . String ( key ) ,
UploadId : multi . UploadId ,
2016-01-22 03:17:53 +01:00
} )
2014-12-19 18:20:07 +01:00
if err != nil {
2016-02-08 23:29:21 +01:00
return nil , parseError ( path , err )
2014-12-19 18:16:51 +01:00
}
2016-02-08 23:29:21 +01:00
var multiSize int64
for _ , part := range resp . Parts {
multiSize += * part . Size
2014-10-25 01:37:25 +02:00
}
2016-02-08 23:29:21 +01:00
return d . newWriter ( key , * multi . UploadId , resp . Parts ) , nil
2014-10-25 01:37:25 +02:00
}
2016-02-08 23:29:21 +01:00
return nil , storagedriver . PathNotFoundError { Path : path }
2014-10-25 01:37:25 +02:00
}
2014-12-19 18:16:51 +01:00
// Stat retrieves the FileInfo for the given path, including the current size
// in bytes and the creation time.
2015-04-28 00:58:58 +02:00
func ( d * driver ) Stat ( ctx context . Context , path string ) ( storagedriver . FileInfo , error ) {
2016-01-22 03:17:53 +01:00
resp , err := d . S3 . ListObjects ( & s3 . ListObjectsInput {
Bucket : aws . String ( d . Bucket ) ,
Prefix : aws . String ( d . s3Path ( path ) ) ,
MaxKeys : aws . Int64 ( 1 ) ,
} )
2014-10-25 01:37:25 +02:00
if err != nil {
2014-12-19 18:16:51 +01:00
return nil , err
2014-10-25 01:37:25 +02:00
}
2014-12-19 18:16:51 +01:00
fi := storagedriver . FileInfoFields {
Path : path ,
2014-10-25 01:37:25 +02:00
}
2016-01-22 03:17:53 +01:00
if len ( resp . Contents ) == 1 {
if * resp . Contents [ 0 ] . Key != d . s3Path ( path ) {
2014-12-19 18:16:51 +01:00
fi . IsDir = true
} else {
fi . IsDir = false
2016-01-22 03:17:53 +01:00
fi . Size = * resp . Contents [ 0 ] . Size
fi . ModTime = * resp . Contents [ 0 ] . LastModified
2014-12-19 18:16:51 +01:00
}
2016-01-22 03:17:53 +01:00
} else if len ( resp . CommonPrefixes ) == 1 {
2014-12-19 18:16:51 +01:00
fi . IsDir = true
} else {
return nil , storagedriver . PathNotFoundError { Path : path }
}
return storagedriver . FileInfoInternal { FileInfoFields : fi } , nil
2014-10-25 01:37:25 +02:00
}
2014-12-19 18:16:51 +01:00
// List returns a list of the objects that are direct descendants of the given path.
2015-12-08 20:02:40 +01:00
func ( d * driver ) List ( ctx context . Context , opath string ) ( [ ] string , error ) {
path := opath
2014-12-19 18:16:51 +01:00
if path != "/" && path [ len ( path ) - 1 ] != '/' {
2014-11-07 21:58:48 +01:00
path = path + "/"
2014-11-04 01:20:38 +01:00
}
2015-02-20 01:31:34 +01:00
// This is to cover for the cases when the rootDirectory of the driver is either "" or "/".
// In those cases, there is no root prefix to replace and we must actually add a "/" to all
// results in order to keep them as valid paths as recognized by storagedriver.PathRegexp
prefix := ""
if d . s3Path ( "" ) == "" {
prefix = "/"
}
2016-01-22 03:17:53 +01:00
resp , err := d . S3 . ListObjects ( & s3 . ListObjectsInput {
Bucket : aws . String ( d . Bucket ) ,
Prefix : aws . String ( d . s3Path ( path ) ) ,
Delimiter : aws . String ( "/" ) ,
MaxKeys : aws . Int64 ( listMax ) ,
} )
2014-10-25 01:37:25 +02:00
if err != nil {
2015-12-08 20:02:40 +01:00
return nil , parseError ( opath , err )
2015-11-24 23:23:12 +01:00
}
2014-10-25 01:37:25 +02:00
files := [ ] string { }
directories := [ ] string { }
2014-11-04 01:20:38 +01:00
for {
2016-01-22 03:17:53 +01:00
for _ , key := range resp . Contents {
files = append ( files , strings . Replace ( * key . Key , d . s3Path ( "" ) , prefix , 1 ) )
2014-10-25 01:37:25 +02:00
}
2016-01-22 03:17:53 +01:00
for _ , commonPrefix := range resp . CommonPrefixes {
commonPrefix := * commonPrefix . Prefix
2015-02-20 01:31:34 +01:00
directories = append ( directories , strings . Replace ( commonPrefix [ 0 : len ( commonPrefix ) - 1 ] , d . s3Path ( "" ) , prefix , 1 ) )
2014-10-25 01:37:25 +02:00
}
2016-01-22 03:17:53 +01:00
if * resp . IsTruncated {
resp , err = d . S3 . ListObjects ( & s3 . ListObjectsInput {
Bucket : aws . String ( d . Bucket ) ,
Prefix : aws . String ( d . s3Path ( path ) ) ,
Delimiter : aws . String ( "/" ) ,
MaxKeys : aws . Int64 ( listMax ) ,
Marker : resp . NextMarker ,
} )
2014-11-04 01:20:38 +01:00
if err != nil {
return nil , err
}
2014-10-25 01:37:25 +02:00
} else {
2014-11-04 01:20:38 +01:00
break
2014-10-25 01:37:25 +02:00
}
}
2015-12-08 20:02:40 +01:00
if opath != "/" {
if len ( files ) == 0 && len ( directories ) == 0 {
// Treat empty response as missing directory, since we don't actually
// have directories in s3.
return nil , storagedriver . PathNotFoundError { Path : opath }
}
}
2014-10-25 01:37:25 +02:00
return append ( files , directories ... ) , nil
}
2014-11-18 00:44:07 +01:00
// Move moves an object stored at sourcePath to destPath, removing the original
// object.
2015-04-28 00:58:58 +02:00
func ( d * driver ) Move ( ctx context . Context , sourcePath string , destPath string ) error {
2014-10-25 01:37:25 +02:00
/* This is terrible, but aws doesn't have an actual move. */
2016-01-22 03:17:53 +01:00
_ , err := d . S3 . CopyObject ( & s3 . CopyObjectInput {
Bucket : aws . String ( d . Bucket ) ,
Key : aws . String ( d . s3Path ( destPath ) ) ,
ContentType : d . getContentType ( ) ,
ACL : d . getACL ( ) ,
ServerSideEncryption : d . getEncryptionMode ( ) ,
2016-03-10 01:52:59 +01:00
SSEKMSKeyId : d . getSSEKMSKeyID ( ) ,
2016-01-22 03:17:53 +01:00
StorageClass : d . getStorageClass ( ) ,
CopySource : aws . String ( d . Bucket + "/" + d . s3Path ( sourcePath ) ) ,
} )
2014-10-25 01:37:25 +02:00
if err != nil {
2014-12-19 18:16:51 +01:00
return parseError ( sourcePath , err )
2014-10-25 01:37:25 +02:00
}
2015-04-28 00:58:58 +02:00
return d . Delete ( ctx , sourcePath )
2014-10-25 01:37:25 +02:00
}
2014-11-18 00:44:07 +01:00
// Delete recursively deletes all objects stored at "path" and its subpaths.
2015-04-28 00:58:58 +02:00
func ( d * driver ) Delete ( ctx context . Context , path string ) error {
2016-01-22 03:17:53 +01:00
resp , err := d . S3 . ListObjects ( & s3 . ListObjectsInput {
Bucket : aws . String ( d . Bucket ) ,
Prefix : aws . String ( d . s3Path ( path ) ) ,
} )
if err != nil || len ( resp . Contents ) == 0 {
2014-11-13 02:19:19 +01:00
return storagedriver . PathNotFoundError { Path : path }
2014-10-25 01:37:25 +02:00
}
2016-01-22 03:17:53 +01:00
s3Objects := make ( [ ] * s3 . ObjectIdentifier , 0 , listMax )
2014-10-25 01:37:25 +02:00
2016-01-22 03:17:53 +01:00
for len ( resp . Contents ) > 0 {
for _ , key := range resp . Contents {
s3Objects = append ( s3Objects , & s3 . ObjectIdentifier {
Key : key . Key ,
} )
2014-10-25 01:37:25 +02:00
}
2016-01-22 03:17:53 +01:00
_ , err := d . S3 . DeleteObjects ( & s3 . DeleteObjectsInput {
Bucket : aws . String ( d . Bucket ) ,
Delete : & s3 . Delete {
Objects : s3Objects ,
Quiet : aws . Bool ( false ) ,
} ,
} )
2014-10-25 01:37:25 +02:00
if err != nil {
return nil
}
2016-01-22 03:17:53 +01:00
resp , err = d . S3 . ListObjects ( & s3 . ListObjectsInput {
Bucket : aws . String ( d . Bucket ) ,
Prefix : aws . String ( d . s3Path ( path ) ) ,
} )
2014-10-25 01:37:25 +02:00
if err != nil {
return err
}
}
return nil
}
2015-01-07 17:31:38 +01:00
// URLFor returns a URL which may be used to retrieve the content stored at the given path.
// May return an UnsupportedMethodErr in certain StorageDriver implementations.
2015-04-28 00:58:58 +02:00
func ( d * driver ) URLFor ( ctx context . Context , path string , options map [ string ] interface { } ) ( string , error ) {
2015-01-14 20:31:11 +01:00
methodString := "GET"
method , ok := options [ "method" ]
if ok {
methodString , ok = method . ( string )
if ! ok || ( methodString != "GET" && methodString != "HEAD" ) {
2015-11-02 22:23:53 +01:00
return "" , storagedriver . ErrUnsupportedMethod { }
2015-01-14 20:31:11 +01:00
}
}
2016-01-22 03:17:53 +01:00
expiresIn := 20 * time . Minute
2015-01-09 02:45:21 +01:00
expires , ok := options [ "expiry" ]
2015-01-09 02:10:32 +01:00
if ok {
et , ok := expires . ( time . Time )
if ok {
2016-01-22 03:17:53 +01:00
expiresIn = et . Sub ( time . Now ( ) )
2015-01-09 02:10:32 +01:00
}
}
2016-01-22 03:17:53 +01:00
var req * request . Request
switch methodString {
case "GET" :
req , _ = d . S3 . GetObjectRequest ( & s3 . GetObjectInput {
Bucket : aws . String ( d . Bucket ) ,
Key : aws . String ( d . s3Path ( path ) ) ,
} )
case "HEAD" :
req , _ = d . S3 . HeadObjectRequest ( & s3 . HeadObjectInput {
Bucket : aws . String ( d . Bucket ) ,
Key : aws . String ( d . s3Path ( path ) ) ,
} )
default :
panic ( "unreachable" )
}
return req . Presign ( expiresIn )
2015-01-07 17:31:38 +01:00
}
2015-02-04 01:54:52 +01:00
func ( d * driver ) s3Path ( path string ) string {
2015-01-24 01:46:43 +01:00
return strings . TrimLeft ( strings . TrimRight ( d . RootDirectory , "/" ) + path , "/" )
2014-12-19 18:16:51 +01:00
}
2014-10-25 01:37:25 +02:00
2015-04-07 01:23:31 +02:00
// S3BucketKey returns the s3 bucket key for the given storage driver path.
func ( d * Driver ) S3BucketKey ( path string ) string {
return d . StorageDriver . ( * driver ) . s3Path ( path )
}
2014-12-19 18:16:51 +01:00
func parseError ( path string , err error ) error {
2016-01-22 03:17:53 +01:00
if s3Err , ok := err . ( awserr . Error ) ; ok && s3Err . Code ( ) == "NoSuchKey" {
2014-12-19 18:16:51 +01:00
return storagedriver . PathNotFoundError { Path : path }
2014-10-25 01:37:25 +02:00
}
2014-12-19 18:16:51 +01:00
return err
2014-10-25 01:37:25 +02:00
}
2016-01-22 03:17:53 +01:00
func ( d * driver ) getEncryptionMode ( ) * string {
2016-03-10 01:52:59 +01:00
if ! d . Encrypt {
return nil
}
if d . KeyID == "" {
2016-01-22 03:17:53 +01:00
return aws . String ( "AES256" )
}
2016-03-10 01:52:59 +01:00
return aws . String ( "aws:kms" )
}
func ( d * driver ) getSSEKMSKeyID ( ) * string {
if d . KeyID != "" {
return aws . String ( d . KeyID )
}
2016-01-22 03:17:53 +01:00
return nil
2014-10-25 01:37:25 +02:00
}
2016-01-22 03:17:53 +01:00
func ( d * driver ) getContentType ( ) * string {
return aws . String ( "application/octet-stream" )
2014-10-25 01:37:25 +02:00
}
2016-01-22 03:17:53 +01:00
func ( d * driver ) getACL ( ) * string {
return aws . String ( "private" )
2014-10-25 01:37:25 +02:00
}
2016-01-22 03:17:53 +01:00
func ( d * driver ) getStorageClass ( ) * string {
return aws . String ( d . StorageClass )
2014-10-25 01:37:25 +02:00
}
2015-04-23 00:07:18 +02:00
2016-02-08 23:29:21 +01:00
// writer attempts to upload parts to S3 in a buffered fashion where the last
// part is at least as large as the chunksize, so the multipart upload could be
// cleanly resumed in the future. This is violated if Close is called after less
// than a full chunk is written.
type writer struct {
driver * driver
key string
uploadID string
parts [ ] * s3 . Part
size int64
readyPart [ ] byte
pendingPart [ ] byte
closed bool
committed bool
cancelled bool
2015-04-23 00:07:18 +02:00
}
2016-02-08 23:29:21 +01:00
func ( d * driver ) newWriter ( key , uploadID string , parts [ ] * s3 . Part ) storagedriver . FileWriter {
var size int64
for _ , part := range parts {
size += * part . Size
}
return & writer {
driver : d ,
key : key ,
uploadID : uploadID ,
parts : parts ,
size : size ,
}
}
2016-04-23 12:13:15 +02:00
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 }
2016-02-08 23:29:21 +01:00
func ( w * writer ) Write ( p [ ] byte ) ( int , error ) {
if w . closed {
return 0 , fmt . Errorf ( "already closed" )
} else if w . committed {
return 0 , fmt . Errorf ( "already committed" )
} else if w . cancelled {
return 0 , fmt . Errorf ( "already cancelled" )
}
// If the last written part is smaller than minChunkSize, we need to make a
// new multipart upload :sadface:
if len ( w . parts ) > 0 && int ( * w . parts [ len ( w . parts ) - 1 ] . Size ) < minChunkSize {
2016-04-23 12:13:15 +02:00
var completedUploadedParts completedParts
2016-02-08 23:29:21 +01:00
for _ , part := range w . parts {
2016-04-23 12:13:15 +02:00
completedUploadedParts = append ( completedUploadedParts , & s3 . CompletedPart {
2016-02-08 23:29:21 +01:00
ETag : part . ETag ,
PartNumber : part . PartNumber ,
} )
}
2016-04-23 12:13:15 +02:00
sort . Sort ( completedUploadedParts )
2016-02-08 23:29:21 +01:00
_ , err := w . driver . S3 . CompleteMultipartUpload ( & s3 . CompleteMultipartUploadInput {
Bucket : aws . String ( w . driver . Bucket ) ,
Key : aws . String ( w . key ) ,
UploadId : aws . String ( w . uploadID ) ,
MultipartUpload : & s3 . CompletedMultipartUpload {
2016-04-23 12:13:15 +02:00
Parts : completedUploadedParts ,
2016-02-08 23:29:21 +01:00
} ,
} )
if err != nil {
w . driver . S3 . AbortMultipartUpload ( & s3 . AbortMultipartUploadInput {
Bucket : aws . String ( w . driver . Bucket ) ,
Key : aws . String ( w . key ) ,
UploadId : aws . String ( w . uploadID ) ,
} )
return 0 , err
}
resp , err := w . driver . S3 . CreateMultipartUpload ( & s3 . CreateMultipartUploadInput {
Bucket : aws . String ( w . driver . Bucket ) ,
Key : aws . String ( w . key ) ,
ContentType : w . driver . getContentType ( ) ,
ACL : w . driver . getACL ( ) ,
ServerSideEncryption : w . driver . getEncryptionMode ( ) ,
StorageClass : w . driver . getStorageClass ( ) ,
} )
if err != nil {
return 0 , err
}
w . uploadID = * resp . UploadId
// If the entire written file is smaller than minChunkSize, we need to make
// a new part from scratch :double sad face:
if w . size < minChunkSize {
resp , err := w . driver . S3 . GetObject ( & s3 . GetObjectInput {
Bucket : aws . String ( w . driver . Bucket ) ,
Key : aws . String ( w . key ) ,
} )
defer resp . Body . Close ( )
if err != nil {
return 0 , err
}
w . parts = nil
w . readyPart , err = ioutil . ReadAll ( resp . Body )
if err != nil {
return 0 , err
}
} else {
// Otherwise we can use the old file as the new first part
copyPartResp , err := w . driver . S3 . UploadPartCopy ( & s3 . UploadPartCopyInput {
Bucket : aws . String ( w . driver . Bucket ) ,
CopySource : aws . String ( w . driver . Bucket + "/" + w . key ) ,
Key : aws . String ( w . key ) ,
PartNumber : aws . Int64 ( 1 ) ,
UploadId : resp . UploadId ,
} )
if err != nil {
return 0 , err
}
w . parts = [ ] * s3 . Part {
{
ETag : copyPartResp . CopyPartResult . ETag ,
PartNumber : aws . Int64 ( 1 ) ,
Size : aws . Int64 ( w . size ) ,
} ,
}
}
}
var n int
for len ( p ) > 0 {
// If no parts are ready to write, fill up the first part
if neededBytes := int ( w . driver . ChunkSize ) - len ( w . readyPart ) ; neededBytes > 0 {
if len ( p ) >= neededBytes {
w . readyPart = append ( w . readyPart , p [ : neededBytes ] ... )
n += neededBytes
p = p [ neededBytes : ]
} else {
w . readyPart = append ( w . readyPart , p ... )
n += len ( p )
p = nil
}
}
if neededBytes := int ( w . driver . ChunkSize ) - len ( w . pendingPart ) ; neededBytes > 0 {
if len ( p ) >= neededBytes {
w . pendingPart = append ( w . pendingPart , p [ : neededBytes ] ... )
n += neededBytes
p = p [ neededBytes : ]
err := w . flushPart ( )
if err != nil {
w . size += int64 ( n )
return n , err
}
} else {
w . pendingPart = append ( w . pendingPart , p ... )
n += len ( p )
p = nil
}
}
}
w . size += int64 ( n )
return n , nil
}
func ( w * writer ) Size ( ) int64 {
return w . size
}
func ( w * writer ) Close ( ) error {
if w . closed {
return fmt . Errorf ( "already closed" )
}
w . closed = true
return w . flushPart ( )
}
func ( w * writer ) Cancel ( ) error {
if w . closed {
return fmt . Errorf ( "already closed" )
} else if w . committed {
return fmt . Errorf ( "already committed" )
}
w . cancelled = true
_ , err := w . driver . S3 . AbortMultipartUpload ( & s3 . AbortMultipartUploadInput {
Bucket : aws . String ( w . driver . Bucket ) ,
Key : aws . String ( w . key ) ,
UploadId : aws . String ( w . uploadID ) ,
} )
return err
}
func ( w * writer ) Commit ( ) error {
if w . closed {
return fmt . Errorf ( "already closed" )
} else if w . committed {
return fmt . Errorf ( "already committed" )
} else if w . cancelled {
return fmt . Errorf ( "already cancelled" )
}
err := w . flushPart ( )
if err != nil {
return err
}
w . committed = true
2016-04-23 12:13:15 +02:00
var completedUploadedParts completedParts
2016-02-08 23:29:21 +01:00
for _ , part := range w . parts {
2016-04-23 12:13:15 +02:00
completedUploadedParts = append ( completedUploadedParts , & s3 . CompletedPart {
2016-02-08 23:29:21 +01:00
ETag : part . ETag ,
PartNumber : part . PartNumber ,
} )
}
2016-04-23 12:13:15 +02:00
sort . Sort ( completedUploadedParts )
2016-02-08 23:29:21 +01:00
_ , err = w . driver . S3 . CompleteMultipartUpload ( & s3 . CompleteMultipartUploadInput {
Bucket : aws . String ( w . driver . Bucket ) ,
Key : aws . String ( w . key ) ,
UploadId : aws . String ( w . uploadID ) ,
MultipartUpload : & s3 . CompletedMultipartUpload {
2016-04-23 12:13:15 +02:00
Parts : completedUploadedParts ,
2016-02-08 23:29:21 +01:00
} ,
} )
if err != nil {
w . driver . S3 . AbortMultipartUpload ( & s3 . AbortMultipartUploadInput {
Bucket : aws . String ( w . driver . Bucket ) ,
Key : aws . String ( w . key ) ,
UploadId : aws . String ( w . uploadID ) ,
} )
return err
}
return nil
}
// flushPart flushes buffers to write a part to S3.
// Only called by Write (with both buffers full) and Close/Commit (always)
func ( w * writer ) flushPart ( ) error {
if len ( w . readyPart ) == 0 && len ( w . pendingPart ) == 0 {
// nothing to write
return nil
}
if len ( w . pendingPart ) < int ( w . driver . ChunkSize ) {
// closing with a small pending part
// combine ready and pending to avoid writing a small part
w . readyPart = append ( w . readyPart , w . pendingPart ... )
w . pendingPart = nil
}
partNumber := aws . Int64 ( int64 ( len ( w . parts ) + 1 ) )
resp , err := w . driver . S3 . UploadPart ( & s3 . UploadPartInput {
Bucket : aws . String ( w . driver . Bucket ) ,
Key : aws . String ( w . key ) ,
PartNumber : partNumber ,
UploadId : aws . String ( w . uploadID ) ,
Body : bytes . NewReader ( w . readyPart ) ,
} )
if err != nil {
return err
}
w . parts = append ( w . parts , & s3 . Part {
ETag : resp . ETag ,
PartNumber : partNumber ,
Size : aws . Int64 ( int64 ( len ( w . readyPart ) ) ) ,
} )
w . readyPart = w . pendingPart
w . pendingPart = nil
return nil
2015-04-23 00:07:18 +02:00
}