services/datamanager/summary/trash_list.go

   1 // Code for generating trash lists
   2
   3 package summary
   4
   5 import (
   6         "errors"
   7         "fmt"
   8         "git.curoverse.com/arvados.git/sdk/go/keepclient"
   9         "git.curoverse.com/arvados.git/services/datamanager/keep"
  10         "time"
  11 )
  12
  13 // BuildTrashLists builds list of blocks to be sent to trash queue
  14 func BuildTrashLists(kc *keepclient.KeepClient,
  15         keepServerInfo *keep.ReadServers,
  16         keepBlocksNotInCollections BlockSet) (m map[string]keep.TrashList, err error) {
  17
  18         // Servers that are writeable
  19         writableServers := map[string]struct{}{}
  20         for _, url := range kc.WritableLocalRoots() {
  21                 writableServers[url] = struct{}{}
  22         }
  23
  24         _ttl, err := kc.Arvados.Discovery("blobSignatureTtl")
  25         if err != nil {
  26                 return nil, errors.New(fmt.Sprintf("Failed to get blobSignatureTtl, can't build trash lists: %v", err))
  27         }
  28
  29         ttl := int64(_ttl.(float64))
  30
  31         // expire unreferenced blocks more than "ttl" seconds old.
  32         expiry := time.Now().UTC().UnixNano() - ttl*1e9
  33
  34         return buildTrashListsInternal(writableServers, keepServerInfo, expiry, keepBlocksNotInCollections), nil
  35 }
  36
  37 func buildTrashListsInternal(writableServers map[string]struct{},
  38         keepServerInfo *keep.ReadServers,
  39         expiry int64,
  40         keepBlocksNotInCollections BlockSet) (m map[string]keep.TrashList) {
  41
  42         m = make(map[string]keep.TrashList)
  43
  44         for block := range keepBlocksNotInCollections {
  45                 for _, blockOnServer := range keepServerInfo.BlockToServers[block] {
  46                         if blockOnServer.Mtime >= expiry {
  47                                 continue
  48                         }
  49
  50                         // block is older than expire cutoff
  51                         srv := keepServerInfo.KeepServerIndexToAddress[blockOnServer.ServerIndex].String()
  52
  53                         if _, writable := writableServers[srv]; !writable {
  54                                 continue
  55                         }
  56
  57                         m[srv] = append(m[srv], keep.TrashRequest{Locator: block.Digest.String(), BlockMtime: blockOnServer.Mtime})
  58                 }
  59         }
  60         return
  61
  62 }