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 main
   6
   7 import (
   8         "bytes"
   9         "context"
  10         "encoding/json"
  11         "errors"
  12         "fmt"
  13         "io"
  14         "io/ioutil"
  15         "os"
  16         "strings"
  17         "sync"
  18         "syscall"
  19         "time"
  20
  21         "git.curoverse.com/arvados.git/sdk/go/arvados"
  22         "git.curoverse.com/arvados.git/sdk/go/ctxlog"
  23         "github.com/prometheus/client_golang/prometheus"
  24         "github.com/sirupsen/logrus"
  25         check "gopkg.in/check.v1"
  26 )
  27
  28 type TestableUnixVolume struct {
  29         UnixVolume
  30         t TB
  31 }
  32
  33 // PutRaw writes a Keep block directly into a UnixVolume, even if
  34 // the volume is readonly.
  35 func (v *TestableUnixVolume) PutRaw(locator string, data []byte) {
  36         defer func(orig bool) {
  37                 v.volume.ReadOnly = orig
  38         }(v.volume.ReadOnly)
  39         v.volume.ReadOnly = false
  40         err := v.Put(context.Background(), locator, data)
  41         if err != nil {
  42                 v.t.Fatal(err)
  43         }
  44 }
  45
  46 func (v *TestableUnixVolume) TouchWithDate(locator string, lastPut time.Time) {
  47         err := syscall.Utime(v.blockPath(locator), &syscall.Utimbuf{lastPut.Unix(), lastPut.Unix()})
  48         if err != nil {
  49                 v.t.Fatal(err)
  50         }
  51 }
  52
  53 func (v *TestableUnixVolume) Teardown() {
  54         if err := os.RemoveAll(v.Root); err != nil {
  55                 v.t.Error(err)
  56         }
  57 }
  58
  59 func (v *TestableUnixVolume) ReadWriteOperationLabelValues() (r, w string) {
  60         return "open", "create"
  61 }
  62
  63 var _ = check.Suite(&UnixVolumeSuite{})
  64
  65 type UnixVolumeSuite struct {
  66         cluster *arvados.Cluster
  67         volumes []*TestableUnixVolume
  68         metrics *volumeMetricsVecs
  69 }
  70
  71 func (s *UnixVolumeSuite) SetUpTest(c *check.C) {
  72         s.cluster = testCluster(c)
  73         s.metrics = newVolumeMetricsVecs(prometheus.NewRegistry())
  74 }
  75
  76 func (s *UnixVolumeSuite) TearDownTest(c *check.C) {
  77         for _, v := range s.volumes {
  78                 v.Teardown()
  79         }
  80 }
  81
  82 func (s *UnixVolumeSuite) newTestableUnixVolume(c *check.C, cluster *arvados.Cluster, volume arvados.Volume, metrics *volumeMetricsVecs, serialize bool) *TestableUnixVolume {
  83         d, err := ioutil.TempDir("", "volume_test")
  84         c.Check(err, check.IsNil)
  85         var locker sync.Locker
  86         if serialize {
  87                 locker = &sync.Mutex{}
  88         }
  89         v := &TestableUnixVolume{
  90                 UnixVolume: UnixVolume{
  91                         Root:    d,
  92                         locker:  locker,
  93                         cluster: cluster,
  94                         logger:  ctxlog.TestLogger(c),
  95                         volume:  volume,
  96                         metrics: metrics,
  97                 },
  98                 t: c,
  99         }
 100         c.Check(v.check(), check.IsNil)
 101         s.volumes = append(s.volumes, v)
 102         return v
 103 }
 104
 105 // serialize = false; readonly = false
 106 func (s *UnixVolumeSuite) TestUnixVolumeWithGenericTests(c *check.C) {
 107         DoGenericVolumeTests(c, false, func(t TB, cluster *arvados.Cluster, volume arvados.Volume, logger logrus.FieldLogger, metrics *volumeMetricsVecs) TestableVolume {
 108                 return s.newTestableUnixVolume(c, cluster, volume, metrics, false)
 109         })
 110 }
 111
 112 // serialize = false; readonly = true
 113 func (s *UnixVolumeSuite) TestUnixVolumeWithGenericTestsReadOnly(c *check.C) {
 114         DoGenericVolumeTests(c, true, func(t TB, cluster *arvados.Cluster, volume arvados.Volume, logger logrus.FieldLogger, metrics *volumeMetricsVecs) TestableVolume {
 115                 return s.newTestableUnixVolume(c, cluster, volume, metrics, true)
 116         })
 117 }
 118
 119 // serialize = true; readonly = false
 120 func (s *UnixVolumeSuite) TestUnixVolumeWithGenericTestsSerialized(c *check.C) {
 121         DoGenericVolumeTests(c, false, func(t TB, cluster *arvados.Cluster, volume arvados.Volume, logger logrus.FieldLogger, metrics *volumeMetricsVecs) TestableVolume {
 122                 return s.newTestableUnixVolume(c, cluster, volume, metrics, false)
 123         })
 124 }
 125
 126 // serialize = true; readonly = true
 127 func (s *UnixVolumeSuite) TestUnixVolumeHandlersWithGenericVolumeTests(c *check.C) {
 128         DoGenericVolumeTests(c, true, func(t TB, cluster *arvados.Cluster, volume arvados.Volume, logger logrus.FieldLogger, metrics *volumeMetricsVecs) TestableVolume {
 129                 return s.newTestableUnixVolume(c, cluster, volume, metrics, true)
 130         })
 131 }
 132
 133 func (s *UnixVolumeSuite) TestGetNotFound(c *check.C) {
 134         v := s.newTestableUnixVolume(c, s.cluster, arvados.Volume{Replication: 1}, s.metrics, false)
 135         defer v.Teardown()
 136         v.Put(context.Background(), TestHash, TestBlock)
 137
 138         buf := make([]byte, BlockSize)
 139         n, err := v.Get(context.Background(), TestHash2, buf)
 140         switch {
 141         case os.IsNotExist(err):
 142                 break
 143         case err == nil:
 144                 c.Errorf("Read should have failed, returned %+q", buf[:n])
 145         default:
 146                 c.Errorf("Read expected ErrNotExist, got: %s", err)
 147         }
 148 }
 149
 150 func (s *UnixVolumeSuite) TestPut(c *check.C) {
 151         v := s.newTestableUnixVolume(c, s.cluster, arvados.Volume{Replication: 1}, s.metrics, false)
 152         defer v.Teardown()
 153
 154         err := v.Put(context.Background(), TestHash, TestBlock)
 155         if err != nil {
 156                 c.Error(err)
 157         }
 158         p := fmt.Sprintf("%s/%s/%s", v.Root, TestHash[:3], TestHash)
 159         if buf, err := ioutil.ReadFile(p); err != nil {
 160                 c.Error(err)
 161         } else if bytes.Compare(buf, TestBlock) != 0 {
 162                 c.Errorf("Write should have stored %s, did store %s",
 163                         string(TestBlock), string(buf))
 164         }
 165 }
 166
 167 func (s *UnixVolumeSuite) TestPutBadVolume(c *check.C) {
 168         v := s.newTestableUnixVolume(c, s.cluster, arvados.Volume{Replication: 1}, s.metrics, false)
 169         defer v.Teardown()
 170
 171         os.Chmod(v.Root, 000)
 172         err := v.Put(context.Background(), TestHash, TestBlock)
 173         if err == nil {
 174                 c.Error("Write should have failed")
 175         }
 176 }
 177
 178 func (s *UnixVolumeSuite) TestUnixVolumeReadonly(c *check.C) {
 179         v := s.newTestableUnixVolume(c, s.cluster, arvados.Volume{ReadOnly: true, Replication: 1}, s.metrics, false)
 180         defer v.Teardown()
 181
 182         v.PutRaw(TestHash, TestBlock)
 183
 184         buf := make([]byte, BlockSize)
 185         _, err := v.Get(context.Background(), TestHash, buf)
 186         if err != nil {
 187                 c.Errorf("got err %v, expected nil", err)
 188         }
 189
 190         err = v.Put(context.Background(), TestHash, TestBlock)
 191         if err != MethodDisabledError {
 192                 c.Errorf("got err %v, expected MethodDisabledError", err)
 193         }
 194
 195         err = v.Touch(TestHash)
 196         if err != MethodDisabledError {
 197                 c.Errorf("got err %v, expected MethodDisabledError", err)
 198         }
 199
 200         err = v.Trash(TestHash)
 201         if err != MethodDisabledError {
 202                 c.Errorf("got err %v, expected MethodDisabledError", err)
 203         }
 204 }
 205
 206 func (s *UnixVolumeSuite) TestIsFull(c *check.C) {
 207         v := s.newTestableUnixVolume(c, s.cluster, arvados.Volume{Replication: 1}, s.metrics, false)
 208         defer v.Teardown()
 209
 210         fullPath := v.Root + "/full"
 211         now := fmt.Sprintf("%d", time.Now().Unix())
 212         os.Symlink(now, fullPath)
 213         if !v.IsFull() {
 214                 c.Errorf("%s: claims not to be full", v)
 215         }
 216         os.Remove(fullPath)
 217
 218         // Test with an expired /full link.
 219         expired := fmt.Sprintf("%d", time.Now().Unix()-3605)
 220         os.Symlink(expired, fullPath)
 221         if v.IsFull() {
 222                 c.Errorf("%s: should no longer be full", v)
 223         }
 224 }
 225
 226 func (s *UnixVolumeSuite) TestNodeStatus(c *check.C) {
 227         v := s.newTestableUnixVolume(c, s.cluster, arvados.Volume{Replication: 1}, s.metrics, false)
 228         defer v.Teardown()
 229
 230         // Get node status and make a basic sanity check.
 231         volinfo := v.Status()
 232         if volinfo.MountPoint != v.Root {
 233                 c.Errorf("GetNodeStatus mount_point %s, expected %s", volinfo.MountPoint, v.Root)
 234         }
 235         if volinfo.DeviceNum == 0 {
 236                 c.Errorf("uninitialized device_num in %v", volinfo)
 237         }
 238         if volinfo.BytesFree == 0 {
 239                 c.Errorf("uninitialized bytes_free in %v", volinfo)
 240         }
 241         if volinfo.BytesUsed == 0 {
 242                 c.Errorf("uninitialized bytes_used in %v", volinfo)
 243         }
 244 }
 245
 246 func (s *UnixVolumeSuite) TestUnixVolumeGetFuncWorkerError(c *check.C) {
 247         v := s.newTestableUnixVolume(c, s.cluster, arvados.Volume{Replication: 1}, s.metrics, false)
 248         defer v.Teardown()
 249
 250         v.Put(context.Background(), TestHash, TestBlock)
 251         mockErr := errors.New("Mock error")
 252         err := v.getFunc(context.Background(), v.blockPath(TestHash), func(rdr io.Reader) error {
 253                 return mockErr
 254         })
 255         if err != mockErr {
 256                 c.Errorf("Got %v, expected %v", err, mockErr)
 257         }
 258 }
 259
 260 func (s *UnixVolumeSuite) TestUnixVolumeGetFuncFileError(c *check.C) {
 261         v := s.newTestableUnixVolume(c, s.cluster, arvados.Volume{Replication: 1}, s.metrics, false)
 262         defer v.Teardown()
 263
 264         funcCalled := false
 265         err := v.getFunc(context.Background(), v.blockPath(TestHash), func(rdr io.Reader) error {
 266                 funcCalled = true
 267                 return nil
 268         })
 269         if err == nil {
 270                 c.Errorf("Expected error opening non-existent file")
 271         }
 272         if funcCalled {
 273                 c.Errorf("Worker func should not have been called")
 274         }
 275 }
 276
 277 func (s *UnixVolumeSuite) TestUnixVolumeGetFuncWorkerWaitsOnMutex(c *check.C) {
 278         v := s.newTestableUnixVolume(c, s.cluster, arvados.Volume{Replication: 1}, s.metrics, false)
 279         defer v.Teardown()
 280
 281         v.Put(context.Background(), TestHash, TestBlock)
 282
 283         mtx := NewMockMutex()
 284         v.locker = mtx
 285
 286         funcCalled := make(chan struct{})
 287         go v.getFunc(context.Background(), v.blockPath(TestHash), func(rdr io.Reader) error {
 288                 funcCalled <- struct{}{}
 289                 return nil
 290         })
 291         select {
 292         case mtx.AllowLock <- struct{}{}:
 293         case <-funcCalled:
 294                 c.Fatal("Function was called before mutex was acquired")
 295         case <-time.After(5 * time.Second):
 296                 c.Fatal("Timed out before mutex was acquired")
 297         }
 298         select {
 299         case <-funcCalled:
 300         case mtx.AllowUnlock <- struct{}{}:
 301                 c.Fatal("Mutex was released before function was called")
 302         case <-time.After(5 * time.Second):
 303                 c.Fatal("Timed out waiting for funcCalled")
 304         }
 305         select {
 306         case mtx.AllowUnlock <- struct{}{}:
 307         case <-time.After(5 * time.Second):
 308                 c.Fatal("Timed out waiting for getFunc() to release mutex")
 309         }
 310 }
 311
 312 func (s *UnixVolumeSuite) TestUnixVolumeCompare(c *check.C) {
 313         v := s.newTestableUnixVolume(c, s.cluster, arvados.Volume{Replication: 1}, s.metrics, false)
 314         defer v.Teardown()
 315
 316         v.Put(context.Background(), TestHash, TestBlock)
 317         err := v.Compare(context.Background(), TestHash, TestBlock)
 318         if err != nil {
 319                 c.Errorf("Got err %q, expected nil", err)
 320         }
 321
 322         err = v.Compare(context.Background(), TestHash, []byte("baddata"))
 323         if err != CollisionError {
 324                 c.Errorf("Got err %q, expected %q", err, CollisionError)
 325         }
 326
 327         v.Put(context.Background(), TestHash, []byte("baddata"))
 328         err = v.Compare(context.Background(), TestHash, TestBlock)
 329         if err != DiskHashError {
 330                 c.Errorf("Got err %q, expected %q", err, DiskHashError)
 331         }
 332
 333         p := fmt.Sprintf("%s/%s/%s", v.Root, TestHash[:3], TestHash)
 334         os.Chmod(p, 000)
 335         err = v.Compare(context.Background(), TestHash, TestBlock)
 336         if err == nil || strings.Index(err.Error(), "permission denied") < 0 {
 337                 c.Errorf("Got err %q, expected %q", err, "permission denied")
 338         }
 339 }
 340
 341 func (s *UnixVolumeSuite) TestUnixVolumeContextCancelPut(c *check.C) {
 342         v := s.newTestableUnixVolume(c, s.cluster, arvados.Volume{Replication: 1}, s.metrics, true)
 343         defer v.Teardown()
 344         v.locker.Lock()
 345         ctx, cancel := context.WithCancel(context.Background())
 346         go func() {
 347                 time.Sleep(50 * time.Millisecond)
 348                 cancel()
 349                 time.Sleep(50 * time.Millisecond)
 350                 v.locker.Unlock()
 351         }()
 352         err := v.Put(ctx, TestHash, TestBlock)
 353         if err != context.Canceled {
 354                 c.Errorf("Put() returned %s -- expected short read / canceled", err)
 355         }
 356 }
 357
 358 func (s *UnixVolumeSuite) TestUnixVolumeContextCancelGet(c *check.C) {
 359         v := s.newTestableUnixVolume(c, s.cluster, arvados.Volume{Replication: 1}, s.metrics, false)
 360         defer v.Teardown()
 361         bpath := v.blockPath(TestHash)
 362         v.PutRaw(TestHash, TestBlock)
 363         os.Remove(bpath)
 364         err := syscall.Mkfifo(bpath, 0600)
 365         if err != nil {
 366                 c.Fatalf("Mkfifo %s: %s", bpath, err)
 367         }
 368         defer os.Remove(bpath)
 369         ctx, cancel := context.WithCancel(context.Background())
 370         go func() {
 371                 time.Sleep(50 * time.Millisecond)
 372                 cancel()
 373         }()
 374         buf := make([]byte, len(TestBlock))
 375         n, err := v.Get(ctx, TestHash, buf)
 376         if n == len(TestBlock) || err != context.Canceled {
 377                 c.Errorf("Get() returned %d, %s -- expected short read / canceled", n, err)
 378         }
 379 }
 380
 381 func (s *UnixVolumeSuite) TestStats(c *check.C) {
 382         vol := s.newTestableUnixVolume(c, s.cluster, arvados.Volume{Replication: 1}, s.metrics, false)
 383         stats := func() string {
 384                 buf, err := json.Marshal(vol.InternalStats())
 385                 c.Check(err, check.IsNil)
 386                 return string(buf)
 387         }
 388
 389         c.Check(stats(), check.Matches, `.*"StatOps":1,.*`) // (*UnixVolume)check() calls Stat() once
 390         c.Check(stats(), check.Matches, `.*"Errors":0,.*`)
 391
 392         loc := "acbd18db4cc2f85cedef654fccc4a4d8"
 393         _, err := vol.Get(context.Background(), loc, make([]byte, 3))
 394         c.Check(err, check.NotNil)
 395         c.Check(stats(), check.Matches, `.*"StatOps":[^0],.*`)
 396         c.Check(stats(), check.Matches, `.*"Errors":[^0],.*`)
 397         c.Check(stats(), check.Matches, `.*"\*os\.PathError":[^0].*`)
 398         c.Check(stats(), check.Matches, `.*"InBytes":0,.*`)
 399         c.Check(stats(), check.Matches, `.*"OpenOps":0,.*`)
 400         c.Check(stats(), check.Matches, `.*"CreateOps":0,.*`)
 401
 402         err = vol.Put(context.Background(), loc, []byte("foo"))
 403         c.Check(err, check.IsNil)
 404         c.Check(stats(), check.Matches, `.*"OutBytes":3,.*`)
 405         c.Check(stats(), check.Matches, `.*"CreateOps":1,.*`)
 406         c.Check(stats(), check.Matches, `.*"OpenOps":0,.*`)
 407         c.Check(stats(), check.Matches, `.*"UtimesOps":0,.*`)
 408
 409         err = vol.Touch(loc)
 410         c.Check(err, check.IsNil)
 411         c.Check(stats(), check.Matches, `.*"FlockOps":1,.*`)
 412         c.Check(stats(), check.Matches, `.*"OpenOps":1,.*`)
 413         c.Check(stats(), check.Matches, `.*"UtimesOps":1,.*`)
 414
 415         _, err = vol.Get(context.Background(), loc, make([]byte, 3))
 416         c.Check(err, check.IsNil)
 417         err = vol.Compare(context.Background(), loc, []byte("foo"))
 418         c.Check(err, check.IsNil)
 419         c.Check(stats(), check.Matches, `.*"InBytes":6,.*`)
 420         c.Check(stats(), check.Matches, `.*"OpenOps":3,.*`)
 421
 422         err = vol.Trash(loc)
 423         c.Check(err, check.IsNil)
 424         c.Check(stats(), check.Matches, `.*"FlockOps":2,.*`)
 425 }