refactor as procedural
[arvados.git] / services / keepstore / trash_worker.go
1 package main
2
3 import (
4         "errors"
5         "time"
6
7         "git.curoverse.com/arvados.git/sdk/go/arvados"
8         log "github.com/Sirupsen/logrus"
9 )
10
11 // RunTrashWorker is used by Keepstore to initiate trash worker channel goroutine.
12 //      The channel will process trash list.
13 //              For each (next) trash request:
14 //      Delete the block indicated by the trash request Locator
15 //              Repeat
16 //
17 func RunTrashWorker(trashq *WorkQueue) {
18         for item := range trashq.NextItem {
19                 trashRequest := item.(TrashRequest)
20                 TrashItem(trashRequest)
21                 trashq.DoneItem <- struct{}{}
22         }
23 }
24
25 // TrashItem deletes the indicated block from every writable volume.
26 func TrashItem(trashRequest TrashRequest) {
27         reqMtime := time.Unix(0, trashRequest.BlockMtime)
28         if time.Since(reqMtime) < theConfig.BlobSignatureTTL.Duration() {
29                 log.Printf("WARNING: data manager asked to delete a %v old block %v (BlockMtime %d = %v), but my blobSignatureTTL is %v! Skipping.",
30                         arvados.Duration(time.Since(reqMtime)),
31                         trashRequest.Locator,
32                         trashRequest.BlockMtime,
33                         reqMtime,
34                         theConfig.BlobSignatureTTL)
35                 return
36         }
37
38         for _, volume := range KeepVM.AllWritable() {
39                 mtime, err := volume.Mtime(trashRequest.Locator)
40                 if err != nil {
41                         log.Printf("%v Delete(%v): %v", volume, trashRequest.Locator, err)
42                         continue
43                 }
44                 if trashRequest.BlockMtime != mtime.UnixNano() {
45                         log.Printf("%v Delete(%v): stored mtime %v does not match trash list value %v", volume, trashRequest.Locator, mtime.UnixNano(), trashRequest.BlockMtime)
46                         continue
47                 }
48
49                 if !theConfig.EnableDelete {
50                         err = errors.New("did not delete block because EnableDelete is false")
51                 } else {
52                         err = volume.Trash(trashRequest.Locator)
53                 }
54
55                 if err != nil {
56                         log.Printf("%v Delete(%v): %v", volume, trashRequest.Locator, err)
57                 } else {
58                         log.Printf("%v Delete(%v) OK", volume, trashRequest.Locator)
59                 }
60         }
61 }