X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/fa785db309d7b53905d327d6bfcab6445537a75f..7af0535d3b0d7960152b06b7211c26bfd7b208cb:/sdk/go/keepclient/keepclient.go diff --git a/sdk/go/keepclient/keepclient.go b/sdk/go/keepclient/keepclient.go index e65baa3b3f..3bc6f4afcd 100644 --- a/sdk/go/keepclient/keepclient.go +++ b/sdk/go/keepclient/keepclient.go @@ -2,11 +2,13 @@ // // 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 ( "bytes" + "context" "crypto/md5" "errors" "fmt" @@ -20,12 +22,12 @@ import ( "sync" "time" + "git.arvados.org/arvados.git/sdk/go/arvados" "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 ( @@ -67,11 +69,11 @@ type ErrNotFound struct { multipleResponseError } -type InsufficientReplicasError error +type InsufficientReplicasError struct{ error } -type OversizeBlockError error +type OversizeBlockError struct{ error } -var ErrOversizeBlock = OversizeBlockError(errors.New("Exceeded maximum block size (" + strconv.Itoa(BLOCKSIZE) + ")")) +var ErrOversizeBlock = OversizeBlockError{error: errors.New("Exceeded maximum block size (" + strconv.Itoa(BLOCKSIZE) + ")")} var MissingArvadosApiHost = errors.New("Missing required environment variable ARVADOS_API_HOST") var MissingArvadosApiToken = errors.New("Missing required environment variable ARVADOS_API_TOKEN") var InvalidLocatorError = errors.New("Invalid locator") @@ -82,14 +84,18 @@ 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" + XKeepReplicasStored = "X-Keep-Replicas-Stored" + XKeepStorageClasses = "X-Keep-Storage-Classes" + XKeepStorageClassesConfirmed = "X-Keep-Storage-Classes-Confirmed" +) 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 @@ -139,7 +145,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 @@ -148,23 +154,12 @@ func New(arv *arvadosclient.ArvadosClient) *KeepClient { // Returns an InsufficientReplicasError if 0 <= replicas < // kc.Wants_replicas. func (kc *KeepClient) PutHR(hash string, r io.Reader, dataBytes int64) (string, int, error) { - // Buffer for reads from 'r' - var bufsize int - if dataBytes > 0 { - if dataBytes > BLOCKSIZE { - return "", 0, ErrOversizeBlock - } - bufsize = int(dataBytes) - } else { - bufsize = BLOCKSIZE - } - - buf := asyncbuf.NewBuffer(make([]byte, 0, bufsize)) - go func() { - _, err := io.Copy(buf, HashCheckingReader{r, md5.New(), hash}) - buf.CloseWithError(err) - }() - return kc.putReplicas(hash, buf.NewReader, dataBytes) + resp, err := kc.BlockWrite(context.Background(), arvados.BlockWriteOptions{ + Hash: hash, + Reader: r, + DataSize: int(dataBytes), + }) + return resp.Locator, resp.Replicas, err } // PutHB writes a block to Keep. The hash of the bytes is given in @@ -172,16 +167,21 @@ func (kc *KeepClient) PutHR(hash string, r io.Reader, dataBytes int64) (string, // // Return values are the same as for PutHR. func (kc *KeepClient) PutHB(hash string, buf []byte) (string, int, error) { - newReader := func() io.Reader { return bytes.NewBuffer(buf) } - return kc.putReplicas(hash, newReader, int64(len(buf))) + resp, err := kc.BlockWrite(context.Background(), arvados.BlockWriteOptions{ + Hash: hash, + Data: buf, + }) + return resp.Locator, resp.Replicas, err } // PutB writes a block to Keep. It computes the hash itself. // // Return values are the same as for PutHR. func (kc *KeepClient) PutB(buffer []byte) (string, int, error) { - hash := fmt.Sprintf("%x", md5.Sum(buffer)) - return kc.PutHB(hash, buffer) + resp, err := kc.BlockWrite(context.Background(), arvados.BlockWriteOptions{ + Data: buffer, + }) + return resp.Locator, resp.Replicas, err } // PutR writes a block to Keep. It first reads all data from r into a buffer @@ -216,7 +216,7 @@ func (kc *KeepClient) getOrHead(method string, locator string, header http.Heade var errs []string - tries_remaining := 1 + kc.Retries + triesRemaining := 1 + kc.Retries serversToTry := kc.getSortedRoots(locator) @@ -225,8 +225,8 @@ func (kc *KeepClient) getOrHead(method string, locator string, header http.Heade var retryList []string - for tries_remaining > 0 { - tries_remaining -= 1 + for triesRemaining > 0 { + triesRemaining-- retryList = nil for _, host := range serversToTry { @@ -332,7 +332,7 @@ func (kc *KeepClient) LocalLocator(locator string) (string, error) { return loc, nil } -// Get() retrieves a block, given a locator. Returns a reader, the +// 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. // @@ -344,13 +344,13 @@ func (kc *KeepClient) Get(locator string) (io.ReadCloser, int64, string, error) 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. @@ -415,7 +415,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() @@ -424,7 +424,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() @@ -433,7 +433,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() @@ -500,6 +500,11 @@ func (kc *KeepClient) ClearBlockCache() { kc.cache().Clear() } +func (kc *KeepClient) SetStorageClasses(sc []string) { + // make a copy so the caller can't mess with it. + kc.StorageClasses = append([]string{}, sc...) +} + var ( // There are four global http.Client objects for the four // possible permutations of TLS behavior (verify/skip-verify)