Merge branch 'thehyve/fix-crunch-documentation' Fix a typo in Crunch Dispatch install...
[arvados.git] / sdk / go / keepclient / keepclient.go
1 // Copyright (C) The Arvados Authors. All rights reserved.
2 //
3 // SPDX-License-Identifier: Apache-2.0
4
5 /* Provides low-level Get/Put primitives for accessing Arvados Keep blocks. */
6 package keepclient
7
8 import (
9         "bytes"
10         "crypto/md5"
11         "errors"
12         "fmt"
13         "io"
14         "io/ioutil"
15         "net"
16         "net/http"
17         "regexp"
18         "strconv"
19         "strings"
20         "sync"
21         "time"
22
23         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
24         "git.curoverse.com/arvados.git/sdk/go/asyncbuf"
25         "git.curoverse.com/arvados.git/sdk/go/httpserver"
26 )
27
28 // A Keep "block" is 64MB.
29 const BLOCKSIZE = 64 * 1024 * 1024
30
31 var (
32         DefaultRequestTimeout      = 20 * time.Second
33         DefaultConnectTimeout      = 2 * time.Second
34         DefaultTLSHandshakeTimeout = 4 * time.Second
35         DefaultKeepAlive           = 180 * time.Second
36
37         DefaultProxyRequestTimeout      = 300 * time.Second
38         DefaultProxyConnectTimeout      = 30 * time.Second
39         DefaultProxyTLSHandshakeTimeout = 10 * time.Second
40         DefaultProxyKeepAlive           = 120 * time.Second
41 )
42
43 // Error interface with an error and boolean indicating whether the error is temporary
44 type Error interface {
45         error
46         Temporary() bool
47 }
48
49 // multipleResponseError is of type Error
50 type multipleResponseError struct {
51         error
52         isTemp bool
53 }
54
55 func (e *multipleResponseError) Temporary() bool {
56         return e.isTemp
57 }
58
59 // BlockNotFound is a multipleResponseError where isTemp is false
60 var BlockNotFound = &ErrNotFound{multipleResponseError{
61         error:  errors.New("Block not found"),
62         isTemp: false,
63 }}
64
65 // ErrNotFound is a multipleResponseError where isTemp can be true or false
66 type ErrNotFound struct {
67         multipleResponseError
68 }
69
70 type InsufficientReplicasError error
71
72 type OversizeBlockError error
73
74 var ErrOversizeBlock = OversizeBlockError(errors.New("Exceeded maximum block size (" + strconv.Itoa(BLOCKSIZE) + ")"))
75 var MissingArvadosApiHost = errors.New("Missing required environment variable ARVADOS_API_HOST")
76 var MissingArvadosApiToken = errors.New("Missing required environment variable ARVADOS_API_TOKEN")
77 var InvalidLocatorError = errors.New("Invalid locator")
78
79 // ErrNoSuchKeepServer is returned when GetIndex is invoked with a UUID with no matching keep server
80 var ErrNoSuchKeepServer = errors.New("No keep server matching the given UUID is found")
81
82 // ErrIncompleteIndex is returned when the Index response does not end with a new empty line
83 var ErrIncompleteIndex = errors.New("Got incomplete index")
84
85 const X_Keep_Desired_Replicas = "X-Keep-Desired-Replicas"
86 const X_Keep_Replicas_Stored = "X-Keep-Replicas-Stored"
87
88 type HTTPClient interface {
89         Do(*http.Request) (*http.Response, error)
90 }
91
92 // Information about Arvados and Keep servers.
93 type KeepClient struct {
94         Arvados            *arvadosclient.ArvadosClient
95         Want_replicas      int
96         localRoots         map[string]string
97         writableLocalRoots map[string]string
98         gatewayRoots       map[string]string
99         lock               sync.RWMutex
100         HTTPClient         HTTPClient
101         Retries            int
102         BlockCache         *BlockCache
103         RequestID          string
104
105         // set to 1 if all writable services are of disk type, otherwise 0
106         replicasPerService int
107
108         // Any non-disk typed services found in the list of keepservers?
109         foundNonDiskSvc bool
110
111         // Disable automatic discovery of keep services
112         disableDiscovery bool
113 }
114
115 // MakeKeepClient creates a new KeepClient, calls
116 // DiscoverKeepServices(), and returns when the client is ready to
117 // use.
118 func MakeKeepClient(arv *arvadosclient.ArvadosClient) (*KeepClient, error) {
119         kc := New(arv)
120         return kc, kc.discoverServices()
121 }
122
123 // New creates a new KeepClient. Service discovery will occur on the
124 // next read/write operation.
125 func New(arv *arvadosclient.ArvadosClient) *KeepClient {
126         defaultReplicationLevel := 2
127         value, err := arv.Discovery("defaultCollectionReplication")
128         if err == nil {
129                 v, ok := value.(float64)
130                 if ok && v > 0 {
131                         defaultReplicationLevel = int(v)
132                 }
133         }
134         return &KeepClient{
135                 Arvados:       arv,
136                 Want_replicas: defaultReplicationLevel,
137                 Retries:       2,
138         }
139 }
140
141 // Put a block given the block hash, a reader, and the number of bytes
142 // to read from the reader (which must be between 0 and BLOCKSIZE).
143 //
144 // Returns the locator for the written block, the number of replicas
145 // written, and an error.
146 //
147 // Returns an InsufficientReplicasError if 0 <= replicas <
148 // kc.Wants_replicas.
149 func (kc *KeepClient) PutHR(hash string, r io.Reader, dataBytes int64) (string, int, error) {
150         // Buffer for reads from 'r'
151         var bufsize int
152         if dataBytes > 0 {
153                 if dataBytes > BLOCKSIZE {
154                         return "", 0, ErrOversizeBlock
155                 }
156                 bufsize = int(dataBytes)
157         } else {
158                 bufsize = BLOCKSIZE
159         }
160
161         buf := asyncbuf.NewBuffer(make([]byte, 0, bufsize))
162         go func() {
163                 _, err := io.Copy(buf, HashCheckingReader{r, md5.New(), hash})
164                 buf.CloseWithError(err)
165         }()
166         return kc.putReplicas(hash, buf.NewReader, dataBytes)
167 }
168
169 // PutHB writes a block to Keep. The hash of the bytes is given in
170 // hash, and the data is given in buf.
171 //
172 // Return values are the same as for PutHR.
173 func (kc *KeepClient) PutHB(hash string, buf []byte) (string, int, error) {
174         newReader := func() io.Reader { return bytes.NewBuffer(buf) }
175         return kc.putReplicas(hash, newReader, int64(len(buf)))
176 }
177
178 // PutB writes a block to Keep. It computes the hash itself.
179 //
180 // Return values are the same as for PutHR.
181 func (kc *KeepClient) PutB(buffer []byte) (string, int, error) {
182         hash := fmt.Sprintf("%x", md5.Sum(buffer))
183         return kc.PutHB(hash, buffer)
184 }
185
186 // PutR writes a block to Keep. It first reads all data from r into a buffer
187 // in order to compute the hash.
188 //
189 // Return values are the same as for PutHR.
190 //
191 // If the block hash and data size are known, PutHR is more efficient.
192 func (kc *KeepClient) PutR(r io.Reader) (locator string, replicas int, err error) {
193         if buffer, err := ioutil.ReadAll(r); err != nil {
194                 return "", 0, err
195         } else {
196                 return kc.PutB(buffer)
197         }
198 }
199
200 func (kc *KeepClient) getOrHead(method string, locator string) (io.ReadCloser, int64, string, error) {
201         if strings.HasPrefix(locator, "d41d8cd98f00b204e9800998ecf8427e+0") {
202                 return ioutil.NopCloser(bytes.NewReader(nil)), 0, "", nil
203         }
204
205         reqid := kc.getRequestID()
206
207         var expectLength int64
208         if parts := strings.SplitN(locator, "+", 3); len(parts) < 2 {
209                 expectLength = -1
210         } else if n, err := strconv.ParseInt(parts[1], 10, 64); err != nil {
211                 expectLength = -1
212         } else {
213                 expectLength = n
214         }
215
216         var errs []string
217
218         tries_remaining := 1 + kc.Retries
219
220         serversToTry := kc.getSortedRoots(locator)
221
222         numServers := len(serversToTry)
223         count404 := 0
224
225         var retryList []string
226
227         for tries_remaining > 0 {
228                 tries_remaining -= 1
229                 retryList = nil
230
231                 for _, host := range serversToTry {
232                         url := host + "/" + locator
233
234                         req, err := http.NewRequest(method, url, nil)
235                         if err != nil {
236                                 errs = append(errs, fmt.Sprintf("%s: %v", url, err))
237                                 continue
238                         }
239                         req.Header.Add("Authorization", "OAuth2 "+kc.Arvados.ApiToken)
240                         req.Header.Add("X-Request-Id", reqid)
241                         resp, err := kc.httpClient().Do(req)
242                         if err != nil {
243                                 // Probably a network error, may be transient,
244                                 // can try again.
245                                 errs = append(errs, fmt.Sprintf("%s: %v", url, err))
246                                 retryList = append(retryList, host)
247                                 continue
248                         }
249                         if resp.StatusCode != http.StatusOK {
250                                 var respbody []byte
251                                 respbody, _ = ioutil.ReadAll(&io.LimitedReader{R: resp.Body, N: 4096})
252                                 resp.Body.Close()
253                                 errs = append(errs, fmt.Sprintf("%s: HTTP %d %q",
254                                         url, resp.StatusCode, bytes.TrimSpace(respbody)))
255
256                                 if resp.StatusCode == 408 ||
257                                         resp.StatusCode == 429 ||
258                                         resp.StatusCode >= 500 {
259                                         // Timeout, too many requests, or other
260                                         // server side failure, transient
261                                         // error, can try again.
262                                         retryList = append(retryList, host)
263                                 } else if resp.StatusCode == 404 {
264                                         count404++
265                                 }
266                                 continue
267                         }
268                         if expectLength < 0 {
269                                 if resp.ContentLength < 0 {
270                                         resp.Body.Close()
271                                         return nil, 0, "", fmt.Errorf("error reading %q: no size hint, no Content-Length header in response", locator)
272                                 }
273                                 expectLength = resp.ContentLength
274                         } else if resp.ContentLength >= 0 && expectLength != resp.ContentLength {
275                                 resp.Body.Close()
276                                 return nil, 0, "", fmt.Errorf("error reading %q: size hint %d != Content-Length %d", locator, expectLength, resp.ContentLength)
277                         }
278                         // Success
279                         if method == "GET" {
280                                 return HashCheckingReader{
281                                         Reader: resp.Body,
282                                         Hash:   md5.New(),
283                                         Check:  locator[0:32],
284                                 }, expectLength, url, nil
285                         } else {
286                                 resp.Body.Close()
287                                 return nil, expectLength, url, nil
288                         }
289                 }
290                 serversToTry = retryList
291         }
292         DebugPrintf("DEBUG: %s %s failed: %v", method, locator, errs)
293
294         var err error
295         if count404 == numServers {
296                 err = BlockNotFound
297         } else {
298                 err = &ErrNotFound{multipleResponseError{
299                         error:  fmt.Errorf("%s %s failed: %v", method, locator, errs),
300                         isTemp: len(serversToTry) > 0,
301                 }}
302         }
303         return nil, 0, "", err
304 }
305
306 // Get() retrieves a block, given a locator. Returns a reader, the
307 // expected data length, the URL the block is being fetched from, and
308 // an error.
309 //
310 // If the block checksum does not match, the final Read() on the
311 // reader returned by this method will return a BadChecksum error
312 // instead of EOF.
313 func (kc *KeepClient) Get(locator string) (io.ReadCloser, int64, string, error) {
314         return kc.getOrHead("GET", locator)
315 }
316
317 // ReadAt() retrieves a portion of block from the cache if it's
318 // present, otherwise from the network.
319 func (kc *KeepClient) ReadAt(locator string, p []byte, off int) (int, error) {
320         return kc.cache().ReadAt(kc, locator, p, off)
321 }
322
323 // Ask() verifies that a block with the given hash is available and
324 // readable, according to at least one Keep service. Unlike Get, it
325 // does not retrieve the data or verify that the data content matches
326 // the hash specified by the locator.
327 //
328 // Returns the data size (content length) reported by the Keep service
329 // and the URI reporting the data size.
330 func (kc *KeepClient) Ask(locator string) (int64, string, error) {
331         _, size, url, err := kc.getOrHead("HEAD", locator)
332         return size, url, err
333 }
334
335 // GetIndex retrieves a list of blocks stored on the given server whose hashes
336 // begin with the given prefix. The returned reader will return an error (other
337 // than EOF) if the complete index cannot be retrieved.
338 //
339 // This is meant to be used only by system components and admin tools.
340 // It will return an error unless the client is using a "data manager token"
341 // recognized by the Keep services.
342 func (kc *KeepClient) GetIndex(keepServiceUUID, prefix string) (io.Reader, error) {
343         url := kc.LocalRoots()[keepServiceUUID]
344         if url == "" {
345                 return nil, ErrNoSuchKeepServer
346         }
347
348         url += "/index"
349         if prefix != "" {
350                 url += "/" + prefix
351         }
352
353         req, err := http.NewRequest("GET", url, nil)
354         if err != nil {
355                 return nil, err
356         }
357
358         req.Header.Add("Authorization", "OAuth2 "+kc.Arvados.ApiToken)
359         req.Header.Set("X-Request-Id", kc.getRequestID())
360         resp, err := kc.httpClient().Do(req)
361         if err != nil {
362                 return nil, err
363         }
364
365         defer resp.Body.Close()
366
367         if resp.StatusCode != http.StatusOK {
368                 return nil, fmt.Errorf("Got http status code: %d", resp.StatusCode)
369         }
370
371         var respBody []byte
372         respBody, err = ioutil.ReadAll(resp.Body)
373         if err != nil {
374                 return nil, err
375         }
376
377         // Got index; verify that it is complete
378         // The response should be "\n" if no locators matched the prefix
379         // Else, it should be a list of locators followed by a blank line
380         if !bytes.Equal(respBody, []byte("\n")) && !bytes.HasSuffix(respBody, []byte("\n\n")) {
381                 return nil, ErrIncompleteIndex
382         }
383
384         // Got complete index; strip the trailing newline and send
385         return bytes.NewReader(respBody[0 : len(respBody)-1]), nil
386 }
387
388 // LocalRoots() returns the map of local (i.e., disk and proxy) Keep
389 // services: uuid -> baseURI.
390 func (kc *KeepClient) LocalRoots() map[string]string {
391         kc.discoverServices()
392         kc.lock.RLock()
393         defer kc.lock.RUnlock()
394         return kc.localRoots
395 }
396
397 // GatewayRoots() returns the map of Keep remote gateway services:
398 // uuid -> baseURI.
399 func (kc *KeepClient) GatewayRoots() map[string]string {
400         kc.discoverServices()
401         kc.lock.RLock()
402         defer kc.lock.RUnlock()
403         return kc.gatewayRoots
404 }
405
406 // WritableLocalRoots() returns the map of writable local Keep services:
407 // uuid -> baseURI.
408 func (kc *KeepClient) WritableLocalRoots() map[string]string {
409         kc.discoverServices()
410         kc.lock.RLock()
411         defer kc.lock.RUnlock()
412         return kc.writableLocalRoots
413 }
414
415 // SetServiceRoots disables service discovery and updates the
416 // localRoots and gatewayRoots maps, without disrupting operations
417 // that are already in progress.
418 //
419 // The supplied maps must not be modified after calling
420 // SetServiceRoots.
421 func (kc *KeepClient) SetServiceRoots(locals, writables, gateways map[string]string) {
422         kc.disableDiscovery = true
423         kc.setServiceRoots(locals, writables, gateways)
424 }
425
426 func (kc *KeepClient) setServiceRoots(locals, writables, gateways map[string]string) {
427         kc.lock.Lock()
428         defer kc.lock.Unlock()
429         kc.localRoots = locals
430         kc.writableLocalRoots = writables
431         kc.gatewayRoots = gateways
432 }
433
434 // getSortedRoots returns a list of base URIs of Keep services, in the
435 // order they should be attempted in order to retrieve content for the
436 // given locator.
437 func (kc *KeepClient) getSortedRoots(locator string) []string {
438         var found []string
439         for _, hint := range strings.Split(locator, "+") {
440                 if len(hint) < 7 || hint[0:2] != "K@" {
441                         // Not a service hint.
442                         continue
443                 }
444                 if len(hint) == 7 {
445                         // +K@abcde means fetch from proxy at
446                         // keep.abcde.arvadosapi.com
447                         found = append(found, "https://keep."+hint[2:]+".arvadosapi.com")
448                 } else if len(hint) == 29 {
449                         // +K@abcde-abcde-abcdeabcdeabcde means fetch
450                         // from gateway with given uuid
451                         if gwURI, ok := kc.GatewayRoots()[hint[2:]]; ok {
452                                 found = append(found, gwURI)
453                         }
454                         // else this hint is no use to us; carry on.
455                 }
456         }
457         // After trying all usable service hints, fall back to local roots.
458         found = append(found, NewRootSorter(kc.LocalRoots(), locator[0:32]).GetSortedRoots()...)
459         return found
460 }
461
462 func (kc *KeepClient) cache() *BlockCache {
463         if kc.BlockCache != nil {
464                 return kc.BlockCache
465         } else {
466                 return DefaultBlockCache
467         }
468 }
469
470 func (kc *KeepClient) ClearBlockCache() {
471         kc.cache().Clear()
472 }
473
474 var (
475         // There are four global http.Client objects for the four
476         // possible permutations of TLS behavior (verify/skip-verify)
477         // and timeout settings (proxy/non-proxy).
478         defaultClient = map[bool]map[bool]HTTPClient{
479                 // defaultClient[false] is used for verified TLS reqs
480                 false: {},
481                 // defaultClient[true] is used for unverified
482                 // (insecure) TLS reqs
483                 true: {},
484         }
485         defaultClientMtx sync.Mutex
486 )
487
488 // httpClient returns the HTTPClient field if it's not nil, otherwise
489 // whichever of the four global http.Client objects is suitable for
490 // the current environment (i.e., TLS verification on/off, keep
491 // services are/aren't proxies).
492 func (kc *KeepClient) httpClient() HTTPClient {
493         if kc.HTTPClient != nil {
494                 return kc.HTTPClient
495         }
496         defaultClientMtx.Lock()
497         defer defaultClientMtx.Unlock()
498         if c, ok := defaultClient[kc.Arvados.ApiInsecure][kc.foundNonDiskSvc]; ok {
499                 return c
500         }
501
502         var requestTimeout, connectTimeout, keepAlive, tlsTimeout time.Duration
503         if kc.foundNonDiskSvc {
504                 // Use longer timeouts when connecting to a proxy,
505                 // because this usually means the intervening network
506                 // is slower.
507                 requestTimeout = DefaultProxyRequestTimeout
508                 connectTimeout = DefaultProxyConnectTimeout
509                 tlsTimeout = DefaultProxyTLSHandshakeTimeout
510                 keepAlive = DefaultProxyKeepAlive
511         } else {
512                 requestTimeout = DefaultRequestTimeout
513                 connectTimeout = DefaultConnectTimeout
514                 tlsTimeout = DefaultTLSHandshakeTimeout
515                 keepAlive = DefaultKeepAlive
516         }
517
518         transport, ok := http.DefaultTransport.(*http.Transport)
519         if ok {
520                 copy := *transport
521                 transport = &copy
522         } else {
523                 // Evidently the application has replaced
524                 // http.DefaultTransport with a different type, so we
525                 // need to build our own from scratch using the Go 1.8
526                 // defaults.
527                 transport = &http.Transport{
528                         MaxIdleConns:          100,
529                         IdleConnTimeout:       90 * time.Second,
530                         ExpectContinueTimeout: time.Second,
531                 }
532         }
533         transport.DialContext = (&net.Dialer{
534                 Timeout:   connectTimeout,
535                 KeepAlive: keepAlive,
536                 DualStack: true,
537         }).DialContext
538         transport.TLSHandshakeTimeout = tlsTimeout
539         transport.TLSClientConfig = arvadosclient.MakeTLSConfig(kc.Arvados.ApiInsecure)
540         c := &http.Client{
541                 Timeout:   requestTimeout,
542                 Transport: transport,
543         }
544         defaultClient[kc.Arvados.ApiInsecure][kc.foundNonDiskSvc] = c
545         return c
546 }
547
548 var reqIDGen = httpserver.IDGenerator{Prefix: "req-"}
549
550 func (kc *KeepClient) getRequestID() string {
551         if kc.RequestID != "" {
552                 return kc.RequestID
553         } else {
554                 return reqIDGen.Next()
555         }
556 }
557
558 type Locator struct {
559         Hash  string
560         Size  int      // -1 if data size is not known
561         Hints []string // Including the size hint, if any
562 }
563
564 func (loc *Locator) String() string {
565         s := loc.Hash
566         if len(loc.Hints) > 0 {
567                 s = s + "+" + strings.Join(loc.Hints, "+")
568         }
569         return s
570 }
571
572 var locatorMatcher = regexp.MustCompile("^([0-9a-f]{32})([+](.*))?$")
573
574 func MakeLocator(path string) (*Locator, error) {
575         sm := locatorMatcher.FindStringSubmatch(path)
576         if sm == nil {
577                 return nil, InvalidLocatorError
578         }
579         loc := Locator{Hash: sm[1], Size: -1}
580         if sm[2] != "" {
581                 loc.Hints = strings.Split(sm[3], "+")
582         } else {
583                 loc.Hints = []string{}
584         }
585         if len(loc.Hints) > 0 {
586                 if size, err := strconv.Atoi(loc.Hints[0]); err == nil {
587                         loc.Size = size
588                 }
589         }
590         return &loc, nil
591 }