X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/54377a7bacc182ace0bb8b55a812e0a9fee5ced8..8a6bd5b851e0e947b74710c4c2dd1ea59dce606a:/sdk/go/keepclient/support.go

diff --git a/sdk/go/keepclient/support.go b/sdk/go/keepclient/support.go
index 6acaf64baa..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,16 +101,16 @@ 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}
 	}
 }
@@ -218,6 +206,7 @@ func (kc *KeepClient) httpBlockWrite(ctx context.Context, req arvados.BlockWrite
 		replicasPerThread = req.Replicas
 	}
 
+	delay := delayCalculator{InitialMaxDelay: kc.RetryDelay}
 	retriesRemaining := req.Attempts
 	var retryServers []string
 
@@ -252,7 +241,7 @@ func (kc *KeepClient) httpBlockWrite(ctx context.Context, req arvados.BlockWrite
 			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++
@@ -269,7 +258,7 @@ func (kc *KeepClient) httpBlockWrite(ctx context.Context, req arvados.BlockWrite
 				}
 			}
 
-			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
 			}
@@ -306,14 +295,17 @@ func (kc *KeepClient) httpBlockWrite(ctx context.Context, req arvados.BlockWrite
 			}
 
 			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
@@ -345,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
+}