Kubernetes源码学习笔记------如何实现优雅重试

func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error

func (r *Reflector) list(stopCh <-chan struct{}) error

func (r *Reflector) watch(w watch.Interface, stopCh <-chan struct{}, resyncerrc chan error) error

utilnet.IsConnectionRefused(err) 

apierrors.IsTooManyRequests(err)

apierrors.IsInternalError(err) && retry.ShouldRetry()

case <-stopCh:
    return nil
case <-r.backoffManager.Backoff().C():
    continue

func (b *backoffManager) Backoff() clock.Timer {
	b.lock.Lock()
	defer b.lock.Unlock()
	if b.timer == nil {
		b.timer = b.clock.NewTimer(b.Step())
	} else {
		b.timer.Reset(b.Step())
	}
	return b.timer
}

package main

import (
	"errors"
	"fmt"
	"math/rand"
	"sync"
	"time"

	"k8s.io/utils/clock"
)

// BackoffManager defines the retry management interface.
type BackoffManager interface {
	Backoff() clock.Timer
}

// backoffManager implements retry management.
type backoffManager struct {
	duration time.Duration // Initial retry interval
	factor   float64       // Exponential backoff factor
	jitter   float64       // Jitter factor
	steps    int           // Maximum retry attempts (not used in this example)
	cap      time.Duration // Maximum retry interval

	lock  sync.Mutex
	timer clock.Timer
	clock clock.Clock
}

// NewBackoffManager creates a new retry manager.
func NewBackoffManager(initial time.Duration, maxInterval time.Duration, factor float64, jitter float64, clock clock.Clock) BackoffManager {
	return &backoffManager{
		duration: initial,
		factor:   factor,
		jitter:   jitter,
		cap:      maxInterval,
		clock:    clock,
	}
}

// Backoff implements the retry logic.
func (b *backoffManager) Backoff() clock.Timer {
	b.lock.Lock()
	defer b.lock.Unlock()

	duration := b.duration
	if b.jitter > 0.0 {
		duration = addJitter(duration, b.jitter)
	}

	if b.factor > 0.0 {
		b.duration = time.Duration(float64(b.duration) * b.factor)
		if b.cap > 0 && b.duration > b.cap {
			b.duration = b.cap
		}
	}

	if b.timer == nil {
		b.timer = b.clock.NewTimer(duration)
	} else {
		b.timer.Reset(duration)
	}
	return b.timer
}

func addJitter(duration time.Duration, jitter float64) time.Duration {
	multiplier := 1.0 + rand.Float64()*jitter
	return time.Duration(float64(duration) * multiplier)
}

func unstableOperation() error {
	// Simulate a randomly failing operation.
	if rand.Float64() < 0.7 {
		return errors.New("service temporarily unavailable")
	}
	return nil
}

func main() {

	stopCh := make(chan struct{})

	backoffMgr := NewBackoffManager(
		100*time.Millisecond, // Initial retry interval
		5*time.Second,        // Maximum retry interval
		2.0,                  // Exponential factor
		0.1,                  // Jitter factor
		clock.RealClock{},    // Use real clock
	)

	go func() {
		time.Sleep(10 * time.Second)
		close(stopCh)
	}()

	attempts := 0
	maxAttempts := 10

	fmt.Println("Starting retry operation...")
	for attempts < maxAttempts {
		attempts++
		err := unstableOperation()
		if err == nil {
			fmt.Printf("Operation successful! Retried %d times\n", attempts-1)
			return
		}

		fmt.Printf("Attempt %d failed: %v\n", attempts, err)

		// Implement retry logic
		select {
		case <-stopCh:
			fmt.Println("Stop signal received, exiting retry loop")
			return
		case <-backoffMgr.Backoff().C():
			fmt.Println("Continuing retry after backoff...")
			continue
		}
	}

	fmt.Printf("Reached maximum retry attempts %d, operation ultimately failed\n", maxAttempts)
}

Starting retry operation...
Attempt 1 failed: service temporarily unavailable
Continuing retry after backoff...
Attempt 2 failed: service temporarily unavailable
Continuing retry after backoff...
Operation successful! Retried 2 times

go func() {
    time.Sleep(10 * time.Second)
    close(stopCh)
}()

Starting retry operation...
Attempt 1 failed: service temporarily unavailable
Continuing retry after backoff...
Attempt 2 failed: service temporarily unavailable
Continuing retry after backoff...
Attempt 3 failed: service temporarily unavailable
Continuing retry after backoff...
Attempt 4 failed: service temporarily unavailable
Continuing retry after backoff...
Attempt 5 failed: service temporarily unavailable
Continuing retry after backoff...
Attempt 6 failed: service temporarily unavailable
Continuing retry after backoff...
Attempt 7 failed: service temporarily unavailable
Stop signal received, exiting retry loop

func (r *retryWithDeadlineImpl) ShouldRetry() bool {
	if r.maxRetryDuration <= time.Duration(0) {
		return false
	}

	if r.clock.Now().Sub(r.firstErrorTime) <= r.maxRetryDuration {
		return true
	}

	r.reset()
	return false
}