services/keep-balance/collection.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package main
   6
   7 import (
   8         "fmt"
   9         "time"
  10
  11         "git.curoverse.com/arvados.git/sdk/go/arvados"
  12 )
  13
  14 func countCollections(c *arvados.Client, params arvados.ResourceListParams) (int, error) {
  15         var page arvados.CollectionList
  16         var zero int
  17         params.Limit = &zero
  18         params.Count = "exact"
  19         err := c.RequestAndDecode(&page, "GET", "arvados/v1/collections", nil, params)
  20         return page.ItemsAvailable, err
  21 }
  22
  23 // EachCollection calls f once for every readable
  24 // collection. EachCollection stops if it encounters an error, such as
  25 // f returning a non-nil error.
  26 //
  27 // The progress function is called periodically with done (number of
  28 // times f has been called) and total (number of times f is expected
  29 // to be called).
  30 //
  31 // If pageSize > 0 it is used as the maximum page size in each API
  32 // call; otherwise the maximum allowed page size is requested.
  33 func EachCollection(c *arvados.Client, pageSize int, f func(arvados.Collection) error, progress func(done, total int)) error {
  34         if progress == nil {
  35                 progress = func(_, _ int) {}
  36         }
  37
  38         expectCount, err := countCollections(c, arvados.ResourceListParams{
  39                 IncludeTrash:       true,
  40                 IncludeOldVersions: true,
  41         })
  42         if err != nil {
  43                 return err
  44         }
  45
  46         limit := pageSize
  47         if limit <= 0 {
  48                 // Use the maximum page size the server allows
  49                 limit = 1<<31 - 1
  50         }
  51         params := arvados.ResourceListParams{
  52                 Limit:              &limit,
  53                 Order:              "modified_at, uuid",
  54                 Count:              "none",
  55                 Select:             []string{"uuid", "unsigned_manifest_text", "modified_at", "portable_data_hash", "replication_desired"},
  56                 IncludeTrash:       true,
  57                 IncludeOldVersions: true,
  58         }
  59         var last arvados.Collection
  60         var filterTime time.Time
  61         callCount := 0
  62         gettingExactTimestamp := false
  63         for {
  64                 progress(callCount, expectCount)
  65                 var page arvados.CollectionList
  66                 err := c.RequestAndDecode(&page, "GET", "arvados/v1/collections", nil, params)
  67                 if err != nil {
  68                         return err
  69                 }
  70                 for _, coll := range page.Items {
  71                         if last.ModifiedAt != nil && *last.ModifiedAt == *coll.ModifiedAt && last.UUID >= coll.UUID {
  72                                 continue
  73                         }
  74                         callCount++
  75                         err = f(coll)
  76                         if err != nil {
  77                                 return err
  78                         }
  79                         last = coll
  80                 }
  81                 if len(page.Items) == 0 && !gettingExactTimestamp {
  82                         break
  83                 } else if last.ModifiedAt == nil {
  84                         return fmt.Errorf("BUG: Last collection on the page (%s) has no modified_at timestamp; cannot make progress", last.UUID)
  85                 } else if len(page.Items) > 0 && *last.ModifiedAt == filterTime {
  86                         // If we requested time>=X and never got a
  87                         // time>X then we might not have received all
  88                         // items with time==X yet. Switch to
  89                         // gettingExactTimestamp mode (if we're not
  90                         // there already), advancing our UUID
  91                         // threshold with each request, until we get
  92                         // an empty page.
  93                         gettingExactTimestamp = true
  94                         params.Filters = []arvados.Filter{{
  95                                 Attr:     "modified_at",
  96                                 Operator: "=",
  97                                 Operand:  filterTime,
  98                         }, {
  99                                 Attr:     "uuid",
 100                                 Operator: ">",
 101                                 Operand:  last.UUID,
 102                         }}
 103                 } else if gettingExactTimestamp {
 104                         // This must be an empty page (in this mode,
 105                         // an unequal timestamp is impossible) so we
 106                         // can start getting pages of newer
 107                         // collections.
 108                         gettingExactTimestamp = false
 109                         params.Filters = []arvados.Filter{{
 110                                 Attr:     "modified_at",
 111                                 Operator: ">",
 112                                 Operand:  filterTime,
 113                         }}
 114                 } else {
 115                         // In the normal case, we know we have seen
 116                         // all collections with modtime<filterTime,
 117                         // but we might not have seen all that have
 118                         // modtime=filterTime. Hence we use >= instead
 119                         // of > and skip the obvious overlapping item,
 120                         // i.e., the last item on the previous
 121                         // page. In some edge cases this can return
 122                         // collections we have already seen, but
 123                         // avoiding that would add overhead in the
 124                         // overwhelmingly common cases, so we don't
 125                         // bother.
 126                         filterTime = *last.ModifiedAt
 127                         params.Filters = []arvados.Filter{{
 128                                 Attr:     "modified_at",
 129                                 Operator: ">=",
 130                                 Operand:  filterTime,
 131                         }, {
 132                                 Attr:     "uuid",
 133                                 Operator: "!=",
 134                                 Operand:  last.UUID,
 135                         }}
 136                 }
 137         }
 138         progress(callCount, expectCount)
 139
 140         if checkCount, err := countCollections(c, arvados.ResourceListParams{
 141                 Filters: []arvados.Filter{{
 142                         Attr:     "modified_at",
 143                         Operator: "<=",
 144                         Operand:  filterTime}},
 145                 IncludeTrash:       true,
 146                 IncludeOldVersions: true,
 147         }); err != nil {
 148                 return err
 149         } else if callCount < checkCount {
 150                 return fmt.Errorf("Retrieved %d collections with modtime <= T=%q, but server now reports there are %d collections with modtime <= T", callCount, filterTime, checkCount)
 151         }
 152
 153         return nil
 154 }