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)
 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 close(errorChan)
 219         defer close(success)
 220         defer close(sem)
 221         defer cancelFunc()
 222
 223         for remoteID := range h.handler.Cluster.RemoteClusters {
 224                 if remoteID == h.handler.Cluster.ClusterID {
 225                         // No need to query local cluster again
 226                         continue
 227                 }
 228
 229                 wg.Add(1)
 230                 go func(remote string) {
 231                         defer wg.Done()
 232                         // blocks until it can put a value into the
 233                         // channel (which has a max queue capacity)
 234                         sem <- true
 235                         select {
 236                         case <-sharedContext.Done():
 237                                 return
 238                         default:
 239                         }
 240
 241                         resp, err := h.handler.remoteClusterRequest(remote, req)
 242                         wasSuccess := false
 243                         defer func() {
 244                                 if resp != nil && !wasSuccess {
 245                                         resp.Body.Close()
 246                                 }
 247                         }()
 248                         if err != nil {
 249                                 errorChan <- err
 250                                 return
 251                         }
 252                         if resp.StatusCode != http.StatusOK {
 253                                 errorChan <- HTTPError{resp.Status, resp.StatusCode}
 254                                 return
 255                         }
 256                         select {
 257                         case <-sharedContext.Done():
 258                                 return
 259                         default:
 260                         }
 261
 262                         newResponse, err := rewriteSignatures(remote, pdh, resp, nil)
 263                         if err != nil {
 264                                 errorChan <- err
 265                                 return
 266                         }
 267                         select {
 268                         case <-sharedContext.Done():
 269                         case success <- newResponse:
 270                                 wasSuccess = true
 271                         }
 272                         <-sem
 273                 }(remoteID)
 274         }
 275         go func() {
 276                 wg.Wait()
 277                 cancelFunc()
 278         }()
 279
 280         var errors []string
 281         errorCode := http.StatusNotFound
 282
 283         for {
 284                 select {
 285                 case newResp = <-success:
 286                         h.handler.proxy.ForwardResponse(w, newResp, nil)
 287                         return
 288                 case err := <-errorChan:
 289                         if httperr, ok := err.(HTTPError); ok {
 290                                 if httperr.Code != http.StatusNotFound {
 291                                         errorCode = http.StatusBadGateway
 292                                 }
 293                         }
 294                         errors = append(errors, err.Error())
 295                 case <-sharedContext.Done():
 296                         httpserver.Errors(w, errors, errorCode)
 297                         return
 298                 }
 299         }
 300 }