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 }