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