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