closes #9514
[arvados.git] / services / keep-balance / collection.go
1 package main
2
3 import (
4         "fmt"
5         "time"
6
7         "git.curoverse.com/arvados.git/sdk/go/arvados"
8 )
9
10 func countCollections(c *arvados.Client, params arvados.ResourceListParams) (int, error) {
11         var page arvados.CollectionList
12         var zero int
13         params.Limit = &zero
14         err := c.RequestAndDecode(&page, "GET", "arvados/v1/collections", nil, params)
15         return page.ItemsAvailable, err
16 }
17
18 // EachCollection calls f once for every readable
19 // collection. EachCollection stops if it encounters an error, such as
20 // f returning a non-nil error.
21 //
22 // The progress function is called periodically with done (number of
23 // times f has been called) and total (number of times f is expected
24 // to be called).
25 //
26 // If pageSize > 0 it is used as the maximum page size in each API
27 // call; otherwise the maximum allowed page size is requested.
28 func EachCollection(c *arvados.Client, pageSize int, f func(arvados.Collection) error, progress func(done, total int)) error {
29         if progress == nil {
30                 progress = func(_, _ int) {}
31         }
32
33         expectCount, err := countCollections(c, arvados.ResourceListParams{})
34         if err != nil {
35                 return err
36         }
37
38         limit := pageSize
39         if limit <= 0 {
40                 // Use the maximum page size the server allows
41                 limit = 1<<31 - 1
42         }
43         params := arvados.ResourceListParams{
44                 Limit:  &limit,
45                 Order:  "modified_at, uuid",
46                 Select: []string{"uuid", "manifest_text", "modified_at", "portable_data_hash", "replication_desired"},
47         }
48         var last arvados.Collection
49         var filterTime time.Time
50         callCount := 0
51         for {
52                 progress(callCount, expectCount)
53                 var page arvados.CollectionList
54                 err := c.RequestAndDecode(&page, "GET", "arvados/v1/collections", nil, params)
55                 if err != nil {
56                         return err
57                 }
58                 for _, coll := range page.Items {
59                         if last.ModifiedAt != nil && *last.ModifiedAt == *coll.ModifiedAt && last.UUID >= coll.UUID {
60                                 continue
61                         }
62                         callCount++
63                         err = f(coll)
64                         if err != nil {
65                                 return err
66                         }
67                         last = coll
68                 }
69                 if last.ModifiedAt == nil || *last.ModifiedAt == filterTime {
70                         if page.ItemsAvailable > len(page.Items) {
71                                 // TODO: use "mtime=X && UUID>Y"
72                                 // filters to get all collections with
73                                 // this timestamp, then use "mtime>X"
74                                 // to get the next timestamp.
75                                 return fmt.Errorf("BUG: Received an entire page with the same modified_at timestamp (%v), cannot make progress", filterTime)
76                         }
77                         break
78                 }
79                 filterTime = *last.ModifiedAt
80                 params.Filters = []arvados.Filter{{
81                         Attr:     "modified_at",
82                         Operator: ">=",
83                         Operand:  filterTime,
84                 }, {
85                         Attr:     "uuid",
86                         Operator: "!=",
87                         Operand:  last.UUID,
88                 }}
89         }
90         progress(callCount, expectCount)
91
92         if checkCount, err := countCollections(c, arvados.ResourceListParams{Filters: []arvados.Filter{{
93                 Attr:     "modified_at",
94                 Operator: "<=",
95                 Operand:  filterTime}}}); err != nil {
96                 return err
97         } else if callCount < checkCount {
98                 return fmt.Errorf("Retrieved %d collections with modtime <= T=%q, but server now reports there are %d collections with modtime <= T", callCount, filterTime, checkCount)
99         }
100
101         return nil
102 }