Merge pull request #21892 from jayunit100/backoff-flake-fix

pkg/client/unversioned/request.go

@@ -645,7 +645,7 @@ func (r *Request) Watch() (watch.Interface, error) {
 	if client == nil {
 		client = http.DefaultClient
 	}
-	time.Sleep(r.backoffMgr.CalculateBackoff(r.URL()))
+	r.backoffMgr.Sleep(r.backoffMgr.CalculateBackoff(r.URL()))
 	resp, err := client.Do(req)
 	updateURLMetrics(r, resp, err)
 	if r.baseURL != nil {
@@ -710,7 +710,7 @@ func (r *Request) Stream() (io.ReadCloser, error) {
 	if client == nil {
 		client = http.DefaultClient
 	}
-	time.Sleep(r.backoffMgr.CalculateBackoff(r.URL()))
+	r.backoffMgr.Sleep(r.backoffMgr.CalculateBackoff(r.URL()))
 	resp, err := client.Do(req)
 	updateURLMetrics(r, resp, err)
 	if r.baseURL != nil {
@@ -792,7 +792,7 @@ func (r *Request) request(fn func(*http.Request, *http.Response)) error {
 		}
 		req.Header = r.headers
 
-		time.Sleep(r.backoffMgr.CalculateBackoff(r.URL()))
+		r.backoffMgr.Sleep(r.backoffMgr.CalculateBackoff(r.URL()))
 		resp, err := client.Do(req)
 		updateURLMetrics(r, resp, err)
 		if err != nil {
@@ -812,7 +812,7 @@ func (r *Request) request(fn func(*http.Request, *http.Response)) error {
 			retries++
 			if seconds, wait := checkWait(resp); wait && retries < maxRetries {
 				glog.V(4).Infof("Got a Retry-After %s response for attempt %d to %v", seconds, retries, url)
-				time.Sleep(time.Duration(seconds) * time.Second)
+				r.backoffMgr.Sleep(time.Duration(seconds) * time.Second)
 				return false
 			}
 			fn(req, resp)

pkg/client/unversioned/request_test.go

@@ -767,17 +767,26 @@ func TestBackoffLifecycle(t *testing.T) {
 	// which are used in the server implementation returning StatusOK above.
 	seconds := []int{0, 1, 2, 4, 8, 0, 1, 2, 4, 0}
 	request := c.Verb("POST").Prefix("backofftest").Suffix("abc")
+	clock := util.FakeClock{}
 	request.backoffMgr = &URLBackoff{
-		Backoff: util.NewBackOff(
+		// Use a fake backoff here to avoid flakes and speed the test up.
+		Backoff: util.NewFakeBackOff(
 			time.Duration(1)*time.Second,
-			time.Duration(200)*time.Second)}
+			time.Duration(200)*time.Second,
+			&clock,
+		)}
+
 	for _, sec := range seconds {
-		start := time.Now()
+		thisBackoff := request.backoffMgr.CalculateBackoff(request.URL())
+		t.Logf("Current backoff %v", thisBackoff)
+		if thisBackoff != time.Duration(sec)*time.Second {
+			t.Errorf("Backoff is %v instead of %v", thisBackoff, sec)
+		}
+		now := clock.Now()
 		request.DoRaw()
-		finish := time.Since(start)
-		t.Logf("%v finished in %v", sec, finish)
-		if finish < time.Duration(sec)*time.Second || finish >= time.Duration(sec+5)*time.Second {
-			t.Fatalf("%v not in range %v", finish, sec)
+		elapsed := clock.Since(now)
+		if clock.Since(now) != thisBackoff {
+			t.Errorf("CalculatedBackoff not honored by clock: Expected time of %v, but got %v ", thisBackoff, elapsed)
 		}
 	}
 }

pkg/client/unversioned/urlbackoff.go

@@ -35,6 +35,7 @@ var maxResponseCode = 499
 type BackoffManager interface {
 	UpdateBackoff(actualUrl *url.URL, err error, responseCode int)
 	CalculateBackoff(actualUrl *url.URL) time.Duration
+	Sleep(d time.Duration)
 }
 
 // URLBackoff struct implements the semantics on top of Backoff which
@@ -54,6 +55,9 @@ func (n *NoBackoff) UpdateBackoff(actualUrl *url.URL, err error, responseCode in
 func (n *NoBackoff) CalculateBackoff(actualUrl *url.URL) time.Duration {
 	return 0 * time.Second
 }
+func (n *NoBackoff) Sleep(d time.Duration) {
+	return
+}
 
 // Disable makes the backoff trivial, i.e., sets it to zero.  This might be used
 // by tests which want to run 1000s of mock requests without slowing down.
@@ -80,7 +84,7 @@ func (b *URLBackoff) baseUrlKey(rawurl *url.URL) string {
 func (b *URLBackoff) UpdateBackoff(actualUrl *url.URL, err error, responseCode int) {
 	// range for retry counts that we store is [0,13]
 	if responseCode > maxResponseCode || serverIsOverloadedSet.Has(responseCode) {
-		b.Backoff.Next(b.baseUrlKey(actualUrl), time.Now())
+		b.Backoff.Next(b.baseUrlKey(actualUrl), b.Backoff.Clock.Now())
 		return
 	} else if responseCode >= 300 || err != nil {
 		glog.V(4).Infof("Client is returning errors: code %v, error %v", responseCode, err)
@@ -95,3 +99,7 @@ func (b *URLBackoff) UpdateBackoff(actualUrl *url.URL, err error, responseCode i
 func (b *URLBackoff) CalculateBackoff(actualUrl *url.URL) time.Duration {
 	return b.Backoff.Get(b.baseUrlKey(actualUrl))
 }
+
+func (b *URLBackoff) Sleep(d time.Duration) {
+	b.Backoff.Clock.Sleep(d)
+}

pkg/util/backoff.go

@@ -36,6 +36,15 @@ type Backoff struct {
 	perItemBackoff  map[string]*backoffEntry
 }
 
+func NewFakeBackOff(initial, max time.Duration, tc *FakeClock) *Backoff {
+	return &Backoff{
+		perItemBackoff:  map[string]*backoffEntry{},
+		Clock:           tc,
+		defaultDuration: initial,
+		maxDuration:     max,
+	}
+}
+
 func NewBackOff(initial, max time.Duration) *Backoff {
 	return &Backoff{
 		perItemBackoff:  map[string]*backoffEntry{},

pkg/util/backoff_test.go

@@ -21,15 +21,6 @@ import (
 	"time"
 )
 
-func NewFakeBackOff(initial, max time.Duration, tc *FakeClock) *Backoff {
-	return &Backoff{
-		perItemBackoff:  map[string]*backoffEntry{},
-		Clock:           tc,
-		defaultDuration: initial,
-		maxDuration:     max,
-	}
-}
-
 func TestSlowBackoff(t *testing.T) {
 	id := "_idSlow"
 	tc := NewFakeClock(time.Now())

pkg/util/clock.go

@@ -27,6 +27,7 @@ type Clock interface {
 	Now() time.Time
 	Since(time.Time) time.Duration
 	After(d time.Duration) <-chan time.Time
+	Sleep(d time.Duration)
 }
 
 var (
@@ -53,6 +54,10 @@ func (RealClock) After(d time.Duration) <-chan time.Time {
 	return time.After(d)
 }
 
+func (RealClock) Sleep(d time.Duration) {
+	time.Sleep(d)
+}
+
 // FakeClock implements Clock, but returns an arbitrary time.
 type FakeClock struct {
 	lock sync.RWMutex
@@ -137,6 +142,10 @@ func (f *FakeClock) HasWaiters() bool {
 	return len(f.waiters) > 0
 }
 
+func (f *FakeClock) Sleep(d time.Duration) {
+	f.Step(d)
+}
+
 // IntervalClock implements Clock, but each invocation of Now steps the clock forward the specified duration
 type IntervalClock struct {
 	Time     time.Time
@@ -159,3 +168,7 @@ func (i *IntervalClock) Since(ts time.Time) time.Duration {
 func (*IntervalClock) After(d time.Duration) <-chan time.Time {
 	panic("IntervalClock doesn't implement After")
 }
+
+func (*IntervalClock) Sleep(d time.Duration) {
+	panic("IntervalClock doesn't implement Sleep")
+}

pkg/util/clock_test.go

@@ -37,6 +37,16 @@ func TestFakeClock(t *testing.T) {
 	}
 }
 
+func TestFakeClockSleep(t *testing.T) {
+	startTime := time.Now()
+	tc := NewFakeClock(startTime)
+	tc.Sleep(time.Duration(1) * time.Hour)
+	now := tc.Now()
+	if now.Sub(startTime) != time.Hour {
+		t.Errorf("Fake sleep failed, expected time to advance by one hour, instead, its %v", now.Sub(startTime))
+	}
+}
+
 func TestFakeAfter(t *testing.T) {
 	tc := NewFakeClock(time.Now())
 	if tc.HasWaiters() {