X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/e26e7f413351efae4a2ec679cc2e234dc1a5020c..6bf9e1a4b5640f3cdd057810f0c9b8a945bb88bd:/sdk/go/keepclient/keepclient.go diff --git a/sdk/go/keepclient/keepclient.go b/sdk/go/keepclient/keepclient.go index 620bdbec4e..21913ff967 100644 --- a/sdk/go/keepclient/keepclient.go +++ b/sdk/go/keepclient/keepclient.go @@ -2,7 +2,8 @@ // // SPDX-License-Identifier: Apache-2.0 -/* Provides low-level Get/Put primitives for accessing Arvados Keep blocks. */ +// Package keepclient provides low-level Get/Put primitives for accessing +// Arvados Keep blocks. package keepclient import ( @@ -20,11 +21,12 @@ import ( "sync" "time" - "git.curoverse.com/arvados.git/sdk/go/arvadosclient" - "git.curoverse.com/arvados.git/sdk/go/asyncbuf" + "git.arvados.org/arvados.git/sdk/go/arvadosclient" + "git.arvados.org/arvados.git/sdk/go/asyncbuf" + "git.arvados.org/arvados.git/sdk/go/httpserver" ) -// A Keep "block" is 64MB. +// BLOCKSIZE defines the length of a Keep "block", which is 64MB. const BLOCKSIZE = 64 * 1024 * 1024 var ( @@ -81,14 +83,14 @@ var ErrNoSuchKeepServer = errors.New("No keep server matching the given UUID is // ErrIncompleteIndex is returned when the Index response does not end with a new empty line var ErrIncompleteIndex = errors.New("Got incomplete index") -const X_Keep_Desired_Replicas = "X-Keep-Desired-Replicas" -const X_Keep_Replicas_Stored = "X-Keep-Replicas-Stored" +const XKeepDesiredReplicas = "X-Keep-Desired-Replicas" +const XKeepReplicasStored = "X-Keep-Replicas-Stored" type HTTPClient interface { Do(*http.Request) (*http.Response, error) } -// Information about Arvados and Keep servers. +// KeepClient holds information about Arvados and Keep servers. type KeepClient struct { Arvados *arvadosclient.ArvadosClient Want_replicas int @@ -99,6 +101,8 @@ type KeepClient struct { HTTPClient HTTPClient Retries int BlockCache *BlockCache + RequestID string + StorageClasses []string // set to 1 if all writable services are of disk type, otherwise 0 replicasPerService int @@ -136,7 +140,7 @@ func New(arv *arvadosclient.ArvadosClient) *KeepClient { } } -// Put a block given the block hash, a reader, and the number of bytes +// PutHR puts a block given the block hash, a reader, and the number of bytes // to read from the reader (which must be between 0 and BLOCKSIZE). // // Returns the locator for the written block, the number of replicas @@ -188,18 +192,20 @@ func (kc *KeepClient) PutB(buffer []byte) (string, int, error) { // // If the block hash and data size are known, PutHR is more efficient. func (kc *KeepClient) PutR(r io.Reader) (locator string, replicas int, err error) { - if buffer, err := ioutil.ReadAll(r); err != nil { + buffer, err := ioutil.ReadAll(r) + if err != nil { return "", 0, err - } else { - return kc.PutB(buffer) } + return kc.PutB(buffer) } -func (kc *KeepClient) getOrHead(method string, locator string) (io.ReadCloser, int64, string, error) { +func (kc *KeepClient) getOrHead(method string, locator string, header http.Header) (io.ReadCloser, int64, string, http.Header, error) { if strings.HasPrefix(locator, "d41d8cd98f00b204e9800998ecf8427e+0") { - return ioutil.NopCloser(bytes.NewReader(nil)), 0, "", nil + return ioutil.NopCloser(bytes.NewReader(nil)), 0, "", nil, nil } + reqid := kc.getRequestID() + var expectLength int64 if parts := strings.SplitN(locator, "+", 3); len(parts) < 2 { expectLength = -1 @@ -211,7 +217,7 @@ func (kc *KeepClient) getOrHead(method string, locator string) (io.ReadCloser, i var errs []string - tries_remaining := 1 + kc.Retries + triesRemaining := 1 + kc.Retries serversToTry := kc.getSortedRoots(locator) @@ -220,8 +226,8 @@ func (kc *KeepClient) getOrHead(method string, locator string) (io.ReadCloser, i var retryList []string - for tries_remaining > 0 { - tries_remaining -= 1 + for triesRemaining > 0 { + triesRemaining-- retryList = nil for _, host := range serversToTry { @@ -232,7 +238,15 @@ func (kc *KeepClient) getOrHead(method string, locator string) (io.ReadCloser, i errs = append(errs, fmt.Sprintf("%s: %v", url, err)) continue } - req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", kc.Arvados.ApiToken)) + for k, v := range header { + req.Header[k] = append([]string(nil), v...) + } + if req.Header.Get("Authorization") == "" { + req.Header.Set("Authorization", "OAuth2 "+kc.Arvados.ApiToken) + } + if req.Header.Get("X-Request-Id") == "" { + req.Header.Set("X-Request-Id", reqid) + } resp, err := kc.httpClient().Do(req) if err != nil { // Probably a network error, may be transient, @@ -263,12 +277,12 @@ func (kc *KeepClient) getOrHead(method string, locator string) (io.ReadCloser, i if expectLength < 0 { if resp.ContentLength < 0 { resp.Body.Close() - return nil, 0, "", fmt.Errorf("error reading %q: no size hint, no Content-Length header in response", locator) + return nil, 0, "", nil, fmt.Errorf("error reading %q: no size hint, no Content-Length header in response", locator) } expectLength = resp.ContentLength } else if resp.ContentLength >= 0 && expectLength != resp.ContentLength { resp.Body.Close() - return nil, 0, "", fmt.Errorf("error reading %q: size hint %d != Content-Length %d", locator, expectLength, resp.ContentLength) + return nil, 0, "", nil, fmt.Errorf("error reading %q: size hint %d != Content-Length %d", locator, expectLength, resp.ContentLength) } // Success if method == "GET" { @@ -276,11 +290,10 @@ func (kc *KeepClient) getOrHead(method string, locator string) (io.ReadCloser, i Reader: resp.Body, Hash: md5.New(), Check: locator[0:32], - }, expectLength, url, nil - } else { - resp.Body.Close() - return nil, expectLength, url, nil + }, expectLength, url, resp.Header, nil } + resp.Body.Close() + return nil, expectLength, url, resp.Header, nil } serversToTry = retryList } @@ -295,10 +308,32 @@ func (kc *KeepClient) getOrHead(method string, locator string) (io.ReadCloser, i isTemp: len(serversToTry) > 0, }} } - return nil, 0, "", err + return nil, 0, "", nil, err } -// Get() retrieves a block, given a locator. Returns a reader, the +// LocalLocator returns a locator equivalent to the one supplied, but +// with a valid signature from the local cluster. If the given locator +// already has a local signature, it is returned unchanged. +func (kc *KeepClient) LocalLocator(locator string) (string, error) { + if !strings.Contains(locator, "+R") { + // Either it has +A, or it's unsigned and we assume + // it's a local locator on a site with signatures + // disabled. + return locator, nil + } + sighdr := fmt.Sprintf("local, time=%s", time.Now().UTC().Format(time.RFC3339)) + _, _, url, hdr, err := kc.getOrHead("HEAD", locator, http.Header{"X-Keep-Signature": []string{sighdr}}) + if err != nil { + return "", err + } + loc := hdr.Get("X-Keep-Locator") + if loc == "" { + return "", fmt.Errorf("missing X-Keep-Locator header in HEAD response from %s", url) + } + return loc, nil +} + +// Get retrieves a block, given a locator. Returns a reader, the // expected data length, the URL the block is being fetched from, and // an error. // @@ -306,16 +341,17 @@ func (kc *KeepClient) getOrHead(method string, locator string) (io.ReadCloser, i // reader returned by this method will return a BadChecksum error // instead of EOF. func (kc *KeepClient) Get(locator string) (io.ReadCloser, int64, string, error) { - return kc.getOrHead("GET", locator) + rdr, size, url, _, err := kc.getOrHead("GET", locator, nil) + return rdr, size, url, err } -// ReadAt() retrieves a portion of block from the cache if it's +// ReadAt retrieves a portion of block from the cache if it's // present, otherwise from the network. func (kc *KeepClient) ReadAt(locator string, p []byte, off int) (int, error) { return kc.cache().ReadAt(kc, locator, p, off) } -// Ask() verifies that a block with the given hash is available and +// Ask verifies that a block with the given hash is available and // readable, according to at least one Keep service. Unlike Get, it // does not retrieve the data or verify that the data content matches // the hash specified by the locator. @@ -323,7 +359,7 @@ func (kc *KeepClient) ReadAt(locator string, p []byte, off int) (int, error) { // Returns the data size (content length) reported by the Keep service // and the URI reporting the data size. func (kc *KeepClient) Ask(locator string) (int64, string, error) { - _, size, url, err := kc.getOrHead("HEAD", locator) + _, size, url, _, err := kc.getOrHead("HEAD", locator, nil) return size, url, err } @@ -350,7 +386,8 @@ func (kc *KeepClient) GetIndex(keepServiceUUID, prefix string) (io.Reader, error return nil, err } - req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", kc.Arvados.ApiToken)) + req.Header.Add("Authorization", "OAuth2 "+kc.Arvados.ApiToken) + req.Header.Set("X-Request-Id", kc.getRequestID()) resp, err := kc.httpClient().Do(req) if err != nil { return nil, err @@ -379,7 +416,7 @@ func (kc *KeepClient) GetIndex(keepServiceUUID, prefix string) (io.Reader, error return bytes.NewReader(respBody[0 : len(respBody)-1]), nil } -// LocalRoots() returns the map of local (i.e., disk and proxy) Keep +// LocalRoots returns the map of local (i.e., disk and proxy) Keep // services: uuid -> baseURI. func (kc *KeepClient) LocalRoots() map[string]string { kc.discoverServices() @@ -388,7 +425,7 @@ func (kc *KeepClient) LocalRoots() map[string]string { return kc.localRoots } -// GatewayRoots() returns the map of Keep remote gateway services: +// GatewayRoots returns the map of Keep remote gateway services: // uuid -> baseURI. func (kc *KeepClient) GatewayRoots() map[string]string { kc.discoverServices() @@ -397,7 +434,7 @@ func (kc *KeepClient) GatewayRoots() map[string]string { return kc.gatewayRoots } -// WritableLocalRoots() returns the map of writable local Keep services: +// WritableLocalRoots returns the map of writable local Keep services: // uuid -> baseURI. func (kc *KeepClient) WritableLocalRoots() map[string]string { kc.discoverServices() @@ -456,9 +493,8 @@ func (kc *KeepClient) getSortedRoots(locator string) []string { func (kc *KeepClient) cache() *BlockCache { if kc.BlockCache != nil { return kc.BlockCache - } else { - return DefaultBlockCache } + return DefaultBlockCache } func (kc *KeepClient) ClearBlockCache() { @@ -509,36 +545,40 @@ func (kc *KeepClient) httpClient() HTTPClient { keepAlive = DefaultKeepAlive } - transport, ok := http.DefaultTransport.(*http.Transport) - if ok { - copy := *transport - transport = © - } else { - // Evidently the application has replaced - // http.DefaultTransport with a different type, so we - // need to build our own from scratch using the Go 1.8 - // defaults. - transport = &http.Transport{ + c := &http.Client{ + Timeout: requestTimeout, + // It's not safe to copy *http.DefaultTransport + // because it has a mutex (which might be locked) + // protecting a private map (which might not be nil). + // So we build our own, using the Go 1.12 default + // values, ignoring any changes the application has + // made to http.DefaultTransport. + Transport: &http.Transport{ + DialContext: (&net.Dialer{ + Timeout: connectTimeout, + KeepAlive: keepAlive, + DualStack: true, + }).DialContext, MaxIdleConns: 100, IdleConnTimeout: 90 * time.Second, - ExpectContinueTimeout: time.Second, - } - } - transport.DialContext = (&net.Dialer{ - Timeout: connectTimeout, - KeepAlive: keepAlive, - DualStack: true, - }).DialContext - transport.TLSHandshakeTimeout = tlsTimeout - transport.TLSClientConfig = arvadosclient.MakeTLSConfig(kc.Arvados.ApiInsecure) - c := &http.Client{ - Timeout: requestTimeout, - Transport: transport, + TLSHandshakeTimeout: tlsTimeout, + ExpectContinueTimeout: 1 * time.Second, + TLSClientConfig: arvadosclient.MakeTLSConfig(kc.Arvados.ApiInsecure), + }, } defaultClient[kc.Arvados.ApiInsecure][kc.foundNonDiskSvc] = c return c } +var reqIDGen = httpserver.IDGenerator{Prefix: "req-"} + +func (kc *KeepClient) getRequestID() string { + if kc.RequestID != "" { + return kc.RequestID + } + return reqIDGen.Next() +} + type Locator struct { Hash string Size int // -1 if data size is not known