sdk/go/keepclient/keepclient.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: Apache-2.0
   4
   5 // Package keepclient provides low-level Get/Put primitives for accessing
   6 // Arvados Keep blocks.
   7 package keepclient
   8
   9 import (
  10         "bufio"
  11         "bytes"
  12         "context"
  13         "crypto/md5"
  14         "errors"
  15         "fmt"
  16         "io"
  17         "io/ioutil"
  18         "net"
  19         "net/http"
  20         "os"
  21         "path/filepath"
  22         "regexp"
  23         "strconv"
  24         "strings"
  25         "sync"
  26         "time"
  27
  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"
  31 )
  32
  33 // BLOCKSIZE defines the length of a Keep "block", which is 64MB.
  34 const BLOCKSIZE = 64 * 1024 * 1024
  35
  36 var (
  37         DefaultRequestTimeout      = 20 * time.Second
  38         DefaultConnectTimeout      = 2 * time.Second
  39         DefaultTLSHandshakeTimeout = 4 * time.Second
  40         DefaultKeepAlive           = 180 * time.Second
  41
  42         DefaultProxyRequestTimeout      = 300 * time.Second
  43         DefaultProxyConnectTimeout      = 30 * time.Second
  44         DefaultProxyTLSHandshakeTimeout = 10 * time.Second
  45         DefaultProxyKeepAlive           = 120 * time.Second
  46
  47         rootCacheDir = "/var/cache/arvados/keep"
  48         userCacheDir = ".cache/arvados/keep" // relative to HOME
  49 )
  50
  51 // Error interface with an error and boolean indicating whether the error is temporary
  52 type Error interface {
  53         error
  54         Temporary() bool
  55 }
  56
  57 // multipleResponseError is of type Error
  58 type multipleResponseError struct {
  59         error
  60         isTemp bool
  61 }
  62
  63 func (e *multipleResponseError) Temporary() bool {
  64         return e.isTemp
  65 }
  66
  67 // BlockNotFound is a multipleResponseError where isTemp is false
  68 var BlockNotFound = &ErrNotFound{multipleResponseError{
  69         error:  errors.New("Block not found"),
  70         isTemp: false,
  71 }}
  72
  73 // ErrNotFound is a multipleResponseError where isTemp can be true or false
  74 type ErrNotFound struct {
  75         multipleResponseError
  76 }
  77
  78 type InsufficientReplicasError struct{ error }
  79
  80 type OversizeBlockError struct{ error }
  81
  82 var ErrOversizeBlock = OversizeBlockError{error: errors.New("Exceeded maximum block size (" + strconv.Itoa(BLOCKSIZE) + ")")}
  83 var MissingArvadosApiHost = errors.New("Missing required environment variable ARVADOS_API_HOST")
  84 var MissingArvadosApiToken = errors.New("Missing required environment variable ARVADOS_API_TOKEN")
  85 var InvalidLocatorError = errors.New("Invalid locator")
  86
  87 // ErrNoSuchKeepServer is returned when GetIndex is invoked with a UUID with no matching keep server
  88 var ErrNoSuchKeepServer = errors.New("No keep server matching the given UUID is found")
  89
  90 // ErrIncompleteIndex is returned when the Index response does not end with a new empty line
  91 var ErrIncompleteIndex = errors.New("Got incomplete index")
  92
  93 const (
  94         XKeepDesiredReplicas         = "X-Keep-Desired-Replicas"
  95         XKeepReplicasStored          = "X-Keep-Replicas-Stored"
  96         XKeepStorageClasses          = "X-Keep-Storage-Classes"
  97         XKeepStorageClassesConfirmed = "X-Keep-Storage-Classes-Confirmed"
  98 )
  99
 100 type HTTPClient interface {
 101         Do(*http.Request) (*http.Response, error)
 102 }
 103
 104 const DiskCacheDisabled = arvados.ByteSizeOrPercent(1)
 105
 106 // KeepClient holds information about Arvados and Keep servers.
 107 type KeepClient struct {
 108         Arvados               *arvadosclient.ArvadosClient
 109         Want_replicas         int
 110         localRoots            map[string]string
 111         writableLocalRoots    map[string]string
 112         gatewayRoots          map[string]string
 113         lock                  sync.RWMutex
 114         HTTPClient            HTTPClient
 115         Retries               int
 116         RequestID             string
 117         StorageClasses        []string
 118         DefaultStorageClasses []string                  // Set by cluster's exported config
 119         DiskCacheSize         arvados.ByteSizeOrPercent // See also DiskCacheDisabled
 120
 121         // set to 1 if all writable services are of disk type, otherwise 0
 122         replicasPerService int
 123
 124         // Any non-disk typed services found in the list of keepservers?
 125         foundNonDiskSvc bool
 126
 127         // Disable automatic discovery of keep services
 128         disableDiscovery bool
 129
 130         gatewayStack arvados.KeepGateway
 131 }
 132
 133 func (kc *KeepClient) Clone() *KeepClient {
 134         kc.lock.Lock()
 135         defer kc.lock.Unlock()
 136         return &KeepClient{
 137                 Arvados:               kc.Arvados,
 138                 Want_replicas:         kc.Want_replicas,
 139                 localRoots:            kc.localRoots,
 140                 writableLocalRoots:    kc.writableLocalRoots,
 141                 gatewayRoots:          kc.gatewayRoots,
 142                 HTTPClient:            kc.HTTPClient,
 143                 Retries:               kc.Retries,
 144                 RequestID:             kc.RequestID,
 145                 StorageClasses:        kc.StorageClasses,
 146                 DefaultStorageClasses: kc.DefaultStorageClasses,
 147                 DiskCacheSize:         kc.DiskCacheSize,
 148                 replicasPerService:    kc.replicasPerService,
 149                 foundNonDiskSvc:       kc.foundNonDiskSvc,
 150                 disableDiscovery:      kc.disableDiscovery,
 151         }
 152 }
 153
 154 func (kc *KeepClient) loadDefaultClasses() error {
 155         scData, err := kc.Arvados.ClusterConfig("StorageClasses")
 156         if err != nil {
 157                 return err
 158         }
 159         classes := scData.(map[string]interface{})
 160         for scName := range classes {
 161                 scConf, _ := classes[scName].(map[string]interface{})
 162                 isDefault, ok := scConf["Default"].(bool)
 163                 if ok && isDefault {
 164                         kc.DefaultStorageClasses = append(kc.DefaultStorageClasses, scName)
 165                 }
 166         }
 167         return nil
 168 }
 169
 170 // MakeKeepClient creates a new KeepClient, loads default storage classes, calls
 171 // DiscoverKeepServices(), and returns when the client is ready to
 172 // use.
 173 func MakeKeepClient(arv *arvadosclient.ArvadosClient) (*KeepClient, error) {
 174         kc := New(arv)
 175         return kc, kc.discoverServices()
 176 }
 177
 178 // New creates a new KeepClient. Service discovery will occur on the
 179 // next read/write operation.
 180 func New(arv *arvadosclient.ArvadosClient) *KeepClient {
 181         defaultReplicationLevel := 2
 182         value, err := arv.Discovery("defaultCollectionReplication")
 183         if err == nil {
 184                 v, ok := value.(float64)
 185                 if ok && v > 0 {
 186                         defaultReplicationLevel = int(v)
 187                 }
 188         }
 189         kc := &KeepClient{
 190                 Arvados:       arv,
 191                 Want_replicas: defaultReplicationLevel,
 192                 Retries:       2,
 193         }
 194         err = kc.loadDefaultClasses()
 195         if err != nil {
 196                 DebugPrintf("DEBUG: Unable to load the default storage classes cluster config")
 197         }
 198         return kc
 199 }
 200
 201 // PutHR puts a block given the block hash, a reader, and the number of bytes
 202 // to read from the reader (which must be between 0 and BLOCKSIZE).
 203 //
 204 // Returns the locator for the written block, the number of replicas
 205 // written, and an error.
 206 //
 207 // Returns an InsufficientReplicasError if 0 <= replicas <
 208 // kc.Wants_replicas.
 209 func (kc *KeepClient) PutHR(hash string, r io.Reader, dataBytes int64) (string, int, error) {
 210         resp, err := kc.BlockWrite(context.Background(), arvados.BlockWriteOptions{
 211                 Hash:     hash,
 212                 Reader:   r,
 213                 DataSize: int(dataBytes),
 214         })
 215         return resp.Locator, resp.Replicas, err
 216 }
 217
 218 // PutHB writes a block to Keep. The hash of the bytes is given in
 219 // hash, and the data is given in buf.
 220 //
 221 // Return values are the same as for PutHR.
 222 func (kc *KeepClient) PutHB(hash string, buf []byte) (string, int, error) {
 223         resp, err := kc.BlockWrite(context.Background(), arvados.BlockWriteOptions{
 224                 Hash: hash,
 225                 Data: buf,
 226         })
 227         return resp.Locator, resp.Replicas, err
 228 }
 229
 230 // PutB writes a block to Keep. It computes the hash itself.
 231 //
 232 // Return values are the same as for PutHR.
 233 func (kc *KeepClient) PutB(buffer []byte) (string, int, error) {
 234         resp, err := kc.BlockWrite(context.Background(), arvados.BlockWriteOptions{
 235                 Data: buffer,
 236         })
 237         return resp.Locator, resp.Replicas, err
 238 }
 239
 240 // PutR writes a block to Keep. It first reads all data from r into a buffer
 241 // in order to compute the hash.
 242 //
 243 // Return values are the same as for PutHR.
 244 //
 245 // If the block hash and data size are known, PutHR is more efficient.
 246 func (kc *KeepClient) PutR(r io.Reader) (locator string, replicas int, err error) {
 247         buffer, err := ioutil.ReadAll(r)
 248         if err != nil {
 249                 return "", 0, err
 250         }
 251         return kc.PutB(buffer)
 252 }
 253
 254 func (kc *KeepClient) getOrHead(method string, locator string, header http.Header) (io.ReadCloser, int64, string, http.Header, error) {
 255         if strings.HasPrefix(locator, "d41d8cd98f00b204e9800998ecf8427e+0") {
 256                 return ioutil.NopCloser(bytes.NewReader(nil)), 0, "", nil, nil
 257         }
 258
 259         reqid := kc.getRequestID()
 260
 261         var expectLength int64
 262         if parts := strings.SplitN(locator, "+", 3); len(parts) < 2 {
 263                 expectLength = -1
 264         } else if n, err := strconv.ParseInt(parts[1], 10, 64); err != nil {
 265                 expectLength = -1
 266         } else {
 267                 expectLength = n
 268         }
 269
 270         var errs []string
 271
 272         triesRemaining := 1 + kc.Retries
 273
 274         serversToTry := kc.getSortedRoots(locator)
 275
 276         numServers := len(serversToTry)
 277         count404 := 0
 278
 279         var retryList []string
 280
 281         for triesRemaining > 0 {
 282                 triesRemaining--
 283                 retryList = nil
 284
 285                 for _, host := range serversToTry {
 286                         url := host + "/" + locator
 287
 288                         req, err := http.NewRequest(method, url, nil)
 289                         if err != nil {
 290                                 errs = append(errs, fmt.Sprintf("%s: %v", url, err))
 291                                 continue
 292                         }
 293                         for k, v := range header {
 294                                 req.Header[k] = append([]string(nil), v...)
 295                         }
 296                         if req.Header.Get("Authorization") == "" {
 297                                 req.Header.Set("Authorization", "OAuth2 "+kc.Arvados.ApiToken)
 298                         }
 299                         if req.Header.Get("X-Request-Id") == "" {
 300                                 req.Header.Set("X-Request-Id", reqid)
 301                         }
 302                         resp, err := kc.httpClient().Do(req)
 303                         if err != nil {
 304                                 // Probably a network error, may be transient,
 305                                 // can try again.
 306                                 errs = append(errs, fmt.Sprintf("%s: %v", url, err))
 307                                 retryList = append(retryList, host)
 308                                 continue
 309                         }
 310                         if resp.StatusCode != http.StatusOK {
 311                                 var respbody []byte
 312                                 respbody, _ = ioutil.ReadAll(&io.LimitedReader{R: resp.Body, N: 4096})
 313                                 resp.Body.Close()
 314                                 errs = append(errs, fmt.Sprintf("%s: HTTP %d %q",
 315                                         url, resp.StatusCode, bytes.TrimSpace(respbody)))
 316
 317                                 if resp.StatusCode == 408 ||
 318                                         resp.StatusCode == 429 ||
 319                                         resp.StatusCode >= 500 {
 320                                         // Timeout, too many requests, or other
 321                                         // server side failure, transient
 322                                         // error, can try again.
 323                                         retryList = append(retryList, host)
 324                                 } else if resp.StatusCode == 404 {
 325                                         count404++
 326                                 }
 327                                 continue
 328                         }
 329                         if expectLength < 0 {
 330                                 if resp.ContentLength < 0 {
 331                                         resp.Body.Close()
 332                                         return nil, 0, "", nil, fmt.Errorf("error reading %q: no size hint, no Content-Length header in response", locator)
 333                                 }
 334                                 expectLength = resp.ContentLength
 335                         } else if resp.ContentLength >= 0 && expectLength != resp.ContentLength {
 336                                 resp.Body.Close()
 337                                 return nil, 0, "", nil, fmt.Errorf("error reading %q: size hint %d != Content-Length %d", locator, expectLength, resp.ContentLength)
 338                         }
 339                         // Success
 340                         if method == "GET" {
 341                                 return HashCheckingReader{
 342                                         Reader: resp.Body,
 343                                         Hash:   md5.New(),
 344                                         Check:  locator[0:32],
 345                                 }, expectLength, url, resp.Header, nil
 346                         }
 347                         resp.Body.Close()
 348                         return nil, expectLength, url, resp.Header, nil
 349                 }
 350                 serversToTry = retryList
 351         }
 352         DebugPrintf("DEBUG: %s %s failed: %v", method, locator, errs)
 353
 354         var err error
 355         if count404 == numServers {
 356                 err = BlockNotFound
 357         } else {
 358                 err = &ErrNotFound{multipleResponseError{
 359                         error:  fmt.Errorf("%s %s failed: %v", method, locator, errs),
 360                         isTemp: len(serversToTry) > 0,
 361                 }}
 362         }
 363         return nil, 0, "", nil, err
 364 }
 365
 366 // attempt to create dir/subdir/ and its parents, up to but not
 367 // including dir itself, using mode 0700.
 368 func makedirs(dir, subdir string) {
 369         for _, part := range strings.Split(subdir, string(os.PathSeparator)) {
 370                 dir = filepath.Join(dir, part)
 371                 os.Mkdir(dir, 0700)
 372         }
 373 }
 374
 375 // upstreamGateway creates/returns the KeepGateway stack used to read
 376 // and write data: a disk-backed cache on top of an http backend.
 377 func (kc *KeepClient) upstreamGateway() arvados.KeepGateway {
 378         kc.lock.Lock()
 379         defer kc.lock.Unlock()
 380         if kc.gatewayStack != nil {
 381                 return kc.gatewayStack
 382         }
 383         var cachedir string
 384         if os.Geteuid() == 0 {
 385                 cachedir = rootCacheDir
 386                 makedirs("/", cachedir)
 387         } else {
 388                 home := "/" + os.Getenv("HOME")
 389                 makedirs(home, userCacheDir)
 390                 cachedir = filepath.Join(home, userCacheDir)
 391         }
 392         backend := &keepViaHTTP{kc}
 393         if kc.DiskCacheSize == DiskCacheDisabled {
 394                 kc.gatewayStack = backend
 395         } else {
 396                 kc.gatewayStack = &arvados.DiskCache{
 397                         Dir:         cachedir,
 398                         MaxSize:     kc.DiskCacheSize,
 399                         KeepGateway: backend,
 400                 }
 401         }
 402         return kc.gatewayStack
 403 }
 404
 405 // LocalLocator returns a locator equivalent to the one supplied, but
 406 // with a valid signature from the local cluster. If the given locator
 407 // already has a local signature, it is returned unchanged.
 408 func (kc *KeepClient) LocalLocator(locator string) (string, error) {
 409         return kc.upstreamGateway().LocalLocator(locator)
 410 }
 411
 412 // Get retrieves the specified block from the local cache or a backend
 413 // server. Returns a reader, the expected data length (or -1 if not
 414 // known), and an error.
 415 //
 416 // The third return value (formerly a source URL in previous versions)
 417 // is an empty string.
 418 //
 419 // If the block checksum does not match, the final Read() on the
 420 // reader returned by this method will return a BadChecksum error
 421 // instead of EOF.
 422 //
 423 // New code should use BlockRead and/or ReadAt instead of Get.
 424 func (kc *KeepClient) Get(locator string) (io.ReadCloser, int64, string, error) {
 425         loc, err := MakeLocator(locator)
 426         if err != nil {
 427                 return nil, 0, "", err
 428         }
 429         pr, pw := io.Pipe()
 430         go func() {
 431                 n, err := kc.BlockRead(context.Background(), arvados.BlockReadOptions{
 432                         Locator: locator,
 433                         WriteTo: pw,
 434                 })
 435                 if err != nil {
 436                         pw.CloseWithError(err)
 437                 } else if loc.Size >= 0 && n != loc.Size {
 438                         pw.CloseWithError(fmt.Errorf("expected block size %d but read %d bytes", loc.Size, n))
 439                 } else {
 440                         pw.Close()
 441                 }
 442         }()
 443         // Wait for the first byte to arrive, so that, if there's an
 444         // error before we receive any data, we can return the error
 445         // directly, instead of indirectly via a reader that returns
 446         // an error.
 447         bufr := bufio.NewReader(pr)
 448         _, err = bufr.Peek(1)
 449         if err != nil && err != io.EOF {
 450                 pr.CloseWithError(err)
 451                 return nil, 0, "", err
 452         }
 453         if err == io.EOF && (loc.Size == 0 || loc.Hash == "d41d8cd98f00b204e9800998ecf8427e") {
 454                 // In the special case of the zero-length block, EOF
 455                 // error from Peek() is normal.
 456                 return pr, 0, "", nil
 457         }
 458         return struct {
 459                 io.Reader
 460                 io.Closer
 461         }{
 462                 Reader: bufr,
 463                 Closer: pr,
 464         }, int64(loc.Size), "", err
 465 }
 466
 467 // BlockRead retrieves a block from the cache if it's present, otherwise
 468 // from the network.
 469 func (kc *KeepClient) BlockRead(ctx context.Context, opts arvados.BlockReadOptions) (int, error) {
 470         return kc.upstreamGateway().BlockRead(ctx, opts)
 471 }
 472
 473 // ReadAt retrieves a portion of block from the cache if it's
 474 // present, otherwise from the network.
 475 func (kc *KeepClient) ReadAt(locator string, p []byte, off int) (int, error) {
 476         return kc.upstreamGateway().ReadAt(locator, p, off)
 477 }
 478
 479 // BlockWrite writes a full block to upstream servers and saves a copy
 480 // in the local cache.
 481 func (kc *KeepClient) BlockWrite(ctx context.Context, req arvados.BlockWriteOptions) (arvados.BlockWriteResponse, error) {
 482         return kc.upstreamGateway().BlockWrite(ctx, req)
 483 }
 484
 485 // Ask verifies that a block with the given hash is available and
 486 // readable, according to at least one Keep service. Unlike Get, it
 487 // does not retrieve the data or verify that the data content matches
 488 // the hash specified by the locator.
 489 //
 490 // Returns the data size (content length) reported by the Keep service
 491 // and the URI reporting the data size.
 492 func (kc *KeepClient) Ask(locator string) (int64, string, error) {
 493         _, size, url, _, err := kc.getOrHead("HEAD", locator, nil)
 494         return size, url, err
 495 }
 496
 497 // GetIndex retrieves a list of blocks stored on the given server whose hashes
 498 // begin with the given prefix. The returned reader will return an error (other
 499 // than EOF) if the complete index cannot be retrieved.
 500 //
 501 // This is meant to be used only by system components and admin tools.
 502 // It will return an error unless the client is using a "data manager token"
 503 // recognized by the Keep services.
 504 func (kc *KeepClient) GetIndex(keepServiceUUID, prefix string) (io.Reader, error) {
 505         url := kc.LocalRoots()[keepServiceUUID]
 506         if url == "" {
 507                 return nil, ErrNoSuchKeepServer
 508         }
 509
 510         url += "/index"
 511         if prefix != "" {
 512                 url += "/" + prefix
 513         }
 514
 515         req, err := http.NewRequest("GET", url, nil)
 516         if err != nil {
 517                 return nil, err
 518         }
 519
 520         req.Header.Add("Authorization", "OAuth2 "+kc.Arvados.ApiToken)
 521         req.Header.Set("X-Request-Id", kc.getRequestID())
 522         resp, err := kc.httpClient().Do(req)
 523         if err != nil {
 524                 return nil, err
 525         }
 526
 527         defer resp.Body.Close()
 528
 529         if resp.StatusCode != http.StatusOK {
 530                 return nil, fmt.Errorf("Got http status code: %d", resp.StatusCode)
 531         }
 532
 533         var respBody []byte
 534         respBody, err = ioutil.ReadAll(resp.Body)
 535         if err != nil {
 536                 return nil, err
 537         }
 538
 539         // Got index; verify that it is complete
 540         // The response should be "\n" if no locators matched the prefix
 541         // Else, it should be a list of locators followed by a blank line
 542         if !bytes.Equal(respBody, []byte("\n")) && !bytes.HasSuffix(respBody, []byte("\n\n")) {
 543                 return nil, ErrIncompleteIndex
 544         }
 545
 546         // Got complete index; strip the trailing newline and send
 547         return bytes.NewReader(respBody[0 : len(respBody)-1]), nil
 548 }
 549
 550 // LocalRoots returns the map of local (i.e., disk and proxy) Keep
 551 // services: uuid -> baseURI.
 552 func (kc *KeepClient) LocalRoots() map[string]string {
 553         kc.discoverServices()
 554         kc.lock.RLock()
 555         defer kc.lock.RUnlock()
 556         return kc.localRoots
 557 }
 558
 559 // GatewayRoots returns the map of Keep remote gateway services:
 560 // uuid -> baseURI.
 561 func (kc *KeepClient) GatewayRoots() map[string]string {
 562         kc.discoverServices()
 563         kc.lock.RLock()
 564         defer kc.lock.RUnlock()
 565         return kc.gatewayRoots
 566 }
 567
 568 // WritableLocalRoots returns the map of writable local Keep services:
 569 // uuid -> baseURI.
 570 func (kc *KeepClient) WritableLocalRoots() map[string]string {
 571         kc.discoverServices()
 572         kc.lock.RLock()
 573         defer kc.lock.RUnlock()
 574         return kc.writableLocalRoots
 575 }
 576
 577 // SetServiceRoots disables service discovery and updates the
 578 // localRoots and gatewayRoots maps, without disrupting operations
 579 // that are already in progress.
 580 //
 581 // The supplied maps must not be modified after calling
 582 // SetServiceRoots.
 583 func (kc *KeepClient) SetServiceRoots(locals, writables, gateways map[string]string) {
 584         kc.disableDiscovery = true
 585         kc.setServiceRoots(locals, writables, gateways)
 586 }
 587
 588 func (kc *KeepClient) setServiceRoots(locals, writables, gateways map[string]string) {
 589         kc.lock.Lock()
 590         defer kc.lock.Unlock()
 591         kc.localRoots = locals
 592         kc.writableLocalRoots = writables
 593         kc.gatewayRoots = gateways
 594 }
 595
 596 // getSortedRoots returns a list of base URIs of Keep services, in the
 597 // order they should be attempted in order to retrieve content for the
 598 // given locator.
 599 func (kc *KeepClient) getSortedRoots(locator string) []string {
 600         var found []string
 601         for _, hint := range strings.Split(locator, "+") {
 602                 if len(hint) < 7 || hint[0:2] != "K@" {
 603                         // Not a service hint.
 604                         continue
 605                 }
 606                 if len(hint) == 7 {
 607                         // +K@abcde means fetch from proxy at
 608                         // keep.abcde.arvadosapi.com
 609                         found = append(found, "https://keep."+hint[2:]+".arvadosapi.com")
 610                 } else if len(hint) == 29 {
 611                         // +K@abcde-abcde-abcdeabcdeabcde means fetch
 612                         // from gateway with given uuid
 613                         if gwURI, ok := kc.GatewayRoots()[hint[2:]]; ok {
 614                                 found = append(found, gwURI)
 615                         }
 616                         // else this hint is no use to us; carry on.
 617                 }
 618         }
 619         // After trying all usable service hints, fall back to local roots.
 620         found = append(found, NewRootSorter(kc.LocalRoots(), locator[0:32]).GetSortedRoots()...)
 621         return found
 622 }
 623
 624 func (kc *KeepClient) SetStorageClasses(sc []string) {
 625         // make a copy so the caller can't mess with it.
 626         kc.StorageClasses = append([]string{}, sc...)
 627 }
 628
 629 var (
 630         // There are four global http.Client objects for the four
 631         // possible permutations of TLS behavior (verify/skip-verify)
 632         // and timeout settings (proxy/non-proxy).
 633         defaultClient = map[bool]map[bool]HTTPClient{
 634                 // defaultClient[false] is used for verified TLS reqs
 635                 false: {},
 636                 // defaultClient[true] is used for unverified
 637                 // (insecure) TLS reqs
 638                 true: {},
 639         }
 640         defaultClientMtx sync.Mutex
 641 )
 642
 643 // httpClient returns the HTTPClient field if it's not nil, otherwise
 644 // whichever of the four global http.Client objects is suitable for
 645 // the current environment (i.e., TLS verification on/off, keep
 646 // services are/aren't proxies).
 647 func (kc *KeepClient) httpClient() HTTPClient {
 648         if kc.HTTPClient != nil {
 649                 return kc.HTTPClient
 650         }
 651         defaultClientMtx.Lock()
 652         defer defaultClientMtx.Unlock()
 653         if c, ok := defaultClient[kc.Arvados.ApiInsecure][kc.foundNonDiskSvc]; ok {
 654                 return c
 655         }
 656
 657         var requestTimeout, connectTimeout, keepAlive, tlsTimeout time.Duration
 658         if kc.foundNonDiskSvc {
 659                 // Use longer timeouts when connecting to a proxy,
 660                 // because this usually means the intervening network
 661                 // is slower.
 662                 requestTimeout = DefaultProxyRequestTimeout
 663                 connectTimeout = DefaultProxyConnectTimeout
 664                 tlsTimeout = DefaultProxyTLSHandshakeTimeout
 665                 keepAlive = DefaultProxyKeepAlive
 666         } else {
 667                 requestTimeout = DefaultRequestTimeout
 668                 connectTimeout = DefaultConnectTimeout
 669                 tlsTimeout = DefaultTLSHandshakeTimeout
 670                 keepAlive = DefaultKeepAlive
 671         }
 672
 673         c := &http.Client{
 674                 Timeout: requestTimeout,
 675                 // It's not safe to copy *http.DefaultTransport
 676                 // because it has a mutex (which might be locked)
 677                 // protecting a private map (which might not be nil).
 678                 // So we build our own, using the Go 1.12 default
 679                 // values, ignoring any changes the application has
 680                 // made to http.DefaultTransport.
 681                 Transport: &http.Transport{
 682                         DialContext: (&net.Dialer{
 683                                 Timeout:   connectTimeout,
 684                                 KeepAlive: keepAlive,
 685                                 DualStack: true,
 686                         }).DialContext,
 687                         MaxIdleConns:          100,
 688                         IdleConnTimeout:       90 * time.Second,
 689                         TLSHandshakeTimeout:   tlsTimeout,
 690                         ExpectContinueTimeout: 1 * time.Second,
 691                         TLSClientConfig:       arvadosclient.MakeTLSConfig(kc.Arvados.ApiInsecure),
 692                 },
 693         }
 694         defaultClient[kc.Arvados.ApiInsecure][kc.foundNonDiskSvc] = c
 695         return c
 696 }
 697
 698 var reqIDGen = httpserver.IDGenerator{Prefix: "req-"}
 699
 700 func (kc *KeepClient) getRequestID() string {
 701         if kc.RequestID != "" {
 702                 return kc.RequestID
 703         }
 704         return reqIDGen.Next()
 705 }
 706
 707 type Locator struct {
 708         Hash  string
 709         Size  int      // -1 if data size is not known
 710         Hints []string // Including the size hint, if any
 711 }
 712
 713 func (loc *Locator) String() string {
 714         s := loc.Hash
 715         if len(loc.Hints) > 0 {
 716                 s = s + "+" + strings.Join(loc.Hints, "+")
 717         }
 718         return s
 719 }
 720
 721 var locatorMatcher = regexp.MustCompile("^([0-9a-f]{32})([+](.*))?$")
 722
 723 func MakeLocator(path string) (*Locator, error) {
 724         sm := locatorMatcher.FindStringSubmatch(path)
 725         if sm == nil {
 726                 return nil, InvalidLocatorError
 727         }
 728         loc := Locator{Hash: sm[1], Size: -1}
 729         if sm[2] != "" {
 730                 loc.Hints = strings.Split(sm[3], "+")
 731         } else {
 732                 loc.Hints = []string{}
 733         }
 734         if len(loc.Hints) > 0 {
 735                 if size, err := strconv.Atoi(loc.Hints[0]); err == nil {
 736                         loc.Size = size
 737                 }
 738         }
 739         return &loc, nil
 740 }