X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/00a96a5bd0203a1330963aaa3b7c3f5a2f970438..8a27fe370239ecb8e50d53f46b45ed61203a35ca:/services/keepstore/volume_generic_test.go diff --git a/services/keepstore/volume_generic_test.go b/services/keepstore/volume_generic_test.go index be83151461..0dd34e3af1 100644 --- a/services/keepstore/volume_generic_test.go +++ b/services/keepstore/volume_generic_test.go @@ -1,16 +1,27 @@ -package main +// Copyright (C) The Arvados Authors. All rights reserved. +// +// SPDX-License-Identifier: AGPL-3.0 + +package keepstore import ( "bytes" + "context" "crypto/md5" "fmt" "os" "regexp" "sort" + "strconv" "strings" "time" - "git.curoverse.com/arvados.git/sdk/go/arvadostest" + "git.arvados.org/arvados.git/sdk/go/arvados" + "git.arvados.org/arvados.git/sdk/go/arvadostest" + "git.arvados.org/arvados.git/sdk/go/ctxlog" + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + "github.com/sirupsen/logrus" ) type TB interface { @@ -28,87 +39,120 @@ type TB interface { // A TestableVolumeFactory returns a new TestableVolume. The factory // function, and the TestableVolume it returns, can use "t" to write // logs, fail the current test, etc. -type TestableVolumeFactory func(t TB) TestableVolume +type TestableVolumeFactory func(t TB, cluster *arvados.Cluster, volume arvados.Volume, logger logrus.FieldLogger, metrics *volumeMetricsVecs) TestableVolume // DoGenericVolumeTests runs a set of tests that every TestableVolume // is expected to pass. It calls factory to create a new TestableVolume // for each test case, to avoid leaking state between tests. -func DoGenericVolumeTests(t TB, factory TestableVolumeFactory) { - testGet(t, factory) - testGetNoSuchBlock(t, factory) +func DoGenericVolumeTests(t TB, readonly bool, factory TestableVolumeFactory) { + var s genericVolumeSuite + s.volume.ReadOnly = readonly + + s.testGet(t, factory) + s.testGetNoSuchBlock(t, factory) + + s.testCompareNonexistent(t, factory) + s.testCompareSameContent(t, factory, TestHash, TestBlock) + s.testCompareSameContent(t, factory, EmptyHash, EmptyBlock) + s.testCompareWithCollision(t, factory, TestHash, TestBlock, []byte("baddata")) + s.testCompareWithCollision(t, factory, TestHash, TestBlock, EmptyBlock) + s.testCompareWithCollision(t, factory, EmptyHash, EmptyBlock, TestBlock) + s.testCompareWithCorruptStoredData(t, factory, TestHash, TestBlock, []byte("baddata")) + s.testCompareWithCorruptStoredData(t, factory, TestHash, TestBlock, EmptyBlock) + s.testCompareWithCorruptStoredData(t, factory, EmptyHash, EmptyBlock, []byte("baddata")) + + if !readonly { + s.testPutBlockWithSameContent(t, factory, TestHash, TestBlock) + s.testPutBlockWithSameContent(t, factory, EmptyHash, EmptyBlock) + s.testPutBlockWithDifferentContent(t, factory, arvadostest.MD5CollisionMD5, arvadostest.MD5CollisionData[0], arvadostest.MD5CollisionData[1]) + s.testPutBlockWithDifferentContent(t, factory, arvadostest.MD5CollisionMD5, EmptyBlock, arvadostest.MD5CollisionData[0]) + s.testPutBlockWithDifferentContent(t, factory, arvadostest.MD5CollisionMD5, arvadostest.MD5CollisionData[0], EmptyBlock) + s.testPutBlockWithDifferentContent(t, factory, EmptyHash, EmptyBlock, arvadostest.MD5CollisionData[0]) + s.testPutMultipleBlocks(t, factory) + + s.testPutAndTouch(t, factory) + } + s.testTouchNoSuchBlock(t, factory) - testCompareNonexistent(t, factory) - testCompareSameContent(t, factory, TestHash, TestBlock) - testCompareSameContent(t, factory, EmptyHash, EmptyBlock) - testCompareWithCollision(t, factory, TestHash, TestBlock, []byte("baddata")) - testCompareWithCollision(t, factory, TestHash, TestBlock, EmptyBlock) - testCompareWithCollision(t, factory, EmptyHash, EmptyBlock, TestBlock) - testCompareWithCorruptStoredData(t, factory, TestHash, TestBlock, []byte("baddata")) - testCompareWithCorruptStoredData(t, factory, TestHash, TestBlock, EmptyBlock) - testCompareWithCorruptStoredData(t, factory, EmptyHash, EmptyBlock, []byte("baddata")) + s.testMtimeNoSuchBlock(t, factory) - testPutBlockWithSameContent(t, factory, TestHash, TestBlock) - testPutBlockWithSameContent(t, factory, EmptyHash, EmptyBlock) - testPutBlockWithDifferentContent(t, factory, arvadostest.MD5CollisionMD5, arvadostest.MD5CollisionData[0], arvadostest.MD5CollisionData[1]) - testPutBlockWithDifferentContent(t, factory, arvadostest.MD5CollisionMD5, EmptyBlock, arvadostest.MD5CollisionData[0]) - testPutBlockWithDifferentContent(t, factory, arvadostest.MD5CollisionMD5, arvadostest.MD5CollisionData[0], EmptyBlock) - testPutBlockWithDifferentContent(t, factory, EmptyHash, EmptyBlock, arvadostest.MD5CollisionData[0]) - testPutMultipleBlocks(t, factory) + s.testIndexTo(t, factory) - testPutAndTouch(t, factory) - testTouchNoSuchBlock(t, factory) + if !readonly { + s.testDeleteNewBlock(t, factory) + s.testDeleteOldBlock(t, factory) + } + s.testDeleteNoSuchBlock(t, factory) - testMtimeNoSuchBlock(t, factory) + s.testStatus(t, factory) - testIndexTo(t, factory) + s.testMetrics(t, readonly, factory) - testDeleteNewBlock(t, factory) - testDeleteOldBlock(t, factory) - testDeleteNoSuchBlock(t, factory) + s.testString(t, factory) - testStatus(t, factory) + if readonly { + s.testUpdateReadOnly(t, factory) + } - testString(t, factory) + s.testGetConcurrent(t, factory) + if !readonly { + s.testPutConcurrent(t, factory) - testUpdateReadOnly(t, factory) + s.testPutFullBlock(t, factory) + } - testGetConcurrent(t, factory) - testPutConcurrent(t, factory) + s.testTrashUntrash(t, readonly, factory) + s.testTrashEmptyTrashUntrash(t, factory) +} - testPutFullBlock(t, factory) +type genericVolumeSuite struct { + cluster *arvados.Cluster + volume arvados.Volume + logger logrus.FieldLogger + metrics *volumeMetricsVecs + registry *prometheus.Registry +} + +func (s *genericVolumeSuite) setup(t TB) { + s.cluster = testCluster(t) + s.logger = ctxlog.TestLogger(t) + s.registry = prometheus.NewRegistry() + s.metrics = newVolumeMetricsVecs(s.registry) +} - testTrashUntrash(t, factory) - testTrashEmptyTrashUntrash(t, factory) - testTrashUntrashWithEmptyTrashGoroutine(t, factory) +func (s *genericVolumeSuite) newVolume(t TB, factory TestableVolumeFactory) TestableVolume { + return factory(t, s.cluster, s.volume, s.logger, s.metrics) } // Put a test block, get it and verify content // Test should pass for both writable and read-only volumes -func testGet(t TB, factory TestableVolumeFactory) { - v := factory(t) +func (s *genericVolumeSuite) testGet(t TB, factory TestableVolumeFactory) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() v.PutRaw(TestHash, TestBlock) - buf, err := v.Get(TestHash) + buf := make([]byte, BlockSize) + n, err := v.Get(context.Background(), TestHash, buf) if err != nil { t.Fatal(err) } - bufs.Put(buf) - - if bytes.Compare(buf, TestBlock) != 0 { + if bytes.Compare(buf[:n], TestBlock) != 0 { t.Errorf("expected %s, got %s", string(TestBlock), string(buf)) } } // Invoke get on a block that does not exist in volume; should result in error // Test should pass for both writable and read-only volumes -func testGetNoSuchBlock(t TB, factory TestableVolumeFactory) { - v := factory(t) +func (s *genericVolumeSuite) testGetNoSuchBlock(t TB, factory TestableVolumeFactory) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() - if _, err := v.Get(TestHash2); err == nil { + buf := make([]byte, BlockSize) + if _, err := v.Get(context.Background(), TestHash2, buf); err == nil { t.Errorf("Expected error while getting non-existing block %v", TestHash2) } } @@ -116,11 +160,12 @@ func testGetNoSuchBlock(t TB, factory TestableVolumeFactory) { // Compare() should return os.ErrNotExist if the block does not exist. // Otherwise, writing new data causes CompareAndTouch() to generate // error logs even though everything is working fine. -func testCompareNonexistent(t TB, factory TestableVolumeFactory) { - v := factory(t) +func (s *genericVolumeSuite) testCompareNonexistent(t TB, factory TestableVolumeFactory) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() - err := v.Compare(TestHash, TestBlock) + err := v.Compare(context.Background(), TestHash, TestBlock) if err != os.ErrNotExist { t.Errorf("Got err %T %q, expected os.ErrNotExist", err, err) } @@ -128,14 +173,15 @@ func testCompareNonexistent(t TB, factory TestableVolumeFactory) { // Put a test block and compare the locator with same content // Test should pass for both writable and read-only volumes -func testCompareSameContent(t TB, factory TestableVolumeFactory, testHash string, testData []byte) { - v := factory(t) +func (s *genericVolumeSuite) testCompareSameContent(t TB, factory TestableVolumeFactory, testHash string, testData []byte) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() v.PutRaw(testHash, testData) // Compare the block locator with same content - err := v.Compare(testHash, testData) + err := v.Compare(context.Background(), testHash, testData) if err != nil { t.Errorf("Got err %q, expected nil", err) } @@ -146,14 +192,15 @@ func testCompareSameContent(t TB, factory TestableVolumeFactory, testHash string // testHash = md5(testDataA). // // Test should pass for both writable and read-only volumes -func testCompareWithCollision(t TB, factory TestableVolumeFactory, testHash string, testDataA, testDataB []byte) { - v := factory(t) +func (s *genericVolumeSuite) testCompareWithCollision(t TB, factory TestableVolumeFactory, testHash string, testDataA, testDataB []byte) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() v.PutRaw(testHash, testDataA) // Compare the block locator with different content; collision - err := v.Compare(TestHash, testDataB) + err := v.Compare(context.Background(), TestHash, testDataB) if err == nil { t.Errorf("Got err nil, expected error due to collision") } @@ -163,13 +210,14 @@ func testCompareWithCollision(t TB, factory TestableVolumeFactory, testHash stri // corrupted. Requires testHash = md5(testDataA) != md5(testDataB). // // Test should pass for both writable and read-only volumes -func testCompareWithCorruptStoredData(t TB, factory TestableVolumeFactory, testHash string, testDataA, testDataB []byte) { - v := factory(t) +func (s *genericVolumeSuite) testCompareWithCorruptStoredData(t TB, factory TestableVolumeFactory, testHash string, testDataA, testDataB []byte) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() v.PutRaw(TestHash, testDataB) - err := v.Compare(testHash, testDataA) + err := v.Compare(context.Background(), testHash, testDataA) if err == nil || err == CollisionError { t.Errorf("Got err %+v, expected non-collision error", err) } @@ -177,20 +225,17 @@ func testCompareWithCorruptStoredData(t TB, factory TestableVolumeFactory, testH // Put a block and put again with same content // Test is intended for only writable volumes -func testPutBlockWithSameContent(t TB, factory TestableVolumeFactory, testHash string, testData []byte) { - v := factory(t) +func (s *genericVolumeSuite) testPutBlockWithSameContent(t TB, factory TestableVolumeFactory, testHash string, testData []byte) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() - if v.Writable() == false { - return - } - - err := v.Put(testHash, testData) + err := v.Put(context.Background(), testHash, testData) if err != nil { t.Errorf("Got err putting block %q: %q, expected nil", TestBlock, err) } - err = v.Put(testHash, testData) + err = v.Put(context.Background(), testHash, testData) if err != nil { t.Errorf("Got err putting block second time %q: %q, expected nil", TestBlock, err) } @@ -198,90 +243,80 @@ func testPutBlockWithSameContent(t TB, factory TestableVolumeFactory, testHash s // Put a block and put again with different content // Test is intended for only writable volumes -func testPutBlockWithDifferentContent(t TB, factory TestableVolumeFactory, testHash string, testDataA, testDataB []byte) { - v := factory(t) +func (s *genericVolumeSuite) testPutBlockWithDifferentContent(t TB, factory TestableVolumeFactory, testHash string, testDataA, testDataB []byte) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() - if v.Writable() == false { - return - } - v.PutRaw(testHash, testDataA) - putErr := v.Put(testHash, testDataB) - buf, getErr := v.Get(testHash) + putErr := v.Put(context.Background(), testHash, testDataB) + buf := make([]byte, BlockSize) + n, getErr := v.Get(context.Background(), testHash, buf) if putErr == nil { // Put must not return a nil error unless it has // overwritten the existing data. - if bytes.Compare(buf, testDataB) != 0 { - t.Errorf("Put succeeded but Get returned %+q, expected %+q", buf, testDataB) + if bytes.Compare(buf[:n], testDataB) != 0 { + t.Errorf("Put succeeded but Get returned %+q, expected %+q", buf[:n], testDataB) } } else { // It is permissible for Put to fail, but it must // leave us with either the original data, the new // data, or nothing at all. - if getErr == nil && bytes.Compare(buf, testDataA) != 0 && bytes.Compare(buf, testDataB) != 0 { - t.Errorf("Put failed but Get returned %+q, which is neither %+q nor %+q", buf, testDataA, testDataB) + if getErr == nil && bytes.Compare(buf[:n], testDataA) != 0 && bytes.Compare(buf[:n], testDataB) != 0 { + t.Errorf("Put failed but Get returned %+q, which is neither %+q nor %+q", buf[:n], testDataA, testDataB) } } - if getErr == nil { - bufs.Put(buf) - } } // Put and get multiple blocks // Test is intended for only writable volumes -func testPutMultipleBlocks(t TB, factory TestableVolumeFactory) { - v := factory(t) +func (s *genericVolumeSuite) testPutMultipleBlocks(t TB, factory TestableVolumeFactory) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() - if v.Writable() == false { - return - } - - err := v.Put(TestHash, TestBlock) + err := v.Put(context.Background(), TestHash, TestBlock) if err != nil { t.Errorf("Got err putting block %q: %q, expected nil", TestBlock, err) } - err = v.Put(TestHash2, TestBlock2) + err = v.Put(context.Background(), TestHash2, TestBlock2) if err != nil { t.Errorf("Got err putting block %q: %q, expected nil", TestBlock2, err) } - err = v.Put(TestHash3, TestBlock3) + err = v.Put(context.Background(), TestHash3, TestBlock3) if err != nil { t.Errorf("Got err putting block %q: %q, expected nil", TestBlock3, err) } - data, err := v.Get(TestHash) + data := make([]byte, BlockSize) + n, err := v.Get(context.Background(), TestHash, data) if err != nil { t.Error(err) } else { - if bytes.Compare(data, TestBlock) != 0 { - t.Errorf("Block present, but got %+q, expected %+q", data, TestBlock) + if bytes.Compare(data[:n], TestBlock) != 0 { + t.Errorf("Block present, but got %+q, expected %+q", data[:n], TestBlock) } - bufs.Put(data) } - data, err = v.Get(TestHash2) + n, err = v.Get(context.Background(), TestHash2, data) if err != nil { t.Error(err) } else { - if bytes.Compare(data, TestBlock2) != 0 { - t.Errorf("Block present, but got %+q, expected %+q", data, TestBlock2) + if bytes.Compare(data[:n], TestBlock2) != 0 { + t.Errorf("Block present, but got %+q, expected %+q", data[:n], TestBlock2) } - bufs.Put(data) } - data, err = v.Get(TestHash3) + n, err = v.Get(context.Background(), TestHash3, data) if err != nil { t.Error(err) } else { - if bytes.Compare(data, TestBlock3) != 0 { - t.Errorf("Block present, but to %+q, expected %+q", data, TestBlock3) + if bytes.Compare(data[:n], TestBlock3) != 0 { + t.Errorf("Block present, but to %+q, expected %+q", data[:n], TestBlock3) } - bufs.Put(data) } } @@ -289,15 +324,12 @@ func testPutMultipleBlocks(t TB, factory TestableVolumeFactory) { // Test that when applying PUT to a block that already exists, // the block's modification time is updated. // Test is intended for only writable volumes -func testPutAndTouch(t TB, factory TestableVolumeFactory) { - v := factory(t) +func (s *genericVolumeSuite) testPutAndTouch(t TB, factory TestableVolumeFactory) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() - if v.Writable() == false { - return - } - - if err := v.Put(TestHash, TestBlock); err != nil { + if err := v.Put(context.Background(), TestHash, TestBlock); err != nil { t.Error(err) } @@ -317,7 +349,7 @@ func testPutAndTouch(t TB, factory TestableVolumeFactory) { } // Write the same block again. - if err := v.Put(TestHash, TestBlock); err != nil { + if err := v.Put(context.Background(), TestHash, TestBlock); err != nil { t.Error(err) } @@ -331,8 +363,9 @@ func testPutAndTouch(t TB, factory TestableVolumeFactory) { // Touching a non-existing block should result in error. // Test should pass for both writable and read-only volumes -func testTouchNoSuchBlock(t TB, factory TestableVolumeFactory) { - v := factory(t) +func (s *genericVolumeSuite) testTouchNoSuchBlock(t TB, factory TestableVolumeFactory) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() if err := v.Touch(TestHash); err == nil { @@ -342,8 +375,9 @@ func testTouchNoSuchBlock(t TB, factory TestableVolumeFactory) { // Invoking Mtime on a non-existing block should result in error. // Test should pass for both writable and read-only volumes -func testMtimeNoSuchBlock(t TB, factory TestableVolumeFactory) { - v := factory(t) +func (s *genericVolumeSuite) testMtimeNoSuchBlock(t TB, factory TestableVolumeFactory) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() if _, err := v.Mtime("12345678901234567890123456789012"); err == nil { @@ -356,14 +390,27 @@ func testMtimeNoSuchBlock(t TB, factory TestableVolumeFactory) { // * with a prefix // * with no such prefix // Test should pass for both writable and read-only volumes -func testIndexTo(t TB, factory TestableVolumeFactory) { - v := factory(t) +func (s *genericVolumeSuite) testIndexTo(t TB, factory TestableVolumeFactory) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() + // minMtime and maxMtime are the minimum and maximum + // acceptable values the index can report for our test + // blocks. 1-second precision is acceptable. + minMtime := time.Now().UTC().UnixNano() + minMtime -= minMtime % 1e9 + v.PutRaw(TestHash, TestBlock) v.PutRaw(TestHash2, TestBlock2) v.PutRaw(TestHash3, TestBlock3) + maxMtime := time.Now().UTC().UnixNano() + if maxMtime%1e9 > 0 { + maxMtime -= maxMtime % 1e9 + maxMtime += 1e9 + } + // Blocks whose names aren't Keep hashes should be omitted from // index v.PutRaw("fffffffffnotreallyahashfffffffff", nil) @@ -376,15 +423,21 @@ func testIndexTo(t TB, factory TestableVolumeFactory) { indexRows := strings.Split(string(buf.Bytes()), "\n") sort.Strings(indexRows) sortedIndex := strings.Join(indexRows, "\n") - m, err := regexp.MatchString( - `^\n`+TestHash+`\+\d+ \d+\n`+ - TestHash3+`\+\d+ \d+\n`+ - TestHash2+`\+\d+ \d+$`, - sortedIndex) - if err != nil { - t.Error(err) - } else if !m { + m := regexp.MustCompile( + `^\n` + TestHash + `\+\d+ (\d+)\n` + + TestHash3 + `\+\d+ \d+\n` + + TestHash2 + `\+\d+ \d+$`, + ).FindStringSubmatch(sortedIndex) + if m == nil { t.Errorf("Got index %q for empty prefix", sortedIndex) + } else { + mtime, err := strconv.ParseInt(m[1], 10, 64) + if err != nil { + t.Error(err) + } else if mtime < minMtime || mtime > maxMtime { + t.Errorf("got %d for TestHash timestamp, expected %d <= t <= %d", + mtime, minMtime, maxMtime) + } } for _, prefix := range []string{"f", "f15", "f15ac"} { @@ -401,7 +454,7 @@ func testIndexTo(t TB, factory TestableVolumeFactory) { for _, prefix := range []string{"zero", "zip", "zilch"} { buf = new(bytes.Buffer) - v.IndexTo(prefix, buf) + err := v.IndexTo(prefix, buf) if err != nil { t.Errorf("Got error on IndexTo with no such prefix %v", err.Error()) } else if buf.Len() != 0 { @@ -413,58 +466,73 @@ func testIndexTo(t TB, factory TestableVolumeFactory) { // Calling Delete() for a block immediately after writing it (not old enough) // should neither delete the data nor return an error. // Test is intended for only writable volumes -func testDeleteNewBlock(t TB, factory TestableVolumeFactory) { - v := factory(t) +func (s *genericVolumeSuite) testDeleteNewBlock(t TB, factory TestableVolumeFactory) { + s.setup(t) + s.cluster.Collections.BlobSigningTTL.Set("5m") + v := s.newVolume(t, factory) defer v.Teardown() - blobSignatureTTL = 300 * time.Second - - if v.Writable() == false { - return - } - v.Put(TestHash, TestBlock) + v.Put(context.Background(), TestHash, TestBlock) if err := v.Trash(TestHash); err != nil { t.Error(err) } - data, err := v.Get(TestHash) + data := make([]byte, BlockSize) + n, err := v.Get(context.Background(), TestHash, data) if err != nil { t.Error(err) - } else { - if bytes.Compare(data, TestBlock) != 0 { - t.Errorf("Got data %+q, expected %+q", data, TestBlock) - } - bufs.Put(data) + } else if bytes.Compare(data[:n], TestBlock) != 0 { + t.Errorf("Got data %+q, expected %+q", data[:n], TestBlock) } } // Calling Delete() for a block with a timestamp older than -// blobSignatureTTL seconds in the past should delete the data. -// Test is intended for only writable volumes -func testDeleteOldBlock(t TB, factory TestableVolumeFactory) { - v := factory(t) +// BlobSigningTTL seconds in the past should delete the data. Test is +// intended for only writable volumes +func (s *genericVolumeSuite) testDeleteOldBlock(t TB, factory TestableVolumeFactory) { + s.setup(t) + s.cluster.Collections.BlobSigningTTL.Set("5m") + v := s.newVolume(t, factory) defer v.Teardown() - blobSignatureTTL = 300 * time.Second - if v.Writable() == false { - return - } - - v.Put(TestHash, TestBlock) - v.TouchWithDate(TestHash, time.Now().Add(-2*blobSignatureTTL)) + v.Put(context.Background(), TestHash, TestBlock) + v.TouchWithDate(TestHash, time.Now().Add(-2*s.cluster.Collections.BlobSigningTTL.Duration())) if err := v.Trash(TestHash); err != nil { t.Error(err) } - if _, err := v.Get(TestHash); err == nil || !os.IsNotExist(err) { + data := make([]byte, BlockSize) + if _, err := v.Get(context.Background(), TestHash, data); err == nil || !os.IsNotExist(err) { t.Errorf("os.IsNotExist(%v) should have been true", err) } + + _, err := v.Mtime(TestHash) + if err == nil || !os.IsNotExist(err) { + t.Fatalf("os.IsNotExist(%v) should have been true", err) + } + + err = v.Compare(context.Background(), TestHash, TestBlock) + if err == nil || !os.IsNotExist(err) { + t.Fatalf("os.IsNotExist(%v) should have been true", err) + } + + indexBuf := new(bytes.Buffer) + v.IndexTo("", indexBuf) + if strings.Contains(string(indexBuf.Bytes()), TestHash) { + t.Fatalf("Found trashed block in IndexTo") + } + + err = v.Touch(TestHash) + if err == nil || !os.IsNotExist(err) { + t.Fatalf("os.IsNotExist(%v) should have been true", err) + } } // Calling Delete() for a block that does not exist should result in error. // Test should pass for both writable and read-only volumes -func testDeleteNoSuchBlock(t TB, factory TestableVolumeFactory) { - v := factory(t) +func (s *genericVolumeSuite) testDeleteNoSuchBlock(t TB, factory TestableVolumeFactory) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() if err := v.Trash(TestHash2); err == nil { @@ -474,8 +542,9 @@ func testDeleteNoSuchBlock(t TB, factory TestableVolumeFactory) { // Invoke Status and verify that VolumeStatus is returned // Test should pass for both writable and read-only volumes -func testStatus(t TB, factory TestableVolumeFactory) { - v := factory(t) +func (s *genericVolumeSuite) testStatus(t TB, factory TestableVolumeFactory) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() // Get node status and make a basic sanity check. @@ -493,10 +562,84 @@ func testStatus(t TB, factory TestableVolumeFactory) { } } +func getValueFrom(cv *prometheus.CounterVec, lbls prometheus.Labels) float64 { + c, _ := cv.GetMetricWith(lbls) + pb := &dto.Metric{} + c.Write(pb) + return pb.GetCounter().GetValue() +} + +func (s *genericVolumeSuite) testMetrics(t TB, readonly bool, factory TestableVolumeFactory) { + var err error + + s.setup(t) + v := s.newVolume(t, factory) + defer v.Teardown() + + opsC, _, ioC := s.metrics.getCounterVecsFor(prometheus.Labels{"device_id": v.GetDeviceID()}) + + if ioC == nil { + t.Error("ioBytes CounterVec is nil") + return + } + + if getValueFrom(ioC, prometheus.Labels{"direction": "out"})+ + getValueFrom(ioC, prometheus.Labels{"direction": "in"}) > 0 { + t.Error("ioBytes counter should be zero") + } + + if opsC == nil { + t.Error("opsCounter CounterVec is nil") + return + } + + var c, writeOpCounter, readOpCounter float64 + + readOpType, writeOpType := v.ReadWriteOperationLabelValues() + writeOpCounter = getValueFrom(opsC, prometheus.Labels{"operation": writeOpType}) + readOpCounter = getValueFrom(opsC, prometheus.Labels{"operation": readOpType}) + + // Test Put if volume is writable + if !readonly { + err = v.Put(context.Background(), TestHash, TestBlock) + if err != nil { + t.Errorf("Got err putting block %q: %q, expected nil", TestBlock, err) + } + // Check that the write operations counter increased + c = getValueFrom(opsC, prometheus.Labels{"operation": writeOpType}) + if c <= writeOpCounter { + t.Error("Operation(s) not counted on Put") + } + // Check that bytes counter is > 0 + if getValueFrom(ioC, prometheus.Labels{"direction": "out"}) == 0 { + t.Error("ioBytes{direction=out} counter shouldn't be zero") + } + } else { + v.PutRaw(TestHash, TestBlock) + } + + buf := make([]byte, BlockSize) + _, err = v.Get(context.Background(), TestHash, buf) + if err != nil { + t.Fatal(err) + } + + // Check that the operations counter increased + c = getValueFrom(opsC, prometheus.Labels{"operation": readOpType}) + if c <= readOpCounter { + t.Error("Operation(s) not counted on Get") + } + // Check that the bytes "in" counter is > 0 + if getValueFrom(ioC, prometheus.Labels{"direction": "in"}) == 0 { + t.Error("ioBytes{direction=in} counter shouldn't be zero") + } +} + // Invoke String for the volume; expect non-empty result // Test should pass for both writable and read-only volumes -func testString(t TB, factory TestableVolumeFactory) { - v := factory(t) +func (s *genericVolumeSuite) testString(t TB, factory TestableVolumeFactory) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() if id := v.String(); len(id) == 0 { @@ -506,28 +649,26 @@ func testString(t TB, factory TestableVolumeFactory) { // Putting, updating, touching, and deleting blocks from a read-only volume result in error. // Test is intended for only read-only volumes -func testUpdateReadOnly(t TB, factory TestableVolumeFactory) { - v := factory(t) +func (s *genericVolumeSuite) testUpdateReadOnly(t TB, factory TestableVolumeFactory) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() - if v.Writable() == true { - return - } - v.PutRaw(TestHash, TestBlock) + buf := make([]byte, BlockSize) // Get from read-only volume should succeed - _, err := v.Get(TestHash) + _, err := v.Get(context.Background(), TestHash, buf) if err != nil { t.Errorf("got err %v, expected nil", err) } // Put a new block to read-only volume should result in error - err = v.Put(TestHash2, TestBlock2) + err = v.Put(context.Background(), TestHash2, TestBlock2) if err == nil { t.Errorf("Expected error when putting block in a read-only volume") } - _, err = v.Get(TestHash2) + _, err = v.Get(context.Background(), TestHash2, buf) if err == nil { t.Errorf("Expected error when getting block whose put in read-only volume failed") } @@ -545,7 +686,7 @@ func testUpdateReadOnly(t TB, factory TestableVolumeFactory) { } // Overwriting an existing block in read-only volume should result in error - err = v.Put(TestHash, TestBlock) + err = v.Put(context.Background(), TestHash, TestBlock) if err == nil { t.Errorf("Expected error when putting block in a read-only volume") } @@ -553,8 +694,9 @@ func testUpdateReadOnly(t TB, factory TestableVolumeFactory) { // Launch concurrent Gets // Test should pass for both writable and read-only volumes -func testGetConcurrent(t TB, factory TestableVolumeFactory) { - v := factory(t) +func (s *genericVolumeSuite) testGetConcurrent(t TB, factory TestableVolumeFactory) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() v.PutRaw(TestHash, TestBlock) @@ -562,61 +704,58 @@ func testGetConcurrent(t TB, factory TestableVolumeFactory) { v.PutRaw(TestHash3, TestBlock3) sem := make(chan int) - go func(sem chan int) { - buf, err := v.Get(TestHash) + go func() { + buf := make([]byte, BlockSize) + n, err := v.Get(context.Background(), TestHash, buf) if err != nil { t.Errorf("err1: %v", err) } - bufs.Put(buf) - if bytes.Compare(buf, TestBlock) != 0 { - t.Errorf("buf should be %s, is %s", string(TestBlock), string(buf)) + if bytes.Compare(buf[:n], TestBlock) != 0 { + t.Errorf("buf should be %s, is %s", string(TestBlock), string(buf[:n])) } sem <- 1 - }(sem) + }() - go func(sem chan int) { - buf, err := v.Get(TestHash2) + go func() { + buf := make([]byte, BlockSize) + n, err := v.Get(context.Background(), TestHash2, buf) if err != nil { t.Errorf("err2: %v", err) } - bufs.Put(buf) - if bytes.Compare(buf, TestBlock2) != 0 { - t.Errorf("buf should be %s, is %s", string(TestBlock2), string(buf)) + if bytes.Compare(buf[:n], TestBlock2) != 0 { + t.Errorf("buf should be %s, is %s", string(TestBlock2), string(buf[:n])) } sem <- 1 - }(sem) + }() - go func(sem chan int) { - buf, err := v.Get(TestHash3) + go func() { + buf := make([]byte, BlockSize) + n, err := v.Get(context.Background(), TestHash3, buf) if err != nil { t.Errorf("err3: %v", err) } - bufs.Put(buf) - if bytes.Compare(buf, TestBlock3) != 0 { - t.Errorf("buf should be %s, is %s", string(TestBlock3), string(buf)) + if bytes.Compare(buf[:n], TestBlock3) != 0 { + t.Errorf("buf should be %s, is %s", string(TestBlock3), string(buf[:n])) } sem <- 1 - }(sem) + }() // Wait for all goroutines to finish - for done := 0; done < 3; { - done += <-sem + for done := 0; done < 3; done++ { + <-sem } } // Launch concurrent Puts // Test is intended for only writable volumes -func testPutConcurrent(t TB, factory TestableVolumeFactory) { - v := factory(t) +func (s *genericVolumeSuite) testPutConcurrent(t TB, factory TestableVolumeFactory) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() - if v.Writable() == false { - return - } - sem := make(chan int) go func(sem chan int) { - err := v.Put(TestHash, TestBlock) + err := v.Put(context.Background(), TestHash, TestBlock) if err != nil { t.Errorf("err1: %v", err) } @@ -624,7 +763,7 @@ func testPutConcurrent(t TB, factory TestableVolumeFactory) { }(sem) go func(sem chan int) { - err := v.Put(TestHash2, TestBlock2) + err := v.Put(context.Background(), TestHash2, TestBlock2) if err != nil { t.Errorf("err2: %v", err) } @@ -632,7 +771,7 @@ func testPutConcurrent(t TB, factory TestableVolumeFactory) { }(sem) go func(sem chan int) { - err := v.Put(TestHash3, TestBlock3) + err := v.Put(context.Background(), TestHash3, TestBlock3) if err != nil { t.Errorf("err3: %v", err) } @@ -640,105 +779,96 @@ func testPutConcurrent(t TB, factory TestableVolumeFactory) { }(sem) // Wait for all goroutines to finish - for done := 0; done < 3; { - done += <-sem + for done := 0; done < 3; done++ { + <-sem } // Double check that we actually wrote the blocks we expected to write. - buf, err := v.Get(TestHash) + buf := make([]byte, BlockSize) + n, err := v.Get(context.Background(), TestHash, buf) if err != nil { t.Errorf("Get #1: %v", err) } - bufs.Put(buf) - if bytes.Compare(buf, TestBlock) != 0 { - t.Errorf("Get #1: expected %s, got %s", string(TestBlock), string(buf)) + if bytes.Compare(buf[:n], TestBlock) != 0 { + t.Errorf("Get #1: expected %s, got %s", string(TestBlock), string(buf[:n])) } - buf, err = v.Get(TestHash2) + n, err = v.Get(context.Background(), TestHash2, buf) if err != nil { t.Errorf("Get #2: %v", err) } - bufs.Put(buf) - if bytes.Compare(buf, TestBlock2) != 0 { - t.Errorf("Get #2: expected %s, got %s", string(TestBlock2), string(buf)) + if bytes.Compare(buf[:n], TestBlock2) != 0 { + t.Errorf("Get #2: expected %s, got %s", string(TestBlock2), string(buf[:n])) } - buf, err = v.Get(TestHash3) + n, err = v.Get(context.Background(), TestHash3, buf) if err != nil { t.Errorf("Get #3: %v", err) } - bufs.Put(buf) - if bytes.Compare(buf, TestBlock3) != 0 { - t.Errorf("Get #3: expected %s, got %s", string(TestBlock3), string(buf)) + if bytes.Compare(buf[:n], TestBlock3) != 0 { + t.Errorf("Get #3: expected %s, got %s", string(TestBlock3), string(buf[:n])) } } // Write and read back a full size block -func testPutFullBlock(t TB, factory TestableVolumeFactory) { - v := factory(t) +func (s *genericVolumeSuite) testPutFullBlock(t TB, factory TestableVolumeFactory) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() - if !v.Writable() { - return - } - wdata := make([]byte, BlockSize) wdata[0] = 'a' wdata[BlockSize-1] = 'z' hash := fmt.Sprintf("%x", md5.Sum(wdata)) - err := v.Put(hash, wdata) + err := v.Put(context.Background(), hash, wdata) if err != nil { t.Fatal(err) } - rdata, err := v.Get(hash) + buf := make([]byte, BlockSize) + n, err := v.Get(context.Background(), hash, buf) if err != nil { t.Error(err) - } else { - defer bufs.Put(rdata) } - if bytes.Compare(rdata, wdata) != 0 { - t.Error("rdata != wdata") + if bytes.Compare(buf[:n], wdata) != 0 { + t.Error("buf %+q != wdata %+q", buf[:n], wdata) } } -// With trashLifetime != 0, perform: +// With BlobTrashLifetime != 0, perform: // Trash an old block - which either raises ErrNotImplemented or succeeds // Untrash - which either raises ErrNotImplemented or succeeds // Get - which must succeed -func testTrashUntrash(t TB, factory TestableVolumeFactory) { - v := factory(t) +func (s *genericVolumeSuite) testTrashUntrash(t TB, readonly bool, factory TestableVolumeFactory) { + s.setup(t) + s.cluster.Collections.BlobTrashLifetime.Set("1h") + v := s.newVolume(t, factory) defer v.Teardown() - defer func() { - trashLifetime = 0 * time.Second - }() - - trashLifetime = 3600 * time.Second // put block and backdate it v.PutRaw(TestHash, TestBlock) - v.TouchWithDate(TestHash, time.Now().Add(-2*blobSignatureTTL)) + v.TouchWithDate(TestHash, time.Now().Add(-2*s.cluster.Collections.BlobSigningTTL.Duration())) - buf, err := v.Get(TestHash) + buf := make([]byte, BlockSize) + n, err := v.Get(context.Background(), TestHash, buf) if err != nil { t.Fatal(err) } - if bytes.Compare(buf, TestBlock) != 0 { - t.Errorf("Got data %+q, expected %+q", buf, TestBlock) + if bytes.Compare(buf[:n], TestBlock) != 0 { + t.Errorf("Got data %+q, expected %+q", buf[:n], TestBlock) } - bufs.Put(buf) // Trash err = v.Trash(TestHash) - if v.Writable() == false { + if readonly { if err != MethodDisabledError { - t.Error(err) + t.Fatal(err) } } else if err != nil { if err != ErrNotImplemented { - t.Error(err) + t.Fatal(err) } } else { - _, err = v.Get(TestHash) + _, err = v.Get(context.Background(), TestHash, buf) if err == nil || !os.IsNotExist(err) { t.Errorf("os.IsNotExist(%v) should have been true", err) } @@ -751,128 +881,153 @@ func testTrashUntrash(t TB, factory TestableVolumeFactory) { } // Get the block - after trash and untrash sequence - buf, err = v.Get(TestHash) + n, err = v.Get(context.Background(), TestHash, buf) if err != nil { t.Fatal(err) } - if bytes.Compare(buf, TestBlock) != 0 { - t.Errorf("Got data %+q, expected %+q", buf, TestBlock) + if bytes.Compare(buf[:n], TestBlock) != 0 { + t.Errorf("Got data %+q, expected %+q", buf[:n], TestBlock) } - bufs.Put(buf) } -func testTrashEmptyTrashUntrash(t TB, factory TestableVolumeFactory) { - v := factory(t) +func (s *genericVolumeSuite) testTrashEmptyTrashUntrash(t TB, factory TestableVolumeFactory) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() - defer func() { - trashLifetime = 0 * time.Second - }() - // First set of tests + checkGet := func() error { + buf := make([]byte, BlockSize) + n, err := v.Get(context.Background(), TestHash, buf) + if err != nil { + return err + } + if bytes.Compare(buf[:n], TestBlock) != 0 { + t.Fatalf("Got data %+q, expected %+q", buf[:n], TestBlock) + } - // With trashLifetime = 1h, test trash/untrash cycle. - trashLifetime = 1 * time.Hour + _, err = v.Mtime(TestHash) + if err != nil { + return err + } + + err = v.Compare(context.Background(), TestHash, TestBlock) + if err != nil { + return err + } + + indexBuf := new(bytes.Buffer) + v.IndexTo("", indexBuf) + if !strings.Contains(string(indexBuf.Bytes()), TestHash) { + return os.ErrNotExist + } + + return nil + } + + // First set: EmptyTrash before reaching the trash deadline. + + s.cluster.Collections.BlobTrashLifetime.Set("1h") - // put block and backdate it v.PutRaw(TestHash, TestBlock) - v.TouchWithDate(TestHash, time.Now().Add(-2*blobSignatureTTL)) + v.TouchWithDate(TestHash, time.Now().Add(-2*s.cluster.Collections.BlobSigningTTL.Duration())) - buf, err := v.Get(TestHash) + err := checkGet() if err != nil { t.Fatal(err) } - if bytes.Compare(buf, TestBlock) != 0 { - t.Fatalf("Got data %+q, expected %+q", buf, TestBlock) - } - bufs.Put(buf) - // Trash it + // Trash the block err = v.Trash(TestHash) if err == MethodDisabledError || err == ErrNotImplemented { + // Skip the trash tests for read-only volumes, and + // volume types that don't support + // BlobTrashLifetime>0. return } - buf, err = v.Get(TestHash) + err = checkGet() + if err == nil || !os.IsNotExist(err) { + t.Fatalf("os.IsNotExist(%v) should have been true", err) + } + + err = v.Touch(TestHash) if err == nil || !os.IsNotExist(err) { t.Fatalf("os.IsNotExist(%v) should have been true", err) } - // Empty trash; the block is still within trashLifetime and hence is not emptied v.EmptyTrash() - // Untrash will hence rescue it + // Even after emptying the trash, we can untrash our block + // because the deadline hasn't been reached. err = v.Untrash(TestHash) if err != nil { t.Fatal(err) } - // Get block will find it - buf, err = v.Get(TestHash) + err = checkGet() if err != nil { t.Fatal(err) } - if bytes.Compare(buf, TestBlock) != 0 { - t.Fatalf("Got data %+q, expected %+q", buf, TestBlock) + + err = v.Touch(TestHash) + if err != nil { + t.Fatal(err) } - bufs.Put(buf) - // Untrash again; should fail + // Because we Touch'ed, need to backdate again for next set of tests + v.TouchWithDate(TestHash, time.Now().Add(-2*s.cluster.Collections.BlobSigningTTL.Duration())) + + // If the only block in the trash has already been untrashed, + // most volumes will fail a subsequent Untrash with a 404, but + // it's also acceptable for Untrash to succeed. err = v.Untrash(TestHash) - if err == nil || !os.IsNotExist(err) { - t.Fatalf("os.IsNotExist(%v) should have been true", err) + if err != nil && !os.IsNotExist(err) { + t.Fatalf("Expected success or os.IsNotExist(), but got: %v", err) } - buf, err = v.Get(TestHash) + // The additional Untrash should not interfere with our + // already-untrashed copy. + err = checkGet() if err != nil { t.Fatal(err) } - if bytes.Compare(buf, TestBlock) != 0 { - t.Fatalf("Got data %+q, expected %+q", buf, TestBlock) - } - bufs.Put(buf) - // Second set of tests + // Untrash might have updated the timestamp, so backdate again + v.TouchWithDate(TestHash, time.Now().Add(-2*s.cluster.Collections.BlobSigningTTL.Duration())) + + // Second set: EmptyTrash after the trash deadline has passed. - // With trashLifetime = 1ns, test trash/untrash cycle. - trashLifetime = 1 * time.Nanosecond + s.cluster.Collections.BlobTrashLifetime.Set("1ns") - // Trash it err = v.Trash(TestHash) if err != nil { t.Fatal(err) } - buf, err = v.Get(TestHash) + err = checkGet() if err == nil || !os.IsNotExist(err) { t.Fatalf("os.IsNotExist(%v) should have been true", err) } - // Untrash + // Even though 1ns has passed, we can untrash because we + // haven't called EmptyTrash yet. err = v.Untrash(TestHash) if err != nil { t.Fatal(err) } - - // Get block will find it - buf, err = v.Get(TestHash) + err = checkGet() if err != nil { t.Fatal(err) } - if bytes.Compare(buf, TestBlock) != 0 { - t.Fatalf("Got data %+q, expected %+q", buf, TestBlock) - } - bufs.Put(buf) - // Trash it again - err = v.Trash(TestHash) - if err == MethodDisabledError || err == ErrNotImplemented { - return - } - buf, err = v.Get(TestHash) + // Trash it again, and this time call EmptyTrash so it really + // goes away. + // (In Azure volumes, un/trash changes Mtime, so first backdate again) + v.TouchWithDate(TestHash, time.Now().Add(-2*s.cluster.Collections.BlobSigningTTL.Duration())) + _ = v.Trash(TestHash) + err = checkGet() if err == nil || !os.IsNotExist(err) { - t.Errorf("os.IsNotExist(%v) should have been true", err) + t.Fatalf("os.IsNotExist(%v) should have been true", err) } - - // Empty trash will empty it v.EmptyTrash() // Untrash won't find it @@ -882,142 +1037,70 @@ func testTrashEmptyTrashUntrash(t TB, factory TestableVolumeFactory) { } // Get block won't find it - buf, err = v.Get(TestHash) + err = checkGet() if err == nil || !os.IsNotExist(err) { t.Fatalf("os.IsNotExist(%v) should have been true", err) } - // Third set of tests + // Third set: If the same data block gets written again after + // being trashed, and then the trash gets emptied, the newer + // un-trashed copy doesn't get deleted along with it. - // Still with trashLifetime = 1ns: put, trash, put one more, trash etc - // put block and backdate it v.PutRaw(TestHash, TestBlock) - v.TouchWithDate(TestHash, time.Now().Add(-2*blobSignatureTTL)) + v.TouchWithDate(TestHash, time.Now().Add(-2*s.cluster.Collections.BlobSigningTTL.Duration())) - // Trash + s.cluster.Collections.BlobTrashLifetime.Set("1ns") err = v.Trash(TestHash) - if err == MethodDisabledError || err == ErrNotImplemented { - return + if err != nil { + t.Fatal(err) } - buf, err = v.Get(TestHash) + err = checkGet() if err == nil || !os.IsNotExist(err) { t.Fatalf("os.IsNotExist(%v) should have been true", err) } - // put again - err = v.Put(TestHash, TestBlock) - if err != nil { - t.Fatal(err) - } - v.TouchWithDate(TestHash, time.Now().Add(-2*blobSignatureTTL)) + v.PutRaw(TestHash, TestBlock) + v.TouchWithDate(TestHash, time.Now().Add(-2*s.cluster.Collections.BlobSigningTTL.Duration())) - // Empty trash will empty the trashed block but the second one is untouched + // EmptyTrash should not delete the untrashed copy. v.EmptyTrash() - - // Get block should work because of the second block - buf, err = v.Get(TestHash) + err = checkGet() if err != nil { t.Fatal(err) } - if bytes.Compare(buf, TestBlock) != 0 { - t.Fatalf("Got data %+q, expected %+q", buf, TestBlock) - } - bufs.Put(buf) - // set trashLifetime to one hour - trashLifetime = 1 * time.Hour + // Fourth set: If the same data block gets trashed twice with + // different deadlines A and C, and then the trash is emptied + // at intermediate time B (A < B < C), it is still possible to + // untrash the block whose deadline is "C". - // trash block - err = v.Trash(TestHash) - if err != nil { - t.Fatal(err) - } - - // Empty trash won't empty this second block which is still within trashLifetime - v.EmptyTrash() - - // Untrash; the second block which is still within trashLifetime will be rescued - err = v.Untrash(TestHash) - if err != nil { - t.Fatal(err) - } + v.PutRaw(TestHash, TestBlock) + v.TouchWithDate(TestHash, time.Now().Add(-2*s.cluster.Collections.BlobSigningTTL.Duration())) - // Get block should work because of the second block - buf, err = v.Get(TestHash) + s.cluster.Collections.BlobTrashLifetime.Set("1ns") + err = v.Trash(TestHash) if err != nil { t.Fatal(err) } - if bytes.Compare(buf, TestBlock) != 0 { - t.Fatalf("Got data %+q, expected %+q", buf, TestBlock) - } - bufs.Put(buf) -} -// With trashLifetime = 1ns, perform: -// Run emptyTrash goroutine -// Trash an old block - which either raises ErrNotImplemented or succeeds -// Untrash - after emptyTrash goroutine ticks, and hence does not actually untrash -// Get - which must fail to find the block -func testTrashUntrashWithEmptyTrashGoroutine(t TB, factory TestableVolumeFactory) { - v := factory(t) - defer v.Teardown() - - doneEmptyingTrash := make(chan bool) - defer func() { - trashLifetime = 0 * time.Second - doneEmptyingTrash <- true - }() - - volumes = append(volumes, v) - - trashLifetime = 1 * time.Nanosecond - trashCheckInterval = 1 * time.Nanosecond - - go emptyTrash(doneEmptyingTrash, trashCheckInterval) - - // Trash old block and untrash a little after first trashCheckInterval v.PutRaw(TestHash, TestBlock) - v.TouchWithDate(TestHash, time.Now().Add(-2*blobSignatureTTL)) + v.TouchWithDate(TestHash, time.Now().Add(-2*s.cluster.Collections.BlobSigningTTL.Duration())) - buf, err := v.Get(TestHash) + s.cluster.Collections.BlobTrashLifetime.Set("1h") + err = v.Trash(TestHash) if err != nil { t.Fatal(err) } - if bytes.Compare(buf, TestBlock) != 0 { - t.Fatalf("Got data %+q, expected %+q", buf, TestBlock) - } - bufs.Put(buf) - - // Trash - err = v.Trash(TestHash) - if err == MethodDisabledError || err == ErrNotImplemented { - return - } - _, err = v.Get(TestHash) - if err == nil || !os.IsNotExist(err) { - t.Fatalf("os.IsNotExist(%v) should have been true", err) - } - - time.Sleep(2 * time.Nanosecond) - - // Untrash + // EmptyTrash should not prevent us from recovering the + // time.Hour ("C") trash + v.EmptyTrash() err = v.Untrash(TestHash) if err != nil { t.Fatal(err) } - - // Get is expected to fail due to EmptyTrash before Untrash - // It is still found on readonly volumes - buf, err = v.Get(TestHash) + err = checkGet() if err != nil { - if !os.IsNotExist(err) { - t.Errorf("os.IsNotExist(%v) should have been true", err) - } - } else { - if bytes.Compare(buf, TestBlock) != 0 { - t.Errorf("Got data %+q, expected %+q", buf, TestBlock) - } - bufs.Put(buf) + t.Fatal(err) } }