lib/controller/federation.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package controller
   6
   7 import (
   8         "bufio"
   9         "bytes"
  10         "context"
  11         "crypto/md5"
  12         "database/sql"
  13         "encoding/json"
  14         "fmt"
  15         "io"
  16         "io/ioutil"
  17         "net/http"
  18         "net/url"
  19         "regexp"
  20         "strings"
  21         "sync"
  22
  23         "git.curoverse.com/arvados.git/sdk/go/arvados"
  24         "git.curoverse.com/arvados.git/sdk/go/auth"
  25         "git.curoverse.com/arvados.git/sdk/go/httpserver"
  26         "git.curoverse.com/arvados.git/sdk/go/keepclient"
  27 )
  28
  29 var pathPattern = `^/arvados/v1/%s(/([0-9a-z]{5})-%s-[0-9a-z]{15})?(.*)$`
  30 var wfRe = regexp.MustCompile(fmt.Sprintf(pathPattern, "workflows", "7fd4e"))
  31 var containersRe = regexp.MustCompile(fmt.Sprintf(pathPattern, "containers", "dz642"))
  32 var containerRequestsRe = regexp.MustCompile(fmt.Sprintf(pathPattern, "container_requests", "xvhdp"))
  33 var collectionRe = regexp.MustCompile(fmt.Sprintf(pathPattern, "collections", "4zz18"))
  34 var collectionByPDHRe = regexp.MustCompile(`^/arvados/v1/collections/([0-9a-fA-F]{32}\+[0-9]+)+$`)
  35
  36 type genericFederatedRequestHandler struct {
  37         next    http.Handler
  38         handler *Handler
  39         matcher *regexp.Regexp
  40 }
  41
  42 type collectionFederatedRequestHandler struct {
  43         next    http.Handler
  44         handler *Handler
  45 }
  46
  47 func (h *Handler) remoteClusterRequest(remoteID string, w http.ResponseWriter, req *http.Request, filter ResponseFilter) {
  48         remote, ok := h.Cluster.RemoteClusters[remoteID]
  49         if !ok {
  50                 httpserver.Error(w, "no proxy available for cluster "+remoteID, http.StatusNotFound)
  51                 return
  52         }
  53         scheme := remote.Scheme
  54         if scheme == "" {
  55                 scheme = "https"
  56         }
  57         err := h.saltAuthToken(req, remoteID)
  58         if err != nil {
  59                 httpserver.Error(w, err.Error(), http.StatusBadRequest)
  60                 return
  61         }
  62         urlOut := &url.URL{
  63                 Scheme:   scheme,
  64                 Host:     remote.Host,
  65                 Path:     req.URL.Path,
  66                 RawPath:  req.URL.RawPath,
  67                 RawQuery: req.URL.RawQuery,
  68         }
  69         client := h.secureClient
  70         if remote.Insecure {
  71                 client = h.insecureClient
  72         }
  73         h.proxy.Do(w, req, urlOut, client, filter)
  74 }
  75
  76 // loadParamsFromForm expects a request with
  77 // application/x-www-form-urlencoded body.  It parses the query, adds
  78 // the query parameters to "params", and replaces the request body
  79 // with a buffer holding the original body contents so it can be
  80 // re-read by downstream proxy steps.
  81 func loadParamsFromForm(req *http.Request, params url.Values) error {
  82         body, err := ioutil.ReadAll(req.Body)
  83         if err != nil {
  84                 return err
  85         }
  86         req.Body = ioutil.NopCloser(bytes.NewBuffer(body))
  87         var v2 url.Values
  88         if v2, err = url.ParseQuery(string(body)); err != nil {
  89                 return err
  90         }
  91         for k, v := range v2 {
  92                 params[k] = append(params[k], v...)
  93         }
  94         return nil
  95 }
  96
  97 // loadParamsFromForm expects a request with application/json body.
  98 // It parses the body, populates "loadInto", and replaces the request
  99 // body with a buffer holding the original body contents so it can be
 100 // re-read by downstream proxy steps.
 101 func loadParamsFromJson(req *http.Request, loadInto interface{}) error {
 102         var cl int64
 103         if req.ContentLength > 0 {
 104                 cl = req.ContentLength
 105         }
 106         postBody := bytes.NewBuffer(make([]byte, 0, cl))
 107         defer req.Body.Close()
 108
 109         rdr := io.TeeReader(req.Body, postBody)
 110
 111         err := json.NewDecoder(rdr).Decode(loadInto)
 112         if err != nil {
 113                 return err
 114         }
 115         req.Body = ioutil.NopCloser(postBody)
 116         return nil
 117 }
 118
 119 type multiClusterQueryResponseCollector struct {
 120         responses []interface{}
 121         error     error
 122         kind      string
 123         clusterID string
 124 }
 125
 126 func (c *multiClusterQueryResponseCollector) collectResponse(resp *http.Response,
 127         requestError error) (newResponse *http.Response, err error) {
 128         if requestError != nil {
 129                 c.error = requestError
 130                 return nil, nil
 131         }
 132
 133         defer resp.Body.Close()
 134         loadInto := make(map[string]interface{})
 135         err = json.NewDecoder(resp.Body).Decode(&loadInto)
 136
 137         if err == nil {
 138                 if resp.StatusCode != http.StatusOK {
 139                         c.error = fmt.Errorf("error fetching from %v (%v): %v", c.clusterID, resp.Status, loadInto["errors"])
 140                 } else {
 141                         c.responses = loadInto["items"].([]interface{})
 142                         c.kind, _ = loadInto["kind"].(string)
 143                 }
 144         } else {
 145                 c.error = fmt.Errorf("error fetching from %v (%v): %v", c.clusterID, resp.Status, err)
 146         }
 147
 148         return nil, nil
 149 }
 150
 151 func (h *genericFederatedRequestHandler) remoteQueryUUIDs(w http.ResponseWriter,
 152         req *http.Request, params url.Values,
 153         clusterID string, uuids []string) (rp []interface{}, kind string, err error) {
 154
 155         found := make(map[string]bool)
 156         for len(uuids) > 0 {
 157                 var remoteReq http.Request
 158                 remoteReq.Header = req.Header
 159                 remoteReq.Method = "POST"
 160                 remoteReq.URL = &url.URL{Path: req.URL.Path}
 161                 remoteParams := make(url.Values)
 162                 remoteParams["_method"] = []string{"GET"}
 163                 remoteParams["count"] = []string{"none"}
 164                 if len(params["select"]) != 0 {
 165                         remoteParams["select"] = params["select"]
 166                 }
 167                 content, err := json.Marshal(uuids)
 168                 if err != nil {
 169                         return nil, "", err
 170                 }
 171                 remoteParams["filters"] = []string{fmt.Sprintf(`[["uuid", "in", %s]]`, content)}
 172                 enc := remoteParams.Encode()
 173                 remoteReq.Body = ioutil.NopCloser(bytes.NewBufferString(enc))
 174
 175                 rc := multiClusterQueryResponseCollector{clusterID: clusterID}
 176
 177                 if clusterID == h.handler.Cluster.ClusterID {
 178                         h.handler.localClusterRequest(w, &remoteReq,
 179                                 rc.collectResponse)
 180                 } else {
 181                         h.handler.remoteClusterRequest(clusterID, w, &remoteReq,
 182                                 rc.collectResponse)
 183                 }
 184                 if rc.error != nil {
 185                         return nil, "", rc.error
 186                 }
 187
 188                 kind = rc.kind
 189
 190                 if len(rc.responses) == 0 {
 191                         // We got zero responses, no point in doing
 192                         // another query.
 193                         return rp, kind, nil
 194                 }
 195
 196                 rp = append(rp, rc.responses...)
 197
 198                 // Go through the responses and determine what was
 199                 // returned.  If there are remaining items, loop
 200                 // around and do another request with just the
 201                 // stragglers.
 202                 for _, i := range rc.responses {
 203                         m, ok := i.(map[string]interface{})
 204                         if ok {
 205                                 uuid, ok := m["uuid"].(string)
 206                                 if ok {
 207                                         found[uuid] = true
 208                                 }
 209                         }
 210                 }
 211
 212                 l := []string{}
 213                 for _, u := range uuids {
 214                         if !found[u] {
 215                                 l = append(l, u)
 216                         }
 217                 }
 218                 uuids = l
 219         }
 220
 221         return rp, kind, nil
 222 }
 223
 224 func (h *genericFederatedRequestHandler) handleMultiClusterQuery(w http.ResponseWriter, req *http.Request,
 225         params url.Values, clusterId *string) bool {
 226
 227         var filters [][]interface{}
 228         err := json.Unmarshal([]byte(params["filters"][0]), &filters)
 229         if err != nil {
 230                 httpserver.Error(w, err.Error(), http.StatusBadRequest)
 231                 return true
 232         }
 233
 234         // Split the list of uuids by prefix
 235         queryClusters := make(map[string][]string)
 236         expectCount := 0
 237         for _, f1 := range filters {
 238                 if len(f1) != 3 {
 239                         return false
 240                 }
 241                 lhs, ok := f1[0].(string)
 242                 if ok && lhs == "uuid" {
 243                         op, ok := f1[1].(string)
 244                         if !ok {
 245                                 return false
 246                         }
 247
 248                         if op == "in" {
 249                                 rhs, ok := f1[2].([]interface{})
 250                                 if ok {
 251                                         for _, i := range rhs {
 252                                                 u := i.(string)
 253                                                 *clusterId = u[0:5]
 254                                                 queryClusters[u[0:5]] = append(queryClusters[u[0:5]], u)
 255                                         }
 256                                         expectCount += len(rhs)
 257                                 }
 258                         } else if op == "=" {
 259                                 u, ok := f1[2].(string)
 260                                 if ok {
 261                                         *clusterId = u[0:5]
 262                                         queryClusters[u[0:5]] = append(queryClusters[u[0:5]], u)
 263                                         expectCount += 1
 264                                 }
 265                         } else {
 266                                 return false
 267                         }
 268                 } else {
 269                         return false
 270                 }
 271         }
 272
 273         if len(queryClusters) <= 1 {
 274                 // Query does not search for uuids across multiple
 275                 // clusters.
 276                 return false
 277         }
 278
 279         // Validations
 280         if !(len(params["count"]) == 1 && (params["count"][0] == `none` ||
 281                 params["count"][0] == `"none"`)) {
 282                 httpserver.Error(w, "Federated multi-object query must have 'count=none'", http.StatusBadRequest)
 283                 return true
 284         }
 285         if len(params["limit"]) != 0 || len(params["offset"]) != 0 || len(params["order"]) != 0 {
 286                 httpserver.Error(w, "Federated multi-object may not provide 'limit', 'offset' or 'order'.", http.StatusBadRequest)
 287                 return true
 288         }
 289         if expectCount > h.handler.Cluster.MaxItemsPerResponse {
 290                 httpserver.Error(w, fmt.Sprintf("Federated multi-object request for %v objects which is more than max page size %v.",
 291                         expectCount, h.handler.Cluster.MaxItemsPerResponse), http.StatusBadRequest)
 292                 return true
 293         }
 294         if len(params["select"]) == 1 {
 295                 foundUUID := false
 296                 var selects []interface{}
 297                 err := json.Unmarshal([]byte(params["select"][0]), &selects)
 298                 if err != nil {
 299                         httpserver.Error(w, err.Error(), http.StatusBadRequest)
 300                         return true
 301                 }
 302
 303                 for _, r := range selects {
 304                         if r.(string) == "uuid" {
 305                                 foundUUID = true
 306                                 break
 307                         }
 308                 }
 309                 if !foundUUID {
 310                         httpserver.Error(w, "Federated multi-object request must include 'uuid' in 'select'", http.StatusBadRequest)
 311                         return true
 312                 }
 313         }
 314
 315         // Perform parallel requests to each cluster
 316
 317         // use channel as a semaphore to limit the number of parallel
 318         // requests at a time
 319         sem := make(chan bool, h.handler.Cluster.ParallelRemoteRequests)
 320         defer close(sem)
 321         wg := sync.WaitGroup{}
 322
 323         req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
 324         mtx := sync.Mutex{}
 325         errors := []error{}
 326         var completeResponses []interface{}
 327         var kind string
 328
 329         for k, v := range queryClusters {
 330                 if len(v) == 0 {
 331                         // Nothing to query
 332                         continue
 333                 }
 334
 335                 // blocks until it can put a value into the
 336                 // channel (which has a max queue capacity)
 337                 sem <- true
 338                 wg.Add(1)
 339                 go func(k string, v []string) {
 340                         rp, kn, err := h.remoteQueryUUIDs(w, req, params, k, v)
 341                         mtx.Lock()
 342                         if err == nil {
 343                                 completeResponses = append(completeResponses, rp...)
 344                                 kind = kn
 345                         } else {
 346                                 errors = append(errors, err)
 347                         }
 348                         mtx.Unlock()
 349                         wg.Done()
 350                         <-sem
 351                 }(k, v)
 352         }
 353         wg.Wait()
 354
 355         if len(errors) > 0 {
 356                 var strerr []string
 357                 for _, e := range errors {
 358                         strerr = append(strerr, e.Error())
 359                 }
 360                 httpserver.Errors(w, strerr, http.StatusBadGateway)
 361                 return true
 362         }
 363
 364         w.Header().Set("Content-Type", "application/json")
 365         w.WriteHeader(http.StatusOK)
 366         itemList := make(map[string]interface{})
 367         itemList["items"] = completeResponses
 368         itemList["kind"] = kind
 369         json.NewEncoder(w).Encode(itemList)
 370
 371         return true
 372 }
 373
 374 func (h *genericFederatedRequestHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 375         m := h.matcher.FindStringSubmatch(req.URL.Path)
 376         clusterId := ""
 377
 378         if len(m) > 0 && m[2] != "" {
 379                 clusterId = m[2]
 380         }
 381
 382         // First, parse the query portion of the URL.
 383         var params url.Values
 384         var err error
 385         if params, err = url.ParseQuery(req.URL.RawQuery); err != nil {
 386                 httpserver.Error(w, err.Error(), http.StatusBadRequest)
 387                 return
 388         }
 389
 390         // Next, if appropriate, merge in parameters from the form POST body.
 391         if req.Method == "POST" && req.Header.Get("Content-Type") == "application/x-www-form-urlencoded" {
 392                 if err = loadParamsFromForm(req, params); err != nil {
 393                         httpserver.Error(w, err.Error(), http.StatusBadRequest)
 394                         return
 395                 }
 396         }
 397
 398         // Check if the parameters have an explicit cluster_id
 399         if len(params["cluster_id"]) == 1 {
 400                 clusterId = params["cluster_id"][0]
 401         }
 402
 403         // Handle the POST-as-GET special case (workaround for large
 404         // GET requests that potentially exceed maximum URL length,
 405         // like multi-object queries where the filter has 100s of
 406         // items)
 407         effectiveMethod := req.Method
 408         if req.Method == "POST" && len(params["_method"]) == 1 {
 409                 effectiveMethod = params["_method"][0]
 410         }
 411
 412         if effectiveMethod == "GET" && clusterId == "" && len(params["filters"]) == 1 {
 413                 if h.handleMultiClusterQuery(w, req, params, &clusterId) {
 414                         return
 415                 }
 416         }
 417
 418         if clusterId == "" || clusterId == h.handler.Cluster.ClusterID {
 419                 h.next.ServeHTTP(w, req)
 420         } else {
 421                 h.handler.remoteClusterRequest(clusterId, w, req, nil)
 422         }
 423 }
 424
 425 type rewriteSignaturesClusterId struct {
 426         clusterID  string
 427         expectHash string
 428 }
 429
 430 func (rw rewriteSignaturesClusterId) rewriteSignatures(resp *http.Response, requestError error) (newResponse *http.Response, err error) {
 431         if requestError != nil {
 432                 return resp, requestError
 433         }
 434
 435         if resp.StatusCode != 200 {
 436                 return resp, nil
 437         }
 438
 439         originalBody := resp.Body
 440         defer originalBody.Close()
 441
 442         var col arvados.Collection
 443         err = json.NewDecoder(resp.Body).Decode(&col)
 444         if err != nil {
 445                 return nil, err
 446         }
 447
 448         // rewriting signatures will make manifest text 5-10% bigger so calculate
 449         // capacity accordingly
 450         updatedManifest := bytes.NewBuffer(make([]byte, 0, int(float64(len(col.ManifestText))*1.1)))
 451
 452         hasher := md5.New()
 453         mw := io.MultiWriter(hasher, updatedManifest)
 454         sz := 0
 455
 456         scanner := bufio.NewScanner(strings.NewReader(col.ManifestText))
 457         scanner.Buffer(make([]byte, 1048576), len(col.ManifestText))
 458         for scanner.Scan() {
 459                 line := scanner.Text()
 460                 tokens := strings.Split(line, " ")
 461                 if len(tokens) < 3 {
 462                         return nil, fmt.Errorf("Invalid stream (<3 tokens): %q", line)
 463                 }
 464
 465                 n, err := mw.Write([]byte(tokens[0]))
 466                 if err != nil {
 467                         return nil, fmt.Errorf("Error updating manifest: %v", err)
 468                 }
 469                 sz += n
 470                 for _, token := range tokens[1:] {
 471                         n, err = mw.Write([]byte(" "))
 472                         if err != nil {
 473                                 return nil, fmt.Errorf("Error updating manifest: %v", err)
 474                         }
 475                         sz += n
 476
 477                         m := keepclient.SignedLocatorRe.FindStringSubmatch(token)
 478                         if m != nil {
 479                                 // Rewrite the block signature to be a remote signature
 480                                 _, err = fmt.Fprintf(updatedManifest, "%s%s%s+R%s-%s%s", m[1], m[2], m[3], rw.clusterID, m[5][2:], m[8])
 481                                 if err != nil {
 482                                         return nil, fmt.Errorf("Error updating manifest: %v", err)
 483                                 }
 484
 485                                 // for hash checking, ignore signatures
 486                                 n, err = fmt.Fprintf(hasher, "%s%s", m[1], m[2])
 487                                 if err != nil {
 488                                         return nil, fmt.Errorf("Error updating manifest: %v", err)
 489                                 }
 490                                 sz += n
 491                         } else {
 492                                 n, err = mw.Write([]byte(token))
 493                                 if err != nil {
 494                                         return nil, fmt.Errorf("Error updating manifest: %v", err)
 495                                 }
 496                                 sz += n
 497                         }
 498                 }
 499                 n, err = mw.Write([]byte("\n"))
 500                 if err != nil {
 501                         return nil, fmt.Errorf("Error updating manifest: %v", err)
 502                 }
 503                 sz += n
 504         }
 505
 506         // Check that expected hash is consistent with
 507         // portable_data_hash field of the returned record
 508         if rw.expectHash == "" {
 509                 rw.expectHash = col.PortableDataHash
 510         } else if rw.expectHash != col.PortableDataHash {
 511                 return nil, fmt.Errorf("portable_data_hash %q on returned record did not match expected hash %q ", rw.expectHash, col.PortableDataHash)
 512         }
 513
 514         // Certify that the computed hash of the manifest_text matches our expectation
 515         sum := hasher.Sum(nil)
 516         computedHash := fmt.Sprintf("%x+%v", sum, sz)
 517         if computedHash != rw.expectHash {
 518                 return nil, fmt.Errorf("Computed manifest_text hash %q did not match expected hash %q", computedHash, rw.expectHash)
 519         }
 520
 521         col.ManifestText = updatedManifest.String()
 522
 523         newbody, err := json.Marshal(col)
 524         if err != nil {
 525                 return nil, err
 526         }
 527
 528         buf := bytes.NewBuffer(newbody)
 529         resp.Body = ioutil.NopCloser(buf)
 530         resp.ContentLength = int64(buf.Len())
 531         resp.Header.Set("Content-Length", fmt.Sprintf("%v", buf.Len()))
 532
 533         return resp, nil
 534 }
 535
 536 func filterLocalClusterResponse(resp *http.Response, requestError error) (newResponse *http.Response, err error) {
 537         if requestError != nil {
 538                 return resp, requestError
 539         }
 540
 541         if resp.StatusCode == 404 {
 542                 // Suppress returning this result, because we want to
 543                 // search the federation.
 544                 return nil, nil
 545         }
 546         return resp, nil
 547 }
 548
 549 type searchRemoteClusterForPDH struct {
 550         pdh           string
 551         remoteID      string
 552         mtx           *sync.Mutex
 553         sentResponse  *bool
 554         sharedContext *context.Context
 555         cancelFunc    func()
 556         errors        *[]string
 557         statusCode    *int
 558 }
 559
 560 func (s *searchRemoteClusterForPDH) filterRemoteClusterResponse(resp *http.Response, requestError error) (newResponse *http.Response, err error) {
 561         s.mtx.Lock()
 562         defer s.mtx.Unlock()
 563
 564         if *s.sentResponse {
 565                 // Another request already returned a response
 566                 return nil, nil
 567         }
 568
 569         if requestError != nil {
 570                 *s.errors = append(*s.errors, fmt.Sprintf("Request error contacting %q: %v", s.remoteID, requestError))
 571                 // Record the error and suppress response
 572                 return nil, nil
 573         }
 574
 575         if resp.StatusCode != 200 {
 576                 // Suppress returning unsuccessful result.  Maybe
 577                 // another request will find it.
 578                 // TODO collect and return error responses.
 579                 *s.errors = append(*s.errors, fmt.Sprintf("Response from %q: %v", s.remoteID, resp.Status))
 580                 if resp.StatusCode != 404 {
 581                         // Got a non-404 error response, convert into BadGateway
 582                         *s.statusCode = http.StatusBadGateway
 583                 }
 584                 return nil, nil
 585         }
 586
 587         s.mtx.Unlock()
 588
 589         // This reads the response body.  We don't want to hold the
 590         // lock while doing this because other remote requests could
 591         // also have made it to this point, and we don't want a
 592         // slow response holding the lock to block a faster response
 593         // that is waiting on the lock.
 594         newResponse, err = rewriteSignaturesClusterId{s.remoteID, s.pdh}.rewriteSignatures(resp, nil)
 595
 596         s.mtx.Lock()
 597
 598         if *s.sentResponse {
 599                 // Another request already returned a response
 600                 return nil, nil
 601         }
 602
 603         if err != nil {
 604                 // Suppress returning unsuccessful result.  Maybe
 605                 // another request will be successful.
 606                 *s.errors = append(*s.errors, fmt.Sprintf("Error parsing response from %q: %v", s.remoteID, err))
 607                 return nil, nil
 608         }
 609
 610         // We have a successful response.  Suppress/cancel all the
 611         // other requests/responses.
 612         *s.sentResponse = true
 613         s.cancelFunc()
 614
 615         return newResponse, nil
 616 }
 617
 618 func (h *collectionFederatedRequestHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 619         if req.Method != "GET" {
 620                 // Only handle GET requests right now
 621                 h.next.ServeHTTP(w, req)
 622                 return
 623         }
 624
 625         m := collectionByPDHRe.FindStringSubmatch(req.URL.Path)
 626         if len(m) != 2 {
 627                 // Not a collection PDH GET request
 628                 m = collectionRe.FindStringSubmatch(req.URL.Path)
 629                 clusterId := ""
 630
 631                 if len(m) > 0 {
 632                         clusterId = m[2]
 633                 }
 634
 635                 if clusterId != "" && clusterId != h.handler.Cluster.ClusterID {
 636                         // request for remote collection by uuid
 637                         h.handler.remoteClusterRequest(clusterId, w, req,
 638                                 rewriteSignaturesClusterId{clusterId, ""}.rewriteSignatures)
 639                         return
 640                 }
 641                 // not a collection UUID request, or it is a request
 642                 // for a local UUID, either way, continue down the
 643                 // handler stack.
 644                 h.next.ServeHTTP(w, req)
 645                 return
 646         }
 647
 648         // Request for collection by PDH.  Search the federation.
 649
 650         // First, query the local cluster.
 651         if h.handler.localClusterRequest(w, req, filterLocalClusterResponse) {
 652                 return
 653         }
 654
 655         sharedContext, cancelFunc := context.WithCancel(req.Context())
 656         defer cancelFunc()
 657         req = req.WithContext(sharedContext)
 658
 659         // Create a goroutine for each cluster in the
 660         // RemoteClusters map.  The first valid result gets
 661         // returned to the client.  When that happens, all
 662         // other outstanding requests are cancelled or
 663         // suppressed.
 664         sentResponse := false
 665         mtx := sync.Mutex{}
 666         wg := sync.WaitGroup{}
 667         var errors []string
 668         var errorCode int = 404
 669
 670         // use channel as a semaphore to limit the number of parallel
 671         // requests at a time
 672         sem := make(chan bool, h.handler.Cluster.ParallelRemoteRequests)
 673         defer close(sem)
 674         for remoteID := range h.handler.Cluster.RemoteClusters {
 675                 // blocks until it can put a value into the
 676                 // channel (which has a max queue capacity)
 677                 sem <- true
 678                 if sentResponse {
 679                         break
 680                 }
 681                 search := &searchRemoteClusterForPDH{m[1], remoteID, &mtx, &sentResponse,
 682                         &sharedContext, cancelFunc, &errors, &errorCode}
 683                 wg.Add(1)
 684                 go func() {
 685                         h.handler.remoteClusterRequest(search.remoteID, w, req, search.filterRemoteClusterResponse)
 686                         wg.Done()
 687                         <-sem
 688                 }()
 689         }
 690         wg.Wait()
 691
 692         if sentResponse {
 693                 return
 694         }
 695
 696         // No successful responses, so return the error
 697         httpserver.Errors(w, errors, errorCode)
 698 }
 699
 700 func (h *Handler) setupProxyRemoteCluster(next http.Handler) http.Handler {
 701         mux := http.NewServeMux()
 702         mux.Handle("/arvados/v1/workflows", &genericFederatedRequestHandler{next, h, wfRe})
 703         mux.Handle("/arvados/v1/workflows/", &genericFederatedRequestHandler{next, h, wfRe})
 704         mux.Handle("/arvados/v1/containers", &genericFederatedRequestHandler{next, h, containersRe})
 705         mux.Handle("/arvados/v1/containers/", &genericFederatedRequestHandler{next, h, containersRe})
 706         mux.Handle("/arvados/v1/container_requests", &genericFederatedRequestHandler{next, h, containerRequestsRe})
 707         mux.Handle("/arvados/v1/container_requests/", &genericFederatedRequestHandler{next, h, containerRequestsRe})
 708         mux.Handle("/arvados/v1/collections", next)
 709         mux.Handle("/arvados/v1/collections/", &collectionFederatedRequestHandler{next, h})
 710         mux.Handle("/", next)
 711
 712         return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
 713                 parts := strings.Split(req.Header.Get("Authorization"), "/")
 714                 alreadySalted := (len(parts) == 3 && parts[0] == "Bearer v2" && len(parts[2]) == 40)
 715
 716                 if alreadySalted ||
 717                         strings.Index(req.Header.Get("Via"), "arvados-controller") != -1 {
 718                         // The token is already salted, or this is a
 719                         // request from another instance of
 720                         // arvados-controller.  In either case, we
 721                         // don't want to proxy this query, so just
 722                         // continue down the instance handler stack.
 723                         next.ServeHTTP(w, req)
 724                         return
 725                 }
 726
 727                 mux.ServeHTTP(w, req)
 728         })
 729
 730         return mux
 731 }
 732
 733 type CurrentUser struct {
 734         Authorization arvados.APIClientAuthorization
 735         UUID          string
 736 }
 737
 738 func (h *Handler) validateAPItoken(req *http.Request, user *CurrentUser) error {
 739         db, err := h.db(req)
 740         if err != nil {
 741                 return err
 742         }
 743         return db.QueryRowContext(req.Context(), `SELECT api_client_authorizations.uuid, users.uuid FROM api_client_authorizations JOIN users on api_client_authorizations.user_id=users.id WHERE api_token=$1 AND (expires_at IS NULL OR expires_at > current_timestamp) LIMIT 1`, user.Authorization.APIToken).Scan(&user.Authorization.UUID, &user.UUID)
 744 }
 745
 746 // Extract the auth token supplied in req, and replace it with a
 747 // salted token for the remote cluster.
 748 func (h *Handler) saltAuthToken(req *http.Request, remote string) error {
 749         creds := auth.NewCredentials()
 750         creds.LoadTokensFromHTTPRequest(req)
 751         if len(creds.Tokens) == 0 && req.Header.Get("Content-Type") == "application/x-www-form-encoded" {
 752                 // Override ParseForm's 10MiB limit by ensuring
 753                 // req.Body is a *http.maxBytesReader.
 754                 req.Body = http.MaxBytesReader(nil, req.Body, 1<<28) // 256MiB. TODO: use MaxRequestSize from discovery doc or config.
 755                 if err := creds.LoadTokensFromHTTPRequestBody(req); err != nil {
 756                         return err
 757                 }
 758                 // Replace req.Body with a buffer that re-encodes the
 759                 // form without api_token, in case we end up
 760                 // forwarding the request.
 761                 if req.PostForm != nil {
 762                         req.PostForm.Del("api_token")
 763                 }
 764                 req.Body = ioutil.NopCloser(bytes.NewBufferString(req.PostForm.Encode()))
 765         }
 766         if len(creds.Tokens) == 0 {
 767                 return nil
 768         }
 769         token, err := auth.SaltToken(creds.Tokens[0], remote)
 770         if err == auth.ErrObsoleteToken {
 771                 // If the token exists in our own database, salt it
 772                 // for the remote. Otherwise, assume it was issued by
 773                 // the remote, and pass it through unmodified.
 774                 currentUser := CurrentUser{Authorization: arvados.APIClientAuthorization{APIToken: creds.Tokens[0]}}
 775                 err = h.validateAPItoken(req, &currentUser)
 776                 if err == sql.ErrNoRows {
 777                         // Not ours; pass through unmodified.
 778                         token = currentUser.Authorization.APIToken
 779                 } else if err != nil {
 780                         return err
 781                 } else {
 782                         // Found; make V2 version and salt it.
 783                         token, err = auth.SaltToken(currentUser.Authorization.TokenV2(), remote)
 784                         if err != nil {
 785                                 return err
 786                         }
 787                 }
 788         } else if err != nil {
 789                 return err
 790         }
 791         req.Header.Set("Authorization", "Bearer "+token)
 792
 793         // Remove api_token=... from the the query string, in case we
 794         // end up forwarding the request.
 795         if values, err := url.ParseQuery(req.URL.RawQuery); err != nil {
 796                 return err
 797         } else if _, ok := values["api_token"]; ok {
 798                 delete(values, "api_token")
 799                 req.URL.RawQuery = values.Encode()
 800         }
 801         return nil
 802 }