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 // * there are no filters other than the "uuid = ..." and "uuid in
  72 //   ..." filters mentioned above.
  73 //
  74 // * The maximum possible response size (total number of objects that
  75 //   could potentially be matched by all of the specified filters)
  76 //   exceeds the local cluster's response page size limit.
  77 //
  78 // If the query involves multiple backends but doesn't meet these
  79 // restrictions, an error is returned without calling fn.
  80 //
  81 // Thus, the caller can assume that either:
  82 //
  83 // * splitListRequest() returns an error, or
  84 //
  85 // * fn is called exactly once, or
  86 //
  87 // * fn is called more than once, with options that satisfy the above
  88 //   restrictions.
  89 //
  90 // Each call to fn indicates a single (local or remote) backend and a
  91 // corresponding options argument suitable for sending to that
  92 // backend.
  93 func (conn *Conn) splitListRequest(ctx context.Context, opts arvados.ListOptions, fn func(context.Context, string, arvados.API, arvados.ListOptions) ([]string, error)) error {
  94         cannotSplit := false
  95         var matchAllFilters map[string]bool
  96         for _, f := range opts.Filters {
  97                 matchThisFilter := map[string]bool{}
  98                 if f.Attr != "uuid" {
  99                         cannotSplit = true
 100                         continue
 101                 }
 102                 if f.Operator == "=" {
 103                         if uuid, ok := f.Operand.(string); ok {
 104                                 matchThisFilter[uuid] = true
 105                         } else {
 106                                 return httpErrorf(http.StatusBadRequest, "invalid operand type %T for filter %q", f.Operand, f)
 107                         }
 108                 } else if f.Operator == "in" {
 109                         if operand, ok := f.Operand.([]interface{}); ok {
 110                                 // skip any elements that aren't
 111                                 // strings (thus can't match a UUID,
 112                                 // thus can't affect the response).
 113                                 for _, v := range operand {
 114                                         if uuid, ok := v.(string); ok {
 115                                                 matchThisFilter[uuid] = true
 116                                         }
 117                                 }
 118                         } else if strings, ok := f.Operand.([]string); ok {
 119                                 for _, uuid := range strings {
 120                                         matchThisFilter[uuid] = true
 121                                 }
 122                         } else {
 123                                 return httpErrorf(http.StatusBadRequest, "invalid operand type %T in filter %q", f.Operand, f)
 124                         }
 125                 } else {
 126                         cannotSplit = true
 127                         continue
 128                 }
 129
 130                 if matchAllFilters == nil {
 131                         matchAllFilters = matchThisFilter
 132                 } else {
 133                         // Reduce matchAllFilters to the intersection
 134                         // of matchAllFilters ∩ matchThisFilter.
 135                         for uuid := range matchAllFilters {
 136                                 if !matchThisFilter[uuid] {
 137                                         delete(matchAllFilters, uuid)
 138                                 }
 139                         }
 140                 }
 141         }
 142
 143         // Collate UUIDs in matchAllFilters by remote cluster ID --
 144         // e.g., todoByRemote["aaaaa"]["aaaaa-4zz18-000000000000000"]
 145         // will be true -- and count the total number of UUIDs we're
 146         // filtering on, so we can compare it to our max page size
 147         // limit.
 148         nUUIDs := 0
 149         todoByRemote := map[string]map[string]bool{}
 150         for uuid := range matchAllFilters {
 151                 if len(uuid) != 27 {
 152                         // Cannot match anything, just drop it
 153                 } else {
 154                         if todoByRemote[uuid[:5]] == nil {
 155                                 todoByRemote[uuid[:5]] = map[string]bool{}
 156                         }
 157                         todoByRemote[uuid[:5]][uuid] = true
 158                         nUUIDs++
 159                 }
 160         }
 161
 162         if len(todoByRemote) > 1 {
 163                 if cannotSplit {
 164                         return httpErrorf(http.StatusBadRequest, "cannot execute federated list query with filters other than 'uuid = ...' and 'uuid in [...]'")
 165                 }
 166                 if opts.Count != "none" {
 167                         return httpErrorf(http.StatusBadRequest, "cannot execute federated list query unless count==\"none\"")
 168                 }
 169                 if opts.Limit >= 0 || opts.Offset != 0 || len(opts.Order) > 0 {
 170                         return httpErrorf(http.StatusBadRequest, "cannot execute federated list query with limit, offset, or order parameter")
 171                 }
 172                 if max := conn.cluster.API.MaxItemsPerResponse; nUUIDs > max {
 173                         return httpErrorf(http.StatusBadRequest, "cannot execute federated list query because number of UUIDs (%d) exceeds page size limit %d", nUUIDs, max)
 174                 }
 175                 selectingUUID := false
 176                 for _, attr := range opts.Select {
 177                         if attr == "uuid" {
 178                                 selectingUUID = true
 179                         }
 180                 }
 181                 if opts.Select != nil && !selectingUUID {
 182                         return httpErrorf(http.StatusBadRequest, "cannot execute federated list query with a select parameter that does not include uuid")
 183                 }
 184         }
 185
 186         ctx, cancel := context.WithCancel(ctx)
 187         defer cancel()
 188         errs := make(chan error, len(todoByRemote))
 189         for clusterID, todo := range todoByRemote {
 190                 go func(clusterID string, todo map[string]bool) {
 191                         // This goroutine sends exactly one value to
 192                         // errs.
 193                         batch := make([]string, 0, len(todo))
 194                         for uuid := range todo {
 195                                 batch = append(batch, uuid)
 196                         }
 197
 198                         var backend arvados.API
 199                         if clusterID == conn.cluster.ClusterID {
 200                                 backend = conn.local
 201                         } else if backend = conn.remotes[clusterID]; backend == nil {
 202                                 errs <- httpErrorf(http.StatusNotFound, "cannot execute federated list query: no proxy available for cluster %q", clusterID)
 203                                 return
 204                         }
 205                         remoteOpts := opts
 206                         for len(todo) > 0 {
 207                                 if len(batch) > len(todo) {
 208                                         // Reduce batch to just the todo's
 209                                         batch = batch[:0]
 210                                         for uuid := range todo {
 211                                                 batch = append(batch, uuid)
 212                                         }
 213                                 }
 214                                 remoteOpts.Filters = []arvados.Filter{{"uuid", "in", batch}}
 215
 216                                 done, err := fn(ctx, clusterID, backend, remoteOpts)
 217                                 if err != nil {
 218                                         errs <- err
 219                                         return
 220                                 }
 221                                 progress := false
 222                                 for _, uuid := range done {
 223                                         if _, ok := todo[uuid]; ok {
 224                                                 progress = true
 225                                                 delete(todo, uuid)
 226                                         }
 227                                 }
 228                                 if !progress {
 229                                         errs <- httpErrorf(http.StatusBadGateway, "cannot make progress in federated list query: cluster %q returned none of the requested UUIDs", clusterID)
 230                                         return
 231                                 }
 232                         }
 233                         errs <- nil
 234                 }(clusterID, todo)
 235         }
 236
 237         // Wait for all goroutines to return, then return the first
 238         // non-nil error, if any.
 239         var firstErr error
 240         for range todoByRemote {
 241                 if err := <-errs; err != nil && firstErr == nil {
 242                         firstErr = err
 243                         // Signal to any remaining fn() calls that
 244                         // further effort is futile.
 245                         cancel()
 246                 }
 247         }
 248         return firstErr
 249 }
 250
 251 func httpErrorf(code int, format string, args ...interface{}) error {
 252         return httpserver.ErrorWithStatus(fmt.Errorf(format, args...), code)
 253 }