1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: AGPL-3.0
19 "git.curoverse.com/arvados.git/sdk/go/arvados"
20 "github.com/AdRoll/goamz/s3"
21 "github.com/AdRoll/goamz/s3/s3test"
22 check "gopkg.in/check.v1"
26 TestBucketName = "testbucket"
29 type fakeClock struct {
33 func (c *fakeClock) Now() time.Time {
41 // Deleting isn't safe from races, but if it's turned on
42 // anyway we do expect it to pass the generic volume tests.
46 var _ = check.Suite(&StubbedS3Suite{})
48 type StubbedS3Suite struct {
49 volumes []*TestableS3Volume
52 func (s *StubbedS3Suite) TestGeneric(c *check.C) {
53 DoGenericVolumeTests(c, func(t TB) TestableVolume {
54 // Use a negative raceWindow so s3test's 1-second
55 // timestamp precision doesn't confuse fixRace.
56 return s.newTestableVolume(c, -2*time.Second, false, 2)
60 func (s *StubbedS3Suite) TestGenericReadOnly(c *check.C) {
61 DoGenericVolumeTests(c, func(t TB) TestableVolume {
62 return s.newTestableVolume(c, -2*time.Second, true, 2)
66 func (s *StubbedS3Suite) TestIndex(c *check.C) {
67 v := s.newTestableVolume(c, 0, false, 2)
69 for i := 0; i < 256; i++ {
70 v.PutRaw(fmt.Sprintf("%02x%030x", i, i), []byte{102, 111, 111})
72 for _, spec := range []struct {
81 buf := new(bytes.Buffer)
82 err := v.IndexTo(spec.prefix, buf)
83 c.Check(err, check.IsNil)
85 idx := bytes.SplitAfter(buf.Bytes(), []byte{10})
86 c.Check(len(idx), check.Equals, spec.expectMatch+1)
87 c.Check(len(idx[len(idx)-1]), check.Equals, 0)
91 func (s *StubbedS3Suite) TestStats(c *check.C) {
92 v := s.newTestableVolume(c, 5*time.Minute, false, 2)
93 stats := func() string {
94 buf, err := json.Marshal(v.InternalStats())
95 c.Check(err, check.IsNil)
99 c.Check(stats(), check.Matches, `.*"Ops":0,.*`)
101 loc := "acbd18db4cc2f85cedef654fccc4a4d8"
102 _, err := v.Get(context.Background(), loc, make([]byte, 3))
103 c.Check(err, check.NotNil)
104 c.Check(stats(), check.Matches, `.*"Ops":[^0],.*`)
105 c.Check(stats(), check.Matches, `.*"\*s3.Error 404 [^"]*":[^0].*`)
106 c.Check(stats(), check.Matches, `.*"InBytes":0,.*`)
108 err = v.Put(context.Background(), loc, []byte("foo"))
109 c.Check(err, check.IsNil)
110 c.Check(stats(), check.Matches, `.*"OutBytes":3,.*`)
111 c.Check(stats(), check.Matches, `.*"PutOps":2,.*`)
113 _, err = v.Get(context.Background(), loc, make([]byte, 3))
114 c.Check(err, check.IsNil)
115 _, err = v.Get(context.Background(), loc, make([]byte, 3))
116 c.Check(err, check.IsNil)
117 c.Check(stats(), check.Matches, `.*"InBytes":6,.*`)
120 type blockingHandler struct {
121 requested chan *http.Request
122 unblock chan struct{}
125 func (h *blockingHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
126 if h.requested != nil {
129 if h.unblock != nil {
132 http.Error(w, "nothing here", http.StatusNotFound)
135 func (s *StubbedS3Suite) TestGetContextCancel(c *check.C) {
136 loc := "acbd18db4cc2f85cedef654fccc4a4d8"
137 buf := make([]byte, 3)
139 s.testContextCancel(c, func(ctx context.Context, v *TestableS3Volume) error {
140 _, err := v.Get(ctx, loc, buf)
145 func (s *StubbedS3Suite) TestCompareContextCancel(c *check.C) {
146 loc := "acbd18db4cc2f85cedef654fccc4a4d8"
149 s.testContextCancel(c, func(ctx context.Context, v *TestableS3Volume) error {
150 return v.Compare(ctx, loc, buf)
154 func (s *StubbedS3Suite) TestPutContextCancel(c *check.C) {
155 loc := "acbd18db4cc2f85cedef654fccc4a4d8"
158 s.testContextCancel(c, func(ctx context.Context, v *TestableS3Volume) error {
159 return v.Put(ctx, loc, buf)
163 func (s *StubbedS3Suite) testContextCancel(c *check.C, testFunc func(context.Context, *TestableS3Volume) error) {
164 handler := &blockingHandler{}
165 srv := httptest.NewServer(handler)
168 v := s.newTestableVolume(c, 5*time.Minute, false, 2)
170 vol.Endpoint = srv.URL
171 v = &TestableS3Volume{S3Volume: &vol}
174 ctx, cancel := context.WithCancel(context.Background())
176 handler.requested = make(chan *http.Request)
177 handler.unblock = make(chan struct{})
178 defer close(handler.unblock)
180 doneFunc := make(chan struct{})
182 err := testFunc(ctx, v)
183 c.Check(err, check.Equals, context.Canceled)
187 timeout := time.After(10 * time.Second)
189 // Wait for the stub server to receive a request, meaning
190 // Get() is waiting for an s3 operation.
193 c.Fatal("timed out waiting for test func to call our handler")
195 c.Fatal("test func finished without even calling our handler!")
196 case <-handler.requested:
208 func (s *StubbedS3Suite) TestBackendStates(c *check.C) {
209 defer func(tl, bs arvados.Duration) {
210 theConfig.TrashLifetime = tl
211 theConfig.BlobSignatureTTL = bs
212 }(theConfig.TrashLifetime, theConfig.BlobSignatureTTL)
213 theConfig.TrashLifetime.Set("1h")
214 theConfig.BlobSignatureTTL.Set("1h")
216 v := s.newTestableVolume(c, 5*time.Minute, false, 2)
219 putS3Obj := func(t time.Time, key string, data []byte) {
223 v.serverClock.now = &t
224 v.bucket.Put(key, data, "application/octet-stream", s3ACL, s3.Options{})
229 for _, scenario := range []struct {
236 canGetAfterTrash bool
238 haveTrashAfterEmpty bool
242 "No related objects",
244 false, false, false, false, false, false,
247 // Stored by older version, or there was a
248 // race between EmptyTrash and Put: Trash is a
249 // no-op even though the data object is very
252 t0.Add(-48 * time.Hour), none, none,
253 true, true, true, false, false, false,
256 "Not trash, but old enough to be eligible for trash",
257 t0.Add(-24 * time.Hour), t0.Add(-2 * time.Hour), none,
258 true, true, false, false, false, false,
261 "Not trash, and not old enough to be eligible for trash",
262 t0.Add(-24 * time.Hour), t0.Add(-30 * time.Minute), none,
263 true, true, true, false, false, false,
266 "Trashed + untrashed copies exist, due to recent race between Trash and Put",
267 t0.Add(-24 * time.Hour), t0.Add(-3 * time.Minute), t0.Add(-2 * time.Minute),
268 true, true, true, true, true, false,
271 "Trashed + untrashed copies exist, trash nearly eligible for deletion: prone to Trash race",
272 t0.Add(-24 * time.Hour), t0.Add(-12 * time.Hour), t0.Add(-59 * time.Minute),
273 true, false, true, true, true, false,
276 "Trashed + untrashed copies exist, trash is eligible for deletion: prone to Trash race",
277 t0.Add(-24 * time.Hour), t0.Add(-12 * time.Hour), t0.Add(-61 * time.Minute),
278 true, false, true, true, false, false,
281 "Trashed + untrashed copies exist, due to old race between Put and unfinished Trash: emptying trash is unsafe",
282 t0.Add(-24 * time.Hour), t0.Add(-12 * time.Hour), t0.Add(-12 * time.Hour),
283 true, false, true, true, true, true,
286 "Trashed + untrashed copies exist, used to be unsafe to empty, but since made safe by fixRace+Touch",
287 t0.Add(-time.Second), t0.Add(-time.Second), t0.Add(-12 * time.Hour),
288 true, true, true, true, false, false,
291 "Trashed + untrashed copies exist because Trash operation was interrupted (no race)",
292 t0.Add(-24 * time.Hour), t0.Add(-24 * time.Hour), t0.Add(-12 * time.Hour),
293 true, false, true, true, false, false,
296 "Trash, not yet eligible for deletion",
297 none, t0.Add(-12 * time.Hour), t0.Add(-time.Minute),
298 false, false, false, true, true, false,
301 "Trash, not yet eligible for deletion, prone to races",
302 none, t0.Add(-12 * time.Hour), t0.Add(-59 * time.Minute),
303 false, false, false, true, true, false,
306 "Trash, eligible for deletion",
307 none, t0.Add(-12 * time.Hour), t0.Add(-2 * time.Hour),
308 false, false, false, true, false, false,
311 "Erroneously trashed during a race, detected before TrashLifetime",
312 none, t0.Add(-30 * time.Minute), t0.Add(-29 * time.Minute),
313 true, false, true, true, true, false,
316 "Erroneously trashed during a race, rescue during EmptyTrash despite reaching TrashLifetime",
317 none, t0.Add(-90 * time.Minute), t0.Add(-89 * time.Minute),
318 true, false, true, true, true, false,
321 "Trashed copy exists with no recent/* marker (cause unknown); repair by untrashing",
322 none, none, t0.Add(-time.Minute),
323 false, false, false, true, true, true,
326 c.Log("Scenario: ", scenario.label)
328 // We have a few tests to run for each scenario, and
329 // the tests are expected to change state. By calling
330 // this setup func between tests, we (re)create the
331 // scenario as specified, using a new unique block
332 // locator to prevent interference from previous
335 setupScenario := func() (string, []byte) {
337 blk := []byte(fmt.Sprintf("%d", nextKey))
338 loc := fmt.Sprintf("%x", md5.Sum(blk))
340 putS3Obj(scenario.dataT, loc, blk)
341 putS3Obj(scenario.recentT, "recent/"+loc, nil)
342 putS3Obj(scenario.trashT, "trash/"+loc, blk)
343 v.serverClock.now = &t0
348 loc, blk := setupScenario()
349 buf := make([]byte, len(blk))
350 _, err := v.Get(context.Background(), loc, buf)
351 c.Check(err == nil, check.Equals, scenario.canGet)
353 c.Check(os.IsNotExist(err), check.Equals, true)
356 // Call Trash, then check canTrash and canGetAfterTrash
357 loc, blk = setupScenario()
359 c.Check(err == nil, check.Equals, scenario.canTrash)
360 _, err = v.Get(context.Background(), loc, buf)
361 c.Check(err == nil, check.Equals, scenario.canGetAfterTrash)
363 c.Check(os.IsNotExist(err), check.Equals, true)
366 // Call Untrash, then check canUntrash
367 loc, blk = setupScenario()
369 c.Check(err == nil, check.Equals, scenario.canUntrash)
370 if scenario.dataT != none || scenario.trashT != none {
371 // In all scenarios where the data exists, we
372 // should be able to Get after Untrash --
373 // regardless of timestamps, errors, race
375 _, err = v.Get(context.Background(), loc, buf)
376 c.Check(err, check.IsNil)
379 // Call EmptyTrash, then check haveTrashAfterEmpty and
381 loc, blk = setupScenario()
383 _, err = v.bucket.Head("trash/"+loc, nil)
384 c.Check(err == nil, check.Equals, scenario.haveTrashAfterEmpty)
385 if scenario.freshAfterEmpty {
386 t, err := v.Mtime(loc)
387 c.Check(err, check.IsNil)
388 // new mtime must be current (with an
389 // allowance for 1s timestamp precision)
390 c.Check(t.After(t0.Add(-time.Second)), check.Equals, true)
393 // Check for current Mtime after Put (applies to all
395 loc, blk = setupScenario()
396 err = v.Put(context.Background(), loc, blk)
397 c.Check(err, check.IsNil)
398 t, err := v.Mtime(loc)
399 c.Check(err, check.IsNil)
400 c.Check(t.After(t0.Add(-time.Second)), check.Equals, true)
404 type TestableS3Volume struct {
406 server *s3test.Server
408 serverClock *fakeClock
411 func (s *StubbedS3Suite) newTestableVolume(c *check.C, raceWindow time.Duration, readonly bool, replication int) *TestableS3Volume {
412 clock := &fakeClock{}
413 srv, err := s3test.NewServer(&s3test.Config{Clock: clock})
414 c.Assert(err, check.IsNil)
416 v := &TestableS3Volume{
418 Bucket: TestBucketName,
420 Region: "test-region-1",
421 LocationConstraint: true,
422 RaceWindow: arvados.Duration(raceWindow),
423 S3Replication: replication,
424 UnsafeDelete: s3UnsafeDelete,
433 err = v.bucket.PutBucket(s3.ACL("private"))
434 c.Assert(err, check.IsNil)
438 func (v *TestableS3Volume) Start() error {
439 tmp, err := ioutil.TempFile("", "keepstore")
440 v.c.Assert(err, check.IsNil)
441 defer os.Remove(tmp.Name())
442 _, err = tmp.Write([]byte("xxx\n"))
443 v.c.Assert(err, check.IsNil)
444 v.c.Assert(tmp.Close(), check.IsNil)
446 v.S3Volume.AccessKeyFile = tmp.Name()
447 v.S3Volume.SecretKeyFile = tmp.Name()
449 v.c.Assert(v.S3Volume.Start(), check.IsNil)
453 // PutRaw skips the ContentMD5 test
454 func (v *TestableS3Volume) PutRaw(loc string, block []byte) {
455 err := v.bucket.Put(loc, block, "application/octet-stream", s3ACL, s3.Options{})
457 log.Printf("PutRaw: %s: %+v", loc, err)
459 err = v.bucket.Put("recent/"+loc, nil, "application/octet-stream", s3ACL, s3.Options{})
461 log.Printf("PutRaw: recent/%s: %+v", loc, err)
465 // TouchWithDate turns back the clock while doing a Touch(). We assume
466 // there are no other operations happening on the same s3test server
468 func (v *TestableS3Volume) TouchWithDate(locator string, lastPut time.Time) {
469 v.serverClock.now = &lastPut
470 err := v.bucket.Put("recent/"+locator, nil, "application/octet-stream", s3ACL, s3.Options{})
474 v.serverClock.now = nil
477 func (v *TestableS3Volume) Teardown() {