X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/ef42f0396f5b51fb8f87b2f7e605d50f32f256b4..0a46b75ac1e96a6f5ce3ae797eb4306b352fab36:/sdk/go/keepclient/keepclient.go diff --git a/sdk/go/keepclient/keepclient.go b/sdk/go/keepclient/keepclient.go index 4cd9f544f7..26aa7177e0 100644 --- a/sdk/go/keepclient/keepclient.go +++ b/sdk/go/keepclient/keepclient.go @@ -11,7 +11,6 @@ import ( "git.curoverse.com/arvados.git/sdk/go/streamer" "io" "io/ioutil" - "log" "net/http" "regexp" "strconv" @@ -22,7 +21,33 @@ import ( // A Keep "block" is 64MB. const BLOCKSIZE = 64 * 1024 * 1024 -var BlockNotFound = errors.New("Block not found") +// Error interface with an error and boolean indicating whether the error is temporary +type Error interface { + error + Temporary() bool +} + +// multipleResponseError is of type Error +type multipleResponseError struct { + error + isTemp bool +} + +func (e *multipleResponseError) Temporary() bool { + return e.isTemp +} + +// BlockNotFound is a multipleResponseError where isTemp is false +var BlockNotFound = &ErrNotFound{multipleResponseError{ + error: errors.New("Block not found"), + isTemp: false, +}} + +// ErrNotFound is a multipleResponseError where isTemp can be true or false +type ErrNotFound struct { + multipleResponseError +} + var InsufficientReplicasError = errors.New("Could not write sufficient replicas") var OversizeBlockError = errors.New("Exceeded maximum block size (" + strconv.Itoa(BLOCKSIZE) + ")") var MissingArvadosApiHost = errors.New("Missing required environment variable ARVADOS_API_HOST") @@ -42,15 +67,18 @@ const X_Keep_Replicas_Stored = "X-Keep-Replicas-Stored" type KeepClient struct { Arvados *arvadosclient.ArvadosClient Want_replicas int - Using_proxy bool localRoots *map[string]string writableLocalRoots *map[string]string gatewayRoots *map[string]string lock sync.RWMutex Client *http.Client + Retries int // set to 1 if all writable services are of disk type, otherwise 0 replicasPerService int + + // Any non-disk typed services found in the list of keepservers? + foundNonDiskSvc bool } // MakeKeepClient creates a new KeepClient by contacting the API server to discover Keep servers. @@ -62,12 +90,21 @@ func MakeKeepClient(arv *arvadosclient.ArvadosClient) (*KeepClient, error) { // New func creates a new KeepClient struct. // This func does not discover keep servers. It is the caller's responsibility. func New(arv *arvadosclient.ArvadosClient) *KeepClient { + defaultReplicationLevel := 2 + value, err := arv.Discovery("defaultCollectionReplication") + if err == nil { + v, ok := value.(float64) + if ok && v > 0 { + defaultReplicationLevel = int(v) + } + } + kc := &KeepClient{ Arvados: arv, - Want_replicas: 2, - Using_proxy: false, + Want_replicas: defaultReplicationLevel, Client: &http.Client{Transport: &http.Transport{ TLSClientConfig: &tls.Config{InsecureSkipVerify: arv.ApiInsecure}}}, + Retries: 2, } return kc } @@ -130,6 +167,85 @@ func (kc *KeepClient) PutR(r io.Reader) (locator string, replicas int, err error } } +func (kc *KeepClient) getOrHead(method string, locator string) (io.ReadCloser, int64, string, error) { + var errs []string + + tries_remaining := 1 + kc.Retries + + serversToTry := kc.getSortedRoots(locator) + + numServers := len(serversToTry) + count404 := 0 + + var retryList []string + + for tries_remaining > 0 { + tries_remaining -= 1 + retryList = nil + + for _, host := range serversToTry { + url := host + "/" + locator + + req, err := http.NewRequest(method, url, nil) + if err != nil { + errs = append(errs, fmt.Sprintf("%s: %v", url, err)) + continue + } + req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", kc.Arvados.ApiToken)) + resp, err := kc.Client.Do(req) + if err != nil { + // Probably a network error, may be transient, + // can try again. + errs = append(errs, fmt.Sprintf("%s: %v", url, err)) + retryList = append(retryList, host) + } else if resp.StatusCode != http.StatusOK { + var respbody []byte + respbody, _ = ioutil.ReadAll(&io.LimitedReader{R: resp.Body, N: 4096}) + resp.Body.Close() + errs = append(errs, fmt.Sprintf("%s: HTTP %d %q", + url, resp.StatusCode, bytes.TrimSpace(respbody))) + + if resp.StatusCode == 408 || + resp.StatusCode == 429 || + resp.StatusCode >= 500 { + // Timeout, too many requests, or other + // server side failure, transient + // error, can try again. + retryList = append(retryList, host) + } else if resp.StatusCode == 404 { + count404++ + } + } else { + // Success. + if method == "GET" { + return HashCheckingReader{ + Reader: resp.Body, + Hash: md5.New(), + Check: locator[0:32], + }, resp.ContentLength, url, nil + } else { + resp.Body.Close() + return nil, resp.ContentLength, url, nil + } + } + + } + serversToTry = retryList + } + DebugPrintf("DEBUG: %s %s failed: %v", method, locator, errs) + + var err error + if count404 == numServers { + err = BlockNotFound + } else { + err = &ErrNotFound{multipleResponseError{ + error: fmt.Errorf("%s %s failed: %v", method, locator, errs), + isTemp: len(serversToTry) > 0, + }} + } + return nil, 0, "", err +} + // 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. @@ -138,34 +254,7 @@ func (kc *KeepClient) PutR(r io.Reader) (locator string, replicas int, err error // reader returned by this method will return a BadChecksum error // instead of EOF. func (kc *KeepClient) Get(locator string) (io.ReadCloser, int64, string, error) { - var errs []string - for _, host := range kc.getSortedRoots(locator) { - url := host + "/" + locator - req, err := http.NewRequest("GET", url, nil) - if err != nil { - errs = append(errs, fmt.Sprintf("%s: %v", url, err)) - continue - } - req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", kc.Arvados.ApiToken)) - resp, err := kc.Client.Do(req) - if err != nil { - errs = append(errs, fmt.Sprintf("%s: %v", url, err)) - continue - } else if resp.StatusCode != http.StatusOK { - respbody, _ := ioutil.ReadAll(&io.LimitedReader{resp.Body, 4096}) - resp.Body.Close() - errs = append(errs, fmt.Sprintf("%s: HTTP %d %q", - url, resp.StatusCode, bytes.TrimSpace(respbody))) - continue - } - return HashCheckingReader{ - Reader: resp.Body, - Hash: md5.New(), - Check: locator[0:32], - }, resp.ContentLength, url, nil - } - log.Printf("DEBUG: GET %s failed: %v", locator, errs) - return nil, 0, "", BlockNotFound + return kc.getOrHead("GET", locator) } // Ask() verifies that a block with the given hash is available and @@ -176,18 +265,8 @@ func (kc *KeepClient) Get(locator string) (io.ReadCloser, int64, string, 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) { - for _, host := range kc.getSortedRoots(locator) { - url := host + "/" + locator - req, err := http.NewRequest("HEAD", url, nil) - if err != nil { - continue - } - req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", kc.Arvados.ApiToken)) - if resp, err := kc.Client.Do(req); err == nil && resp.StatusCode == http.StatusOK { - return resp.ContentLength, url, nil - } - } - return 0, "", BlockNotFound + _, size, url, err := kc.getOrHead("HEAD", locator) + return size, url, err } // GetIndex retrieves a list of blocks stored on the given server whose hashes