X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/dc16046dbfdca4a9c0d94971730d220b27e80620..8a6bd5b851e0e947b74710c4c2dd1ea59dce606a:/sdk/go/keepclient/support.go diff --git a/sdk/go/keepclient/support.go b/sdk/go/keepclient/support.go index 8d299815b2..142f1d2151 100644 --- a/sdk/go/keepclient/support.go +++ b/sdk/go/keepclient/support.go @@ -12,28 +12,16 @@ import ( "fmt" "io" "io/ioutil" - "log" + "math/rand" "net/http" - "os" "strconv" "strings" + "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" ) -// DebugPrintf emits debug messages. The easiest way to enable -// keepclient debug messages in your application is to assign -// log.Printf to DebugPrintf. -var DebugPrintf = func(string, ...interface{}) {} - -func init() { - if arvadosclient.StringBool(os.Getenv("ARVADOS_DEBUG")) { - DebugPrintf = log.Printf - } -} - type keepService struct { Uuid string `json:"uuid"` Hostname string `json:"service_host"` @@ -68,7 +56,7 @@ func (kc *KeepClient) uploadToKeepServer(host string, hash string, classesTodo [ var err error var url = fmt.Sprintf("%s/%s", host, hash) if req, err = http.NewRequest("PUT", url, nil); err != nil { - DebugPrintf("DEBUG: [%s] Error creating request PUT %v error: %v", reqid, url, err.Error()) + kc.debugf("[%s] Error creating request: PUT %s error: %s", reqid, url, err) uploadStatusChan <- uploadStatus{err, url, 0, 0, nil, ""} return } @@ -92,7 +80,7 @@ func (kc *KeepClient) uploadToKeepServer(host string, hash string, classesTodo [ var resp *http.Response if resp, err = kc.httpClient().Do(req); err != nil { - DebugPrintf("DEBUG: [%s] Upload failed %v error: %v", reqid, url, err.Error()) + kc.debugf("[%s] Upload failed: %s error: %s", reqid, url, err) uploadStatusChan <- uploadStatus{err, url, 0, 0, nil, err.Error()} return } @@ -104,7 +92,7 @@ func (kc *KeepClient) uploadToKeepServer(host string, hash string, classesTodo [ scc := resp.Header.Get(XKeepStorageClassesConfirmed) classesStored, err := parseStorageClassesConfirmedHeader(scc) if err != nil { - DebugPrintf("DEBUG: [%s] Ignoring invalid %s header %q: %s", reqid, XKeepStorageClassesConfirmed, scc, err) + kc.debugf("[%s] Ignoring invalid %s header %q: %s", reqid, XKeepStorageClassesConfirmed, scc, err) } defer resp.Body.Close() @@ -113,21 +101,21 @@ func (kc *KeepClient) uploadToKeepServer(host string, hash string, classesTodo [ respbody, err2 := ioutil.ReadAll(&io.LimitedReader{R: resp.Body, N: 4096}) response := strings.TrimSpace(string(respbody)) if err2 != nil && err2 != io.EOF { - DebugPrintf("DEBUG: [%s] Upload %v error: %v response: %v", reqid, url, err2.Error(), response) + kc.debugf("[%s] Upload %s error: %s response: %s", reqid, url, err2, response) uploadStatusChan <- uploadStatus{err2, url, resp.StatusCode, rep, classesStored, response} } else if resp.StatusCode == http.StatusOK { - DebugPrintf("DEBUG: [%s] Upload %v success", reqid, url) + kc.debugf("[%s] Upload %s success", reqid, url) uploadStatusChan <- uploadStatus{nil, url, resp.StatusCode, rep, classesStored, response} } else { if resp.StatusCode >= 300 && response == "" { response = resp.Status } - DebugPrintf("DEBUG: [%s] Upload %v error: %v response: %v", reqid, url, resp.StatusCode, response) + kc.debugf("[%s] Upload %s status: %d %s", reqid, url, resp.StatusCode, response) uploadStatusChan <- uploadStatus{errors.New(resp.Status), url, resp.StatusCode, rep, classesStored, response} } } -func (kc *KeepClient) BlockWrite(ctx context.Context, req arvados.BlockWriteOptions) (arvados.BlockWriteResponse, error) { +func (kc *KeepClient) httpBlockWrite(ctx context.Context, req arvados.BlockWriteOptions) (arvados.BlockWriteResponse, error) { var resp arvados.BlockWriteResponse var getReader func() io.Reader if req.Data == nil && req.Reader == nil { @@ -149,8 +137,12 @@ func (kc *KeepClient) BlockWrite(ctx context.Context, req arvados.BlockWriteOpti getReader = func() io.Reader { return bytes.NewReader(req.Data[:req.DataSize]) } } else { buf := asyncbuf.NewBuffer(make([]byte, 0, req.DataSize)) + reader := req.Reader + if req.Hash != "" { + reader = HashCheckingReader{req.Reader, md5.New(), req.Hash} + } go func() { - _, err := io.Copy(buf, HashCheckingReader{req.Reader, md5.New(), req.Hash}) + _, err := io.Copy(buf, reader) buf.CloseWithError(err) }() getReader = buf.NewReader @@ -214,6 +206,7 @@ func (kc *KeepClient) BlockWrite(ctx context.Context, req arvados.BlockWriteOpti replicasPerThread = req.Replicas } + delay := delayCalculator{InitialMaxDelay: kc.RetryDelay} retriesRemaining := req.Attempts var retryServers []string @@ -248,7 +241,7 @@ func (kc *KeepClient) BlockWrite(ctx context.Context, req arvados.BlockWriteOpti for active*replicasPerThread < maxConcurrency { // Start some upload requests if nextServer < len(sv) { - DebugPrintf("DEBUG: [%s] Begin upload %s to %s", req.RequestID, req.Hash, sv[nextServer]) + kc.debugf("[%s] Begin upload %s to %s", req.RequestID, req.Hash, sv[nextServer]) go kc.uploadToKeepServer(sv[nextServer], req.Hash, classesTodo, getReader(), uploadStatusChan, req.DataSize, req.RequestID) nextServer++ active++ @@ -265,7 +258,7 @@ func (kc *KeepClient) BlockWrite(ctx context.Context, req arvados.BlockWriteOpti } } - DebugPrintf("DEBUG: [%s] Replicas remaining to write: %v active uploads: %v", req.RequestID, replicasTodo, active) + kc.debugf("[%s] Replicas remaining to write: %d active uploads: %d", req.RequestID, replicasTodo, active) if active < 1 { break } @@ -302,14 +295,17 @@ func (kc *KeepClient) BlockWrite(ctx context.Context, req arvados.BlockWriteOpti } if status.statusCode == 0 || status.statusCode == 408 || status.statusCode == 429 || - (status.statusCode >= 500 && status.statusCode != 503) { + (status.statusCode >= 500 && status.statusCode != http.StatusInsufficientStorage) { // Timeout, too many requests, or other server side failure - // Do not retry when status code is 503, which means the keep server is full + // (do not auto-retry status 507 "full") retryServers = append(retryServers, status.url[0:strings.LastIndex(status.url, "/")]) } } sv = retryServers + if len(sv) > 0 { + time.Sleep(delay.Next()) + } } return resp, nil @@ -341,3 +337,37 @@ func parseStorageClassesConfirmedHeader(hdr string) (map[string]int, error) { } return classesStored, nil } + +// delayCalculator calculates a series of delays for implementing +// exponential backoff with jitter. The first call to Next() returns +// a random duration between MinimumRetryDelay and the specified +// InitialMaxDelay (or DefaultRetryDelay if 0). The max delay is +// doubled on each subsequent call to Next(), up to 10x the initial +// max delay. +type delayCalculator struct { + InitialMaxDelay time.Duration + n int // number of delays returned so far + nextmax time.Duration + limit time.Duration +} + +func (dc *delayCalculator) Next() time.Duration { + if dc.nextmax <= MinimumRetryDelay { + // initialize + if dc.InitialMaxDelay > 0 { + dc.nextmax = dc.InitialMaxDelay + } else { + dc.nextmax = DefaultRetryDelay + } + dc.limit = 10 * dc.nextmax + } + d := time.Duration(rand.Float64() * float64(dc.nextmax)) + if d < MinimumRetryDelay { + d = MinimumRetryDelay + } + dc.nextmax *= 2 + if dc.nextmax > dc.limit { + dc.nextmax = dc.limit + } + return d +}