21316: Merge commit '1416f698b72de4b09350d9c2fb25c1405c3247bc' into 21316-left-panel...
[arvados.git] / services / keepstore / trash_worker_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         "container/list"
9         "context"
10         "time"
11
12         "git.arvados.org/arvados.git/sdk/go/ctxlog"
13         "github.com/prometheus/client_golang/prometheus"
14         check "gopkg.in/check.v1"
15 )
16
17 type TrashWorkerTestData struct {
18         Locator1    string
19         Block1      []byte
20         BlockMtime1 int64
21
22         Locator2    string
23         Block2      []byte
24         BlockMtime2 int64
25
26         CreateData      bool
27         CreateInVolume1 bool
28
29         UseTrashLifeTime bool
30         DifferentMtimes  bool
31
32         DeleteLocator    string
33         SpecifyMountUUID bool
34
35         ExpectLocator1 bool
36         ExpectLocator2 bool
37 }
38
39 // Delete block that does not exist in any of the keep volumes.
40 // Expect no errors.
41 func (s *HandlerSuite) TestTrashWorkerIntegration_GetNonExistingLocator(c *check.C) {
42         s.cluster.Collections.BlobTrash = true
43         testData := TrashWorkerTestData{
44                 Locator1: "5d41402abc4b2a76b9719d911017c592",
45                 Block1:   []byte("hello"),
46
47                 Locator2: "5d41402abc4b2a76b9719d911017c592",
48                 Block2:   []byte("hello"),
49
50                 CreateData: false,
51
52                 DeleteLocator: "5d41402abc4b2a76b9719d911017c592",
53
54                 ExpectLocator1: false,
55                 ExpectLocator2: false,
56         }
57         s.performTrashWorkerTest(c, testData)
58 }
59
60 // Delete a block that exists on volume 1 of the keep servers. Expect
61 // the second locator in volume 2 to be unaffected.
62 func (s *HandlerSuite) TestTrashWorkerIntegration_LocatorInVolume1(c *check.C) {
63         s.cluster.Collections.BlobTrash = true
64         testData := TrashWorkerTestData{
65                 Locator1: TestHash,
66                 Block1:   TestBlock,
67
68                 Locator2: TestHash2,
69                 Block2:   TestBlock2,
70
71                 CreateData: true,
72
73                 DeleteLocator: TestHash, // first locator
74
75                 ExpectLocator1: false,
76                 ExpectLocator2: true,
77         }
78         s.performTrashWorkerTest(c, testData)
79 }
80
81 // Delete a block that exists on volume 2 of the keep servers. Expect
82 // the first locator in volume 1 to be unaffected.
83 func (s *HandlerSuite) TestTrashWorkerIntegration_LocatorInVolume2(c *check.C) {
84         s.cluster.Collections.BlobTrash = true
85         testData := TrashWorkerTestData{
86                 Locator1: TestHash,
87                 Block1:   TestBlock,
88
89                 Locator2: TestHash2,
90                 Block2:   TestBlock2,
91
92                 CreateData: true,
93
94                 DeleteLocator: TestHash2, // locator 2
95
96                 ExpectLocator1: true,
97                 ExpectLocator2: false,
98         }
99         s.performTrashWorkerTest(c, testData)
100 }
101
102 // Delete a block with matching mtime for locator in both
103 // volumes. Expect locator to be deleted from both volumes.
104 func (s *HandlerSuite) TestTrashWorkerIntegration_LocatorInBothVolumes(c *check.C) {
105         s.cluster.Collections.BlobTrash = true
106         testData := TrashWorkerTestData{
107                 Locator1: TestHash,
108                 Block1:   TestBlock,
109
110                 Locator2: TestHash,
111                 Block2:   TestBlock,
112
113                 CreateData: true,
114
115                 DeleteLocator: TestHash,
116
117                 ExpectLocator1: false,
118                 ExpectLocator2: false,
119         }
120         s.performTrashWorkerTest(c, testData)
121 }
122
123 // Same locator with different Mtimes exists in both volumes. Delete
124 // the second and expect the first to be still around.
125 func (s *HandlerSuite) TestTrashWorkerIntegration_MtimeMatchesForLocator1ButNotForLocator2(c *check.C) {
126         s.cluster.Collections.BlobTrash = true
127         testData := TrashWorkerTestData{
128                 Locator1: TestHash,
129                 Block1:   TestBlock,
130
131                 Locator2: TestHash,
132                 Block2:   TestBlock,
133
134                 CreateData:      true,
135                 DifferentMtimes: true,
136
137                 DeleteLocator: TestHash,
138
139                 ExpectLocator1: true,
140                 ExpectLocator2: false,
141         }
142         s.performTrashWorkerTest(c, testData)
143 }
144
145 // Delete a block that exists on both volumes with matching mtimes,
146 // but specify a MountUUID in the request so it only gets deleted from
147 // the first volume.
148 func (s *HandlerSuite) TestTrashWorkerIntegration_SpecifyMountUUID(c *check.C) {
149         s.cluster.Collections.BlobTrash = true
150         testData := TrashWorkerTestData{
151                 Locator1: TestHash,
152                 Block1:   TestBlock,
153
154                 Locator2: TestHash,
155                 Block2:   TestBlock,
156
157                 CreateData: true,
158
159                 DeleteLocator:    TestHash,
160                 SpecifyMountUUID: true,
161
162                 ExpectLocator1: true,
163                 ExpectLocator2: true,
164         }
165         s.performTrashWorkerTest(c, testData)
166 }
167
168 // Two different locators in volume 1. Delete one of them. Expect the
169 // other unaffected.
170 func (s *HandlerSuite) TestTrashWorkerIntegration_TwoDifferentLocatorsInVolume1(c *check.C) {
171         s.cluster.Collections.BlobTrash = true
172         testData := TrashWorkerTestData{
173                 Locator1: TestHash,
174                 Block1:   TestBlock,
175
176                 Locator2: TestHash2,
177                 Block2:   TestBlock2,
178
179                 CreateData:      true,
180                 CreateInVolume1: true,
181
182                 DeleteLocator: TestHash, // locator 1
183
184                 ExpectLocator1: false,
185                 ExpectLocator2: true,
186         }
187         s.performTrashWorkerTest(c, testData)
188 }
189
190 // Allow default Trash Life time to be used. Thus, the newly created
191 // block will not be deleted because its Mtime is within the trash
192 // life time.
193 func (s *HandlerSuite) TestTrashWorkerIntegration_SameLocatorInTwoVolumesWithDefaultTrashLifeTime(c *check.C) {
194         s.cluster.Collections.BlobTrash = true
195         testData := TrashWorkerTestData{
196                 Locator1: TestHash,
197                 Block1:   TestBlock,
198
199                 Locator2: TestHash2,
200                 Block2:   TestBlock2,
201
202                 CreateData:      true,
203                 CreateInVolume1: true,
204
205                 UseTrashLifeTime: true,
206
207                 DeleteLocator: TestHash, // locator 1
208
209                 // Since trash life time is in effect, block won't be deleted.
210                 ExpectLocator1: true,
211                 ExpectLocator2: true,
212         }
213         s.performTrashWorkerTest(c, testData)
214 }
215
216 // Delete a block with matching mtime for locator in both volumes, but
217 // EnableDelete is false, so block won't be deleted.
218 func (s *HandlerSuite) TestTrashWorkerIntegration_DisabledDelete(c *check.C) {
219         s.cluster.Collections.BlobTrash = false
220         testData := TrashWorkerTestData{
221                 Locator1: TestHash,
222                 Block1:   TestBlock,
223
224                 Locator2: TestHash,
225                 Block2:   TestBlock,
226
227                 CreateData: true,
228
229                 DeleteLocator: TestHash,
230
231                 ExpectLocator1: true,
232                 ExpectLocator2: true,
233         }
234         s.performTrashWorkerTest(c, testData)
235 }
236
237 func (s *HandlerSuite) performTrashWorkerTest(c *check.C, testData TrashWorkerTestData) {
238         c.Assert(s.handler.setup(context.Background(), s.cluster, "", prometheus.NewRegistry(), testServiceURL), check.IsNil)
239         // Replace the router's trashq -- which the worker goroutines
240         // started by setup() are now receiving from -- with a new
241         // one, so we can see what the handler sends to it.
242         trashq := NewWorkQueue()
243         s.handler.Handler.(*router).trashq = trashq
244
245         // Put test content
246         mounts := s.handler.volmgr.AllWritable()
247         if testData.CreateData {
248                 mounts[0].Put(context.Background(), testData.Locator1, testData.Block1)
249                 mounts[0].Put(context.Background(), testData.Locator1+".meta", []byte("metadata"))
250
251                 if testData.CreateInVolume1 {
252                         mounts[0].Put(context.Background(), testData.Locator2, testData.Block2)
253                         mounts[0].Put(context.Background(), testData.Locator2+".meta", []byte("metadata"))
254                 } else {
255                         mounts[1].Put(context.Background(), testData.Locator2, testData.Block2)
256                         mounts[1].Put(context.Background(), testData.Locator2+".meta", []byte("metadata"))
257                 }
258         }
259
260         oldBlockTime := time.Now().Add(-s.cluster.Collections.BlobSigningTTL.Duration() - time.Minute)
261
262         // Create TrashRequest for the test
263         trashRequest := TrashRequest{
264                 Locator:    testData.DeleteLocator,
265                 BlockMtime: oldBlockTime.UnixNano(),
266         }
267         if testData.SpecifyMountUUID {
268                 trashRequest.MountUUID = s.handler.volmgr.Mounts()[0].UUID
269         }
270
271         // Run trash worker and put the trashRequest on trashq
272         trashList := list.New()
273         trashList.PushBack(trashRequest)
274
275         if !testData.UseTrashLifeTime {
276                 // Trash worker would not delete block if its Mtime is
277                 // within trash life time. Back-date the block to
278                 // allow the deletion to succeed.
279                 for _, mnt := range mounts {
280                         mnt.Volume.(*MockVolume).Timestamps[testData.DeleteLocator] = oldBlockTime
281                         if testData.DifferentMtimes {
282                                 oldBlockTime = oldBlockTime.Add(time.Second)
283                         }
284                 }
285         }
286         go RunTrashWorker(s.handler.volmgr, ctxlog.TestLogger(c), s.cluster, trashq)
287
288         // Install gate so all local operations block until we say go
289         gate := make(chan struct{})
290         for _, mnt := range mounts {
291                 mnt.Volume.(*MockVolume).Gate = gate
292         }
293
294         assertStatusItem := func(k string, expect float64) {
295                 if v := getStatusItem(s.handler, "TrashQueue", k); v != expect {
296                         c.Errorf("Got %s %v, expected %v", k, v, expect)
297                 }
298         }
299
300         assertStatusItem("InProgress", 0)
301         assertStatusItem("Queued", 0)
302
303         listLen := trashList.Len()
304         trashq.ReplaceQueue(trashList)
305
306         // Wait for worker to take request(s)
307         expectEqualWithin(c, time.Second, listLen, func() interface{} { return trashq.Status().InProgress })
308
309         // Ensure status.json also reports work is happening
310         assertStatusItem("InProgress", float64(1))
311         assertStatusItem("Queued", float64(listLen-1))
312
313         // Let worker proceed
314         close(gate)
315
316         // Wait for worker to finish
317         expectEqualWithin(c, time.Second, 0, func() interface{} { return trashq.Status().InProgress })
318
319         // Verify Locator1 to be un/deleted as expected
320         buf := make([]byte, BlockSize)
321         size, err := GetBlock(context.Background(), s.handler.volmgr, testData.Locator1, buf, nil)
322         if testData.ExpectLocator1 {
323                 if size == 0 || err != nil {
324                         c.Errorf("Expected Locator1 to be still present: %s", testData.Locator1)
325                 }
326         } else {
327                 if size > 0 || err == nil {
328                         c.Errorf("Expected Locator1 to be deleted: %s", testData.Locator1)
329                 }
330         }
331
332         // Verify Locator2 to be un/deleted as expected
333         if testData.Locator1 != testData.Locator2 {
334                 size, err = GetBlock(context.Background(), s.handler.volmgr, testData.Locator2, buf, nil)
335                 if testData.ExpectLocator2 {
336                         if size == 0 || err != nil {
337                                 c.Errorf("Expected Locator2 to be still present: %s", testData.Locator2)
338                         }
339                 } else {
340                         if size > 0 || err == nil {
341                                 c.Errorf("Expected Locator2 to be deleted: %s", testData.Locator2)
342                         }
343                 }
344         }
345
346         // The DifferentMtimes test puts the same locator in two
347         // different volumes, but only one copy has an Mtime matching
348         // the trash request.
349         if testData.DifferentMtimes {
350                 locatorFoundIn := 0
351                 for _, volume := range s.handler.volmgr.AllReadable() {
352                         buf := make([]byte, BlockSize)
353                         if _, err := volume.Get(context.Background(), testData.Locator1, buf); err == nil {
354                                 locatorFoundIn = locatorFoundIn + 1
355                         }
356                 }
357                 c.Check(locatorFoundIn, check.Equals, 1)
358         }
359 }