Move retry logic into pkg/libs/endpointcheck

We will need the same check in another PR.

Author: tchap

pkg/libs/endpointaccessible/endpoint_accessible_controller.go

@@ -12,7 +12,6 @@ import (
 
 	"github.com/cenkalti/backoff/v4"
 
-	"github.com/openshift/library-go/pkg/operator/resource/retry"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -22,16 +21,12 @@ import (
 	"github.com/openshift/library-go/pkg/controller/factory"
 	"github.com/openshift/library-go/pkg/operator/events"
 	"github.com/openshift/library-go/pkg/operator/v1helpers"
+
+	"github.com/openshift/cluster-authentication-operator/pkg/libs/endpointcheck"
 )
 
-// The following constants are put together so that
-// all attempts fit safely into resyncInterval.
 const (
 	resyncInterval = 1 * time.Minute
-
-	defaultRequestTimeout = 5 * time.Second
-	defaultRetryInterval  = 2 * time.Second
-	defaultAttemptCount   = 3
 )
 
 type endpointAccessibleController struct {
@@ -42,16 +37,10 @@ type endpointAccessibleController struct {
 	availableConditionName    string
 	endpointCheckDisabledFunc EndpointCheckDisabledFunc
 	// httpClient overrides the default TLS client when set; used in tests.
-	httpClient *http.Client
-	// requestTimeout is the per-request context timeout.
-	// Defaults to defaultRequestTimeout when unset.
+	httpClient     *http.Client
 	requestTimeout time.Duration
-	// retryInterval is the sleep duration between retry attempts.
-	// Defaults to defaultRetryInterval when unset.
-	retryInterval time.Duration
-	// attemptCount is the maximum number of fetch+check cycles.
-	// Defaults to defaultAttemptCount when unset.
-	attemptCount int
+	retryInterval  time.Duration
+	attemptCount   uint64
 }
 
 type EndpointListFunc func() ([]string, error)
@@ -125,26 +114,12 @@ func (c *endpointAccessibleController) sync(ctx context.Context, syncCtx factory
 	// Retry the full fetch+check cycle so that stale pod IPs from a rolling
 	// upgrade are replaced with fresh ones as soon as the Endpoints object is
 	// updated between attempts.
-	attempts := c.attemptCount
-	if attempts <= 0 {
-		attempts = defaultAttemptCount
-	}
-	requestTimeout := c.requestTimeout
-	if requestTimeout <= 0 {
-		requestTimeout = defaultRequestTimeout
-	}
-	retryInterval := c.retryInterval
-	if retryInterval <= 0 {
-		retryInterval = defaultRetryInterval
-	}
-
-	// Assemble a check function to use with backoff.Retry.
 	var (
 		endpoints       []string
 		endpointListErr error
 		errs            []error
 	)
-	checkFn := func() error {
+	checkFn := func(ctx context.Context, requestTimeout time.Duration) error {
 		var err error
 		endpoints, err = c.endpointListFn()
 		if err != nil {
@@ -199,17 +174,7 @@ func (c *endpointAccessibleController) sync(ctx context.Context, syncCtx factory
 		return errors.New("no endpoints")
 	}
 
-	// Run checkFn given number of times. Getting a timeout from checkFn causes an immediate retry,
-	// we don't wait another retryInterval before performing the next check.
-	skippableBoff := retry.NewSkippableBackOff(backoff.NewConstantBackOff(retryInterval))
-	boff := backoff.WithContext(backoff.WithMaxRetries(skippableBoff, uint64(attempts-1)), ctx)
-	_ = backoff.Retry(func() error {
-		err := checkFn()
-		if errors.Is(err, context.DeadlineExceeded) {
-			skippableBoff.SkipNext()
-		}
-		return err
-	}, boff)
+	_ = endpointcheck.Check(ctx, c.requestTimeout, c.retryInterval, c.attemptCount, checkFn)
 
 	if err := endpointListErr; err != nil {
 		if apierrors.IsNotFound(err) {

pkg/libs/endpointaccessible/endpoint_accessible_controller_test.go

@@ -7,16 +7,14 @@ import (
 	"net/http/httptest"
 	"sync/atomic"
 	"testing"
-	"testing/synctest"
 	"time"
 
 	clocktesting "k8s.io/utils/clock/testing"
 
 	operatorv1 "github.com/openshift/api/operator/v1"
-	"github.com/openshift/library-go/pkg/operator/v1helpers"
-
 	"github.com/openshift/library-go/pkg/controller/factory"
 	"github.com/openshift/library-go/pkg/operator/events"
+	"github.com/openshift/library-go/pkg/operator/v1helpers"
 )
 
 func newSyncContext(t *testing.T) factory.SyncContext {
@@ -102,73 +100,6 @@ func TestEndpointAccessibleController_sync(t *testing.T) {
 	}
 }
 
-// TestEndpointAccessibleController_sync_retry verifies the retry logic for
-// fast (non-timeout) failures: the controller sleeps for retryInterval between
-// attempts, and either recovers or gives up after attemptCount tries.
-func TestEndpointAccessibleController_sync_retry(t *testing.T) {
-	const attemptCount = 3
-
-	tests := []struct {
-		name       string
-		failFirstN int32 // how many initial requests the server should reject with 500
-		wantErr    bool
-	}{
-		{
-			name:       "succeeds on last attempt",
-			failFirstN: 2,
-			wantErr:    false,
-		},
-		{
-			name:       "fails after all attempts exhausted",
-			failFirstN: 3,
-			wantErr:    true,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			syncCtx := newSyncContext(t)
-			synctest.Test(t, func(t *testing.T) {
-				c := &endpointAccessibleController{
-					operatorClient: v1helpers.NewFakeOperatorClient(&operatorv1.OperatorSpec{}, &operatorv1.OperatorStatus{}, nil),
-					endpointListFn: func() ([]string, error) {
-						return []string{"http://example.com"}, nil
-					},
-					httpClient:     &http.Client{Transport: &failFastTransport{maxFails: tt.failFirstN}},
-					requestTimeout: defaultRequestTimeout,
-					retryInterval:  10 * time.Second,
-					attemptCount:   attemptCount,
-				}
-
-				start := time.Now()
-				done := make(chan error, 1)
-				go func() {
-					done <- c.sync(context.Background(), syncCtx)
-				}()
-
-				// Advance time for each backoff sleep between attempts.
-				backoffs := min(int(tt.failFirstN), attemptCount-1)
-				for range backoffs {
-					synctest.Wait()
-					time.Sleep(c.retryInterval + time.Millisecond)
-				}
-				synctest.Wait()
-
-				err := <-done
-				if (err != nil) != tt.wantErr {
-					t.Errorf("sync() error = %v, wantErr %v", err, tt.wantErr)
-				}
-
-				// Verify that each retry used the backoff sleep.
-				elapsed := time.Since(start)
-				expectedBackoff := time.Duration(backoffs) * c.retryInterval
-				if elapsed < expectedBackoff {
-					t.Errorf("elapsed %v < %v; backoff was skipped for fast failures", elapsed, expectedBackoff)
-				}
-			})
-		})
-	}
-}
-
 // TestEndpointAccessibleController_sync_retryStaleEndpoint verifies that
 // endpointListFn is re-invoked on each retry attempt. This covers the upgrade
 // scenario where Endpoints/EndpointSlices briefly contain a stale pod IP: the
@@ -212,100 +143,3 @@ func TestEndpointAccessibleController_sync_retryStaleEndpoint(t *testing.T) {
 	}
 }
 
-// TestEndpointAccessibleController_sync_requestTimeout verifies that the
-// per-request timeout is enforced, that the retry mechanism handles timed-out
-// requests correctly, and that the backoff sleep is skipped after timeouts
-// (since the requestTimeout already provided sufficient delay).
-func TestEndpointAccessibleController_sync_requestTimeout(t *testing.T) {
-	const attemptCount = 3
-
-	tests := []struct {
-		name      string
-		hangCount int32 // requests that time out before one succeeds
-		wantErr   bool
-	}{
-		{
-			name:      "succeeds after one timed-out retry",
-			hangCount: 1,
-			wantErr:   false,
-		},
-		{
-			name:      "fails after all retries time out",
-			hangCount: 3,
-			wantErr:   true,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			// Create the sync context outside the bubble: factory.NewSyncContext spawns
-			// a background work-queue goroutine that never exits on its own, which would
-			// deadlock the synctest bubble.
-			syncCtx := newSyncContext(t)
-			synctest.Test(t, func(t *testing.T) {
-				c := &endpointAccessibleController{
-					operatorClient: v1helpers.NewFakeOperatorClient(&operatorv1.OperatorSpec{}, &operatorv1.OperatorStatus{}, nil),
-					endpointListFn: func() ([]string, error) {
-						return []string{"http://example.com"}, nil
-					},
-					httpClient:     &http.Client{Transport: &hangingTransport{maxHangs: tt.hangCount}},
-					requestTimeout: defaultRequestTimeout,
-					retryInterval:  10 * time.Second, // large — would be visible if not skipped
-					attemptCount:   attemptCount,
-				}
-
-				start := time.Now()
-				done := make(chan error, 1)
-				go func() {
-					done <- c.sync(context.Background(), syncCtx)
-				}()
-
-				for range tt.hangCount {
-					synctest.Wait()
-					time.Sleep(c.requestTimeout + time.Millisecond)
-				}
-				synctest.Wait()
-
-				err := <-done
-				if (err != nil) != tt.wantErr {
-					t.Errorf("sync() error = %v, wantErr %v", err, tt.wantErr)
-				}
-
-				// Elapsed time should be only hangCount * requestTimeout with no
-				// retryInterval added — backoff is skipped after timeouts.
-				elapsed := time.Since(start)
-				maxExpected := time.Duration(tt.hangCount)*(c.requestTimeout+time.Millisecond) + time.Second
-				if elapsed > maxExpected {
-					t.Errorf("elapsed %v exceeds %v; backoff sleep was not skipped after timeout", elapsed, maxExpected)
-				}
-			})
-		})
-	}
-}
-
-// hangingTransport simulates a slow endpoint: the first maxHangs requests block
-// until their context is canceled; subsequent requests succeed immediately with 200 OK.
-type hangingTransport struct {
-	count    atomic.Int32
-	maxHangs int32
-}
-
-func (h *hangingTransport) RoundTrip(req *http.Request) (*http.Response, error) {
-	if h.count.Add(1) <= h.maxHangs {
-		<-req.Context().Done()
-		return nil, req.Context().Err()
-	}
-	return &http.Response{StatusCode: http.StatusOK, Body: http.NoBody}, nil
-}
-
-// failFastTransport returns 500 for the first maxFails requests, then 200.
-type failFastTransport struct {
-	count    atomic.Int32
-	maxFails int32
-}
-
-func (f *failFastTransport) RoundTrip(req *http.Request) (*http.Response, error) {
-	if f.count.Add(1) <= f.maxFails {
-		return &http.Response{StatusCode: http.StatusInternalServerError, Body: http.NoBody}, nil
-	}
-	return &http.Response{StatusCode: http.StatusOK, Body: http.NoBody}, nil
-}

pkg/libs/endpointcheck/check.go

@@ -0,0 +1,54 @@
+// Package endpointcheck provides a retry-with-backoff loop for endpoint
+// health checks. It skips the backoff sleep after context.DeadlineExceeded
+// errors (since the request timeout already provided sufficient delay) and
+// stops immediately on backoff.Permanent errors.
+package endpointcheck
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/cenkalti/backoff/v4"
+
+	"github.com/openshift/library-go/pkg/operator/resource/retry"
+)
+
+const (
+	DefaultRequestTimeout = 5 * time.Second
+	DefaultRetryInterval  = 2 * time.Second
+	DefaultAttemptCount   = 3
+)
+
+// CheckFunc is called on each attempt with the parent context and the
+// per-request timeout. Return nil on success, a regular error to trigger
+// a retry, or backoff.Permanent(err) to abort immediately.
+type CheckFunc func(ctx context.Context, requestTimeout time.Duration) error
+
+// Check runs checkFn up to attemptCount times, sleeping retryInterval between
+// attempts. Zero-valued parameters fall back to their Default* constants.
+// If checkFn returns context.DeadlineExceeded the next backoff sleep is
+// skipped, since the request timeout already consumed enough wall-clock time.
+func Check(ctx context.Context, requestTimeout time.Duration, retryInterval time.Duration, attemptCount uint64, checkFn CheckFunc) error {
+	if requestTimeout == 0 {
+		requestTimeout = DefaultRequestTimeout
+	}
+	if retryInterval == 0 {
+		retryInterval = DefaultRetryInterval
+	}
+	if attemptCount == 0 {
+		attemptCount = DefaultAttemptCount
+	}
+
+	// Run checkFn given number of times. Getting a timeout from checkFn causes an immediate retry,
+	// we don't wait another retryInterval before performing the next check.
+	skippableBoff := retry.NewSkippableBackOff(backoff.NewConstantBackOff(retryInterval))
+	boff := backoff.WithContext(backoff.WithMaxRetries(skippableBoff, attemptCount-1), ctx)
+	return backoff.Retry(func() error {
+		err := checkFn(ctx, requestTimeout)
+		if errors.Is(err, context.DeadlineExceeded) {
+			skippableBoff.SkipNext()
+		}
+		return err
+	}, boff)
+}