services/keepstore/unix_volume_test.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package keepstore
   6
   7 import (
   8         "bytes"
   9         "context"
  10         "encoding/json"
  11         "errors"
  12         "fmt"
  13         "io"
  14         "io/ioutil"
  15         "os"
  16         "sync"
  17         "syscall"
  18         "time"
  19
  20         "git.arvados.org/arvados.git/sdk/go/ctxlog"
  21         "github.com/prometheus/client_golang/prometheus"
  22         check "gopkg.in/check.v1"
  23 )
  24
  25 type testableUnixVolume struct {
  26         unixVolume
  27         t TB
  28 }
  29
  30 func (v *testableUnixVolume) TouchWithDate(locator string, lastPut time.Time) {
  31         err := syscall.Utime(v.blockPath(locator), &syscall.Utimbuf{Actime: lastPut.Unix(), Modtime: lastPut.Unix()})
  32         if err != nil {
  33                 v.t.Fatal(err)
  34         }
  35 }
  36
  37 func (v *testableUnixVolume) Teardown() {
  38         if err := os.RemoveAll(v.Root); err != nil {
  39                 v.t.Error(err)
  40         }
  41 }
  42
  43 func (v *testableUnixVolume) ReadWriteOperationLabelValues() (r, w string) {
  44         return "open", "create"
  45 }
  46
  47 var _ = check.Suite(&unixVolumeSuite{})
  48
  49 type unixVolumeSuite struct {
  50         params  newVolumeParams
  51         volumes []*testableUnixVolume
  52 }
  53
  54 func (s *unixVolumeSuite) SetUpTest(c *check.C) {
  55         logger := ctxlog.TestLogger(c)
  56         reg := prometheus.NewRegistry()
  57         s.params = newVolumeParams{
  58                 UUID:        "zzzzz-nyw5e-999999999999999",
  59                 Cluster:     testCluster(c),
  60                 Logger:      logger,
  61                 MetricsVecs: newVolumeMetricsVecs(reg),
  62                 BufferPool:  newBufferPool(logger, 8, reg),
  63         }
  64 }
  65
  66 func (s *unixVolumeSuite) TearDownTest(c *check.C) {
  67         for _, v := range s.volumes {
  68                 v.Teardown()
  69         }
  70 }
  71
  72 func (s *unixVolumeSuite) newTestableUnixVolume(c *check.C, params newVolumeParams, serialize bool) *testableUnixVolume {
  73         d, err := ioutil.TempDir("", "volume_test")
  74         c.Check(err, check.IsNil)
  75         var locker sync.Locker
  76         if serialize {
  77                 locker = &sync.Mutex{}
  78         }
  79         v := &testableUnixVolume{
  80                 unixVolume: unixVolume{
  81                         Root:    d,
  82                         locker:  locker,
  83                         uuid:    params.UUID,
  84                         cluster: params.Cluster,
  85                         logger:  params.Logger,
  86                         volume:  params.ConfigVolume,
  87                         metrics: params.MetricsVecs,
  88                 },
  89                 t: c,
  90         }
  91         c.Check(v.check(), check.IsNil)
  92         s.volumes = append(s.volumes, v)
  93         return v
  94 }
  95
  96 func (s *unixVolumeSuite) TestUnixVolumeWithGenericTests(c *check.C) {
  97         DoGenericVolumeTests(c, false, func(t TB, params newVolumeParams) TestableVolume {
  98                 return s.newTestableUnixVolume(c, params, false)
  99         })
 100 }
 101
 102 func (s *unixVolumeSuite) TestUnixVolumeWithGenericTests_ReadOnly(c *check.C) {
 103         DoGenericVolumeTests(c, true, func(t TB, params newVolumeParams) TestableVolume {
 104                 return s.newTestableUnixVolume(c, params, false)
 105         })
 106 }
 107
 108 func (s *unixVolumeSuite) TestUnixVolumeWithGenericTests_Serialized(c *check.C) {
 109         DoGenericVolumeTests(c, false, func(t TB, params newVolumeParams) TestableVolume {
 110                 return s.newTestableUnixVolume(c, params, true)
 111         })
 112 }
 113
 114 func (s *unixVolumeSuite) TestUnixVolumeWithGenericTests_Readonly_Serialized(c *check.C) {
 115         DoGenericVolumeTests(c, true, func(t TB, params newVolumeParams) TestableVolume {
 116                 return s.newTestableUnixVolume(c, params, true)
 117         })
 118 }
 119
 120 func (s *unixVolumeSuite) TestGetNotFound(c *check.C) {
 121         v := s.newTestableUnixVolume(c, s.params, true)
 122         defer v.Teardown()
 123         v.BlockWrite(context.Background(), TestHash, TestBlock)
 124
 125         buf := bytes.NewBuffer(nil)
 126         _, err := v.BlockRead(context.Background(), TestHash2, buf)
 127         switch {
 128         case os.IsNotExist(err):
 129                 break
 130         case err == nil:
 131                 c.Errorf("Read should have failed, returned %+q", buf.Bytes())
 132         default:
 133                 c.Errorf("Read expected ErrNotExist, got: %s", err)
 134         }
 135 }
 136
 137 func (s *unixVolumeSuite) TestPut(c *check.C) {
 138         v := s.newTestableUnixVolume(c, s.params, false)
 139         defer v.Teardown()
 140
 141         err := v.BlockWrite(context.Background(), TestHash, TestBlock)
 142         if err != nil {
 143                 c.Error(err)
 144         }
 145         p := fmt.Sprintf("%s/%s/%s", v.Root, TestHash[:3], TestHash)
 146         if buf, err := ioutil.ReadFile(p); err != nil {
 147                 c.Error(err)
 148         } else if bytes.Compare(buf, TestBlock) != 0 {
 149                 c.Errorf("Write should have stored %s, did store %s",
 150                         string(TestBlock), string(buf))
 151         }
 152 }
 153
 154 func (s *unixVolumeSuite) TestPutBadVolume(c *check.C) {
 155         v := s.newTestableUnixVolume(c, s.params, false)
 156         defer v.Teardown()
 157
 158         err := os.RemoveAll(v.Root)
 159         c.Assert(err, check.IsNil)
 160         err = v.BlockWrite(context.Background(), TestHash, TestBlock)
 161         c.Check(err, check.IsNil)
 162 }
 163
 164 func (s *unixVolumeSuite) TestIsFull(c *check.C) {
 165         v := s.newTestableUnixVolume(c, s.params, false)
 166         defer v.Teardown()
 167
 168         fullPath := v.Root + "/full"
 169         now := fmt.Sprintf("%d", time.Now().Unix())
 170         os.Symlink(now, fullPath)
 171         if !v.isFull() {
 172                 c.Error("volume claims not to be full")
 173         }
 174         os.Remove(fullPath)
 175
 176         // Test with an expired /full link.
 177         expired := fmt.Sprintf("%d", time.Now().Unix()-3605)
 178         os.Symlink(expired, fullPath)
 179         if v.isFull() {
 180                 c.Error("volume should no longer be full")
 181         }
 182 }
 183
 184 func (s *unixVolumeSuite) TestUnixVolumeGetFuncWorkerError(c *check.C) {
 185         v := s.newTestableUnixVolume(c, s.params, false)
 186         defer v.Teardown()
 187
 188         v.BlockWrite(context.Background(), TestHash, TestBlock)
 189         mockErr := errors.New("Mock error")
 190         err := v.getFunc(context.Background(), v.blockPath(TestHash), func(rdr io.Reader) error {
 191                 return mockErr
 192         })
 193         if err != mockErr {
 194                 c.Errorf("Got %v, expected %v", err, mockErr)
 195         }
 196 }
 197
 198 func (s *unixVolumeSuite) TestUnixVolumeGetFuncFileError(c *check.C) {
 199         v := s.newTestableUnixVolume(c, s.params, false)
 200         defer v.Teardown()
 201
 202         funcCalled := false
 203         err := v.getFunc(context.Background(), v.blockPath(TestHash), func(rdr io.Reader) error {
 204                 funcCalled = true
 205                 return nil
 206         })
 207         if err == nil {
 208                 c.Errorf("Expected error opening non-existent file")
 209         }
 210         if funcCalled {
 211                 c.Errorf("Worker func should not have been called")
 212         }
 213 }
 214
 215 func (s *unixVolumeSuite) TestUnixVolumeGetFuncWorkerWaitsOnMutex(c *check.C) {
 216         v := s.newTestableUnixVolume(c, s.params, false)
 217         defer v.Teardown()
 218
 219         v.BlockWrite(context.Background(), TestHash, TestBlock)
 220
 221         mtx := NewMockMutex()
 222         v.locker = mtx
 223
 224         funcCalled := make(chan struct{})
 225         go v.getFunc(context.Background(), v.blockPath(TestHash), func(rdr io.Reader) error {
 226                 funcCalled <- struct{}{}
 227                 return nil
 228         })
 229         select {
 230         case mtx.AllowLock <- struct{}{}:
 231         case <-funcCalled:
 232                 c.Fatal("Function was called before mutex was acquired")
 233         case <-time.After(5 * time.Second):
 234                 c.Fatal("Timed out before mutex was acquired")
 235         }
 236         select {
 237         case <-funcCalled:
 238         case mtx.AllowUnlock <- struct{}{}:
 239                 c.Fatal("Mutex was released before function was called")
 240         case <-time.After(5 * time.Second):
 241                 c.Fatal("Timed out waiting for funcCalled")
 242         }
 243         select {
 244         case mtx.AllowUnlock <- struct{}{}:
 245         case <-time.After(5 * time.Second):
 246                 c.Fatal("Timed out waiting for getFunc() to release mutex")
 247         }
 248 }
 249
 250 type MockMutex struct {
 251         AllowLock   chan struct{}
 252         AllowUnlock chan struct{}
 253 }
 254
 255 func NewMockMutex() *MockMutex {
 256         return &MockMutex{
 257                 AllowLock:   make(chan struct{}),
 258                 AllowUnlock: make(chan struct{}),
 259         }
 260 }
 261
 262 // Lock waits for someone to send to AllowLock.
 263 func (m *MockMutex) Lock() {
 264         <-m.AllowLock
 265 }
 266
 267 // Unlock waits for someone to send to AllowUnlock.
 268 func (m *MockMutex) Unlock() {
 269         <-m.AllowUnlock
 270 }
 271
 272 func (s *unixVolumeSuite) TestUnixVolumeContextCancelBlockWrite(c *check.C) {
 273         v := s.newTestableUnixVolume(c, s.params, true)
 274         defer v.Teardown()
 275         v.locker.Lock()
 276         defer v.locker.Unlock()
 277         ctx, cancel := context.WithCancel(context.Background())
 278         go func() {
 279                 time.Sleep(50 * time.Millisecond)
 280                 cancel()
 281         }()
 282         err := v.BlockWrite(ctx, TestHash, TestBlock)
 283         if err != context.Canceled {
 284                 c.Errorf("BlockWrite() returned %s -- expected short read / canceled", err)
 285         }
 286 }
 287
 288 func (s *unixVolumeSuite) TestUnixVolumeContextCancelBlockRead(c *check.C) {
 289         v := s.newTestableUnixVolume(c, s.params, true)
 290         defer v.Teardown()
 291         err := v.BlockWrite(context.Background(), TestHash, TestBlock)
 292         if err != nil {
 293                 c.Fatal(err)
 294         }
 295         ctx, cancel := context.WithCancel(context.Background())
 296         v.locker.Lock()
 297         defer v.locker.Unlock()
 298         go func() {
 299                 time.Sleep(50 * time.Millisecond)
 300                 cancel()
 301         }()
 302         n, err := v.BlockRead(ctx, TestHash, io.Discard)
 303         if n > 0 || err != context.Canceled {
 304                 c.Errorf("BlockRead() returned %d, %s -- expected short read / canceled", n, err)
 305         }
 306 }
 307
 308 func (s *unixVolumeSuite) TestStats(c *check.C) {
 309         vol := s.newTestableUnixVolume(c, s.params, false)
 310         stats := func() string {
 311                 buf, err := json.Marshal(vol.InternalStats())
 312                 c.Check(err, check.IsNil)
 313                 return string(buf)
 314         }
 315
 316         c.Check(stats(), check.Matches, `.*"StatOps":1,.*`) // (*unixVolume)check() calls Stat() once
 317         c.Check(stats(), check.Matches, `.*"Errors":0,.*`)
 318
 319         _, err := vol.BlockRead(context.Background(), fooHash, io.Discard)
 320         c.Check(err, check.NotNil)
 321         c.Check(stats(), check.Matches, `.*"StatOps":[^0],.*`)
 322         c.Check(stats(), check.Matches, `.*"Errors":[^0],.*`)
 323         c.Check(stats(), check.Matches, `.*"\*(fs|os)\.PathError":[^0].*`) // os.PathError changed to fs.PathError in Go 1.16
 324         c.Check(stats(), check.Matches, `.*"InBytes":0,.*`)
 325         c.Check(stats(), check.Matches, `.*"OpenOps":0,.*`)
 326         c.Check(stats(), check.Matches, `.*"CreateOps":0,.*`)
 327
 328         err = vol.BlockWrite(context.Background(), fooHash, []byte("foo"))
 329         c.Check(err, check.IsNil)
 330         c.Check(stats(), check.Matches, `.*"OutBytes":3,.*`)
 331         c.Check(stats(), check.Matches, `.*"CreateOps":1,.*`)
 332         c.Check(stats(), check.Matches, `.*"OpenOps":0,.*`)
 333         c.Check(stats(), check.Matches, `.*"UtimesOps":1,.*`)
 334
 335         err = vol.BlockTouch(fooHash)
 336         c.Check(err, check.IsNil)
 337         c.Check(stats(), check.Matches, `.*"FlockOps":1,.*`)
 338         c.Check(stats(), check.Matches, `.*"OpenOps":1,.*`)
 339         c.Check(stats(), check.Matches, `.*"UtimesOps":2,.*`)
 340
 341         buf := bytes.NewBuffer(nil)
 342         _, err = vol.BlockRead(context.Background(), fooHash, buf)
 343         c.Check(err, check.IsNil)
 344         c.Check(buf.String(), check.Equals, "foo")
 345         c.Check(stats(), check.Matches, `.*"InBytes":3,.*`)
 346         c.Check(stats(), check.Matches, `.*"OpenOps":2,.*`)
 347
 348         err = vol.BlockTrash(fooHash)
 349         c.Check(err, check.IsNil)
 350         c.Check(stats(), check.Matches, `.*"FlockOps":2,.*`)
 351 }
 352
 353 func (s *unixVolumeSuite) TestSkipUnusedDirs(c *check.C) {
 354         vol := s.newTestableUnixVolume(c, s.params, false)
 355
 356         err := os.Mkdir(vol.unixVolume.Root+"/aaa", 0777)
 357         c.Assert(err, check.IsNil)
 358         err = os.Mkdir(vol.unixVolume.Root+"/.aaa", 0777) // EmptyTrash should not look here
 359         c.Assert(err, check.IsNil)
 360         deleteme := vol.unixVolume.Root + "/aaa/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.trash.1"
 361         err = ioutil.WriteFile(deleteme, []byte{1, 2, 3}, 0777)
 362         c.Assert(err, check.IsNil)
 363         skipme := vol.unixVolume.Root + "/.aaa/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.trash.1"
 364         err = ioutil.WriteFile(skipme, []byte{1, 2, 3}, 0777)
 365         c.Assert(err, check.IsNil)
 366         vol.EmptyTrash()
 367
 368         _, err = os.Stat(skipme)
 369         c.Check(err, check.IsNil)
 370
 371         _, err = os.Stat(deleteme)
 372         c.Check(err, check.NotNil)
 373         c.Check(os.IsNotExist(err), check.Equals, true)
 374 }