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