lib/controller/fed_collections.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         "encoding/json"
  13         "fmt"
  14         "io"
  15         "io/ioutil"
  16         "net/http"
  17         "strings"
  18         "sync"
  19
  20         "git.curoverse.com/arvados.git/sdk/go/arvados"
  21         "git.curoverse.com/arvados.git/sdk/go/httpserver"
  22         "git.curoverse.com/arvados.git/sdk/go/keepclient"
  23 )
  24
  25 type collectionFederatedRequestHandler struct {
  26         next    http.Handler
  27         handler *Handler
  28 }
  29
  30 func rewriteSignatures(clusterID string, expectHash string,
  31         resp *http.Response, requestError error) (newResponse *http.Response, err error) {
  32
  33         if requestError != nil {
  34                 return resp, requestError
  35         }
  36
  37         if resp.StatusCode != http.StatusOK {
  38                 return resp, nil
  39         }
  40
  41         originalBody := resp.Body
  42         defer originalBody.Close()
  43
  44         var col arvados.Collection
  45         err = json.NewDecoder(resp.Body).Decode(&col)
  46         if err != nil {
  47                 return nil, err
  48         }
  49
  50         // rewriting signatures will make manifest text 5-10% bigger so calculate
  51         // capacity accordingly
  52         updatedManifest := bytes.NewBuffer(make([]byte, 0, int(float64(len(col.ManifestText))*1.1)))
  53
  54         hasher := md5.New()
  55         mw := io.MultiWriter(hasher, updatedManifest)
  56         sz := 0
  57
  58         scanner := bufio.NewScanner(strings.NewReader(col.ManifestText))
  59         scanner.Buffer(make([]byte, 1048576), len(col.ManifestText))
  60         for scanner.Scan() {
  61                 line := scanner.Text()
  62                 tokens := strings.Split(line, " ")
  63                 if len(tokens) < 3 {
  64                         return nil, fmt.Errorf("Invalid stream (<3 tokens): %q", line)
  65                 }
  66
  67                 n, err := mw.Write([]byte(tokens[0]))
  68                 if err != nil {
  69                         return nil, fmt.Errorf("Error updating manifest: %v", err)
  70                 }
  71                 sz += n
  72                 for _, token := range tokens[1:] {
  73                         n, err = mw.Write([]byte(" "))
  74                         if err != nil {
  75                                 return nil, fmt.Errorf("Error updating manifest: %v", err)
  76                         }
  77                         sz += n
  78
  79                         m := keepclient.SignedLocatorRe.FindStringSubmatch(token)
  80                         if m != nil {
  81                                 // Rewrite the block signature to be a remote signature
  82                                 _, err = fmt.Fprintf(updatedManifest, "%s%s%s+R%s-%s%s", m[1], m[2], m[3], clusterID, m[5][2:], m[8])
  83                                 if err != nil {
  84                                         return nil, fmt.Errorf("Error updating manifest: %v", err)
  85                                 }
  86
  87                                 // for hash checking, ignore signatures
  88                                 n, err = fmt.Fprintf(hasher, "%s%s", m[1], m[2])
  89                                 if err != nil {
  90                                         return nil, fmt.Errorf("Error updating manifest: %v", err)
  91                                 }
  92                                 sz += n
  93                         } else {
  94                                 n, err = mw.Write([]byte(token))
  95                                 if err != nil {
  96                                         return nil, fmt.Errorf("Error updating manifest: %v", err)
  97                                 }
  98                                 sz += n
  99                         }
 100                 }
 101                 n, err = mw.Write([]byte("\n"))
 102                 if err != nil {
 103                         return nil, fmt.Errorf("Error updating manifest: %v", err)
 104                 }
 105                 sz += n
 106         }
 107
 108         // Check that expected hash is consistent with
 109         // portable_data_hash field of the returned record
 110         if expectHash == "" {
 111                 expectHash = col.PortableDataHash
 112         } else if expectHash != col.PortableDataHash {
 113                 return nil, fmt.Errorf("portable_data_hash %q on returned record did not match expected hash %q ", expectHash, col.PortableDataHash)
 114         }
 115
 116         // Certify that the computed hash of the manifest_text matches our expectation
 117         sum := hasher.Sum(nil)
 118         computedHash := fmt.Sprintf("%x+%v", sum, sz)
 119         if computedHash != expectHash {
 120                 return nil, fmt.Errorf("Computed manifest_text hash %q did not match expected hash %q", computedHash, expectHash)
 121         }
 122
 123         col.ManifestText = updatedManifest.String()
 124
 125         newbody, err := json.Marshal(col)
 126         if err != nil {
 127                 return nil, err
 128         }
 129
 130         buf := bytes.NewBuffer(newbody)
 131         resp.Body = ioutil.NopCloser(buf)
 132         resp.ContentLength = int64(buf.Len())
 133         resp.Header.Set("Content-Length", fmt.Sprintf("%v", buf.Len()))
 134
 135         return resp, nil
 136 }
 137
 138 func filterLocalClusterResponse(resp *http.Response, requestError error) (newResponse *http.Response, err error) {
 139         if requestError != nil {
 140                 return resp, requestError
 141         }
 142
 143         if resp.StatusCode == http.StatusNotFound {
 144                 // Suppress returning this result, because we want to
 145                 // search the federation.
 146                 return nil, nil
 147         }
 148         return resp, nil
 149 }
 150
 151 type searchRemoteClusterForPDH struct {
 152         pdh           string
 153         remoteID      string
 154         mtx           *sync.Mutex
 155         sentResponse  *bool
 156         sharedContext *context.Context
 157         cancelFunc    func()
 158         errors        *[]string
 159         statusCode    *int
 160 }
 161
 162 func (h *collectionFederatedRequestHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 163         if req.Method != "GET" {
 164                 // Only handle GET requests right now
 165                 h.next.ServeHTTP(w, req)
 166                 return
 167         }
 168
 169         m := collectionByPDHRe.FindStringSubmatch(req.URL.Path)
 170         if len(m) != 2 {
 171                 // Not a collection PDH GET request
 172                 m = collectionRe.FindStringSubmatch(req.URL.Path)
 173                 clusterId := ""
 174
 175                 if len(m) > 0 {
 176                         clusterId = m[2]
 177                 }
 178
 179                 if clusterId != "" && clusterId != h.handler.Cluster.ClusterID {
 180                         // request for remote collection by uuid
 181                         resp, cancel, err := h.handler.remoteClusterRequest(clusterId, req)
 182                         if cancel != nil {
 183                                 defer cancel()
 184                         }
 185                         newResponse, err := rewriteSignatures(clusterId, "", resp, err)
 186                         h.handler.proxy.ForwardResponse(w, newResponse, err)
 187                         return
 188                 }
 189                 // not a collection UUID request, or it is a request
 190                 // for a local UUID, either way, continue down the
 191                 // handler stack.
 192                 h.next.ServeHTTP(w, req)
 193                 return
 194         }
 195
 196         // Request for collection by PDH.  Search the federation.
 197
 198         // First, query the local cluster.
 199         resp, localClusterRequestCancel, err := h.handler.localClusterRequest(req)
 200         if localClusterRequestCancel != nil {
 201                 defer localClusterRequestCancel()
 202         }
 203         newResp, err := filterLocalClusterResponse(resp, err)
 204         if newResp != nil || err != nil {
 205                 h.handler.proxy.ForwardResponse(w, newResp, err)
 206                 return
 207         }
 208
 209         // Create a goroutine for each cluster in the
 210         // RemoteClusters map.  The first valid result gets
 211         // returned to the client.  When that happens, all
 212         // other outstanding requests are cancelled
 213         sharedContext, cancelFunc := context.WithCancel(req.Context())
 214         req = req.WithContext(sharedContext)
 215         wg := sync.WaitGroup{}
 216         pdh := m[1]
 217         success := make(chan *http.Response)
 218         errorChan := make(chan error)
 219
 220         // use channel as a semaphore to limit the number of concurrent
 221         // requests at a time
 222         sem := make(chan bool, h.handler.Cluster.RequestLimits.GetMultiClusterRequestConcurrency())
 223
 224         defer close(errorChan)
 225         defer close(success)
 226         defer close(sem)
 227         defer cancelFunc()
 228
 229         for remoteID := range h.handler.Cluster.RemoteClusters {
 230                 if remoteID == h.handler.Cluster.ClusterID {
 231                         // No need to query local cluster again
 232                         continue
 233                 }
 234
 235                 wg.Add(1)
 236                 go func(remote string) {
 237                         defer wg.Done()
 238                         // blocks until it can put a value into the
 239                         // channel (which has a max queue capacity)
 240                         sem <- true
 241                         select {
 242                         case <-sharedContext.Done():
 243                                 return
 244                         default:
 245                         }
 246
 247                         resp, _, err := h.handler.remoteClusterRequest(remote, req)
 248                         wasSuccess := false
 249                         defer func() {
 250                                 if resp != nil && !wasSuccess {
 251                                         resp.Body.Close()
 252                                 }
 253                         }()
 254                         // Don't need to do anything with the cancel
 255                         // function returned by remoteClusterRequest
 256                         // because the context inherits from
 257                         // sharedContext, so when sharedContext is
 258                         // cancelled it should cancel that one as
 259                         // well.
 260                         if err != nil {
 261                                 errorChan <- err
 262                                 return
 263                         }
 264                         if resp.StatusCode != http.StatusOK {
 265                                 errorChan <- HTTPError{resp.Status, resp.StatusCode}
 266                                 return
 267                         }
 268                         select {
 269                         case <-sharedContext.Done():
 270                                 return
 271                         default:
 272                         }
 273
 274                         newResponse, err := rewriteSignatures(remote, pdh, resp, nil)
 275                         if err != nil {
 276                                 errorChan <- err
 277                                 return
 278                         }
 279                         select {
 280                         case <-sharedContext.Done():
 281                         case success <- newResponse:
 282                                 wasSuccess = true
 283                         }
 284                         <-sem
 285                 }(remoteID)
 286         }
 287         go func() {
 288                 wg.Wait()
 289                 cancelFunc()
 290         }()
 291
 292         var errors []string
 293         errorCode := http.StatusNotFound
 294
 295         for {
 296                 select {
 297                 case newResp = <-success:
 298                         h.handler.proxy.ForwardResponse(w, newResp, nil)
 299                         return
 300                 case err := <-errorChan:
 301                         if httperr, ok := err.(HTTPError); ok {
 302                                 if httperr.Code != http.StatusNotFound {
 303                                         errorCode = http.StatusBadGateway
 304                                 }
 305                         }
 306                         errors = append(errors, err.Error())
 307                 case <-sharedContext.Done():
 308                         httpserver.Errors(w, errors, errorCode)
 309                         return
 310                 }
 311         }
 312 }