1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: Apache-2.0
5 // Package keepclient provides low-level Get/Put primitives for accessing
6 // Arvados Keep blocks.
28 "git.arvados.org/arvados.git/sdk/go/arvados"
29 "git.arvados.org/arvados.git/sdk/go/arvadosclient"
30 "git.arvados.org/arvados.git/sdk/go/httpserver"
33 // BLOCKSIZE defines the length of a Keep "block", which is 64MB.
34 const BLOCKSIZE = 64 * 1024 * 1024
37 DefaultRequestTimeout = 20 * time.Second
38 DefaultConnectTimeout = 2 * time.Second
39 DefaultTLSHandshakeTimeout = 4 * time.Second
40 DefaultKeepAlive = 180 * time.Second
42 DefaultProxyRequestTimeout = 300 * time.Second
43 DefaultProxyConnectTimeout = 30 * time.Second
44 DefaultProxyTLSHandshakeTimeout = 10 * time.Second
45 DefaultProxyKeepAlive = 120 * time.Second
47 rootCacheDir = "/var/cache/arvados/keep"
48 userCacheDir = ".cache/arvados/keep" // relative to HOME
51 // Error interface with an error and boolean indicating whether the error is temporary
52 type Error interface {
57 // multipleResponseError is of type Error
58 type multipleResponseError struct {
63 func (e *multipleResponseError) Temporary() bool {
67 // BlockNotFound is a multipleResponseError where isTemp is false
68 var BlockNotFound = &ErrNotFound{multipleResponseError{
69 error: errors.New("Block not found"),
73 // ErrNotFound is a multipleResponseError where isTemp can be true or false
74 type ErrNotFound struct {
78 func (*ErrNotFound) HTTPStatus() int { return http.StatusNotFound }
80 type InsufficientReplicasError struct{ error }
82 type OversizeBlockError struct{ error }
84 var ErrOversizeBlock = OversizeBlockError{error: errors.New("Exceeded maximum block size (" + strconv.Itoa(BLOCKSIZE) + ")")}
85 var MissingArvadosApiHost = errors.New("Missing required environment variable ARVADOS_API_HOST")
86 var MissingArvadosApiToken = errors.New("Missing required environment variable ARVADOS_API_TOKEN")
87 var InvalidLocatorError = errors.New("Invalid locator")
89 // ErrNoSuchKeepServer is returned when GetIndex is invoked with a UUID with no matching keep server
90 var ErrNoSuchKeepServer = errors.New("No keep server matching the given UUID is found")
92 // ErrIncompleteIndex is returned when the Index response does not end with a new empty line
93 var ErrIncompleteIndex = errors.New("Got incomplete index")
96 XKeepDesiredReplicas = "X-Keep-Desired-Replicas"
97 XKeepReplicasStored = "X-Keep-Replicas-Stored"
98 XKeepStorageClasses = "X-Keep-Storage-Classes"
99 XKeepStorageClassesConfirmed = "X-Keep-Storage-Classes-Confirmed"
102 type HTTPClient interface {
103 Do(*http.Request) (*http.Response, error)
106 const DiskCacheDisabled = arvados.ByteSizeOrPercent(1)
108 // KeepClient holds information about Arvados and Keep servers.
109 type KeepClient struct {
110 Arvados *arvadosclient.ArvadosClient
112 localRoots map[string]string
113 writableLocalRoots map[string]string
114 gatewayRoots map[string]string
116 HTTPClient HTTPClient
119 StorageClasses []string
120 DefaultStorageClasses []string // Set by cluster's exported config
121 DiskCacheSize arvados.ByteSizeOrPercent // See also DiskCacheDisabled
123 // set to 1 if all writable services are of disk type, otherwise 0
124 replicasPerService int
126 // Any non-disk typed services found in the list of keepservers?
129 // Disable automatic discovery of keep services
130 disableDiscovery bool
132 gatewayStack arvados.KeepGateway
135 func (kc *KeepClient) Clone() *KeepClient {
137 defer kc.lock.Unlock()
140 Want_replicas: kc.Want_replicas,
141 localRoots: kc.localRoots,
142 writableLocalRoots: kc.writableLocalRoots,
143 gatewayRoots: kc.gatewayRoots,
144 HTTPClient: kc.HTTPClient,
146 RequestID: kc.RequestID,
147 StorageClasses: kc.StorageClasses,
148 DefaultStorageClasses: kc.DefaultStorageClasses,
149 DiskCacheSize: kc.DiskCacheSize,
150 replicasPerService: kc.replicasPerService,
151 foundNonDiskSvc: kc.foundNonDiskSvc,
152 disableDiscovery: kc.disableDiscovery,
156 func (kc *KeepClient) loadDefaultClasses() error {
157 scData, err := kc.Arvados.ClusterConfig("StorageClasses")
161 classes := scData.(map[string]interface{})
162 for scName := range classes {
163 scConf, _ := classes[scName].(map[string]interface{})
164 isDefault, ok := scConf["Default"].(bool)
166 kc.DefaultStorageClasses = append(kc.DefaultStorageClasses, scName)
172 // MakeKeepClient creates a new KeepClient, loads default storage classes, calls
173 // DiscoverKeepServices(), and returns when the client is ready to
175 func MakeKeepClient(arv *arvadosclient.ArvadosClient) (*KeepClient, error) {
177 return kc, kc.discoverServices()
180 // New creates a new KeepClient. Service discovery will occur on the
181 // next read/write operation.
182 func New(arv *arvadosclient.ArvadosClient) *KeepClient {
183 defaultReplicationLevel := 2
184 value, err := arv.Discovery("defaultCollectionReplication")
186 v, ok := value.(float64)
188 defaultReplicationLevel = int(v)
193 Want_replicas: defaultReplicationLevel,
196 err = kc.loadDefaultClasses()
198 DebugPrintf("DEBUG: Unable to load the default storage classes cluster config")
203 // PutHR puts a block given the block hash, a reader, and the number of bytes
204 // to read from the reader (which must be between 0 and BLOCKSIZE).
206 // Returns the locator for the written block, the number of replicas
207 // written, and an error.
209 // Returns an InsufficientReplicasError if 0 <= replicas <
210 // kc.Wants_replicas.
211 func (kc *KeepClient) PutHR(hash string, r io.Reader, dataBytes int64) (string, int, error) {
212 resp, err := kc.BlockWrite(context.Background(), arvados.BlockWriteOptions{
215 DataSize: int(dataBytes),
217 return resp.Locator, resp.Replicas, err
220 // PutHB writes a block to Keep. The hash of the bytes is given in
221 // hash, and the data is given in buf.
223 // Return values are the same as for PutHR.
224 func (kc *KeepClient) PutHB(hash string, buf []byte) (string, int, error) {
225 resp, err := kc.BlockWrite(context.Background(), arvados.BlockWriteOptions{
229 return resp.Locator, resp.Replicas, err
232 // PutB writes a block to Keep. It computes the hash itself.
234 // Return values are the same as for PutHR.
235 func (kc *KeepClient) PutB(buffer []byte) (string, int, error) {
236 resp, err := kc.BlockWrite(context.Background(), arvados.BlockWriteOptions{
239 return resp.Locator, resp.Replicas, err
242 // PutR writes a block to Keep. It first reads all data from r into a buffer
243 // in order to compute the hash.
245 // Return values are the same as for PutHR.
247 // If the block hash and data size are known, PutHR is more efficient.
248 func (kc *KeepClient) PutR(r io.Reader) (locator string, replicas int, err error) {
249 buffer, err := ioutil.ReadAll(r)
253 return kc.PutB(buffer)
256 func (kc *KeepClient) getOrHead(method string, locator string, header http.Header) (io.ReadCloser, int64, string, http.Header, error) {
257 if strings.HasPrefix(locator, "d41d8cd98f00b204e9800998ecf8427e+0") {
258 return ioutil.NopCloser(bytes.NewReader(nil)), 0, "", nil, nil
261 reqid := kc.getRequestID()
263 var expectLength int64
264 if parts := strings.SplitN(locator, "+", 3); len(parts) < 2 {
266 } else if n, err := strconv.ParseInt(parts[1], 10, 64); err != nil {
274 triesRemaining := 1 + kc.Retries
276 serversToTry := kc.getSortedRoots(locator)
278 numServers := len(serversToTry)
281 var retryList []string
283 for triesRemaining > 0 {
287 for _, host := range serversToTry {
288 url := host + "/" + locator
290 req, err := http.NewRequest(method, url, nil)
292 errs = append(errs, fmt.Sprintf("%s: %v", url, err))
295 for k, v := range header {
296 req.Header[k] = append([]string(nil), v...)
298 if req.Header.Get("Authorization") == "" {
299 req.Header.Set("Authorization", "OAuth2 "+kc.Arvados.ApiToken)
301 if req.Header.Get("X-Request-Id") == "" {
302 req.Header.Set("X-Request-Id", reqid)
304 resp, err := kc.httpClient().Do(req)
306 // Probably a network error, may be transient,
308 errs = append(errs, fmt.Sprintf("%s: %v", url, err))
309 retryList = append(retryList, host)
312 if resp.StatusCode != http.StatusOK {
314 respbody, _ = ioutil.ReadAll(&io.LimitedReader{R: resp.Body, N: 4096})
316 errs = append(errs, fmt.Sprintf("%s: HTTP %d %q",
317 url, resp.StatusCode, bytes.TrimSpace(respbody)))
319 if resp.StatusCode == 408 ||
320 resp.StatusCode == 429 ||
321 resp.StatusCode >= 500 {
322 // Timeout, too many requests, or other
323 // server side failure, transient
324 // error, can try again.
325 retryList = append(retryList, host)
326 } else if resp.StatusCode == 404 {
331 if expectLength < 0 {
332 if resp.ContentLength < 0 {
334 return nil, 0, "", nil, fmt.Errorf("error reading %q: no size hint, no Content-Length header in response", locator)
336 expectLength = resp.ContentLength
337 } else if resp.ContentLength >= 0 && expectLength != resp.ContentLength {
339 return nil, 0, "", nil, fmt.Errorf("error reading %q: size hint %d != Content-Length %d", locator, expectLength, resp.ContentLength)
343 return HashCheckingReader{
346 Check: locator[0:32],
347 }, expectLength, url, resp.Header, nil
350 return nil, expectLength, url, resp.Header, nil
352 serversToTry = retryList
354 DebugPrintf("DEBUG: %s %s failed: %v", method, locator, errs)
357 if count404 == numServers {
360 err = &ErrNotFound{multipleResponseError{
361 error: fmt.Errorf("%s %s failed: %v", method, locator, errs),
362 isTemp: len(serversToTry) > 0,
365 return nil, 0, "", nil, err
368 // attempt to create dir/subdir/ and its parents, up to but not
369 // including dir itself, using mode 0700.
370 func makedirs(dir, subdir string) {
371 for _, part := range strings.Split(subdir, string(os.PathSeparator)) {
372 dir = filepath.Join(dir, part)
377 // upstreamGateway creates/returns the KeepGateway stack used to read
378 // and write data: a disk-backed cache on top of an http backend.
379 func (kc *KeepClient) upstreamGateway() arvados.KeepGateway {
381 defer kc.lock.Unlock()
382 if kc.gatewayStack != nil {
383 return kc.gatewayStack
386 if os.Geteuid() == 0 {
387 cachedir = rootCacheDir
388 makedirs("/", cachedir)
390 home := "/" + os.Getenv("HOME")
391 makedirs(home, userCacheDir)
392 cachedir = filepath.Join(home, userCacheDir)
394 backend := &keepViaHTTP{kc}
395 if kc.DiskCacheSize == DiskCacheDisabled {
396 kc.gatewayStack = backend
398 kc.gatewayStack = &arvados.DiskCache{
400 MaxSize: kc.DiskCacheSize,
401 KeepGateway: backend,
404 return kc.gatewayStack
407 // LocalLocator returns a locator equivalent to the one supplied, but
408 // with a valid signature from the local cluster. If the given locator
409 // already has a local signature, it is returned unchanged.
410 func (kc *KeepClient) LocalLocator(locator string) (string, error) {
411 return kc.upstreamGateway().LocalLocator(locator)
414 // Get retrieves the specified block from the local cache or a backend
415 // server. Returns a reader, the expected data length (or -1 if not
416 // known), and an error.
418 // The third return value (formerly a source URL in previous versions)
419 // is an empty string.
421 // If the block checksum does not match, the final Read() on the
422 // reader returned by this method will return a BadChecksum error
425 // New code should use BlockRead and/or ReadAt instead of Get.
426 func (kc *KeepClient) Get(locator string) (io.ReadCloser, int64, string, error) {
427 loc, err := MakeLocator(locator)
429 return nil, 0, "", err
433 n, err := kc.BlockRead(context.Background(), arvados.BlockReadOptions{
438 pw.CloseWithError(err)
439 } else if loc.Size >= 0 && n != loc.Size {
440 pw.CloseWithError(fmt.Errorf("expected block size %d but read %d bytes", loc.Size, n))
445 // Wait for the first byte to arrive, so that, if there's an
446 // error before we receive any data, we can return the error
447 // directly, instead of indirectly via a reader that returns
449 bufr := bufio.NewReader(pr)
450 _, err = bufr.Peek(1)
451 if err != nil && err != io.EOF {
452 pr.CloseWithError(err)
453 return nil, 0, "", err
455 if err == io.EOF && (loc.Size == 0 || loc.Hash == "d41d8cd98f00b204e9800998ecf8427e") {
456 // In the special case of the zero-length block, EOF
457 // error from Peek() is normal.
458 return pr, 0, "", nil
466 }, int64(loc.Size), "", err
469 // BlockRead retrieves a block from the cache if it's present, otherwise
471 func (kc *KeepClient) BlockRead(ctx context.Context, opts arvados.BlockReadOptions) (int, error) {
472 return kc.upstreamGateway().BlockRead(ctx, opts)
475 // ReadAt retrieves a portion of block from the cache if it's
476 // present, otherwise from the network.
477 func (kc *KeepClient) ReadAt(locator string, p []byte, off int) (int, error) {
478 return kc.upstreamGateway().ReadAt(locator, p, off)
481 // BlockWrite writes a full block to upstream servers and saves a copy
482 // in the local cache.
483 func (kc *KeepClient) BlockWrite(ctx context.Context, req arvados.BlockWriteOptions) (arvados.BlockWriteResponse, error) {
484 return kc.upstreamGateway().BlockWrite(ctx, req)
487 // Ask verifies that a block with the given hash is available and
488 // readable, according to at least one Keep service. Unlike Get, it
489 // does not retrieve the data or verify that the data content matches
490 // the hash specified by the locator.
492 // Returns the data size (content length) reported by the Keep service
493 // and the URI reporting the data size.
494 func (kc *KeepClient) Ask(locator string) (int64, string, error) {
495 _, size, url, _, err := kc.getOrHead("HEAD", locator, nil)
496 return size, url, err
499 // GetIndex retrieves a list of blocks stored on the given server whose hashes
500 // begin with the given prefix. The returned reader will return an error (other
501 // than EOF) if the complete index cannot be retrieved.
503 // This is meant to be used only by system components and admin tools.
504 // It will return an error unless the client is using a "data manager token"
505 // recognized by the Keep services.
506 func (kc *KeepClient) GetIndex(keepServiceUUID, prefix string) (io.Reader, error) {
507 url := kc.LocalRoots()[keepServiceUUID]
509 return nil, ErrNoSuchKeepServer
517 req, err := http.NewRequest("GET", url, nil)
522 req.Header.Add("Authorization", "OAuth2 "+kc.Arvados.ApiToken)
523 req.Header.Set("X-Request-Id", kc.getRequestID())
524 resp, err := kc.httpClient().Do(req)
529 defer resp.Body.Close()
531 if resp.StatusCode != http.StatusOK {
532 return nil, fmt.Errorf("Got http status code: %d", resp.StatusCode)
536 respBody, err = ioutil.ReadAll(resp.Body)
541 // Got index; verify that it is complete
542 // The response should be "\n" if no locators matched the prefix
543 // Else, it should be a list of locators followed by a blank line
544 if !bytes.Equal(respBody, []byte("\n")) && !bytes.HasSuffix(respBody, []byte("\n\n")) {
545 return nil, ErrIncompleteIndex
548 // Got complete index; strip the trailing newline and send
549 return bytes.NewReader(respBody[0 : len(respBody)-1]), nil
552 // LocalRoots returns the map of local (i.e., disk and proxy) Keep
553 // services: uuid -> baseURI.
554 func (kc *KeepClient) LocalRoots() map[string]string {
555 kc.discoverServices()
557 defer kc.lock.RUnlock()
561 // GatewayRoots returns the map of Keep remote gateway services:
563 func (kc *KeepClient) GatewayRoots() map[string]string {
564 kc.discoverServices()
566 defer kc.lock.RUnlock()
567 return kc.gatewayRoots
570 // WritableLocalRoots returns the map of writable local Keep services:
572 func (kc *KeepClient) WritableLocalRoots() map[string]string {
573 kc.discoverServices()
575 defer kc.lock.RUnlock()
576 return kc.writableLocalRoots
579 // SetServiceRoots disables service discovery and updates the
580 // localRoots and gatewayRoots maps, without disrupting operations
581 // that are already in progress.
583 // The supplied maps must not be modified after calling
585 func (kc *KeepClient) SetServiceRoots(locals, writables, gateways map[string]string) {
586 kc.disableDiscovery = true
587 kc.setServiceRoots(locals, writables, gateways)
590 func (kc *KeepClient) setServiceRoots(locals, writables, gateways map[string]string) {
592 defer kc.lock.Unlock()
593 kc.localRoots = locals
594 kc.writableLocalRoots = writables
595 kc.gatewayRoots = gateways
598 // getSortedRoots returns a list of base URIs of Keep services, in the
599 // order they should be attempted in order to retrieve content for the
601 func (kc *KeepClient) getSortedRoots(locator string) []string {
603 for _, hint := range strings.Split(locator, "+") {
604 if len(hint) < 7 || hint[0:2] != "K@" {
605 // Not a service hint.
609 // +K@abcde means fetch from proxy at
610 // keep.abcde.arvadosapi.com
611 found = append(found, "https://keep."+hint[2:]+".arvadosapi.com")
612 } else if len(hint) == 29 {
613 // +K@abcde-abcde-abcdeabcdeabcde means fetch
614 // from gateway with given uuid
615 if gwURI, ok := kc.GatewayRoots()[hint[2:]]; ok {
616 found = append(found, gwURI)
618 // else this hint is no use to us; carry on.
621 // After trying all usable service hints, fall back to local roots.
622 found = append(found, NewRootSorter(kc.LocalRoots(), locator[0:32]).GetSortedRoots()...)
626 func (kc *KeepClient) SetStorageClasses(sc []string) {
627 // make a copy so the caller can't mess with it.
628 kc.StorageClasses = append([]string{}, sc...)
632 // There are four global http.Client objects for the four
633 // possible permutations of TLS behavior (verify/skip-verify)
634 // and timeout settings (proxy/non-proxy).
635 defaultClient = map[bool]map[bool]HTTPClient{
636 // defaultClient[false] is used for verified TLS reqs
638 // defaultClient[true] is used for unverified
639 // (insecure) TLS reqs
642 defaultClientMtx sync.Mutex
645 // httpClient returns the HTTPClient field if it's not nil, otherwise
646 // whichever of the four global http.Client objects is suitable for
647 // the current environment (i.e., TLS verification on/off, keep
648 // services are/aren't proxies).
649 func (kc *KeepClient) httpClient() HTTPClient {
650 if kc.HTTPClient != nil {
653 defaultClientMtx.Lock()
654 defer defaultClientMtx.Unlock()
655 if c, ok := defaultClient[kc.Arvados.ApiInsecure][kc.foundNonDiskSvc]; ok {
659 var requestTimeout, connectTimeout, keepAlive, tlsTimeout time.Duration
660 if kc.foundNonDiskSvc {
661 // Use longer timeouts when connecting to a proxy,
662 // because this usually means the intervening network
664 requestTimeout = DefaultProxyRequestTimeout
665 connectTimeout = DefaultProxyConnectTimeout
666 tlsTimeout = DefaultProxyTLSHandshakeTimeout
667 keepAlive = DefaultProxyKeepAlive
669 requestTimeout = DefaultRequestTimeout
670 connectTimeout = DefaultConnectTimeout
671 tlsTimeout = DefaultTLSHandshakeTimeout
672 keepAlive = DefaultKeepAlive
676 Timeout: requestTimeout,
677 // It's not safe to copy *http.DefaultTransport
678 // because it has a mutex (which might be locked)
679 // protecting a private map (which might not be nil).
680 // So we build our own, using the Go 1.12 default
681 // values, ignoring any changes the application has
682 // made to http.DefaultTransport.
683 Transport: &http.Transport{
684 DialContext: (&net.Dialer{
685 Timeout: connectTimeout,
686 KeepAlive: keepAlive,
690 IdleConnTimeout: 90 * time.Second,
691 TLSHandshakeTimeout: tlsTimeout,
692 ExpectContinueTimeout: 1 * time.Second,
693 TLSClientConfig: arvadosclient.MakeTLSConfig(kc.Arvados.ApiInsecure),
696 defaultClient[kc.Arvados.ApiInsecure][kc.foundNonDiskSvc] = c
700 var reqIDGen = httpserver.IDGenerator{Prefix: "req-"}
702 func (kc *KeepClient) getRequestID() string {
703 if kc.RequestID != "" {
706 return reqIDGen.Next()
709 type Locator struct {
711 Size int // -1 if data size is not known
712 Hints []string // Including the size hint, if any
715 func (loc *Locator) String() string {
717 if len(loc.Hints) > 0 {
718 s = s + "+" + strings.Join(loc.Hints, "+")
723 var locatorMatcher = regexp.MustCompile("^([0-9a-f]{32})([+](.*))?$")
725 func MakeLocator(path string) (*Locator, error) {
726 sm := locatorMatcher.FindStringSubmatch(path)
728 return nil, InvalidLocatorError
730 loc := Locator{Hash: sm[1], Size: -1}
732 loc.Hints = strings.Split(sm[3], "+")
734 loc.Hints = []string{}
736 if len(loc.Hints) > 0 {
737 if size, err := strconv.Atoi(loc.Hints[0]); err == nil {