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         "fmt"
  11         "io/ioutil"
  12         "os"
  13         "strings"
  14         "sync"
  15         "time"
  16
  17         "git.arvados.org/arvados.git/sdk/go/arvados"
  18         "git.arvados.org/arvados.git/sdk/go/arvadosclient"
  19         "git.arvados.org/arvados.git/sdk/go/arvadostest"
  20         "git.arvados.org/arvados.git/sdk/go/keepclient"
  21         "github.com/AdRoll/goamz/aws"
  22         "github.com/AdRoll/goamz/s3"
  23         check "gopkg.in/check.v1"
  24 )
  25
  26 type s3stage struct {
  27         arv        *arvados.Client
  28         ac         *arvadosclient.ArvadosClient
  29         kc         *keepclient.KeepClient
  30         proj       arvados.Group
  31         projbucket *s3.Bucket
  32         coll       arvados.Collection
  33         collbucket *s3.Bucket
  34 }
  35
  36 func (s *IntegrationSuite) s3setup(c *check.C) s3stage {
  37         var proj arvados.Group
  38         var coll arvados.Collection
  39         arv := arvados.NewClientFromEnv()
  40         arv.AuthToken = arvadostest.ActiveToken
  41         err := arv.RequestAndDecode(&proj, "POST", "arvados/v1/groups", nil, map[string]interface{}{
  42                 "group": map[string]interface{}{
  43                         "group_class": "project",
  44                         "name":        "keep-web s3 test",
  45                 },
  46                 "ensure_unique_name": true,
  47         })
  48         c.Assert(err, check.IsNil)
  49         err = arv.RequestAndDecode(&coll, "POST", "arvados/v1/collections", nil, map[string]interface{}{"collection": map[string]interface{}{
  50                 "owner_uuid":    proj.UUID,
  51                 "name":          "keep-web s3 test collection",
  52                 "manifest_text": ". d41d8cd98f00b204e9800998ecf8427e+0 0:0:emptyfile\n./emptydir d41d8cd98f00b204e9800998ecf8427e+0 0:0:.\n",
  53         }})
  54         c.Assert(err, check.IsNil)
  55         ac, err := arvadosclient.New(arv)
  56         c.Assert(err, check.IsNil)
  57         kc, err := keepclient.MakeKeepClient(ac)
  58         c.Assert(err, check.IsNil)
  59         fs, err := coll.FileSystem(arv, kc)
  60         c.Assert(err, check.IsNil)
  61         f, err := fs.OpenFile("sailboat.txt", os.O_CREATE|os.O_WRONLY, 0644)
  62         c.Assert(err, check.IsNil)
  63         _, err = f.Write([]byte("⛵\n"))
  64         c.Assert(err, check.IsNil)
  65         err = f.Close()
  66         c.Assert(err, check.IsNil)
  67         err = fs.Sync()
  68         c.Assert(err, check.IsNil)
  69         err = arv.RequestAndDecode(&coll, "GET", "arvados/v1/collections/"+coll.UUID, nil, nil)
  70         c.Assert(err, check.IsNil)
  71
  72         auth := aws.NewAuth(arvadostest.ActiveTokenV2, arvadostest.ActiveTokenV2, "", time.Now().Add(time.Hour))
  73         region := aws.Region{
  74                 Name:       s.testServer.Addr,
  75                 S3Endpoint: "http://" + s.testServer.Addr,
  76         }
  77         client := s3.New(*auth, region)
  78         return s3stage{
  79                 arv:  arv,
  80                 ac:   ac,
  81                 kc:   kc,
  82                 proj: proj,
  83                 projbucket: &s3.Bucket{
  84                         S3:   client,
  85                         Name: proj.UUID,
  86                 },
  87                 coll: coll,
  88                 collbucket: &s3.Bucket{
  89                         S3:   client,
  90                         Name: coll.UUID,
  91                 },
  92         }
  93 }
  94
  95 func (stage s3stage) teardown(c *check.C) {
  96         if stage.coll.UUID != "" {
  97                 err := stage.arv.RequestAndDecode(&stage.coll, "DELETE", "arvados/v1/collections/"+stage.coll.UUID, nil, nil)
  98                 c.Check(err, check.IsNil)
  99         }
 100         if stage.proj.UUID != "" {
 101                 err := stage.arv.RequestAndDecode(&stage.proj, "DELETE", "arvados/v1/groups/"+stage.proj.UUID, nil, nil)
 102                 c.Check(err, check.IsNil)
 103         }
 104 }
 105
 106 func (s *IntegrationSuite) TestS3CollectionGetObject(c *check.C) {
 107         stage := s.s3setup(c)
 108         defer stage.teardown(c)
 109         s.testS3GetObject(c, stage.collbucket, "")
 110 }
 111 func (s *IntegrationSuite) TestS3ProjectGetObject(c *check.C) {
 112         stage := s.s3setup(c)
 113         defer stage.teardown(c)
 114         s.testS3GetObject(c, stage.projbucket, stage.coll.Name+"/")
 115 }
 116 func (s *IntegrationSuite) testS3GetObject(c *check.C, bucket *s3.Bucket, prefix string) {
 117         rdr, err := bucket.GetReader(prefix + "emptyfile")
 118         c.Assert(err, check.IsNil)
 119         buf, err := ioutil.ReadAll(rdr)
 120         c.Check(err, check.IsNil)
 121         c.Check(len(buf), check.Equals, 0)
 122         err = rdr.Close()
 123         c.Check(err, check.IsNil)
 124
 125         rdr, err = bucket.GetReader(prefix + "missingfile")
 126         c.Check(err, check.ErrorMatches, `404 Not Found`)
 127
 128         rdr, err = bucket.GetReader(prefix + "sailboat.txt")
 129         c.Assert(err, check.IsNil)
 130         buf, err = ioutil.ReadAll(rdr)
 131         c.Check(err, check.IsNil)
 132         c.Check(buf, check.DeepEquals, []byte("⛵\n"))
 133         err = rdr.Close()
 134         c.Check(err, check.IsNil)
 135 }
 136
 137 func (s *IntegrationSuite) TestS3CollectionPutObjectSuccess(c *check.C) {
 138         stage := s.s3setup(c)
 139         defer stage.teardown(c)
 140         s.testS3PutObjectSuccess(c, stage.collbucket, "")
 141 }
 142 func (s *IntegrationSuite) TestS3ProjectPutObjectSuccess(c *check.C) {
 143         stage := s.s3setup(c)
 144         defer stage.teardown(c)
 145         s.testS3PutObjectSuccess(c, stage.projbucket, stage.coll.Name+"/")
 146 }
 147 func (s *IntegrationSuite) testS3PutObjectSuccess(c *check.C, bucket *s3.Bucket, prefix string) {
 148         for _, trial := range []struct {
 149                 path string
 150                 size int
 151         }{
 152                 {
 153                         path: "newfile",
 154                         size: 128000000,
 155                 }, {
 156                         path: "newdir/newfile",
 157                         size: 1 << 26,
 158                 }, {
 159                         path: "newdir1/newdir2/newfile",
 160                         size: 0,
 161                 },
 162         } {
 163                 c.Logf("=== %v", trial)
 164
 165                 objname := prefix + trial.path
 166
 167                 _, err := bucket.GetReader(objname)
 168                 c.Assert(err, check.ErrorMatches, `404 Not Found`)
 169
 170                 buf := make([]byte, trial.size)
 171                 rand.Read(buf)
 172
 173                 err = bucket.PutReader(objname, bytes.NewReader(buf), int64(len(buf)), "application/octet-stream", s3.Private, s3.Options{})
 174                 c.Check(err, check.IsNil)
 175
 176                 rdr, err := bucket.GetReader(objname)
 177                 if !c.Check(err, check.IsNil) {
 178                         continue
 179                 }
 180                 buf2, err := ioutil.ReadAll(rdr)
 181                 c.Check(err, check.IsNil)
 182                 c.Check(buf2, check.HasLen, len(buf))
 183                 c.Check(bytes.Equal(buf, buf2), check.Equals, true)
 184         }
 185 }
 186
 187 func (s *IntegrationSuite) TestS3CollectionPutObjectFailure(c *check.C) {
 188         stage := s.s3setup(c)
 189         defer stage.teardown(c)
 190         s.testS3PutObjectFailure(c, stage.collbucket, "")
 191 }
 192 func (s *IntegrationSuite) TestS3ProjectPutObjectFailure(c *check.C) {
 193         stage := s.s3setup(c)
 194         defer stage.teardown(c)
 195         s.testS3PutObjectFailure(c, stage.projbucket, stage.coll.Name+"/")
 196 }
 197 func (s *IntegrationSuite) testS3PutObjectFailure(c *check.C, bucket *s3.Bucket, prefix string) {
 198         var wg sync.WaitGroup
 199         for _, trial := range []struct {
 200                 path string
 201         }{
 202                 {
 203                         path: "emptyfile/newname", // emptyfile exists, see s3setup()
 204                 }, {
 205                         path: "emptyfile/", // emptyfile exists, see s3setup()
 206                 }, {
 207                         path: "emptydir", // dir already exists, see s3setup()
 208                 }, {
 209                         path: "emptydir/",
 210                 }, {
 211                         path: "emptydir//",
 212                 }, {
 213                         path: "newdir/",
 214                 }, {
 215                         path: "newdir//",
 216                 }, {
 217                         path: "/",
 218                 }, {
 219                         path: "//",
 220                 }, {
 221                         path: "foo//bar",
 222                 }, {
 223                         path: "",
 224                 },
 225         } {
 226                 trial := trial
 227                 wg.Add(1)
 228                 go func() {
 229                         defer wg.Done()
 230                         c.Logf("=== %v", trial)
 231
 232                         objname := prefix + trial.path
 233
 234                         buf := make([]byte, 1234)
 235                         rand.Read(buf)
 236
 237                         err := bucket.PutReader(objname, bytes.NewReader(buf), int64(len(buf)), "application/octet-stream", s3.Private, s3.Options{})
 238                         if !c.Check(err, check.ErrorMatches, `400 Bad.*`, check.Commentf("PUT %q should fail", objname)) {
 239                                 return
 240                         }
 241
 242                         if objname != "" && objname != "/" {
 243                                 _, err = bucket.GetReader(objname)
 244                                 c.Check(err, check.ErrorMatches, `404 Not Found`, check.Commentf("GET %q should return 404", objname))
 245                         }
 246                 }()
 247         }
 248         wg.Wait()
 249 }
 250
 251 func (stage *s3stage) writeBigDirs(c *check.C, dirs int, filesPerDir int) {
 252         fs, err := stage.coll.FileSystem(stage.arv, stage.kc)
 253         c.Assert(err, check.IsNil)
 254         for d := 0; d < dirs; d++ {
 255                 dir := fmt.Sprintf("dir%d", d)
 256                 c.Assert(fs.Mkdir(dir, 0755), check.IsNil)
 257                 for i := 0; i < filesPerDir; i++ {
 258                         f, err := fs.OpenFile(fmt.Sprintf("%s/file%d.txt", dir, i), os.O_CREATE|os.O_WRONLY, 0644)
 259                         c.Assert(err, check.IsNil)
 260                         c.Assert(f.Close(), check.IsNil)
 261                 }
 262         }
 263         c.Assert(fs.Sync(), check.IsNil)
 264 }
 265
 266 func (s *IntegrationSuite) TestS3CollectionList(c *check.C) {
 267         stage := s.s3setup(c)
 268         defer stage.teardown(c)
 269
 270         filesPerDir := 1001
 271         stage.writeBigDirs(c, 2, filesPerDir)
 272         s.testS3List(c, stage.collbucket, "", 4000, 2+filesPerDir*2)
 273         s.testS3List(c, stage.collbucket, "", 131, 2+filesPerDir*2)
 274         s.testS3List(c, stage.collbucket, "dir0/", 71, filesPerDir)
 275 }
 276 func (s *IntegrationSuite) testS3List(c *check.C, bucket *s3.Bucket, prefix string, pageSize, expectFiles int) {
 277         expectPageSize := pageSize
 278         if expectPageSize > 1000 {
 279                 expectPageSize = 1000
 280         }
 281         gotKeys := map[string]s3.Key{}
 282         nextMarker := ""
 283         pages := 0
 284         for {
 285                 resp, err := bucket.List(prefix, "", nextMarker, pageSize)
 286                 if !c.Check(err, check.IsNil) {
 287                         break
 288                 }
 289                 c.Check(len(resp.Contents) <= expectPageSize, check.Equals, true)
 290                 if pages++; !c.Check(pages <= (expectFiles/expectPageSize)+1, check.Equals, true) {
 291                         break
 292                 }
 293                 for _, key := range resp.Contents {
 294                         gotKeys[key.Key] = key
 295                 }
 296                 if !resp.IsTruncated {
 297                         c.Check(resp.NextMarker, check.Equals, "")
 298                         break
 299                 }
 300                 if !c.Check(resp.NextMarker, check.Not(check.Equals), "") {
 301                         break
 302                 }
 303                 nextMarker = resp.NextMarker
 304         }
 305         c.Check(len(gotKeys), check.Equals, expectFiles)
 306 }
 307
 308 func (s *IntegrationSuite) TestS3CollectionListRollup(c *check.C) {
 309         stage := s.s3setup(c)
 310         defer stage.teardown(c)
 311
 312         dirs := 2
 313         filesPerDir := 500
 314         stage.writeBigDirs(c, dirs, filesPerDir)
 315         err := stage.collbucket.PutReader("dingbats", &bytes.Buffer{}, 0, "application/octet-stream", s3.Private, s3.Options{})
 316         c.Assert(err, check.IsNil)
 317         var allfiles []string
 318         for marker := ""; ; {
 319                 resp, err := stage.collbucket.List("", "", marker, 20000)
 320                 c.Check(err, check.IsNil)
 321                 for _, key := range resp.Contents {
 322                         if len(allfiles) == 0 || allfiles[len(allfiles)-1] != key.Key {
 323                                 allfiles = append(allfiles, key.Key)
 324                         }
 325                 }
 326                 marker = resp.NextMarker
 327                 if marker == "" {
 328                         break
 329                 }
 330         }
 331         c.Check(allfiles, check.HasLen, dirs*filesPerDir+3)
 332
 333         for _, trial := range []struct {
 334                 prefix    string
 335                 delimiter string
 336                 marker    string
 337         }{
 338                 {"di", "/", ""},
 339                 {"di", "r", ""},
 340                 {"di", "n", ""},
 341                 {"dir0", "/", ""},
 342                 {"dir0", "/", "dir0/file14.txt"},       // no commonprefixes
 343                 {"", "", "dir0/file14.txt"},            // middle page, skip walking dir1
 344                 {"", "", "dir1/file14.txt"},            // middle page, skip walking dir0
 345                 {"", "", "dir1/file498.txt"},           // last page of results
 346                 {"dir1/file", "", "dir1/file498.txt"},  // last page of results, with prefix
 347                 {"dir1/file", "/", "dir1/file498.txt"}, // last page of results, with prefix + delimiter
 348                 {"dir1", "Z", "dir1/file498.txt"},      // delimiter "Z" never appears
 349                 {"dir2", "/", ""},                      // prefix "dir2" does not exist
 350                 {"", "/", ""},
 351         } {
 352                 c.Logf("\n\n=== trial %+v", trial)
 353
 354                 maxKeys := 20
 355                 resp, err := stage.collbucket.List(trial.prefix, trial.delimiter, trial.marker, maxKeys)
 356                 c.Check(err, check.IsNil)
 357                 if resp.IsTruncated && trial.delimiter == "" {
 358                         // goamz List method fills in the missing
 359                         // NextMarker field if resp.IsTruncated, so
 360                         // now we can't really tell whether it was
 361                         // sent by the server or by goamz. In cases
 362                         // where it should be empty but isn't, assume
 363                         // it's goamz's fault.
 364                         resp.NextMarker = ""
 365                 }
 366
 367                 var expectKeys []string
 368                 var expectPrefixes []string
 369                 var expectNextMarker string
 370                 var expectTruncated bool
 371                 for _, key := range allfiles {
 372                         full := len(expectKeys)+len(expectPrefixes) >= maxKeys
 373                         if !strings.HasPrefix(key, trial.prefix) || key < trial.marker {
 374                                 continue
 375                         } else if idx := strings.Index(key[len(trial.prefix):], trial.delimiter); trial.delimiter != "" && idx >= 0 {
 376                                 prefix := key[:len(trial.prefix)+idx+1]
 377                                 if len(expectPrefixes) > 0 && expectPrefixes[len(expectPrefixes)-1] == prefix {
 378                                         // same prefix as previous key
 379                                 } else if full {
 380                                         expectNextMarker = key
 381                                         expectTruncated = true
 382                                 } else {
 383                                         expectPrefixes = append(expectPrefixes, prefix)
 384                                 }
 385                         } else if full {
 386                                 if trial.delimiter != "" {
 387                                         expectNextMarker = key
 388                                 }
 389                                 expectTruncated = true
 390                                 break
 391                         } else {
 392                                 expectKeys = append(expectKeys, key)
 393                         }
 394                 }
 395
 396                 var gotKeys []string
 397                 for _, key := range resp.Contents {
 398                         gotKeys = append(gotKeys, key.Key)
 399                 }
 400                 var gotPrefixes []string
 401                 for _, prefix := range resp.CommonPrefixes {
 402                         gotPrefixes = append(gotPrefixes, prefix)
 403                 }
 404                 c.Check(gotKeys, check.DeepEquals, expectKeys)
 405                 c.Check(gotPrefixes, check.DeepEquals, expectPrefixes)
 406                 c.Check(resp.NextMarker, check.Equals, expectNextMarker)
 407                 c.Check(resp.IsTruncated, check.Equals, expectTruncated)
 408                 c.Logf("=== trial %+v keys %q prefixes %q nextMarker %q", trial, gotKeys, gotPrefixes, resp.NextMarker)
 409         }
 410 }