//
// 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"
"sync"
"time"
- "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
- "git.curoverse.com/arvados.git/sdk/go/asyncbuf"
- "git.curoverse.com/arvados.git/sdk/go/httpserver"
+ "git.arvados.org/arvados.git/sdk/go/arvados"
+ "git.arvados.org/arvados.git/sdk/go/arvadosclient"
+ "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 (
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")
// 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
- localRoots map[string]string
- writableLocalRoots map[string]string
- gatewayRoots map[string]string
- lock sync.RWMutex
- HTTPClient HTTPClient
- Retries int
- BlockCache *BlockCache
- RequestID string
- StorageClasses []string
+ Arvados *arvadosclient.ArvadosClient
+ Want_replicas int
+ localRoots map[string]string
+ writableLocalRoots map[string]string
+ gatewayRoots map[string]string
+ lock sync.RWMutex
+ HTTPClient HTTPClient
+ Retries int
+ BlockCache *BlockCache
+ RequestID string
+ StorageClasses []string
+ DefaultStorageClasses []string // Set by cluster's exported config
// set to 1 if all writable services are of disk type, otherwise 0
replicasPerService int
disableDiscovery bool
}
-// MakeKeepClient creates a new KeepClient, calls
+func (kc *KeepClient) loadDefaultClasses() error {
+ scData, err := kc.Arvados.ClusterConfig("StorageClasses")
+ if err != nil {
+ return err
+ }
+ classes := scData.(map[string]interface{})
+ for scName := range classes {
+ scConf, _ := classes[scName].(map[string]interface{})
+ isDefault, ok := scConf["Default"].(bool)
+ if ok && isDefault {
+ kc.DefaultStorageClasses = append(kc.DefaultStorageClasses, scName)
+ }
+ }
+ return nil
+}
+
+// MakeKeepClient creates a new KeepClient, loads default storage classes, calls
// DiscoverKeepServices(), and returns when the client is ready to
// use.
func MakeKeepClient(arv *arvadosclient.ArvadosClient) (*KeepClient, error) {
defaultReplicationLevel = int(v)
}
}
- return &KeepClient{
+ kc := &KeepClient{
Arvados: arv,
Want_replicas: defaultReplicationLevel,
Retries: 2,
}
+ err = kc.loadDefaultClasses()
+ if err != nil {
+ DebugPrintf("DEBUG: Unable to load the default storage classes cluster config")
+ }
+ return kc
}
-// 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
// 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
//
// 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
//
// 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, header http.Header) (io.ReadCloser, int64, string, http.Header, error) {
var errs []string
- tries_remaining := 1 + kc.Retries
+ triesRemaining := 1 + kc.Retries
serversToTry := kc.getSortedRoots(locator)
var retryList []string
- for tries_remaining > 0 {
- tries_remaining -= 1
+ for triesRemaining > 0 {
+ triesRemaining--
retryList = nil
for _, host := range serversToTry {
Hash: md5.New(),
Check: locator[0:32],
}, expectLength, url, resp.Header, nil
- } else {
- resp.Body.Close()
- return nil, expectLength, url, resp.Header, nil
}
+ resp.Body.Close()
+ return nil, expectLength, url, resp.Header, nil
}
serversToTry = retryList
}
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.
//
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.
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()
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()
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()
func (kc *KeepClient) cache() *BlockCache {
if kc.BlockCache != nil {
return kc.BlockCache
- } else {
- return DefaultBlockCache
}
+ return DefaultBlockCache
}
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)
// 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.10 default
+ // 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{
MaxIdleConns: 100,
IdleConnTimeout: 90 * time.Second,
TLSHandshakeTimeout: tlsTimeout,
- ExpectContinueTimeout: time.Second,
+ ExpectContinueTimeout: 1 * time.Second,
TLSClientConfig: arvadosclient.MakeTLSConfig(kc.Arvados.ApiInsecure),
},
}
func (kc *KeepClient) getRequestID() string {
if kc.RequestID != "" {
return kc.RequestID
- } else {
- return reqIDGen.Next()
}
+ return reqIDGen.Next()
}
type Locator struct {