//
// SPDX-License-Identifier: AGPL-3.0
-package main
+package keepstore
import (
"bytes"
"io"
"io/ioutil"
"os"
- "strings"
"sync"
"syscall"
- "testing"
"time"
- "github.com/ghodss/yaml"
+ "git.arvados.org/arvados.git/sdk/go/ctxlog"
"github.com/prometheus/client_golang/prometheus"
check "gopkg.in/check.v1"
)
-type TestableUnixVolume struct {
- UnixVolume
+type testableUnixVolume struct {
+ unixVolume
t TB
}
-func NewTestableUnixVolume(t TB, serialize bool, readonly bool) *TestableUnixVolume {
- d, err := ioutil.TempDir("", "volume_test")
+func (v *testableUnixVolume) TouchWithDate(locator string, lastPut time.Time) {
+ err := syscall.Utime(v.blockPath(locator), &syscall.Utimbuf{Actime: lastPut.Unix(), Modtime: lastPut.Unix()})
if err != nil {
- t.Fatal(err)
- }
- var locker sync.Locker
- if serialize {
- locker = &sync.Mutex{}
- }
- return &TestableUnixVolume{
- UnixVolume: UnixVolume{
- Root: d,
- ReadOnly: readonly,
- locker: locker,
- },
- t: t,
+ v.t.Fatal(err)
}
}
-// PutRaw writes a Keep block directly into a UnixVolume, even if
-// the volume is readonly.
-func (v *TestableUnixVolume) PutRaw(locator string, data []byte) {
- defer func(orig bool) {
- v.ReadOnly = orig
- }(v.ReadOnly)
- v.ReadOnly = false
- err := v.Put(context.Background(), locator, data)
- if err != nil {
- v.t.Fatal(err)
+func (v *testableUnixVolume) Teardown() {
+ if err := os.RemoveAll(v.Root); err != nil {
+ v.t.Error(err)
}
}
-func (v *TestableUnixVolume) TouchWithDate(locator string, lastPut time.Time) {
- err := syscall.Utime(v.blockPath(locator), &syscall.Utimbuf{lastPut.Unix(), lastPut.Unix()})
- if err != nil {
- v.t.Fatal(err)
- }
+func (v *testableUnixVolume) ReadWriteOperationLabelValues() (r, w string) {
+ return "open", "create"
}
-func (v *TestableUnixVolume) Teardown() {
- if err := os.RemoveAll(v.Root); err != nil {
- v.t.Fatal(err)
+var _ = check.Suite(&unixVolumeSuite{})
+
+type unixVolumeSuite struct {
+ params newVolumeParams
+ volumes []*testableUnixVolume
+}
+
+func (s *unixVolumeSuite) SetUpTest(c *check.C) {
+ logger := ctxlog.TestLogger(c)
+ reg := prometheus.NewRegistry()
+ s.params = newVolumeParams{
+ UUID: "zzzzz-nyw5e-999999999999999",
+ Cluster: testCluster(c),
+ Logger: logger,
+ MetricsVecs: newVolumeMetricsVecs(reg),
+ BufferPool: newBufferPool(logger, 8, reg),
}
}
-func (v *TestableUnixVolume) GetMetricsVecs() (opsCounters, errCounters, ioBytes *prometheus.CounterVec) {
- opsCounters = v.os.stats.opsCounters
- errCounters = v.os.stats.errCounters
- ioBytes = v.os.stats.ioBytes
- return
+func (s *unixVolumeSuite) TearDownTest(c *check.C) {
+ for _, v := range s.volumes {
+ v.Teardown()
+ }
}
-// serialize = false; readonly = false
-func TestUnixVolumeWithGenericTests(t *testing.T) {
- DoGenericVolumeTests(t, func(t TB) TestableVolume {
- return NewTestableUnixVolume(t, false, false)
- })
+func (s *unixVolumeSuite) newTestableUnixVolume(c *check.C, params newVolumeParams, serialize bool) *testableUnixVolume {
+ d, err := ioutil.TempDir("", "volume_test")
+ c.Check(err, check.IsNil)
+ var locker sync.Locker
+ if serialize {
+ locker = &sync.Mutex{}
+ }
+ v := &testableUnixVolume{
+ unixVolume: unixVolume{
+ Root: d,
+ locker: locker,
+ uuid: params.UUID,
+ cluster: params.Cluster,
+ logger: params.Logger,
+ volume: params.ConfigVolume,
+ metrics: params.MetricsVecs,
+ bufferPool: params.BufferPool,
+ },
+ t: c,
+ }
+ c.Check(v.check(), check.IsNil)
+ s.volumes = append(s.volumes, v)
+ return v
}
-// serialize = false; readonly = true
-func TestUnixVolumeWithGenericTestsReadOnly(t *testing.T) {
- DoGenericVolumeTests(t, func(t TB) TestableVolume {
- return NewTestableUnixVolume(t, false, true)
+func (s *unixVolumeSuite) TestUnixVolumeWithGenericTests(c *check.C) {
+ DoGenericVolumeTests(c, false, func(t TB, params newVolumeParams) TestableVolume {
+ return s.newTestableUnixVolume(c, params, false)
})
}
-// serialize = true; readonly = false
-func TestUnixVolumeWithGenericTestsSerialized(t *testing.T) {
- DoGenericVolumeTests(t, func(t TB) TestableVolume {
- return NewTestableUnixVolume(t, true, false)
+func (s *unixVolumeSuite) TestUnixVolumeWithGenericTests_ReadOnly(c *check.C) {
+ DoGenericVolumeTests(c, true, func(t TB, params newVolumeParams) TestableVolume {
+ return s.newTestableUnixVolume(c, params, false)
})
}
-// serialize = false; readonly = false
-func TestUnixVolumeHandlersWithGenericVolumeTests(t *testing.T) {
- DoHandlersWithGenericVolumeTests(t, func(t TB) (*RRVolumeManager, []TestableVolume) {
- vols := make([]Volume, 2)
- testableUnixVols := make([]TestableVolume, 2)
-
- for i := range vols {
- v := NewTestableUnixVolume(t, false, false)
- vols[i] = v
- testableUnixVols[i] = v
- }
-
- return MakeRRVolumeManager(vols), testableUnixVols
+func (s *unixVolumeSuite) TestUnixVolumeWithGenericTests_Serialized(c *check.C) {
+ DoGenericVolumeTests(c, false, func(t TB, params newVolumeParams) TestableVolume {
+ return s.newTestableUnixVolume(c, params, true)
})
}
-func TestReplicationDefault1(t *testing.T) {
- v := &UnixVolume{
- Root: "/",
- ReadOnly: true,
- }
- metrics := newVolumeMetricsVecs(prometheus.NewRegistry())
- if err := v.Start(metrics); err != nil {
- t.Error(err)
- }
- if got := v.Replication(); got != 1 {
- t.Errorf("Replication() returned %d, expected 1 if no config given", got)
- }
+func (s *unixVolumeSuite) TestUnixVolumeWithGenericTests_Readonly_Serialized(c *check.C) {
+ DoGenericVolumeTests(c, true, func(t TB, params newVolumeParams) TestableVolume {
+ return s.newTestableUnixVolume(c, params, true)
+ })
}
-func TestGetNotFound(t *testing.T) {
- v := NewTestableUnixVolume(t, false, false)
+func (s *unixVolumeSuite) TestGetNotFound(c *check.C) {
+ v := s.newTestableUnixVolume(c, s.params, true)
defer v.Teardown()
- v.Put(context.Background(), TestHash, TestBlock)
+ v.BlockWrite(context.Background(), TestHash, TestBlock)
- buf := make([]byte, BlockSize)
- n, err := v.Get(context.Background(), TestHash2, buf)
+ buf := bytes.NewBuffer(nil)
+ _, err := v.BlockRead(context.Background(), TestHash2, buf)
switch {
case os.IsNotExist(err):
break
case err == nil:
- t.Errorf("Read should have failed, returned %+q", buf[:n])
+ c.Errorf("Read should have failed, returned %+q", buf.Bytes())
default:
- t.Errorf("Read expected ErrNotExist, got: %s", err)
+ c.Errorf("Read expected ErrNotExist, got: %s", err)
}
}
-func TestPut(t *testing.T) {
- v := NewTestableUnixVolume(t, false, false)
+func (s *unixVolumeSuite) TestPut(c *check.C) {
+ v := s.newTestableUnixVolume(c, s.params, false)
defer v.Teardown()
- err := v.Put(context.Background(), TestHash, TestBlock)
+ err := v.BlockWrite(context.Background(), TestHash, TestBlock)
if err != nil {
- t.Error(err)
+ c.Error(err)
}
p := fmt.Sprintf("%s/%s/%s", v.Root, TestHash[:3], TestHash)
if buf, err := ioutil.ReadFile(p); err != nil {
- t.Error(err)
+ c.Error(err)
} else if bytes.Compare(buf, TestBlock) != 0 {
- t.Errorf("Write should have stored %s, did store %s",
+ c.Errorf("Write should have stored %s, did store %s",
string(TestBlock), string(buf))
}
}
-func TestPutBadVolume(t *testing.T) {
- v := NewTestableUnixVolume(t, false, false)
- defer v.Teardown()
-
- os.Chmod(v.Root, 000)
- err := v.Put(context.Background(), TestHash, TestBlock)
- if err == nil {
- t.Error("Write should have failed")
- }
-}
-
-func TestUnixVolumeReadonly(t *testing.T) {
- v := NewTestableUnixVolume(t, false, true)
+func (s *unixVolumeSuite) TestPutBadVolume(c *check.C) {
+ v := s.newTestableUnixVolume(c, s.params, false)
defer v.Teardown()
- v.PutRaw(TestHash, TestBlock)
-
- buf := make([]byte, BlockSize)
- _, err := v.Get(context.Background(), TestHash, buf)
- if err != nil {
- t.Errorf("got err %v, expected nil", err)
- }
-
- err = v.Put(context.Background(), TestHash, TestBlock)
- if err != MethodDisabledError {
- t.Errorf("got err %v, expected MethodDisabledError", err)
- }
-
- err = v.Touch(TestHash)
- if err != MethodDisabledError {
- t.Errorf("got err %v, expected MethodDisabledError", err)
- }
-
- err = v.Trash(TestHash)
- if err != MethodDisabledError {
- t.Errorf("got err %v, expected MethodDisabledError", err)
- }
+ err := os.RemoveAll(v.Root)
+ c.Assert(err, check.IsNil)
+ err = v.BlockWrite(context.Background(), TestHash, TestBlock)
+ c.Check(err, check.IsNil)
}
-func TestIsFull(t *testing.T) {
- v := NewTestableUnixVolume(t, false, false)
+func (s *unixVolumeSuite) TestIsFull(c *check.C) {
+ v := s.newTestableUnixVolume(c, s.params, false)
defer v.Teardown()
fullPath := v.Root + "/full"
now := fmt.Sprintf("%d", time.Now().Unix())
os.Symlink(now, fullPath)
- if !v.IsFull() {
- t.Errorf("%s: claims not to be full", v)
+ if !v.isFull() {
+ c.Error("volume claims not to be full")
}
os.Remove(fullPath)
// Test with an expired /full link.
expired := fmt.Sprintf("%d", time.Now().Unix()-3605)
os.Symlink(expired, fullPath)
- if v.IsFull() {
- t.Errorf("%s: should no longer be full", v)
+ if v.isFull() {
+ c.Error("volume should no longer be full")
}
}
-func TestNodeStatus(t *testing.T) {
- v := NewTestableUnixVolume(t, false, false)
+func (s *unixVolumeSuite) TestUnixVolumeGetFuncWorkerError(c *check.C) {
+ v := s.newTestableUnixVolume(c, s.params, false)
defer v.Teardown()
- // Get node status and make a basic sanity check.
- volinfo := v.Status()
- if volinfo.MountPoint != v.Root {
- t.Errorf("GetNodeStatus mount_point %s, expected %s", volinfo.MountPoint, v.Root)
- }
- if volinfo.DeviceNum == 0 {
- t.Errorf("uninitialized device_num in %v", volinfo)
- }
- if volinfo.BytesFree == 0 {
- t.Errorf("uninitialized bytes_free in %v", volinfo)
- }
- if volinfo.BytesUsed == 0 {
- t.Errorf("uninitialized bytes_used in %v", volinfo)
- }
-}
-
-func TestUnixVolumeGetFuncWorkerError(t *testing.T) {
- v := NewTestableUnixVolume(t, false, false)
- defer v.Teardown()
-
- v.Put(context.Background(), TestHash, TestBlock)
+ v.BlockWrite(context.Background(), TestHash, TestBlock)
mockErr := errors.New("Mock error")
err := v.getFunc(context.Background(), v.blockPath(TestHash), func(rdr io.Reader) error {
return mockErr
})
if err != mockErr {
- t.Errorf("Got %v, expected %v", err, mockErr)
+ c.Errorf("Got %v, expected %v", err, mockErr)
}
}
-func TestUnixVolumeGetFuncFileError(t *testing.T) {
- v := NewTestableUnixVolume(t, false, false)
+func (s *unixVolumeSuite) TestUnixVolumeGetFuncFileError(c *check.C) {
+ v := s.newTestableUnixVolume(c, s.params, false)
defer v.Teardown()
funcCalled := false
return nil
})
if err == nil {
- t.Errorf("Expected error opening non-existent file")
+ c.Errorf("Expected error opening non-existent file")
}
if funcCalled {
- t.Errorf("Worker func should not have been called")
+ c.Errorf("Worker func should not have been called")
}
}
-func TestUnixVolumeGetFuncWorkerWaitsOnMutex(t *testing.T) {
- v := NewTestableUnixVolume(t, false, false)
+func (s *unixVolumeSuite) TestUnixVolumeGetFuncWorkerWaitsOnMutex(c *check.C) {
+ v := s.newTestableUnixVolume(c, s.params, false)
defer v.Teardown()
- v.Put(context.Background(), TestHash, TestBlock)
+ v.BlockWrite(context.Background(), TestHash, TestBlock)
mtx := NewMockMutex()
v.locker = mtx
select {
case mtx.AllowLock <- struct{}{}:
case <-funcCalled:
- t.Fatal("Function was called before mutex was acquired")
+ c.Fatal("Function was called before mutex was acquired")
case <-time.After(5 * time.Second):
- t.Fatal("Timed out before mutex was acquired")
+ c.Fatal("Timed out before mutex was acquired")
}
select {
case <-funcCalled:
case mtx.AllowUnlock <- struct{}{}:
- t.Fatal("Mutex was released before function was called")
+ c.Fatal("Mutex was released before function was called")
case <-time.After(5 * time.Second):
- t.Fatal("Timed out waiting for funcCalled")
+ c.Fatal("Timed out waiting for funcCalled")
}
select {
case mtx.AllowUnlock <- struct{}{}:
case <-time.After(5 * time.Second):
- t.Fatal("Timed out waiting for getFunc() to release mutex")
+ c.Fatal("Timed out waiting for getFunc() to release mutex")
}
}
-func TestUnixVolumeCompare(t *testing.T) {
- v := NewTestableUnixVolume(t, false, false)
- defer v.Teardown()
-
- v.Put(context.Background(), TestHash, TestBlock)
- err := v.Compare(context.Background(), TestHash, TestBlock)
- if err != nil {
- t.Errorf("Got err %q, expected nil", err)
- }
+type MockMutex struct {
+ AllowLock chan struct{}
+ AllowUnlock chan struct{}
+}
- err = v.Compare(context.Background(), TestHash, []byte("baddata"))
- if err != CollisionError {
- t.Errorf("Got err %q, expected %q", err, CollisionError)
+func NewMockMutex() *MockMutex {
+ return &MockMutex{
+ AllowLock: make(chan struct{}),
+ AllowUnlock: make(chan struct{}),
}
+}
- v.Put(context.Background(), TestHash, []byte("baddata"))
- err = v.Compare(context.Background(), TestHash, TestBlock)
- if err != DiskHashError {
- t.Errorf("Got err %q, expected %q", err, DiskHashError)
- }
+// Lock waits for someone to send to AllowLock.
+func (m *MockMutex) Lock() {
+ <-m.AllowLock
+}
- p := fmt.Sprintf("%s/%s/%s", v.Root, TestHash[:3], TestHash)
- os.Chmod(p, 000)
- err = v.Compare(context.Background(), TestHash, TestBlock)
- if err == nil || strings.Index(err.Error(), "permission denied") < 0 {
- t.Errorf("Got err %q, expected %q", err, "permission denied")
- }
+// Unlock waits for someone to send to AllowUnlock.
+func (m *MockMutex) Unlock() {
+ <-m.AllowUnlock
}
-func TestUnixVolumeContextCancelPut(t *testing.T) {
- v := NewTestableUnixVolume(t, true, false)
+func (s *unixVolumeSuite) TestUnixVolumeContextCancelBlockWrite(c *check.C) {
+ v := s.newTestableUnixVolume(c, s.params, true)
defer v.Teardown()
v.locker.Lock()
+ defer v.locker.Unlock()
ctx, cancel := context.WithCancel(context.Background())
go func() {
time.Sleep(50 * time.Millisecond)
cancel()
- time.Sleep(50 * time.Millisecond)
- v.locker.Unlock()
}()
- err := v.Put(ctx, TestHash, TestBlock)
+ err := v.BlockWrite(ctx, TestHash, TestBlock)
if err != context.Canceled {
- t.Errorf("Put() returned %s -- expected short read / canceled", err)
+ c.Errorf("BlockWrite() returned %s -- expected short read / canceled", err)
}
}
-func TestUnixVolumeContextCancelGet(t *testing.T) {
- v := NewTestableUnixVolume(t, false, false)
+func (s *unixVolumeSuite) TestUnixVolumeContextCancelBlockRead(c *check.C) {
+ v := s.newTestableUnixVolume(c, s.params, true)
defer v.Teardown()
- bpath := v.blockPath(TestHash)
- v.PutRaw(TestHash, TestBlock)
- os.Remove(bpath)
- err := syscall.Mkfifo(bpath, 0600)
+ err := v.BlockWrite(context.Background(), TestHash, TestBlock)
if err != nil {
- t.Fatalf("Mkfifo %s: %s", bpath, err)
+ c.Fatal(err)
}
- defer os.Remove(bpath)
ctx, cancel := context.WithCancel(context.Background())
+ v.locker.Lock()
+ defer v.locker.Unlock()
go func() {
time.Sleep(50 * time.Millisecond)
cancel()
}()
- buf := make([]byte, len(TestBlock))
- n, err := v.Get(ctx, TestHash, buf)
- if n == len(TestBlock) || err != context.Canceled {
- t.Errorf("Get() returned %d, %s -- expected short read / canceled", n, err)
- }
-}
-
-var _ = check.Suite(&UnixVolumeSuite{})
-
-type UnixVolumeSuite struct {
- volume *TestableUnixVolume
-}
-
-func (s *UnixVolumeSuite) TearDownTest(c *check.C) {
- if s.volume != nil {
- s.volume.Teardown()
+ n, err := v.BlockRead(ctx, TestHash, io.Discard)
+ if n > 0 || err != context.Canceled {
+ c.Errorf("BlockRead() returned %d, %s -- expected short read / canceled", n, err)
}
}
-func (s *UnixVolumeSuite) TestStats(c *check.C) {
- s.volume = NewTestableUnixVolume(c, false, false)
+func (s *unixVolumeSuite) TestStats(c *check.C) {
+ vol := s.newTestableUnixVolume(c, s.params, false)
stats := func() string {
- buf, err := json.Marshal(s.volume.InternalStats())
+ buf, err := json.Marshal(vol.InternalStats())
c.Check(err, check.IsNil)
return string(buf)
}
- c.Check(stats(), check.Matches, `.*"StatOps":0,.*`)
+ c.Check(stats(), check.Matches, `.*"StatOps":1,.*`) // (*unixVolume)check() calls Stat() once
c.Check(stats(), check.Matches, `.*"Errors":0,.*`)
- loc := "acbd18db4cc2f85cedef654fccc4a4d8"
- _, err := s.volume.Get(context.Background(), loc, make([]byte, 3))
+ _, err := vol.BlockRead(context.Background(), fooHash, io.Discard)
c.Check(err, check.NotNil)
c.Check(stats(), check.Matches, `.*"StatOps":[^0],.*`)
c.Check(stats(), check.Matches, `.*"Errors":[^0],.*`)
- c.Check(stats(), check.Matches, `.*"\*os\.PathError":[^0].*`)
+ c.Check(stats(), check.Matches, `.*"\*(fs|os)\.PathError":[^0].*`) // os.PathError changed to fs.PathError in Go 1.16
c.Check(stats(), check.Matches, `.*"InBytes":0,.*`)
c.Check(stats(), check.Matches, `.*"OpenOps":0,.*`)
c.Check(stats(), check.Matches, `.*"CreateOps":0,.*`)
- err = s.volume.Put(context.Background(), loc, []byte("foo"))
+ err = vol.BlockWrite(context.Background(), fooHash, []byte("foo"))
c.Check(err, check.IsNil)
c.Check(stats(), check.Matches, `.*"OutBytes":3,.*`)
c.Check(stats(), check.Matches, `.*"CreateOps":1,.*`)
c.Check(stats(), check.Matches, `.*"OpenOps":0,.*`)
- c.Check(stats(), check.Matches, `.*"UtimesOps":0,.*`)
+ c.Check(stats(), check.Matches, `.*"UtimesOps":1,.*`)
- err = s.volume.Touch(loc)
+ err = vol.BlockTouch(fooHash)
c.Check(err, check.IsNil)
c.Check(stats(), check.Matches, `.*"FlockOps":1,.*`)
c.Check(stats(), check.Matches, `.*"OpenOps":1,.*`)
- c.Check(stats(), check.Matches, `.*"UtimesOps":1,.*`)
+ c.Check(stats(), check.Matches, `.*"UtimesOps":2,.*`)
- _, err = s.volume.Get(context.Background(), loc, make([]byte, 3))
- c.Check(err, check.IsNil)
- err = s.volume.Compare(context.Background(), loc, []byte("foo"))
+ buf := bytes.NewBuffer(nil)
+ _, err = vol.BlockRead(context.Background(), fooHash, buf)
c.Check(err, check.IsNil)
- c.Check(stats(), check.Matches, `.*"InBytes":6,.*`)
- c.Check(stats(), check.Matches, `.*"OpenOps":3,.*`)
+ c.Check(buf.String(), check.Equals, "foo")
+ c.Check(stats(), check.Matches, `.*"InBytes":3,.*`)
+ c.Check(stats(), check.Matches, `.*"OpenOps":2,.*`)
- err = s.volume.Trash(loc)
+ err = vol.BlockTrash(fooHash)
c.Check(err, check.IsNil)
c.Check(stats(), check.Matches, `.*"FlockOps":2,.*`)
}
-func (s *UnixVolumeSuite) TestConfig(c *check.C) {
- var cfg Config
- err := yaml.Unmarshal([]byte(`
-Volumes:
- - Type: Directory
- StorageClasses: ["class_a", "class_b"]
-`), &cfg)
-
+func (s *unixVolumeSuite) TestSkipUnusedDirs(c *check.C) {
+ vol := s.newTestableUnixVolume(c, s.params, false)
+
+ err := os.Mkdir(vol.unixVolume.Root+"/aaa", 0777)
+ c.Assert(err, check.IsNil)
+ err = os.Mkdir(vol.unixVolume.Root+"/.aaa", 0777) // EmptyTrash should not look here
+ c.Assert(err, check.IsNil)
+ deleteme := vol.unixVolume.Root + "/aaa/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.trash.1"
+ err = ioutil.WriteFile(deleteme, []byte{1, 2, 3}, 0777)
+ c.Assert(err, check.IsNil)
+ skipme := vol.unixVolume.Root + "/.aaa/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.trash.1"
+ err = ioutil.WriteFile(skipme, []byte{1, 2, 3}, 0777)
+ c.Assert(err, check.IsNil)
+ vol.EmptyTrash()
+
+ _, err = os.Stat(skipme)
c.Check(err, check.IsNil)
- c.Check(cfg.Volumes[0].GetStorageClasses(), check.DeepEquals, []string{"class_a", "class_b"})
+
+ _, err = os.Stat(deleteme)
+ c.Check(err, check.NotNil)
+ c.Check(os.IsNotExist(err), check.Equals, true)
}
projects / arvados.git / blobdiff