X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/cb48eb95d516b1d4a1ffe18be34703005531117a..6b9cb6715bb7c04b2a425bc28b41a51f8a273a7a:/services/keepstore/volume_generic_test.go diff --git a/services/keepstore/volume_generic_test.go b/services/keepstore/volume_generic_test.go index 6dca74eaf2..16084058b7 100644 --- a/services/keepstore/volume_generic_test.go +++ b/services/keepstore/volume_generic_test.go @@ -1,161 +1,166 @@ -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" - "testing" + "sync" "time" + + "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 { + Error(args ...interface{}) + Errorf(format string, args ...interface{}) + Fail() + FailNow() + Failed() bool + Fatal(args ...interface{}) + Fatalf(format string, args ...interface{}) + Log(args ...interface{}) + Logf(format string, args ...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 *testing.T) TestableVolume +type TestableVolumeFactory func(t TB, params newVolumeParams) 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 *testing.T, factory TestableVolumeFactory) { - testGet(t, factory) - testGetNoSuchBlock(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")) - - testPutBlockWithSameContent(t, factory, TestHash, TestBlock) - testPutBlockWithSameContent(t, factory, EmptyHash, EmptyBlock) - testPutBlockWithDifferentContent(t, factory, TestHash, TestBlock, TestBlock2) - testPutBlockWithDifferentContent(t, factory, TestHash, EmptyBlock, TestBlock) - testPutBlockWithDifferentContent(t, factory, TestHash, TestBlock, EmptyBlock) - testPutBlockWithDifferentContent(t, factory, EmptyHash, EmptyBlock, TestBlock) - testPutMultipleBlocks(t, factory) - - testPutAndTouch(t, factory) - testTouchNoSuchBlock(t, factory) +func DoGenericVolumeTests(t TB, readonly bool, factory TestableVolumeFactory) { + var s genericVolumeSuite + s.volume.ReadOnly = readonly - testMtimeNoSuchBlock(t, factory) + s.testGet(t, factory) + s.testGetNoSuchBlock(t, factory) - testIndexTo(t, factory) + 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) - testDeleteNewBlock(t, factory) - testDeleteOldBlock(t, factory) - testDeleteNoSuchBlock(t, factory) - - testStatus(t, factory) - - testString(t, factory) - - testUpdateReadOnly(t, factory) - - testGetConcurrent(t, factory) - testPutConcurrent(t, factory) -} + s.testPutAndTouch(t, factory) + } + s.testTouchNoSuchBlock(t, factory) -// Put a test block, get it and verify content -// Test should pass for both writable and read-only volumes -func testGet(t *testing.T, factory TestableVolumeFactory) { - v := factory(t) - defer v.Teardown() + s.testMtimeNoSuchBlock(t, factory) - v.PutRaw(TestHash, TestBlock) + s.testIndex(t, factory) - buf, err := v.Get(TestHash) - if err != nil { - t.Fatal(err) + if !readonly { + s.testDeleteNewBlock(t, factory) + s.testDeleteOldBlock(t, factory) } + s.testDeleteNoSuchBlock(t, factory) - bufs.Put(buf) + s.testMetrics(t, readonly, factory) - if bytes.Compare(buf, TestBlock) != 0 { - t.Errorf("expected %s, got %s", string(TestBlock), string(buf)) + s.testGetConcurrent(t, factory) + if !readonly { + s.testPutConcurrent(t, factory) + s.testPutFullBlock(t, factory) + s.testTrashUntrash(t, readonly, factory) + s.testTrashEmptyTrashUntrash(t, factory) } } -// 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 *testing.T, factory TestableVolumeFactory) { - v := factory(t) - defer v.Teardown() - - if _, err := v.Get(TestHash2); err == nil { - t.Errorf("Expected error while getting non-existing block %v", TestHash2) - } +type genericVolumeSuite struct { + cluster *arvados.Cluster + volume arvados.Volume + logger logrus.FieldLogger + metrics *volumeMetricsVecs + registry *prometheus.Registry + bufferPool *bufferPool } -// Put a test block and compare the locator with same content -// Test should pass for both writable and read-only volumes -func testCompareSameContent(t *testing.T, factory TestableVolumeFactory, testHash string, testData []byte) { - v := factory(t) - defer v.Teardown() - - v.PutRaw(testHash, testData) +func (s *genericVolumeSuite) setup(t TB) { + s.cluster = testCluster(t) + s.logger = ctxlog.TestLogger(t) + s.registry = prometheus.NewRegistry() + s.metrics = newVolumeMetricsVecs(s.registry) + s.bufferPool = newBufferPool(s.logger, 8, s.registry) +} - // Compare the block locator with same content - err := v.Compare(testHash, testData) - if err != nil { - t.Errorf("Got err %q, expected nil", err) - } +func (s *genericVolumeSuite) newVolume(t TB, factory TestableVolumeFactory) TestableVolume { + return factory(t, newVolumeParams{ + UUID: "zzzzz-nyw5e-999999999999999", + Cluster: s.cluster, + ConfigVolume: s.volume, + Logger: s.logger, + MetricsVecs: s.metrics, + BufferPool: s.bufferPool, + }) } -// Test behavior of Compare() when stored data matches expected -// checksum but differs from new data we need to store. Requires -// testHash = md5(testDataA). -// +// Put a test block, get it and verify content // Test should pass for both writable and read-only volumes -func testCompareWithCollision(t *testing.T, factory TestableVolumeFactory, testHash string, testDataA, testDataB []byte) { - 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, testDataA) + err := v.BlockWrite(context.Background(), TestHash, TestBlock) + if err != nil { + t.Error(err) + } - // Compare the block locator with different content; collision - err := v.Compare(TestHash, testDataB) - if err == nil { - t.Errorf("Got err nil, expected error due to collision") + buf := &brbuffer{} + err = v.BlockRead(context.Background(), TestHash, buf) + if err != nil { + t.Error(err) + } + if bytes.Compare(buf.Bytes(), TestBlock) != 0 { + t.Errorf("expected %s, got %s", "foo", buf.String()) } } -// Test behavior of Compare() when stored data has become -// corrupted. Requires testHash = md5(testDataA) != md5(testDataB). -// +// 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 testCompareWithCorruptStoredData(t *testing.T, factory TestableVolumeFactory, testHash string, testDataA, testDataB []byte) { - v := factory(t) +func (s *genericVolumeSuite) testGetNoSuchBlock(t TB, factory TestableVolumeFactory) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() - v.PutRaw(TestHash, testDataB) - - err := v.Compare(testHash, testDataA) - if err == nil || err == CollisionError { - t.Errorf("Got err %+v, expected non-collision error", err) + if err := v.BlockRead(context.Background(), barHash, brdiscard); err == nil { + t.Errorf("Expected error while getting non-existing block %v", barHash) } } // Put a block and put again with same content // Test is intended for only writable volumes -func testPutBlockWithSameContent(t *testing.T, 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.BlockWrite(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.BlockWrite(context.Background(), testHash, testData) if err != nil { t.Errorf("Got err putting block second time %q: %q, expected nil", TestBlock, err) } @@ -163,115 +168,100 @@ func testPutBlockWithSameContent(t *testing.T, factory TestableVolumeFactory, te // Put a block and put again with different content // Test is intended for only writable volumes -func testPutBlockWithDifferentContent(t *testing.T, 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 - } - - err := v.Put(testHash, testDataA) - if err != nil { - t.Errorf("Got err putting block %q: %q, expected nil", testDataA, err) - } + v.BlockWrite(context.Background(), testHash, testDataA) - putErr := v.Put(testHash, testDataB) - buf, getErr := v.Get(testHash) + putErr := v.BlockWrite(context.Background(), testHash, testDataB) + buf := &brbuffer{} + getErr := v.BlockRead(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 { + if buf.String() != string(testDataB) { t.Errorf("Put succeeded but Get returned %+q, expected %+q", buf, 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 { + if getErr == nil && buf.String() != string(testDataA) && buf.String() != string(testDataB) { t.Errorf("Put failed but Get returned %+q, which is neither %+q nor %+q", buf, testDataA, testDataB) } } - if getErr == nil { - bufs.Put(buf) - } } // Put and get multiple blocks // Test is intended for only writable volumes -func testPutMultipleBlocks(t *testing.T, 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.BlockWrite(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.BlockWrite(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.BlockWrite(context.Background(), TestHash3, TestBlock3) if err != nil { t.Errorf("Got err putting block %q: %q, expected nil", TestBlock3, err) } - data, err := v.Get(TestHash) + buf := &brbuffer{} + err = v.BlockRead(context.Background(), TestHash, buf) 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(buf.Bytes(), TestBlock) != 0 { + t.Errorf("Block present, but got %+q, expected %+q", buf, TestBlock) } - bufs.Put(data) } - data, err = v.Get(TestHash2) + buf.Reset() + err = v.BlockRead(context.Background(), TestHash2, buf) 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(buf.Bytes(), TestBlock2) != 0 { + t.Errorf("Block present, but got %+q, expected %+q", buf, TestBlock2) } - bufs.Put(data) } - data, err = v.Get(TestHash3) + buf.Reset() + err = v.BlockRead(context.Background(), TestHash3, buf) 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(buf.Bytes(), TestBlock3) != 0 { + t.Errorf("Block present, but to %+q, expected %+q", buf, TestBlock3) } - bufs.Put(data) } } -// testPutAndTouch -// 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 *testing.T, factory TestableVolumeFactory) { - v := factory(t) +// testPutAndTouch checks that when applying PUT to a block that +// already exists, the block's modification time is updated. Intended +// for only writable volumes. +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.BlockWrite(context.Background(), TestHash, TestBlock); err != nil { t.Error(err) } // We'll verify { t0 < threshold < t1 }, where t0 is the - // existing block's timestamp on disk before Put() and t1 is - // its timestamp after Put(). + // existing block's timestamp on disk before BlockWrite() and t1 is + // its timestamp after BlockWrite(). threshold := time.Now().Add(-time.Second) // Set the stored block's mtime far enough in the past that we @@ -285,7 +275,7 @@ func testPutAndTouch(t *testing.T, factory TestableVolumeFactory) { } // Write the same block again. - if err := v.Put(TestHash, TestBlock); err != nil { + if err := v.BlockWrite(context.Background(), TestHash, TestBlock); err != nil { t.Error(err) } @@ -299,19 +289,21 @@ func testPutAndTouch(t *testing.T, factory TestableVolumeFactory) { // Touching a non-existing block should result in error. // Test should pass for both writable and read-only volumes -func testTouchNoSuchBlock(t *testing.T, 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 { + if err := v.BlockTouch(TestHash); err == nil { t.Error("Expected error when attempted to touch a non-existing block") } } // Invoking Mtime on a non-existing block should result in error. // Test should pass for both writable and read-only volumes -func testMtimeNoSuchBlock(t *testing.T, 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 { @@ -319,38 +311,64 @@ func testMtimeNoSuchBlock(t *testing.T, factory TestableVolumeFactory) { } } -// Put a few blocks and invoke IndexTo with: +// Put a few blocks and invoke Index with: // * no prefix // * with a prefix // * with no such prefix // Test should pass for both writable and read-only volumes -func testIndexTo(t *testing.T, factory TestableVolumeFactory) { - v := factory(t) +func (s *genericVolumeSuite) testIndex(t TB, factory TestableVolumeFactory) { + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() - v.PutRaw(TestHash, TestBlock) - v.PutRaw(TestHash2, TestBlock2) - v.PutRaw(TestHash3, TestBlock3) + // 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.BlockWrite(context.Background(), TestHash, TestBlock) + v.BlockWrite(context.Background(), TestHash2, TestBlock2) + v.BlockWrite(context.Background(), 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.BlockWrite(context.Background(), "fffffffffnotreallyahashfffffffff", nil) + v.BlockWrite(context.Background(), "FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF", nil) + v.BlockWrite(context.Background(), "f0000000000000000000000000000000f", nil) + v.BlockWrite(context.Background(), "f00", nil) buf := new(bytes.Buffer) - v.IndexTo("", buf) + v.Index(context.Background(), "", buf) 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"} { buf = new(bytes.Buffer) - v.IndexTo(prefix, buf) + v.Index(context.Background(), prefix, buf) m, err := regexp.MatchString(`^`+TestHash2+`\+\d+ \d+\n$`, string(buf.Bytes())) if err != nil { @@ -362,11 +380,11 @@ func testIndexTo(t *testing.T, factory TestableVolumeFactory) { for _, prefix := range []string{"zero", "zip", "zilch"} { buf = new(bytes.Buffer) - v.IndexTo(prefix, buf) + err := v.Index(context.Background(), prefix, buf) if err != nil { - t.Errorf("Got error on IndexTo with no such prefix %v", err.Error()) + t.Errorf("Got error on Index with no such prefix %v", err.Error()) } else if buf.Len() != 0 { - t.Errorf("Expected empty list for IndexTo with no such prefix %s", prefix) + t.Errorf("Expected empty list for Index with no such prefix %s", prefix) } } } @@ -374,262 +392,522 @@ func testIndexTo(t *testing.T, 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 *testing.T, 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.BlockWrite(context.Background(), TestHash, TestBlock) - v.Put(TestHash, TestBlock) - - if err := v.Delete(TestHash); err != nil { + if err := v.BlockTrash(TestHash); err != nil { t.Error(err) } - data, err := v.Get(TestHash) + buf := &brbuffer{} + err := v.BlockRead(context.Background(), TestHash, buf) 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 buf.String() != string(TestBlock) { + t.Errorf("Got data %+q, expected %+q", buf.String(), 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 *testing.T, 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.BlockWrite(context.Background(), TestHash, TestBlock) + v.TouchWithDate(TestHash, time.Now().Add(-2*s.cluster.Collections.BlobSigningTTL.Duration())) + + if err := v.BlockTrash(TestHash); err != nil { + t.Error(err) + } + if err := v.BlockRead(context.Background(), TestHash, brdiscard); err == nil || !os.IsNotExist(err) { + t.Errorf("os.IsNotExist(%v) should have been true", err) } - v.Put(TestHash, TestBlock) - v.TouchWithDate(TestHash, time.Now().Add(-2*blobSignatureTTL)) + _, err := v.Mtime(TestHash) + if err == nil || !os.IsNotExist(err) { + t.Errorf("os.IsNotExist(%v) should have been true", err) + } - if err := v.Delete(TestHash); err != nil { - t.Error(err) + indexBuf := new(bytes.Buffer) + v.Index(context.Background(), "", indexBuf) + if strings.Contains(string(indexBuf.Bytes()), TestHash) { + t.Errorf("Found trashed block in Index") } - if _, err := v.Get(TestHash); err == nil || !os.IsNotExist(err) { + + err = v.BlockTouch(TestHash) + if err == nil || !os.IsNotExist(err) { t.Errorf("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 *testing.T, 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.Delete(TestHash2); err == nil { + if err := v.BlockTrash(TestHash2); err == nil { t.Errorf("Expected error when attempting to delete a non-existing block") } } -// Invoke Status and verify that VolumeStatus is returned -// Test should pass for both writable and read-only volumes -func testStatus(t *testing.T, factory TestableVolumeFactory) { - v := factory(t) - defer v.Teardown() - - // Get node status and make a basic sanity check. - status := v.Status() - if status.DeviceNum == 0 { - t.Errorf("uninitialized device_num in %v", status) - } - - if status.BytesFree == 0 { - t.Errorf("uninitialized bytes_free in %v", status) - } - - if status.BytesUsed == 0 { - t.Errorf("uninitialized bytes_used in %v", status) - } +func getValueFrom(cv *prometheus.CounterVec, lbls prometheus.Labels) float64 { + c, _ := cv.GetMetricWith(lbls) + pb := &dto.Metric{} + c.Write(pb) + return pb.GetCounter().GetValue() } -// Invoke String for the volume; expect non-empty result -// Test should pass for both writable and read-only volumes -func testString(t *testing.T, factory TestableVolumeFactory) { - v := factory(t) +func (s *genericVolumeSuite) testMetrics(t TB, readonly bool, factory TestableVolumeFactory) { + var err error + + s.setup(t) + v := s.newVolume(t, factory) defer v.Teardown() - if id := v.String(); len(id) == 0 { - t.Error("Got empty string for v.String()") + opsC, _, ioC := s.metrics.getCounterVecsFor(prometheus.Labels{"device_id": v.DeviceID()}) + + if ioC == nil { + t.Error("ioBytes CounterVec is nil") + return } -} -// 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 *testing.T, factory TestableVolumeFactory) { - v := factory(t) - defer v.Teardown() + if getValueFrom(ioC, prometheus.Labels{"direction": "out"})+ + getValueFrom(ioC, prometheus.Labels{"direction": "in"}) > 0 { + t.Error("ioBytes counter should be zero") + } - if v.Writable() == true { + if opsC == nil { + t.Error("opsCounter CounterVec is nil") return } - v.PutRaw(TestHash, TestBlock) + var c, writeOpCounter, readOpCounter float64 - // Get from read-only volume should succeed - _, err := v.Get(TestHash) - if err != nil { - t.Errorf("got err %v, expected nil", err) - } + readOpType, writeOpType := v.ReadWriteOperationLabelValues() + writeOpCounter = getValueFrom(opsC, prometheus.Labels{"operation": writeOpType}) + readOpCounter = getValueFrom(opsC, prometheus.Labels{"operation": readOpType}) - // Put a new block to read-only volume should result in error - err = v.Put(TestHash2, TestBlock2) - if err == nil { - t.Errorf("Expected error when putting block in a read-only volume") - } - _, err = v.Get(TestHash2) - if err == nil { - t.Errorf("Expected error when getting block whose put in read-only volume failed") + // Test Put if volume is writable + if !readonly { + err = v.BlockWrite(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.BlockWrite(context.Background(), TestHash, TestBlock) } - // Touch a block in read-only volume should result in error - err = v.Touch(TestHash) - if err == nil { - t.Errorf("Expected error when touching block in a read-only volume") + err = v.BlockRead(context.Background(), TestHash, brdiscard) + if err != nil { + t.Error(err) } - // Delete a block from a read-only volume should result in error - err = v.Delete(TestHash) - if err == nil { - t.Errorf("Expected error when deleting block from a read-only volume") + // Check that the operations counter increased + c = getValueFrom(opsC, prometheus.Labels{"operation": readOpType}) + if c <= readOpCounter { + t.Error("Operation(s) not counted on Get") } - - // Overwriting an existing block in read-only volume should result in error - err = v.Put(TestHash, TestBlock) - if err == nil { - t.Errorf("Expected error when putting block in a read-only volume") + // 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") } } // Launch concurrent Gets // Test should pass for both writable and read-only volumes -func testGetConcurrent(t *testing.T, 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) - v.PutRaw(TestHash2, TestBlock2) - v.PutRaw(TestHash3, TestBlock3) + v.BlockWrite(context.Background(), TestHash, TestBlock) + v.BlockWrite(context.Background(), TestHash2, TestBlock2) + v.BlockWrite(context.Background(), TestHash3, TestBlock3) sem := make(chan int) - go func(sem chan int) { - buf, err := v.Get(TestHash) + go func() { + buf := &brbuffer{} + err := v.BlockRead(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 buf.String() != string(TestBlock) { + t.Errorf("buf should be %s, is %s", TestBlock, buf) } sem <- 1 - }(sem) + }() - go func(sem chan int) { - buf, err := v.Get(TestHash2) + go func() { + buf := &brbuffer{} + err := v.BlockRead(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 buf.String() != string(TestBlock2) { + t.Errorf("buf should be %s, is %s", TestBlock2, buf) } sem <- 1 - }(sem) + }() - go func(sem chan int) { - buf, err := v.Get(TestHash3) + go func() { + buf := &brbuffer{} + err := v.BlockRead(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 buf.String() != string(TestBlock3) { + t.Errorf("buf should be %s, is %s", TestBlock3, buf) } 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 *testing.T, 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 { + blks := []struct { + hash string + data []byte + }{ + {hash: TestHash, data: TestBlock}, + {hash: TestHash2, data: TestBlock2}, + {hash: TestHash3, data: TestBlock3}, + } + + var wg sync.WaitGroup + for _, blk := range blks { + blk := blk + wg.Add(1) + go func() { + defer wg.Done() + err := v.BlockWrite(context.Background(), blk.hash, blk.data) + if err != nil { + t.Errorf("%s: %v", blk.hash, err) + } + }() + } + wg.Wait() + + // Check that we actually wrote the blocks. + for _, blk := range blks { + buf := &brbuffer{} + err := v.BlockRead(context.Background(), blk.hash, buf) + if err != nil { + t.Errorf("get %s: %v", blk.hash, err) + } else if buf.String() != string(blk.data) { + t.Errorf("get %s: expected %s, got %s", blk.hash, blk.data, buf) + } + } +} + +// Write and read back a full size block +func (s *genericVolumeSuite) testPutFullBlock(t TB, factory TestableVolumeFactory) { + s.setup(t) + v := s.newVolume(t, factory) + defer v.Teardown() + + wdata := make([]byte, BlockSize) + wdata[0] = 'a' + wdata[BlockSize-1] = 'z' + hash := fmt.Sprintf("%x", md5.Sum(wdata)) + err := v.BlockWrite(context.Background(), hash, wdata) + if err != nil { + t.Error(err) + } + + buf := &brbuffer{} + err = v.BlockRead(context.Background(), hash, buf) + if err != nil { + t.Error(err) + } + if buf.String() != string(wdata) { + t.Errorf("buf (len %d) != wdata (len %d)", buf.Len(), len(wdata)) + } +} + +// 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 (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() + + // put block and backdate it + v.BlockWrite(context.Background(), TestHash, TestBlock) + v.TouchWithDate(TestHash, time.Now().Add(-2*s.cluster.Collections.BlobSigningTTL.Duration())) + + buf := &brbuffer{} + err := v.BlockRead(context.Background(), TestHash, buf) + if err != nil { + t.Error(err) + } + if buf.String() != string(TestBlock) { + t.Errorf("Got data %+q, expected %+q", buf, TestBlock) + } + + // Trash + err = v.BlockTrash(TestHash) + if err != nil { + t.Error(err) return } + buf.Reset() + err = v.BlockRead(context.Background(), TestHash, buf) + if err == nil || !os.IsNotExist(err) { + t.Errorf("os.IsNotExist(%v) should have been true", err) + } - sem := make(chan int) - go func(sem chan int) { - err := v.Put(TestHash, TestBlock) + // Untrash + err = v.BlockUntrash(TestHash) + if err != nil { + t.Error(err) + } + + // Get the block - after trash and untrash sequence + buf.Reset() + err = v.BlockRead(context.Background(), TestHash, buf) + if err != nil { + t.Error(err) + } + if buf.String() != string(TestBlock) { + t.Errorf("Got data %+q, expected %+q", buf, TestBlock) + } +} + +func (s *genericVolumeSuite) testTrashEmptyTrashUntrash(t TB, factory TestableVolumeFactory) { + s.setup(t) + v := s.newVolume(t, factory) + defer v.Teardown() + + checkGet := func() error { + buf := &brbuffer{} + err := v.BlockRead(context.Background(), TestHash, buf) if err != nil { - t.Errorf("err1: %v", err) + return err + } + if buf.String() != string(TestBlock) { + t.Errorf("Got data %+q, expected %+q", buf, TestBlock) } - sem <- 1 - }(sem) - go func(sem chan int) { - err := v.Put(TestHash2, TestBlock2) + _, err = v.Mtime(TestHash) if err != nil { - t.Errorf("err2: %v", err) + return err } - sem <- 1 - }(sem) - go func(sem chan int) { - err := v.Put(TestHash3, TestBlock3) - if err != nil { - t.Errorf("err3: %v", err) + indexBuf := new(bytes.Buffer) + v.Index(context.Background(), "", indexBuf) + if !strings.Contains(string(indexBuf.Bytes()), TestHash) { + return os.ErrNotExist } - sem <- 1 - }(sem) - // Wait for all goroutines to finish - for done := 0; done < 3; { - done += <-sem + return nil + } + + // First set: EmptyTrash before reaching the trash deadline. + + s.cluster.Collections.BlobTrashLifetime.Set("1h") + + v.BlockWrite(context.Background(), TestHash, TestBlock) + v.TouchWithDate(TestHash, time.Now().Add(-2*s.cluster.Collections.BlobSigningTTL.Duration())) + + err := checkGet() + if err != nil { + t.Error(err) } - // Double check that we actually wrote the blocks we expected to write. - buf, err := v.Get(TestHash) + // Trash the block + err = v.BlockTrash(TestHash) if err != nil { - t.Errorf("Get #1: %v", err) + t.Error(err) + } + + err = checkGet() + if err == nil || !os.IsNotExist(err) { + t.Errorf("os.IsNotExist(%v) should have been true", err) } - bufs.Put(buf) - if bytes.Compare(buf, TestBlock) != 0 { - t.Errorf("Get #1: expected %s, got %s", string(TestBlock), string(buf)) + + err = v.BlockTouch(TestHash) + if err == nil || !os.IsNotExist(err) { + t.Errorf("os.IsNotExist(%v) should have been true", err) } - buf, err = v.Get(TestHash2) + v.EmptyTrash() + + // Even after emptying the trash, we can untrash our block + // because the deadline hasn't been reached. + err = v.BlockUntrash(TestHash) if err != nil { - t.Errorf("Get #2: %v", err) + t.Error(err) } - bufs.Put(buf) - if bytes.Compare(buf, TestBlock2) != 0 { - t.Errorf("Get #2: expected %s, got %s", string(TestBlock2), string(buf)) + + err = checkGet() + if err != nil { + t.Error(err) } - buf, err = v.Get(TestHash3) + err = v.BlockTouch(TestHash) if err != nil { - t.Errorf("Get #3: %v", err) + t.Error(err) + } + + // 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.BlockUntrash(TestHash) + if err != nil && !os.IsNotExist(err) { + t.Errorf("Expected success or os.IsNotExist(), but got: %v", err) } - bufs.Put(buf) - if bytes.Compare(buf, TestBlock3) != 0 { - t.Errorf("Get #3: expected %s, got %s", string(TestBlock3), string(buf)) + + // The additional Untrash should not interfere with our + // already-untrashed copy. + err = checkGet() + if err != nil { + t.Error(err) + } + + // 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. + + s.cluster.Collections.BlobTrashLifetime.Set("1ns") + + err = v.BlockTrash(TestHash) + if err != nil { + t.Error(err) + } + err = checkGet() + if err == nil || !os.IsNotExist(err) { + t.Errorf("os.IsNotExist(%v) should have been true", err) + } + + // Even though 1ns has passed, we can untrash because we + // haven't called EmptyTrash yet. + err = v.BlockUntrash(TestHash) + if err != nil { + t.Error(err) + } + err = checkGet() + if err != nil { + t.Error(err) + } + + // 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.BlockTrash(TestHash) + err = checkGet() + if err == nil || !os.IsNotExist(err) { + t.Errorf("os.IsNotExist(%v) should have been true", err) + } + v.EmptyTrash() + + // Untrash won't find it + err = v.BlockUntrash(TestHash) + if err == nil || !os.IsNotExist(err) { + t.Errorf("os.IsNotExist(%v) should have been true", err) + } + + // Get block won't find it + err = checkGet() + if err == nil || !os.IsNotExist(err) { + t.Errorf("os.IsNotExist(%v) should have been true", err) + } + + // 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. + + v.BlockWrite(context.Background(), TestHash, TestBlock) + v.TouchWithDate(TestHash, time.Now().Add(-2*s.cluster.Collections.BlobSigningTTL.Duration())) + + s.cluster.Collections.BlobTrashLifetime.Set("1ns") + err = v.BlockTrash(TestHash) + if err != nil { + t.Error(err) + } + err = checkGet() + if err == nil || !os.IsNotExist(err) { + t.Errorf("os.IsNotExist(%v) should have been true", err) + } + + v.BlockWrite(context.Background(), TestHash, TestBlock) + v.TouchWithDate(TestHash, time.Now().Add(-2*s.cluster.Collections.BlobSigningTTL.Duration())) + + // EmptyTrash should not delete the untrashed copy. + v.EmptyTrash() + err = checkGet() + if err != nil { + t.Error(err) + } + + // 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". + + v.BlockWrite(context.Background(), TestHash, TestBlock) + v.TouchWithDate(TestHash, time.Now().Add(-2*s.cluster.Collections.BlobSigningTTL.Duration())) + + s.cluster.Collections.BlobTrashLifetime.Set("1ns") + err = v.BlockTrash(TestHash) + if err != nil { + t.Error(err) + } + + v.BlockWrite(context.Background(), TestHash, TestBlock) + v.TouchWithDate(TestHash, time.Now().Add(-2*s.cluster.Collections.BlobSigningTTL.Duration())) + + s.cluster.Collections.BlobTrashLifetime.Set("1h") + err = v.BlockTrash(TestHash) + if err != nil { + t.Error(err) + } + + // EmptyTrash should not prevent us from recovering the + // time.Hour ("C") trash + v.EmptyTrash() + err = v.BlockUntrash(TestHash) + if err != nil { + t.Error(err) + } + err = checkGet() + if err != nil { + t.Error(err) } }