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, err := h.handler.remoteClusterRequest(clusterId, req)
 182                         newResponse, err := rewriteSignatures(clusterId, "", resp, err)
 183                         h.handler.proxy.ForwardResponse(w, newResponse, err)
 184                         return
 185                 }
 186                 // not a collection UUID request, or it is a request
 187                 // for a local UUID, either way, continue down the
 188                 // handler stack.
 189                 h.next.ServeHTTP(w, req)
 190                 return
 191         }
 192
 193         // Request for collection by PDH.  Search the federation.
 194
 195         // First, query the local cluster.
 196         resp, err := h.handler.localClusterRequest(req)
 197         newResp, err := filterLocalClusterResponse(resp, err)
 198         if newResp != nil || err != nil {
 199                 h.handler.proxy.ForwardResponse(w, newResp, err)
 200                 return
 201         }
 202
 203         // Create a goroutine for each cluster in the
 204         // RemoteClusters map.  The first valid result gets
 205         // returned to the client.  When that happens, all
 206         // other outstanding requests are cancelled
 207         sharedContext, cancelFunc := context.WithCancel(req.Context())
 208         req = req.WithContext(sharedContext)
 209         wg := sync.WaitGroup{}
 210         pdh := m[1]
 211         success := make(chan *http.Response)
 212         errorChan := make(chan error, len(h.handler.Cluster.RemoteClusters))
 213
 214         // use channel as a semaphore to limit the number of concurrent
 215         // requests at a time
 216         sem := make(chan bool, h.handler.Cluster.RequestLimits.GetMultiClusterRequestConcurrency())
 217
 218         defer cancelFunc()
 219
 220         for remoteID := range h.handler.Cluster.RemoteClusters {
 221                 if remoteID == h.handler.Cluster.ClusterID {
 222                         // No need to query local cluster again
 223                         continue
 224                 }
 225
 226                 wg.Add(1)
 227                 go func(remote string) {
 228                         defer wg.Done()
 229                         // blocks until it can put a value into the
 230                         // channel (which has a max queue capacity)
 231                         sem <- true
 232                         select {
 233                         case <-sharedContext.Done():
 234                                 return
 235                         default:
 236                         }
 237
 238                         resp, err := h.handler.remoteClusterRequest(remote, req)
 239                         wasSuccess := false
 240                         defer func() {
 241                                 if resp != nil && !wasSuccess {
 242                                         resp.Body.Close()
 243                                 }
 244                         }()
 245                         if err != nil {
 246                                 errorChan <- err
 247                                 return
 248                         }
 249                         if resp.StatusCode != http.StatusOK {
 250                                 errorChan <- HTTPError{resp.Status, resp.StatusCode}
 251                                 return
 252                         }
 253                         select {
 254                         case <-sharedContext.Done():
 255                                 return
 256                         default:
 257                         }
 258
 259                         newResponse, err := rewriteSignatures(remote, pdh, resp, nil)
 260                         if err != nil {
 261                                 errorChan <- err
 262                                 return
 263                         }
 264                         select {
 265                         case <-sharedContext.Done():
 266                         case success <- newResponse:
 267                                 wasSuccess = true
 268                         }
 269                         <-sem
 270                 }(remoteID)
 271         }
 272         go func() {
 273                 wg.Wait()
 274                 cancelFunc()
 275         }()
 276
 277         errorCode := http.StatusNotFound
 278
 279         for {
 280                 select {
 281                 case newResp = <-success:
 282                         h.handler.proxy.ForwardResponse(w, newResp, nil)
 283                         return
 284                 case <-sharedContext.Done():
 285                         var errors []string
 286                         for len(errorChan) > 0 {
 287                                 err := <-errorChan
 288                                 if httperr, ok := err.(HTTPError); ok {
 289                                         if httperr.Code != http.StatusNotFound {
 290                                                 errorCode = http.StatusBadGateway
 291                                         }
 292                                 }
 293                                 errors = append(errors, err.Error())
 294                         }
 295                         httpserver.Errors(w, errors, errorCode)
 296                         return
 297                 }
 298         }
 299 }