Update custom-metrics-apiserver and metrics-server

vendor/k8s.io/apimachinery/pkg/util/clock/clock.go

@@ -52,23 +52,26 @@ func (RealClock) Since(ts time.Time) time.Duration {
 	return time.Since(ts)
 }
 
-// Same as time.After(d).
+// After is the same as time.After(d).
 func (RealClock) After(d time.Duration) <-chan time.Time {
 	return time.After(d)
 }
 
+// NewTimer returns a new Timer.
 func (RealClock) NewTimer(d time.Duration) Timer {
 	return &realTimer{
 		timer: time.NewTimer(d),
 	}
 }
 
+// NewTicker returns a new Ticker.
 func (RealClock) NewTicker(d time.Duration) Ticker {
 	return &realTicker{
 		ticker: time.NewTicker(d),
 	}
 }
 
+// Sleep pauses the RealClock for duration d.
 func (RealClock) Sleep(d time.Duration) {
 	time.Sleep(d)
 }
@@ -94,12 +97,14 @@ type fakeClockWaiter struct {
 	destChan      chan time.Time
 }
 
+// NewFakePassiveClock returns a new FakePassiveClock.
 func NewFakePassiveClock(t time.Time) *FakePassiveClock {
 	return &FakePassiveClock{
 		time: t,
 	}
 }
 
