13647: Merge branch 'master' into 13647-keepstore-config
[arvados.git] / services / keepstore / 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         "crypto/md5"
11         "errors"
12         "fmt"
13         "io"
14         "os"
15         "strings"
16         "sync"
17         "time"
18
19         "git.curoverse.com/arvados.git/sdk/go/arvados"
20         "github.com/sirupsen/logrus"
21 )
22
23 var (
24         TestBlock       = []byte("The quick brown fox jumps over the lazy dog.")
25         TestHash        = "e4d909c290d0fb1ca068ffaddf22cbd0"
26         TestHashPutResp = "e4d909c290d0fb1ca068ffaddf22cbd0+44\n"
27
28         TestBlock2 = []byte("Pack my box with five dozen liquor jugs.")
29         TestHash2  = "f15ac516f788aec4f30932ffb6395c39"
30
31         TestBlock3 = []byte("Now is the time for all good men to come to the aid of their country.")
32         TestHash3  = "eed29bbffbc2dbe5e5ee0bb71888e61f"
33
34         // BadBlock is used to test collisions and corruption.
35         // It must not match any test hashes.
36         BadBlock = []byte("The magic words are squeamish ossifrage.")
37
38         EmptyHash  = "d41d8cd98f00b204e9800998ecf8427e"
39         EmptyBlock = []byte("")
40 )
41
42 // A TestableVolume allows test suites to manipulate the state of an
43 // underlying Volume, in order to test behavior in cases that are
44 // impractical to achieve with a sequence of normal Volume operations.
45 type TestableVolume interface {
46         Volume
47
48         // [Over]write content for a locator with the given data,
49         // bypassing all constraints like readonly and serialize.
50         PutRaw(locator string, data []byte)
51
52         // Returns the strings that a driver uses to record read/write operations.
53         ReadWriteOperationLabelValues() (r, w string)
54
55         // Specify the value Mtime() should return, until the next
56         // call to Touch, TouchWithDate, or Put.
57         TouchWithDate(locator string, lastPut time.Time)
58
59         // Clean up, delete temporary files.
60         Teardown()
61 }
62
63 func init() {
64         driver["mock"] = newMockVolume
65 }
66
67 // MockVolumes are test doubles for Volumes, used to test handlers.
68 type MockVolume struct {
69         Store      map[string][]byte
70         Timestamps map[string]time.Time
71
72         // Bad volumes return an error for every operation.
73         Bad            bool
74         BadVolumeError error
75
76         // Touchable volumes' Touch() method succeeds for a locator
77         // that has been Put().
78         Touchable bool
79
80         // Gate is a "starting gate", allowing test cases to pause
81         // volume operations long enough to inspect state. Every
82         // operation (except Status) starts by receiving from
83         // Gate. Sending one value unblocks one operation; closing the
84         // channel unblocks all operations. By default, Gate is a
85         // closed channel, so all operations proceed without
86         // blocking. See trash_worker_test.go for an example.
87         Gate chan struct{} `json:"-"`
88
89         cluster *arvados.Cluster
90         volume  arvados.Volume
91         logger  logrus.FieldLogger
92         metrics *volumeMetricsVecs
93         called  map[string]int
94         mutex   sync.Mutex
95 }
96
97 // newMockVolume returns a non-Bad, non-Readonly, Touchable mock
98 // volume.
99 func newMockVolume(cluster *arvados.Cluster, volume arvados.Volume, logger logrus.FieldLogger, metrics *volumeMetricsVecs) (Volume, error) {
100         gate := make(chan struct{})
101         close(gate)
102         return &MockVolume{
103                 Store:      make(map[string][]byte),
104                 Timestamps: make(map[string]time.Time),
105                 Bad:        false,
106                 Touchable:  true,
107                 called:     map[string]int{},
108                 Gate:       gate,
109                 cluster:    cluster,
110                 volume:     volume,
111                 logger:     logger,
112                 metrics:    metrics,
113         }, nil
114 }
115
116 // CallCount returns how many times the named method has been called.
117 func (v *MockVolume) CallCount(method string) int {
118         v.mutex.Lock()
119         defer v.mutex.Unlock()
120         c, ok := v.called[method]
121         if !ok {
122                 return 0
123         }
124         return c
125 }
126
127 func (v *MockVolume) gotCall(method string) {
128         v.mutex.Lock()
129         defer v.mutex.Unlock()
130         if _, ok := v.called[method]; !ok {
131                 v.called[method] = 1
132         } else {
133                 v.called[method]++
134         }
135 }
136
137 func (v *MockVolume) Compare(ctx context.Context, loc string, buf []byte) error {
138         v.gotCall("Compare")
139         <-v.Gate
140         if v.Bad {
141                 return v.BadVolumeError
142         } else if block, ok := v.Store[loc]; ok {
143                 if fmt.Sprintf("%x", md5.Sum(block)) != loc {
144                         return DiskHashError
145                 }
146                 if bytes.Compare(buf, block) != 0 {
147                         return CollisionError
148                 }
149                 return nil
150         } else {
151                 return NotFoundError
152         }
153 }
154
155 func (v *MockVolume) Get(ctx context.Context, loc string, buf []byte) (int, error) {
156         v.gotCall("Get")
157         <-v.Gate
158         if v.Bad {
159                 return 0, v.BadVolumeError
160         } else if block, ok := v.Store[loc]; ok {
161                 copy(buf[:len(block)], block)
162                 return len(block), nil
163         }
164         return 0, os.ErrNotExist
165 }
166
167 func (v *MockVolume) Put(ctx context.Context, loc string, block []byte) error {
168         v.gotCall("Put")
169         <-v.Gate
170         if v.Bad {
171                 return v.BadVolumeError
172         }
173         if v.volume.ReadOnly {
174                 return MethodDisabledError
175         }
176         v.Store[loc] = block
177         return v.Touch(loc)
178 }
179
180 func (v *MockVolume) Touch(loc string) error {
181         v.gotCall("Touch")
182         <-v.Gate
183         if v.volume.ReadOnly {
184                 return MethodDisabledError
185         }
186         if v.Touchable {
187                 v.Timestamps[loc] = time.Now()
188                 return nil
189         }
190         return errors.New("Touch failed")
191 }
192
193 func (v *MockVolume) Mtime(loc string) (time.Time, error) {
194         v.gotCall("Mtime")
195         <-v.Gate
196         var mtime time.Time
197         var err error
198         if v.Bad {
199                 err = v.BadVolumeError
200         } else if t, ok := v.Timestamps[loc]; ok {
201                 mtime = t
202         } else {
203                 err = os.ErrNotExist
204         }
205         return mtime, err
206 }
207
208 func (v *MockVolume) IndexTo(prefix string, w io.Writer) error {
209         v.gotCall("IndexTo")
210         <-v.Gate
211         for loc, block := range v.Store {
212                 if !IsValidLocator(loc) || !strings.HasPrefix(loc, prefix) {
213                         continue
214                 }
215                 _, err := fmt.Fprintf(w, "%s+%d %d\n",
216                         loc, len(block), 123456789)
217                 if err != nil {
218                         return err
219                 }
220         }
221         return nil
222 }
223
224 func (v *MockVolume) Trash(loc string) error {
225         v.gotCall("Delete")
226         <-v.Gate
227         if v.volume.ReadOnly {
228                 return MethodDisabledError
229         }
230         if _, ok := v.Store[loc]; ok {
231                 if time.Since(v.Timestamps[loc]) < time.Duration(v.cluster.Collections.BlobSigningTTL) {
232                         return nil
233                 }
234                 delete(v.Store, loc)
235                 return nil
236         }
237         return os.ErrNotExist
238 }
239
240 func (v *MockVolume) GetDeviceID() string {
241         return "mock-device-id"
242 }
243
244 func (v *MockVolume) Untrash(loc string) error {
245         return nil
246 }
247
248 func (v *MockVolume) Status() *VolumeStatus {
249         var used uint64
250         for _, block := range v.Store {
251                 used = used + uint64(len(block))
252         }
253         return &VolumeStatus{"/bogo", 123, 1000000 - used, used}
254 }
255
256 func (v *MockVolume) String() string {
257         return "[MockVolume]"
258 }
259
260 func (v *MockVolume) EmptyTrash() {
261 }
262
263 func (v *MockVolume) GetStorageClasses() []string {
264         return nil
265 }