services/keep-web/s3_test.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package main
   6
   7 import (
   8         "bytes"
   9         "crypto/rand"
  10         "io/ioutil"
  11         "os"
  12         "sync"
  13         "time"
  14
  15         "git.arvados.org/arvados.git/sdk/go/arvados"
  16         "git.arvados.org/arvados.git/sdk/go/arvadosclient"
  17         "git.arvados.org/arvados.git/sdk/go/arvadostest"
  18         "git.arvados.org/arvados.git/sdk/go/keepclient"
  19         "github.com/AdRoll/goamz/aws"
  20         "github.com/AdRoll/goamz/s3"
  21         check "gopkg.in/check.v1"
  22 )
  23
  24 type s3stage struct {
  25         arv        *arvados.Client
  26         proj       arvados.Group
  27         projbucket *s3.Bucket
  28         coll       arvados.Collection
  29         collbucket *s3.Bucket
  30 }
  31
  32 func (s *IntegrationSuite) s3setup(c *check.C) s3stage {
  33         var proj arvados.Group
  34         var coll arvados.Collection
  35         arv := arvados.NewClientFromEnv()
  36         arv.AuthToken = arvadostest.ActiveToken
  37         err := arv.RequestAndDecode(&proj, "POST", "arvados/v1/groups", nil, map[string]interface{}{
  38                 "group": map[string]interface{}{
  39                         "group_class": "project",
  40                         "name":        "keep-web s3 test",
  41                 },
  42                 "ensure_unique_name": true,
  43         })
  44         c.Assert(err, check.IsNil)
  45         err = arv.RequestAndDecode(&coll, "POST", "arvados/v1/collections", nil, map[string]interface{}{"collection": map[string]interface{}{
  46                 "owner_uuid":    proj.UUID,
  47                 "name":          "keep-web s3 test collection",
  48                 "manifest_text": ". d41d8cd98f00b204e9800998ecf8427e+0 0:0:emptyfile\n./emptydir d41d8cd98f00b204e9800998ecf8427e+0 0:0:.\n",
  49         }})
  50         c.Assert(err, check.IsNil)
  51         ac, err := arvadosclient.New(arv)
  52         c.Assert(err, check.IsNil)
  53         kc, err := keepclient.MakeKeepClient(ac)
  54         c.Assert(err, check.IsNil)
  55         fs, err := coll.FileSystem(arv, kc)
  56         c.Assert(err, check.IsNil)
  57         f, err := fs.OpenFile("sailboat.txt", os.O_CREATE|os.O_WRONLY, 0644)
  58         c.Assert(err, check.IsNil)
  59         _, err = f.Write([]byte("⛵\n"))
  60         c.Assert(err, check.IsNil)
  61         err = f.Close()
  62         c.Assert(err, check.IsNil)
  63         err = fs.Sync()
  64         c.Assert(err, check.IsNil)
  65
  66         auth := aws.NewAuth(arvadostest.ActiveTokenV2, arvadostest.ActiveTokenV2, "", time.Now().Add(time.Hour))
  67         region := aws.Region{
  68                 Name:       s.testServer.Addr,
  69                 S3Endpoint: "http://" + s.testServer.Addr,
  70         }
  71         client := s3.New(*auth, region)
  72         return s3stage{
  73                 arv:  arv,
  74                 proj: proj,
  75                 projbucket: &s3.Bucket{
  76                         S3:   client,
  77                         Name: proj.UUID,
  78                 },
  79                 coll: coll,
  80                 collbucket: &s3.Bucket{
  81                         S3:   client,
  82                         Name: coll.UUID,
  83                 },
  84         }
  85 }
  86
  87 func (stage s3stage) teardown(c *check.C) {
  88         if stage.coll.UUID != "" {
  89                 err := stage.arv.RequestAndDecode(&stage.coll, "DELETE", "arvados/v1/collections/"+stage.coll.UUID, nil, nil)
  90                 c.Check(err, check.IsNil)
  91         }
  92 }
  93
  94 func (s *IntegrationSuite) TestS3CollectionGetObject(c *check.C) {
  95         stage := s.s3setup(c)
  96         defer stage.teardown(c)
  97         s.testS3GetObject(c, stage.collbucket, "")
  98 }
  99 func (s *IntegrationSuite) TestS3ProjectGetObject(c *check.C) {
 100         stage := s.s3setup(c)
 101         defer stage.teardown(c)
 102         s.testS3GetObject(c, stage.projbucket, stage.coll.Name+"/")
 103 }
 104 func (s *IntegrationSuite) testS3GetObject(c *check.C, bucket *s3.Bucket, prefix string) {
 105         rdr, err := bucket.GetReader(prefix + "emptyfile")
 106         c.Assert(err, check.IsNil)
 107         buf, err := ioutil.ReadAll(rdr)
 108         c.Check(err, check.IsNil)
 109         c.Check(len(buf), check.Equals, 0)
 110         err = rdr.Close()
 111         c.Check(err, check.IsNil)
 112
 113         rdr, err = bucket.GetReader(prefix + "missingfile")
 114         c.Check(err, check.ErrorMatches, `404 Not Found`)
 115
 116         rdr, err = bucket.GetReader(prefix + "sailboat.txt")
 117         c.Assert(err, check.IsNil)
 118         buf, err = ioutil.ReadAll(rdr)
 119         c.Check(err, check.IsNil)
 120         c.Check(buf, check.DeepEquals, []byte("⛵\n"))
 121         err = rdr.Close()
 122         c.Check(err, check.IsNil)
 123 }
 124
 125 func (s *IntegrationSuite) TestS3CollectionPutObjectSuccess(c *check.C) {
 126         stage := s.s3setup(c)
 127         defer stage.teardown(c)
 128         s.testS3PutObjectSuccess(c, stage.collbucket, "")
 129 }
 130 func (s *IntegrationSuite) TestS3ProjectPutObjectSuccess(c *check.C) {
 131         stage := s.s3setup(c)
 132         defer stage.teardown(c)
 133         s.testS3PutObjectSuccess(c, stage.projbucket, stage.coll.Name+"/")
 134 }
 135 func (s *IntegrationSuite) testS3PutObjectSuccess(c *check.C, bucket *s3.Bucket, prefix string) {
 136         for _, trial := range []struct {
 137                 path string
 138                 size int
 139         }{
 140                 {
 141                         path: "newfile",
 142                         size: 128000000,
 143                 }, {
 144                         path: "newdir/newfile",
 145                         size: 1 << 26,
 146                 }, {
 147                         path: "newdir1/newdir2/newfile",
 148                         size: 0,
 149                 },
 150         } {
 151                 c.Logf("=== %v", trial)
 152
 153                 objname := prefix + trial.path
 154
 155                 _, err := bucket.GetReader(objname)
 156                 c.Assert(err, check.ErrorMatches, `404 Not Found`)
 157
 158                 buf := make([]byte, trial.size)
 159                 rand.Read(buf)
 160
 161                 err = bucket.PutReader(objname, bytes.NewReader(buf), int64(len(buf)), "application/octet-stream", s3.Private, s3.Options{})
 162                 c.Check(err, check.IsNil)
 163
 164                 rdr, err := bucket.GetReader(objname)
 165                 if !c.Check(err, check.IsNil) {
 166                         continue
 167                 }
 168                 buf2, err := ioutil.ReadAll(rdr)
 169                 c.Check(err, check.IsNil)
 170                 c.Check(buf2, check.HasLen, len(buf))
 171                 c.Check(bytes.Equal(buf, buf2), check.Equals, true)
 172         }
 173 }
 174
 175 func (s *IntegrationSuite) TestS3CollectionPutObjectFailure(c *check.C) {
 176         stage := s.s3setup(c)
 177         defer stage.teardown(c)
 178         s.testS3PutObjectFailure(c, stage.collbucket, "")
 179 }
 180 func (s *IntegrationSuite) TestS3ProjectPutObjectFailure(c *check.C) {
 181         stage := s.s3setup(c)
 182         defer stage.teardown(c)
 183         s.testS3PutObjectFailure(c, stage.projbucket, stage.coll.Name+"/")
 184 }
 185 func (s *IntegrationSuite) testS3PutObjectFailure(c *check.C, bucket *s3.Bucket, prefix string) {
 186         var wg sync.WaitGroup
 187         for _, trial := range []struct {
 188                 path string
 189         }{
 190                 {
 191                         path: "emptyfile/newname", // emptyfile exists, see s3setup()
 192                 }, {
 193                         path: "emptyfile/", // emptyfile exists, see s3setup()
 194                 }, {
 195                         path: "emptydir", // dir already exists, see s3setup()
 196                 }, {
 197                         path: "emptydir/",
 198                 }, {
 199                         path: "emptydir//",
 200                 }, {
 201                         path: "newdir/",
 202                 }, {
 203                         path: "newdir//",
 204                 }, {
 205                         path: "/",
 206                 }, {
 207                         path: "//",
 208                 }, {
 209                         path: "foo//bar",
 210                 }, {
 211                         path: "",
 212                 },
 213         } {
 214                 trial := trial
 215                 wg.Add(1)
 216                 go func() {
 217                         defer wg.Done()
 218                         c.Logf("=== %v", trial)
 219
 220                         objname := prefix + trial.path
 221
 222                         buf := make([]byte, 1234)
 223                         rand.Read(buf)
 224
 225                         err := bucket.PutReader(objname, bytes.NewReader(buf), int64(len(buf)), "application/octet-stream", s3.Private, s3.Options{})
 226                         if !c.Check(err, check.ErrorMatches, `400 Bad.*`, check.Commentf("PUT %q should fail", objname)) {
 227                                 return
 228                         }
 229
 230                         _, err = bucket.GetReader(objname)
 231                         c.Check(err, check.ErrorMatches, `404 Not Found`, check.Commentf("GET %q should return 404", objname))
 232                 }()
 233         }
 234         wg.Wait()
 235 }