Merge branch '14287-federated-list'
[arvados.git] / lib / controller / federation / list.go
1 // Copyright (C) The Arvados Authors. All rights reserved.
2 //
3 // SPDX-License-Identifier: AGPL-3.0
4
5 package federation
6
7 import (
8         "context"
9         "fmt"
10         "net/http"
11         "sort"
12         "sync"
13
14         "git.curoverse.com/arvados.git/sdk/go/arvados"
15         "git.curoverse.com/arvados.git/sdk/go/httpserver"
16 )
17
18 //go:generate go run generate.go
19
20 // CollectionList is used as a template to auto-generate List()
21 // methods for other types; see generate.go.
22
23 func (conn *Conn) CollectionList(ctx context.Context, options arvados.ListOptions) (arvados.CollectionList, error) {
24         var mtx sync.Mutex
25         var merged arvados.CollectionList
26         err := conn.splitListRequest(ctx, options, func(ctx context.Context, _ string, backend arvados.API, options arvados.ListOptions) ([]string, error) {
27                 cl, err := backend.CollectionList(ctx, options)
28                 if err != nil {
29                         return nil, err
30                 }
31                 mtx.Lock()
32                 defer mtx.Unlock()
33                 if len(merged.Items) == 0 {
34                         merged = cl
35                 } else {
36                         merged.Items = append(merged.Items, cl.Items...)
37                 }
38                 uuids := make([]string, 0, len(cl.Items))
39                 for _, item := range cl.Items {
40                         uuids = append(uuids, item.UUID)
41                 }
42                 return uuids, nil
43         })
44         sort.Slice(merged.Items, func(i, j int) bool { return merged.Items[i].UUID < merged.Items[j].UUID })
45         return merged, err
46 }
47
48 // Call fn on one or more local/remote backends if opts indicates a
49 // federation-wide list query, i.e.:
50 //
51 // * There is at least one filter of the form
52 //   ["uuid","in",[a,b,c,...]] or ["uuid","=",a]
53 //
54 // * One or more of the supplied UUIDs (a,b,c,...) has a non-local
55 //   prefix.
56 //
57 // * There are no other filters
58 //
59 // (If opts doesn't indicate a federation-wide list query, fn is just
60 // called once with the local backend.)
61 //
62 // fn is called more than once only if the query meets the following
63 // restrictions:
64 //
65 // * Count=="none"
66 //
67 // * Limit<0
68 //
69 // * len(Order)==0
70 //
71 // * Each filter must be either "uuid = ..." or "uuid in [...]".
72 //
73 // * The maximum possible response size (total number of objects that
74 //   could potentially be matched by all of the specified filters)
75 //   exceeds the local cluster's response page size limit.
76 //
77 // If the query involves multiple backends but doesn't meet these
78 // restrictions, an error is returned without calling fn.
79 //
80 // Thus, the caller can assume that either:
81 //
82 // * splitListRequest() returns an error, or
83 //
84 // * fn is called exactly once, or
85 //
86 // * fn is called more than once, with options that satisfy the above
87 //   restrictions.
88 //
89 // Each call to fn indicates a single (local or remote) backend and a
90 // corresponding options argument suitable for sending to that
91 // backend.
92 func (conn *Conn) splitListRequest(ctx context.Context, opts arvados.ListOptions, fn func(context.Context, string, arvados.API, arvados.ListOptions) ([]string, error)) error {
93         cannotSplit := false
94         var matchAllFilters map[string]bool
95         for _, f := range opts.Filters {
96                 matchThisFilter := map[string]bool{}
97                 if f.Attr != "uuid" {
98                         cannotSplit = true
99                         continue
100                 }
101                 if f.Operator == "=" {
102                         if uuid, ok := f.Operand.(string); ok {
103                                 matchThisFilter[uuid] = true
104                         } else {
105                                 return httpErrorf(http.StatusBadRequest, "invalid operand type %T for filter %q", f.Operand, f)
106                         }
107                 } else if f.Operator == "in" {
108                         if operand, ok := f.Operand.([]interface{}); ok {
109                                 // skip any elements that aren't
110                                 // strings (thus can't match a UUID,
111                                 // thus can't affect the response).
112                                 for _, v := range operand {
113                                         if uuid, ok := v.(string); ok {
114                                                 matchThisFilter[uuid] = true
115                                         }
116                                 }
117                         } else if strings, ok := f.Operand.([]string); ok {
118                                 for _, uuid := range strings {
119                                         matchThisFilter[uuid] = true
120                                 }
121                         } else {
122                                 return httpErrorf(http.StatusBadRequest, "invalid operand type %T in filter %q", f.Operand, f)
123                         }
124                 } else {
125                         cannotSplit = true
126                         continue
127                 }
128
129                 if matchAllFilters == nil {
130                         matchAllFilters = matchThisFilter
131                 } else {
132                         // Reduce matchAllFilters to the intersection
133                         // of matchAllFilters ∩ matchThisFilter.
134                         for uuid := range matchAllFilters {
135                                 if !matchThisFilter[uuid] {
136                                         delete(matchAllFilters, uuid)
137                                 }
138                         }
139                 }
140         }
141
142         // Collate UUIDs in matchAllFilters by remote cluster ID --
143         // e.g., todoByRemote["aaaaa"]["aaaaa-4zz18-000000000000000"]
144         // will be true -- and count the total number of UUIDs we're
145         // filtering on, so we can compare it to our max page size
146         // limit.
147         nUUIDs := 0
148         todoByRemote := map[string]map[string]bool{}
149         for uuid := range matchAllFilters {
150                 if len(uuid) != 27 {
151                         // Cannot match anything, just drop it
152                 } else {
153                         if todoByRemote[uuid[:5]] == nil {
154                                 todoByRemote[uuid[:5]] = map[string]bool{}
155                         }
156                         todoByRemote[uuid[:5]][uuid] = true
157                         nUUIDs++
158                 }
159         }
160
161         if len(todoByRemote) > 1 {
162                 if cannotSplit {
163                         return httpErrorf(http.StatusBadRequest, "cannot execute federated list query: each filter must be either 'uuid = ...' or 'uuid in [...]'")
164                 }
165                 if opts.Count != "none" {
166                         return httpErrorf(http.StatusBadRequest, "cannot execute federated list query unless count==\"none\"")
167                 }
168                 if opts.Limit >= 0 || opts.Offset != 0 || len(opts.Order) > 0 {
169                         return httpErrorf(http.StatusBadRequest, "cannot execute federated list query with limit, offset, or order parameter")
170                 }
171                 if max := conn.cluster.API.MaxItemsPerResponse; nUUIDs > max {
172                         return httpErrorf(http.StatusBadRequest, "cannot execute federated list query because number of UUIDs (%d) exceeds page size limit %d", nUUIDs, max)
173                 }
174                 selectingUUID := false
175                 for _, attr := range opts.Select {
176                         if attr == "uuid" {
177                                 selectingUUID = true
178                         }
179                 }
180                 if opts.Select != nil && !selectingUUID {
181                         return httpErrorf(http.StatusBadRequest, "cannot execute federated list query with a select parameter that does not include uuid")
182                 }
183         }
184
185         ctx, cancel := context.WithCancel(ctx)
186         defer cancel()
187         errs := make(chan error, len(todoByRemote))
188         for clusterID, todo := range todoByRemote {
189                 go func(clusterID string, todo map[string]bool) {
190                         // This goroutine sends exactly one value to
191                         // errs.
192                         batch := make([]string, 0, len(todo))
193                         for uuid := range todo {
194                                 batch = append(batch, uuid)
195                         }
196
197                         var backend arvados.API
198                         if clusterID == conn.cluster.ClusterID {
199                                 backend = conn.local
200                         } else if backend = conn.remotes[clusterID]; backend == nil {
201                                 errs <- httpErrorf(http.StatusNotFound, "cannot execute federated list query: no proxy available for cluster %q", clusterID)
202                                 return
203                         }
204                         remoteOpts := opts
205                         for len(todo) > 0 {
206                                 if len(batch) > len(todo) {
207                                         // Reduce batch to just the todo's
208                                         batch = batch[:0]
209                                         for uuid := range todo {
210                                                 batch = append(batch, uuid)
211                                         }
212                                 }
213                                 remoteOpts.Filters = []arvados.Filter{{"uuid", "in", batch}}
214
215                                 done, err := fn(ctx, clusterID, backend, remoteOpts)
216                                 if err != nil {
217                                         errs <- err
218                                         return
219                                 }
220                                 progress := false
221                                 for _, uuid := range done {
222                                         if _, ok := todo[uuid]; ok {
223                                                 progress = true
224                                                 delete(todo, uuid)
225                                         }
226                                 }
227                                 if !progress {
228                                         errs <- httpErrorf(http.StatusBadGateway, "cannot make progress in federated list query: cluster %q returned none of the requested UUIDs", clusterID)
229                                         return
230                                 }
231                         }
232                         errs <- nil
233                 }(clusterID, todo)
234         }
235
236         // Wait for all goroutines to return, then return the first
237         // non-nil error, if any.
238         var firstErr error
239         for range todoByRemote {
240                 if err := <-errs; err != nil && firstErr == nil {
241                         firstErr = err
242                         // Signal to any remaining fn() calls that
243                         // further effort is futile.
244                         cancel()
245                 }
246         }
247         return firstErr
248 }
249
250 func httpErrorf(code int, format string, args ...interface{}) error {
251         return httpserver.ErrorWithStatus(fmt.Errorf(format, args...), code)
252 }