19234: s3v2 + non-aws: default to us-east-1 signing settings.

[arvados.git] / services / keepstore / s3aws_volume.go

diff --git a/services/keepstore/s3aws_volume.go b/services/keepstore/s3aws_volume.go
index a7801ae262fbe73f30c94b0a0692c8dfff8e440a..d068dde074ea254ef814aea38eefa6f63102d7e3 100644 (file)
--- a/services/keepstore/s3aws_volume.go
+++ b/services/keepstore/s3aws_volume.go
@@ -62,6 +62,8 @@ type s3AWSbucket struct {
 // aws-sdk-go based on the UseAWSS3v2Driver feature flag
 func chooseS3VolumeDriver(cluster *arvados.Cluster, volume arvados.Volume, logger logrus.FieldLogger, metrics *volumeMetricsVecs) (Volume, error) {
        v := &S3Volume{cluster: cluster, volume: volume, metrics: metrics}
+       // Default value will be overriden if it happens to be defined in the config
+       v.S3VolumeDriverParameters.UseAWSS3v2Driver = true
        err := json.Unmarshal(volume.DriverParameters, v)
        if err != nil {
                return nil, err
@@ -112,18 +114,14 @@ func newS3AWSVolume(cluster *arvados.Cluster, volume arvados.Volume, logger logr
 }
 
 func (v *S3AWSVolume) translateError(err error) error {
-       if aerr, ok := err.(awserr.Error); ok {
-               switch aerr.Code() {
-               case "NotFound":
+       if _, ok := err.(*aws.RequestCanceledError); ok {
+               return context.Canceled
+       } else if aerr, ok := err.(awserr.Error); ok {
+               if aerr.Code() == "NotFound" {
                        return os.ErrNotExist
-               case "NoSuchKey":
+               } else if aerr.Code() == "NoSuchKey" {
                        return os.ErrNotExist
                }
-       } else {
-               switch err.(type) {
-               case *aws.RequestCanceledError:
-                       return context.Canceled
-               }
        }
        return err
 }
@@ -186,19 +184,25 @@ func (v *S3AWSVolume) check(ec2metadataHostname string) error {
                        if v.Endpoint != "" && service == "s3" {
                                return aws.Endpoint{
                                        URL:           v.Endpoint,
-                                       SigningRegion: v.Region,
+                                       SigningRegion: region,
                                }, nil
                        } else if service == "ec2metadata" && ec2metadataHostname != "" {
                                return aws.Endpoint{
                                        URL: ec2metadataHostname,
                                }, nil
+                       } else {
+                               return defaultResolver.ResolveEndpoint(service, region)
                        }
-
-                       return defaultResolver.ResolveEndpoint(service, region)
                }
                cfg.EndpointResolver = aws.EndpointResolverFunc(myCustomResolver)
        }
-
+       if v.Region == "" {
+               // Endpoint is already specified (otherwise we would
+               // have errored out above), but Region is also
+               // required by the aws sdk, in order to determine
+               // SignatureVersions.
+               v.Region = "us-east-1"
+       }
        cfg.Region = v.Region
 
        // Zero timeouts mean "wait forever", which is a bad
@@ -472,14 +476,9 @@ func (v *S3AWSVolume) Get(ctx context.Context, loc string, buf []byte) (int, err
        // through 'buf []byte', and we don't want to allocate two buffers for each
        // read request. Instead, use a version of ReadBlock that accepts 'buf []byte'
        // as an input.
-       return v.ReadBlock(ctx, loc, buf)
-}
-
-func (v *S3AWSVolume) ReadBlock(ctx context.Context, loc string, buf []byte) (int, error) {
        key := v.key(loc)
        count, err := v.readWorker(ctx, key, buf)
        if err == nil {
-               v.bucket.stats.TickInBytes(uint64(count))
                return count, err
        }
 
@@ -506,7 +505,6 @@ func (v *S3AWSVolume) ReadBlock(ctx context.Context, loc string, buf []byte) (in
                err = v.translateError(err)
                return 0, err
        }
-       v.bucket.stats.TickInBytes(uint64(count))
        return count, err
 }
 
@@ -526,12 +524,8 @@ func (v *S3AWSVolume) readWorker(ctx context.Context, key string, buf []byte) (i
        v.bucket.stats.TickOps("get")
        v.bucket.stats.Tick(&v.bucket.stats.Ops, &v.bucket.stats.GetOps)
        v.bucket.stats.TickErr(err)
-       if err != nil {
-               return 0, v.translateError(err)
-       }
-       buf = awsBuf.Bytes()
-
-       return int(count), err
+       v.bucket.stats.TickInBytes(uint64(count))
+       return int(count), v.translateError(err)
 }
 
 func (v *S3AWSVolume) writeObject(ctx context.Context, key string, r io.Reader) error {
@@ -551,7 +545,7 @@ func (v *S3AWSVolume) writeObject(ctx context.Context, key string, r io.Reader)
                var contentMD5 string
                md5, err := hex.DecodeString(loc)
                if err != nil {
-                       return err
+                       return v.translateError(err)
                }
                contentMD5 = base64.StdEncoding.EncodeToString(md5)
                uploadInput.ContentMD5 = &contentMD5
@@ -580,22 +574,18 @@ func (v *S3AWSVolume) writeObject(ctx context.Context, key string, r io.Reader)
        v.bucket.stats.Tick(&v.bucket.stats.Ops, &v.bucket.stats.PutOps)
        v.bucket.stats.TickErr(err)
 
-       return err
+       return v.translateError(err)
 }
 
 // Put writes a block.
 func (v *S3AWSVolume) Put(ctx context.Context, loc string, block []byte) error {
        // Do not use putWithPipe here; we want to pass an io.ReadSeeker to the S3
        // sdk to avoid memory allocation there. See #17339 for more information.
-       return v.translateError(v.WriteBlock(ctx, loc, bytes.NewReader(block)))
-}
-
-// WriteBlock implements BlockWriter.
-func (v *S3AWSVolume) WriteBlock(ctx context.Context, loc string, rdr io.Reader) error {
        if v.volume.ReadOnly {
                return MethodDisabledError
        }
 
+       rdr := bytes.NewReader(block)
        r := NewCountingReaderAtSeeker(rdr, v.bucket.stats.TickOutBytes)
        key := v.key(loc)
        err := v.writeObject(ctx, key, r)