+// NewFakeClock returns a new FakeClock
 func NewFakeClock(t time.Time) *FakeClock {
 	return &FakeClock{
 		FakePassiveClock: *NewFakePassiveClock(t),
@@ -120,14 +125,14 @@ func (f *FakePassiveClock) Since(ts time.Time) time.Duration {
 	return f.time.Sub(ts)
 }
 
-// Sets the time.
+// SetTime sets the time on the FakePassiveClock.
 func (f *FakePassiveClock) SetTime(t time.Time) {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 	f.time = t
 }
 
-// Fake version of time.After(d).
+// After is the Fake version of time.After(d).
 func (f *FakeClock) After(d time.Duration) <-chan time.Time {
 	f.lock.Lock()
 	defer f.lock.Unlock()
@@ -140,7 +145,7 @@ func (f *FakeClock) After(d time.Duration) <-chan time.Time {
 	return ch
 }
 
-// Fake version of time.NewTimer(d).
+// NewTimer is the Fake version of time.NewTimer(d).
 func (f *FakeClock) NewTimer(d time.Duration) Timer {
 	f.lock.Lock()
 	defer f.lock.Unlock()
@@ -157,6 +162,7 @@ func (f *FakeClock) NewTimer(d time.Duration) Timer {
 	return timer
 }
 
+// NewTicker returns a new Ticker.
 func (f *FakeClock) NewTicker(d time.Duration) Ticker {
 	f.lock.Lock()
 	defer f.lock.Unlock()
@@ -174,14 +180,14 @@ func (f *FakeClock) NewTicker(d time.Duration) Ticker {
 	}
 }
 
-// Move clock by Duration, notify anyone that's called After, Tick, or NewTimer
+// Step moves clock by Duration, notifies anyone that's called After, Tick, or NewTimer
 func (f *FakeClock) Step(d time.Duration) {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 	f.setTimeLocked(f.time.Add(d))
 }
 
-// Sets the time.
+// SetTime sets the time on a FakeClock.
 func (f *FakeClock) SetTime(t time.Time) {
 	f.lock.Lock()
 	defer f.lock.Unlock()
@@ -219,7 +225,7 @@ func (f *FakeClock) setTimeLocked(t time.Time) {
 	f.waiters = newWaiters
 }
 
-// Returns true if After has been called on f but not yet satisfied (so you can
+// HasWaiters returns true if After has been called on f but not yet satisfied (so you can
 // write race-free tests).
 func (f *FakeClock) HasWaiters() bool {
 	f.lock.RLock()
@@ -227,6 +233,7 @@ func (f *FakeClock) HasWaiters() bool {
 	return len(f.waiters) > 0
 }
 
+// Sleep pauses the FakeClock for duration d.
 func (f *FakeClock) Sleep(d time.Duration) {
 	f.Step(d)
 }
@@ -248,24 +255,25 @@ func (i *IntervalClock) Since(ts time.Time) time.Duration {
 	return i.Time.Sub(ts)
 }
 
-// Unimplemented, will panic.
+// After is currently unimplemented, will panic.
 // TODO: make interval clock use FakeClock so this can be implemented.
 func (*IntervalClock) After(d time.Duration) <-chan time.Time {
 	panic("IntervalClock doesn't implement After")
 }
 
-// Unimplemented, will panic.
+// NewTimer is currently unimplemented, will panic.
 // TODO: make interval clock use FakeClock so this can be implemented.
 func (*IntervalClock) NewTimer(d time.Duration) Timer {
 	panic("IntervalClock doesn't implement NewTimer")
 }
 
-// Unimplemented, will panic.
+// NewTicker is currently unimplemented, will panic.
 // TODO: make interval clock use FakeClock so this can be implemented.
 func (*IntervalClock) NewTicker(d time.Duration) Ticker {
 	panic("IntervalClock doesn't implement NewTicker")
 }
 
+// Sleep is currently unimplemented; will panic.
 func (*IntervalClock) Sleep(d time.Duration) {
 	panic("IntervalClock doesn't implement Sleep")
 }
@@ -355,6 +363,7 @@ func (f *fakeTimer) Reset(d time.Duration) bool {
 	return false
 }
 
+// Ticker defines the Ticker interface
 type Ticker interface {
 	C() <-chan time.Time
 	Stop()

vendor/k8s.io/apimachinery/pkg/util/errors/errors.go

@@ -28,9 +28,14 @@ type MessageCountMap map[string]int
 
 // Aggregate represents an object that contains multiple errors, but does not
 // necessarily have singular semantic meaning.
+// The aggregate can be used with `errors.Is()` to check for the occurrence of
+// a specific error type.
+// Errors.As() is not supported, because the caller presumably cares about a
+// specific error of potentially multiple that match the given type.
 type Aggregate interface {
 	error
 	Errors() []error
+	Is(error) bool
 }
 
 // NewAggregate converts a slice of errors into an Aggregate interface, which
@@ -71,16 +76,17 @@ func (agg aggregate) Error() string {
 	}
 	seenerrs := sets.NewString()
 	result := ""
-	agg.visit(func(err error) {
+	agg.visit(func(err error) bool {
 		msg := err.Error()
 		if seenerrs.Has(msg) {
-			return
+			return false
 		}
 		seenerrs.Insert(msg)
 		if len(seenerrs) > 1 {
 			result += ", "
 		}
 		result += msg
+		return false
 	})
 	if len(seenerrs) == 1 {
 		return result
@@ -88,19 +94,33 @@ func (agg aggregate) Error() string {
 	return "[" + result + "]"
 }
 
-func (agg aggregate) visit(f func(err error)) {
+func (agg aggregate) Is(target error) bool {
+	return agg.visit(func(err error) bool {
+		return errors.Is(err, target)
+	})
+}
+
+func (agg aggregate) visit(f func(err error) bool) bool {
 	for _, err := range agg {
 		switch err := err.(type) {
 		case aggregate:
-			err.visit(f)
+			if match := err.visit(f); match {
+				return match
+			}
 		case Aggregate:
 			for _, nestedErr := range err.Errors() {
-				f(nestedErr)
+				if match := f(nestedErr); match {
+					return match
+				}
 			}
 		default:
-			f(err)
+			if match := f(err); match {
+				return match
+			}
 		}
 	}
+
+	return false
 }
 
 // Errors is part of the Aggregate interface.

vendor/k8s.io/apimachinery/pkg/util/intstr/generated.pb.go

@@ -38,7 +38,7 @@ var _ = math.Inf
 // is compatible with the proto package it is being compiled against.
 // A compilation error at this line likely means your copy of the
 // proto package needs to be updated.
-const _ = proto.GoGoProtoPackageIsVersion2 // please upgrade the proto package
+const _ = proto.GoGoProtoPackageIsVersion3 // please upgrade the proto package
 
 func (m *IntOrString) Reset()      { *m = IntOrString{} }
 func (*IntOrString) ProtoMessage() {}
@@ -289,6 +289,7 @@ func (m *IntOrString) Unmarshal(dAtA []byte) error {
 func skipGenerated(dAtA []byte) (n int, err error) {
 	l := len(dAtA)
 	iNdEx := 0
+	depth := 0
 	for iNdEx < l {
 		var wire uint64
 		for shift := uint(0); ; shift += 7 {
@@ -320,10 +321,8 @@ func skipGenerated(dAtA []byte) (n int, err error) {
 					break
 				}
 			}
-			return iNdEx, nil
 		case 1:
 			iNdEx += 8
-			return iNdEx, nil
 		case 2:
 			var length int
 			for shift := uint(0); ; shift += 7 {
@@ -344,55 +343,30 @@ func skipGenerated(dAtA []byte) (n int, err error) {
 				return 0, ErrInvalidLengthGenerated
 			}
 			iNdEx += length
-			if iNdEx < 0 {
-				return 0, ErrInvalidLengthGenerated
-			}
-			return iNdEx, nil
 		case 3:
-			for {
-				var innerWire uint64
-				var start int = iNdEx
-				for shift := uint(0); ; shift += 7 {
-					if shift >= 64 {
-						return 0, ErrIntOverflowGenerated
-					}
-					if iNdEx >= l {
-						return 0, io.ErrUnexpectedEOF
-					}
-					b := dAtA[iNdEx]
-					iNdEx++
-					innerWire |= (uint64(b) & 0x7F) << shift
-					if b < 0x80 {
-						break
-					}
-				}
-				innerWireType := int(innerWire & 0x7)
-				if innerWireType == 4 {
-					break
-				}
-				next, err := skipGenerated(dAtA[start:])
-				if err != nil {
-					return 0, err
-				}
-				iNdEx = start + next
-				if iNdEx < 0 {
-					return 0, ErrInvalidLengthGenerated
-				}
-			}
-			return iNdEx, nil
+			depth++
 		case 4:
-			return iNdEx, nil
+			if depth == 0 {
+				return 0, ErrUnexpectedEndOfGroupGenerated
+			}
+			depth--
 		case 5:
 			iNdEx += 4
-			return iNdEx, nil
 		default:
 			return 0, fmt.Errorf("proto: illegal wireType %d", wireType)
 		}
+		if iNdEx < 0 {
+			return 0, ErrInvalidLengthGenerated
+		}
+		if depth == 0 {
+			return iNdEx, nil
+		}
 	}
-	panic("unreachable")
+	return 0, io.ErrUnexpectedEOF
 }
 
 var (
-	ErrInvalidLengthGenerated = fmt.Errorf("proto: negative length found during unmarshaling")
-	ErrIntOverflowGenerated   = fmt.Errorf("proto: integer overflow")
+	ErrInvalidLengthGenerated        = fmt.Errorf("proto: negative length found during unmarshaling")
+	ErrIntOverflowGenerated          = fmt.Errorf("proto: integer overflow")
+	ErrUnexpectedEndOfGroupGenerated = fmt.Errorf("proto: unexpected end of group")
 )

vendor/k8s.io/apimachinery/pkg/util/intstr/intstr.go

@@ -97,7 +97,8 @@ func (intstr *IntOrString) String() string {
 }
 
 // IntValue returns the IntVal if type Int, or if
-// it is a String, will attempt a conversion to int.
+// it is a String, will attempt a conversion to int,
+// returning 0 if a parsing error occurs.
 func (intstr *IntOrString) IntValue() int {
 	if intstr.Type == String {
 		i, _ := strconv.Atoi(intstr.StrVal)

vendor/k8s.io/apimachinery/pkg/util/json/json.go

@@ -66,11 +66,36 @@ func Unmarshal(data []byte, v interface{}) error {
 		// If the decode succeeds, post-process the map to convert json.Number objects to int64 or float64
 		return convertSliceNumbers(*v, 0)
 
+	case *interface{}:
+		// Build a decoder from the given data
+		decoder := json.NewDecoder(bytes.NewBuffer(data))
+		// Preserve numbers, rather than casting to float64 automatically
+		decoder.UseNumber()
+		// Run the decode
+		if err := decoder.Decode(v); err != nil {
+			return err
+		}
+		// If the decode succeeds, post-process the map to convert json.Number objects to int64 or float64
+		return convertInterfaceNumbers(v, 0)
+
 	default:
 		return json.Unmarshal(data, v)
 	}
 }
 
+func convertInterfaceNumbers(v *interface{}, depth int) error {
+	var err error
+	switch v2 := (*v).(type) {
+	case json.Number:
+		*v, err = convertNumber(v2)
+	case map[string]interface{}:
+		err = convertMapNumbers(v2, depth+1)
+	case []interface{}:
+		err = convertSliceNumbers(v2, depth+1)
+	}
+	return err
+}
+
 // convertMapNumbers traverses the map, converting any json.Number values to int64 or float64.
 // values which are map[string]interface{} or []interface{} are recursively visited
 func convertMapNumbers(m map[string]interface{}, depth int) error {

vendor/k8s.io/apimachinery/pkg/util/net/http.go

@@ -55,6 +55,12 @@ func JoinPreservingTrailingSlash(elem ...string) string {
 	return result
 }
 
+// IsTimeout returns true if the given error is a network timeout error
+func IsTimeout(err error) bool {
+	neterr, ok := err.(net.Error)
+	return ok && neterr != nil && neterr.Timeout()
+}
+
 // IsProbableEOF returns true if the given error resembles a connection termination
 // scenario that would justify assuming that the watch is empty.
 // These errors are what the Go http stack returns back to us which are general
@@ -206,13 +212,17 @@ func GetHTTPClient(req *http.Request) string {
 	return "unknown"
 }
 
-// SourceIPs splits the comma separated X-Forwarded-For header or returns the X-Real-Ip header or req.RemoteAddr,
-// in that order, ignoring invalid IPs. It returns nil if all of these are empty or invalid.
+// SourceIPs splits the comma separated X-Forwarded-For header and joins it with
+// the X-Real-Ip header and/or req.RemoteAddr, ignoring invalid IPs.
+// The X-Real-Ip is omitted if it's already present in the X-Forwarded-For chain.
+// The req.RemoteAddr is always the last IP in the returned list.
+// It returns nil if all of these are empty or invalid.
 func SourceIPs(req *http.Request) []net.IP {
+	var srcIPs []net.IP
+
 	hdr := req.Header
 	// First check the X-Forwarded-For header for requests via proxy.
 	hdrForwardedFor := hdr.Get("X-Forwarded-For")
-	forwardedForIPs := []net.IP{}
 	if hdrForwardedFor != "" {
 		// X-Forwarded-For can be a csv of IPs in case of multiple proxies.
 		// Use the first valid one.
@@ -220,38 +230,49 @@ func SourceIPs(req *http.Request) []net.IP {
 		for _, part := range parts {
 			ip := net.ParseIP(strings.TrimSpace(part))
 			if ip != nil {
-				forwardedForIPs = append(forwardedForIPs, ip)
+				srcIPs = append(srcIPs, ip)
 			}
 		}
 	}
-	if len(forwardedForIPs) > 0 {
-		return forwardedForIPs
-	}
 
 	// Try the X-Real-Ip header.
 	hdrRealIp := hdr.Get("X-Real-Ip")
 	if hdrRealIp != "" {
 		ip := net.ParseIP(hdrRealIp)
-		if ip != nil {
-			return []net.IP{ip}
+		// Only append the X-Real-Ip if it's not already contained in the X-Forwarded-For chain.
+		if ip != nil && !containsIP(srcIPs, ip) {
+			srcIPs = append(srcIPs, ip)
 		}
 	}
 
-	// Fallback to Remote Address in request, which will give the correct client IP when there is no proxy.
+	// Always include the request Remote Address as it cannot be easily spoofed.
+	var remoteIP net.IP
 	// Remote Address in Go's HTTP server is in the form host:port so we need to split that first.
 	host, _, err := net.SplitHostPort(req.RemoteAddr)
 	if err == nil {
-		if remoteIP := net.ParseIP(host); remoteIP != nil {
-			return []net.IP{remoteIP}
+		remoteIP = net.ParseIP(host)
+	}
+	// Fallback if Remote Address was just IP.
+	if remoteIP == nil {
+		remoteIP = net.ParseIP(req.RemoteAddr)
+	}
+
+	// Don't duplicate remote IP if it's already the last address in the chain.
+	if remoteIP != nil && (len(srcIPs) == 0 || !remoteIP.Equal(srcIPs[len(srcIPs)-1])) {
+		srcIPs = append(srcIPs, remoteIP)
+	}
+
+	return srcIPs
+}
+
+// Checks whether the given IP address is contained in the list of IPs.
+func containsIP(ips []net.IP, ip net.IP) bool {
+	for _, v := range ips {
+		if v.Equal(ip) {
+			return true
 		}
 	}
-
-	// Fallback if Remote Address was just IP.
-	if remoteIP := net.ParseIP(req.RemoteAddr); remoteIP != nil {
-		return []net.IP{remoteIP}
-	}
-
-	return nil
+	return false
 }
 
 // Extracts and returns the clients IP from the given request.

vendor/k8s.io/apimachinery/pkg/util/validation/validation.go

@@ -109,6 +109,44 @@ func IsFullyQualifiedDomainName(fldPath *field.Path, name string) field.ErrorLis
 	return allErrors
 }
 
+// Allowed characters in an HTTP Path as defined by RFC 3986. A HTTP path may
+// contain:
+// * unreserved characters (alphanumeric, '-', '.', '_', '~')
+// * percent-encoded octets
+// * sub-delims ("!", "$", "&", "'", "(", ")", "*", "+", ",", ";", "=")
+// * a colon character (":")
+const httpPathFmt string = `[A-Za-z0-9/\-._~%!$&'()*+,;=:]+`
+
+var httpPathRegexp = regexp.MustCompile("^" + httpPathFmt + "$")
+
+// IsDomainPrefixedPath checks if the given string is a domain-prefixed path
+// (e.g. acme.io/foo). All characters before the first "/" must be a valid
+// subdomain as defined by RFC 1123. All characters trailing the first "/" must
+// be valid HTTP Path characters as defined by RFC 3986.
+func IsDomainPrefixedPath(fldPath *field.Path, dpPath string) field.ErrorList {
+	var allErrs field.ErrorList
+	if len(dpPath) == 0 {
+		return append(allErrs, field.Required(fldPath, ""))
+	}
+
+	segments := strings.SplitN(dpPath, "/", 2)
+	if len(segments) != 2 || len(segments[0]) == 0 || len(segments[1]) == 0 {
+		return append(allErrs, field.Invalid(fldPath, dpPath, "must be a domain-prefixed path (such as \"acme.io/foo\")"))
+	}
+
+	host := segments[0]
+	for _, err := range IsDNS1123Subdomain(host) {
+		allErrs = append(allErrs, field.Invalid(fldPath, host, err))
+	}
+
+	path := segments[1]
+	if !httpPathRegexp.MatchString(path) {
+		return append(allErrs, field.Invalid(fldPath, path, RegexError("Invalid path", httpPathFmt)))
+	}
+
+	return allErrs
+}
+
 const labelValueFmt string = "(" + qualifiedNameFmt + ")?"
 const labelValueErrMsg string = "a valid label must be an empty string or consist of alphanumeric characters, '-', '_' or '.', and must start and end with an alphanumeric character"
 

vendor/k8s.io/apimachinery/pkg/util/wait/wait.go

@@ -19,10 +19,12 @@ package wait
 import (
 	"context"
 	"errors"
+	"math"
 	"math/rand"
 	"sync"
 	"time"
 
+	"k8s.io/apimachinery/pkg/util/clock"
 	"k8s.io/apimachinery/pkg/util/runtime"
 )
 
@@ -128,9 +130,15 @@ func NonSlidingUntilWithContext(ctx context.Context, f func(context.Context), pe
 // Close stopCh to stop. f may not be invoked if stop channel is already
 // closed. Pass NeverStop to if you don't want it stop.
 func JitterUntil(f func(), period time.Duration, jitterFactor float64, sliding bool, stopCh <-chan struct{}) {
-	var t *time.Timer
-	var sawTimeout bool
+	BackoffUntil(f, NewJitteredBackoffManager(period, jitterFactor, &clock.RealClock{}), sliding, stopCh)
+}
 
+// BackoffUntil loops until stop channel is closed, run f every duration given by BackoffManager.
+//
+// If sliding is true, the period is computed after f runs. If it is false then
+// period includes the runtime for f.
+func BackoffUntil(f func(), backoff BackoffManager, sliding bool, stopCh <-chan struct{}) {
+	var t clock.Timer
 	for {
 		select {
 		case <-stopCh:
@@ -138,13 +146,8 @@ func JitterUntil(f func(), period time.Duration, jitterFactor float64, sliding b
 		default:
 		}
 
-		jitteredPeriod := period
-		if jitterFactor > 0.0 {
-			jitteredPeriod = Jitter(period, jitterFactor)
-		}
-
 		if !sliding {
-			t = resetOrReuseTimer(t, jitteredPeriod, sawTimeout)
+			t = backoff.Backoff()
 		}
 
 		func() {
@@ -153,7 +156,7 @@ func JitterUntil(f func(), period time.Duration, jitterFactor float64, sliding b
 		}()
 
 		if sliding {
-			t = resetOrReuseTimer(t, jitteredPeriod, sawTimeout)
+			t = backoff.Backoff()
 		}
 
 		// NOTE: b/c there is no priority selection in golang
@@ -164,8 +167,7 @@ func JitterUntil(f func(), period time.Duration, jitterFactor float64, sliding b
 		select {
 		case <-stopCh:
 			return
-		case <-t.C:
-			sawTimeout = true
+		case <-t.C():
 		}
 	}
 }
@@ -203,6 +205,12 @@ var ErrWaitTimeout = errors.New("timed out waiting for the condition")
 // if the loop should be aborted.
 type ConditionFunc func() (done bool, err error)
 
+// runConditionWithCrashProtection runs a ConditionFunc with crash protection
+func runConditionWithCrashProtection(condition ConditionFunc) (bool, error) {
+	defer runtime.HandleCrash()
+	return condition()
+}
+
 // Backoff holds parameters applied to a Backoff function.
 type Backoff struct {
 	// The initial duration.
@@ -277,6 +285,105 @@ func contextForChannel(parentCh <-chan struct{}) (context.Context, context.Cance
 	return ctx, cancel
 }
 
+// BackoffManager manages backoff with a particular scheme based on its underlying implementation. It provides
+// an interface to return a timer for backoff, and caller shall backoff until Timer.C() drains. If the second Backoff()
+// is called before the timer from the first Backoff() call finishes, the first timer will NOT be drained and result in
+// undetermined behavior.
+// The BackoffManager is supposed to be called in a single-threaded environment.
+type BackoffManager interface {
+	Backoff() clock.Timer
+}
+
+type exponentialBackoffManagerImpl struct {
+	backoff              *Backoff
+	backoffTimer         clock.Timer
+	lastBackoffStart     time.Time
+	initialBackoff       time.Duration
+	backoffResetDuration time.Duration
+	clock                clock.Clock
+}
+
+// NewExponentialBackoffManager returns a manager for managing exponential backoff. Each backoff is jittered and
+// backoff will not exceed the given max. If the backoff is not called within resetDuration, the backoff is reset.
+// This backoff manager is used to reduce load during upstream unhealthiness.
+func NewExponentialBackoffManager(initBackoff, maxBackoff, resetDuration time.Duration, backoffFactor, jitter float64, c clock.Clock) BackoffManager {
+	return &exponentialBackoffManagerImpl{
+		backoff: &Backoff{
+			Duration: initBackoff,
+			Factor:   backoffFactor,
+			Jitter:   jitter,
+
+			// the current impl of wait.Backoff returns Backoff.Duration once steps are used up, which is not
+			// what we ideally need here, we set it to max int and assume we will never use up the steps
+			Steps: math.MaxInt32,
+			Cap:   maxBackoff,
+		},
+		backoffTimer:         nil,
+		initialBackoff:       initBackoff,
+		lastBackoffStart:     c.Now(),
+		backoffResetDuration: resetDuration,
+		clock:                c,
+	}
+}
+
+func (b *exponentialBackoffManagerImpl) getNextBackoff() time.Duration {
+	if b.clock.Now().Sub(b.lastBackoffStart) > b.backoffResetDuration {
+		b.backoff.Steps = math.MaxInt32
+		b.backoff.Duration = b.initialBackoff
+	}
+	b.lastBackoffStart = b.clock.Now()
+	return b.backoff.Step()
+}
+
+// Backoff implements BackoffManager.Backoff, it returns a timer so caller can block on the timer for exponential backoff.
+// The returned timer must be drained before calling Backoff() the second time
+func (b *exponentialBackoffManagerImpl) Backoff() clock.Timer {
+	if b.backoffTimer == nil {
+		b.backoffTimer = b.clock.NewTimer(b.getNextBackoff())
+	} else {
+		b.backoffTimer.Reset(b.getNextBackoff())
+	}
+	return b.backoffTimer
+}
+
+type jitteredBackoffManagerImpl struct {
+	clock        clock.Clock
+	duration     time.Duration
+	jitter       float64
+	backoffTimer clock.Timer
+}
+
+// NewJitteredBackoffManager returns a BackoffManager that backoffs with given duration plus given jitter. If the jitter
+// is negative, backoff will not be jittered.
+func NewJitteredBackoffManager(duration time.Duration, jitter float64, c clock.Clock) BackoffManager {
+	return &jitteredBackoffManagerImpl{
+		clock:        c,
+		duration:     duration,
+		jitter:       jitter,
+		backoffTimer: nil,
+	}
+}
+
+func (j *jitteredBackoffManagerImpl) getNextBackoff() time.Duration {
+	jitteredPeriod := j.duration
+	if j.jitter > 0.0 {
+		jitteredPeriod = Jitter(j.duration, j.jitter)
+	}
+	return jitteredPeriod
+}
+
+// Backoff implements BackoffManager.Backoff, it returns a timer so caller can block on the timer for jittered backoff.
+// The returned timer must be drained before calling Backoff() the second time
+func (j *jitteredBackoffManagerImpl) Backoff() clock.Timer {
+	backoff := j.getNextBackoff()
+	if j.backoffTimer == nil {
+		j.backoffTimer = j.clock.NewTimer(backoff)
+	} else {
+		j.backoffTimer.Reset(backoff)
+	}
+	return j.backoffTimer
+}
+
 // ExponentialBackoff repeats a condition check with exponential backoff.
 //
 // It repeatedly checks the condition and then sleeps, using `backoff.Step()`
@@ -289,7 +396,7 @@ func contextForChannel(parentCh <-chan struct{}) (context.Context, context.Cance
 // In all other cases, ErrWaitTimeout is returned.
 func ExponentialBackoff(backoff Backoff, condition ConditionFunc) error {
 	for backoff.Steps > 0 {
-		if ok, err := condition(); err != nil || ok {
+		if ok, err := runConditionWithCrashProtection(condition); err != nil || ok {
 			return err
 		}
 		if backoff.Steps == 1 {
@@ -335,7 +442,7 @@ func PollImmediate(interval, timeout time.Duration, condition ConditionFunc) err
 }
 
 func pollImmediateInternal(wait WaitFunc, condition ConditionFunc) error {
-	done, err := condition()
+	done, err := runConditionWithCrashProtection(condition)
 	if err != nil {
 		return err
 	}
@@ -364,7 +471,7 @@ func PollInfinite(interval time.Duration, condition ConditionFunc) error {
 // Some intervals may be missed if the condition takes too long or the time
 // window is too short.
 func PollImmediateInfinite(interval time.Duration, condition ConditionFunc) error {
-	done, err := condition()
+	done, err := runConditionWithCrashProtection(condition)
 	if err != nil {
 		return err
 	}
@@ -431,7 +538,7 @@ func WaitFor(wait WaitFunc, fn ConditionFunc, done <-chan struct{}) error {
 	for {
 		select {
 		case _, open := <-c:
-			ok, err := fn()
+			ok, err := runConditionWithCrashProtection(fn)
 			if err != nil {
 				return err
 			}
@@ -497,16 +604,3 @@ func poller(interval, timeout time.Duration) WaitFunc {
 		return ch
 	})
 }
-
-// resetOrReuseTimer avoids allocating a new timer if one is already in use.
-// Not safe for multiple threads.
-func resetOrReuseTimer(t *time.Timer, d time.Duration, sawTimeout bool) *time.Timer {
-	if t == nil {
-		return time.NewTimer(d)
-	}
-	if !t.Stop() && !sawTimeout {
-		<-t.C
-	}
-	t.Reset(d)
-	return t
-}