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.curoverse.com/arvados.git/sdk/go/arvados"
  16         "git.curoverse.com/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 must be 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) > 1 {
 185                 if cannotSplit {
 186                         return httpErrorf(http.StatusBadRequest, "cannot execute federated list query: each filter must be either 'uuid = ...' or 'uuid in [...]'")
 187                 }
 188                 if opts.Count != "none" {
 189                         return httpErrorf(http.StatusBadRequest, "cannot execute federated list query unless count==\"none\"")
 190                 }
 191                 if opts.Limit >= 0 || opts.Offset != 0 || len(opts.Order) > 0 {
 192                         return httpErrorf(http.StatusBadRequest, "cannot execute federated list query with limit, offset, or order parameter")
 193                 }
 194                 if max := conn.cluster.API.MaxItemsPerResponse; nUUIDs > max {
 195                         return httpErrorf(http.StatusBadRequest, "cannot execute federated list query because number of UUIDs (%d) exceeds page size limit %d", nUUIDs, max)
 196                 }
 197                 selectingUUID := false
 198                 for _, attr := range opts.Select {
 199                         if attr == "uuid" {
 200                                 selectingUUID = true
 201                         }
 202                 }
 203                 if opts.Select != nil && !selectingUUID {
 204                         return httpErrorf(http.StatusBadRequest, "cannot execute federated list query with a select parameter that does not include uuid")
 205                 }
 206         }
 207
 208         ctx, cancel := context.WithCancel(ctx)
 209         defer cancel()
 210         errs := make(chan error, len(todoByRemote))
 211         for clusterID, todo := range todoByRemote {
 212                 go func(clusterID string, todo map[string]bool) {
 213                         // This goroutine sends exactly one value to
 214                         // errs.
 215                         batch := make([]string, 0, len(todo))
 216                         for uuid := range todo {
 217                                 batch = append(batch, uuid)
 218                         }
 219
 220                         var backend arvados.API
 221                         if clusterID == conn.cluster.ClusterID {
 222                                 backend = conn.local
 223                         } else if backend = conn.remotes[clusterID]; backend == nil {
 224                                 errs <- httpErrorf(http.StatusNotFound, "cannot execute federated list query: no proxy available for cluster %q", clusterID)
 225                                 return
 226                         }
 227                         remoteOpts := opts
 228                         for len(todo) > 0 {
 229                                 if len(batch) > len(todo) {
 230                                         // Reduce batch to just the todo's
 231                                         batch = batch[:0]
 232                                         for uuid := range todo {
 233                                                 batch = append(batch, uuid)
 234                                         }
 235                                 }
 236                                 remoteOpts.Filters = []arvados.Filter{{"uuid", "in", batch}}
 237
 238                                 done, err := fn(ctx, clusterID, backend, remoteOpts)
 239                                 if err != nil {
 240                                         errs <- httpErrorf(http.StatusBadGateway, err.Error())
 241                                         return
 242                                 }
 243                                 progress := false
 244                                 for _, uuid := range done {
 245                                         if _, ok := todo[uuid]; ok {
 246                                                 progress = true
 247                                                 delete(todo, uuid)
 248                                         }
 249                                 }
 250                                 if len(done) == 0 {
 251                                         // Zero items == no more
 252                                         // results exist, no need to
 253                                         // get another page.
 254                                         break
 255                                 } else if !progress {
 256                                         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))
 257                                         return
 258                                 }
 259                         }
 260                         errs <- nil
 261                 }(clusterID, todo)
 262         }
 263
 264         // Wait for all goroutines to return, then return the first
 265         // non-nil error, if any.
 266         var firstErr error
 267         for range todoByRemote {
 268                 if err := <-errs; err != nil && firstErr == nil {
 269                         firstErr = err
 270                         // Signal to any remaining fn() calls that
 271                         // further effort is futile.
 272                         cancel()
 273                 }
 274         }
 275         return firstErr
 276 }
 277
 278 func httpErrorf(code int, format string, args ...interface{}) error {
 279         return httpserver.ErrorWithStatus(fmt.Errorf(format, args...), code)
 280 }