services/datamanager/summary/trash_list.go

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