fix(operator): back off exponentially on 429 and fail liveness when the loop stalls

charts/ferrvault-operator/README.md

@@ -51,6 +51,7 @@ kubectl delete crd ferrvaultsecrets.ferrvault.com ferrvaultconnections.ferrvault
 | `leaderElection.id` | `ferrvault-operator.ferrvault.com` | Change to run multiple isolated instances in one cluster. |
 | `watchNamespace` | `""` (cluster-wide) | Single namespace scope when set. |
 | `defaultRefreshInterval` | `1h` | Fallback for `FerrVaultSecret.spec.refreshInterval`. |
+| `stallThreshold` | `15m` | Liveness fails when no reconcile completes for this long while FerrVault resources exist. Keep it above the 10m connection probe interval. |
 | `logLevel` | `info` | `debug`, `info`, `warn`, `error`. |
 | `extraArgs` | `[]` | Extra manager CLI flags. |
 | `metrics.enabled` | `true` | |

charts/ferrvault-operator/templates/deployment.yaml

@@ -37,6 +37,7 @@ spec:
             - --metrics-bind-address=:{{ .Values.metrics.port }}
             - --health-probe-bind-address=:{{ .Values.probe.port }}
             - --default-refresh-interval={{ .Values.defaultRefreshInterval }}
+            - --stall-threshold={{ .Values.stallThreshold }}
             {{- if .Values.leaderElection.enabled }}
             - --leader-elect
             - --leader-elect-id={{ .Values.leaderElection.id }}

charts/ferrvault-operator/values.yaml

@@ -29,6 +29,10 @@ watchNamespace: ""
 # --- Runtime flags ---
 # Default refresh interval applied when a FerrVaultSecret omits `spec.refreshInterval`.
 defaultRefreshInterval: 1h
+# Liveness fails when no reconcile has completed for this long while FerrVault
+# resources exist. Keep it above the 10m connection probe interval, which is
+# the shortest reconcile a healthy operator is guaranteed to perform.
+stallThreshold: 15m
 # Log level for the operator's zap logger (debug, info, warn, error).
 logLevel: info
 # Extra command-line flags appended to the manager binary.

cmd/main.go

@@ -44,6 +44,7 @@ func main() {
 		leaderElectionID       string
 		defaultRefreshInterval time.Duration
 		watchNamespace         string
+		stallThreshold         time.Duration
 	)
 
 	flag.StringVar(&metricsAddr, "metrics-bind-address", ":8080",
@@ -59,6 +60,9 @@ func main() {
 		"Fallback refresh interval used when a FerrVaultSecret omits spec.refreshInterval.")
 	flag.StringVar(&watchNamespace, "watch-namespace", "",
 		"Restrict the controller to a single namespace. Empty means cluster-wide.")
+	flag.DurationVar(&stallThreshold, "stall-threshold", 15*time.Minute,
+		"Fail the liveness probe when no reconcile has completed for this long "+
+			"while FerrVault resources exist. Must stay above the connection probe interval.")
 
 	opts := zap.Options{Development: false}
 	opts.BindFlags(flag.CommandLine)
@@ -117,21 +121,24 @@ func main() {
 	}
 
 	broker := controller.NewTokenBroker(mgr.GetClient())
+	heartbeat := controller.NewHeartbeat(time.Now())
 
 	if err := (&controller.FerrVaultSecretReconciler{
 		Client:                 mgr.GetClient(),
 		Scheme:                 mgr.GetScheme(),
 		DefaultRefreshInterval: defaultRefreshInterval,
 		Broker:                 broker,
+		Heartbeat:              heartbeat,
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "FerrVaultSecret")
 		os.Exit(1)
 	}
 
 	if err := (&controller.FerrVaultConnectionReconciler{
-		Client: mgr.GetClient(),
-		Scheme: mgr.GetScheme(),
-		Broker: broker,
+		Client:    mgr.GetClient(),
+		Scheme:    mgr.GetScheme(),
+		Broker:    broker,
+		Heartbeat: heartbeat,
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "FerrVaultConnection")
 		os.Exit(1)
@@ -141,6 +148,11 @@ func main() {
 		setupLog.Error(err, "unable to set up health check")
 		os.Exit(1)
 	}
+	if err := mgr.AddHealthzCheck("reconcile-progress",
+		controller.StallChecker(mgr.GetClient(), heartbeat, stallThreshold)); err != nil {
+		setupLog.Error(err, "unable to set up stall check")
+		os.Exit(1)
+	}
 	if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
 		setupLog.Error(err, "unable to set up readiness check")
 		os.Exit(1)
@@ -149,6 +161,7 @@ func main() {
 	setupLog.Info("starting manager",
 		"watchNamespace", fmtNs(watchNamespace),
 		"defaultRefreshInterval", defaultRefreshInterval,
+		"stallThreshold", stallThreshold,
 	)
 	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
 		setupLog.Error(err, "problem running manager")

internal/controller/ferrvaultconnection_controller.go

@@ -3,6 +3,7 @@ package controller
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
@@ -35,20 +36,31 @@ const connectionInUseRequeue = 30 * time.Second
 
 type FerrVaultConnectionReconciler struct {
 	client.Client
-	Scheme *runtime.Scheme
-	Broker *TokenBroker
+	Scheme    *runtime.Scheme
+	Broker    *TokenBroker
+	Heartbeat *Heartbeat
+
+	backoffOnce sync.Once
+	backoff     *rateLimitBackoff
+}
+
+func (r *FerrVaultConnectionReconciler) rateLimit() *rateLimitBackoff {
+	r.backoffOnce.Do(func() { r.backoff = newRateLimitBackoff() })
+	return r.backoff
 }
 
 // +kubebuilder:rbac:groups=ferrvault.com,resources=ferrvaultconnections,verbs=get;list;watch
 // +kubebuilder:rbac:groups=ferrvault.com,resources=ferrvaultconnections/status,verbs=get;update;patch
 
 func (r *FerrVaultConnectionReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	logger := log.FromContext(ctx).WithValues("ferrvaultconnection", req.NamespacedName)
+	defer func() { r.Heartbeat.Beat(time.Now()) }()
 
 	var conn fvv1alpha1.FerrVaultConnection
 	if err := r.Get(ctx, req.NamespacedName, &conn); err != nil {
 		if apierrors.IsNotFound(err) {
 			DeleteConnectionReady(req.Namespace, req.Name)
+			r.rateLimit().forget(req.NamespacedName)
 			return ctrl.Result{}, nil
 		}
 		return ctrl.Result{}, fmt.Errorf("load FerrVaultConnection: %w", err)
@@ -82,10 +94,12 @@ func (r *FerrVaultConnectionReconciler) Reconcile(ctx context.Context, req ctrl.
 	// connexion en échec garde la cause de son échec. Seul l'horodatage du
 	// dernier contrôle serait trompeur, et il n'est pas touché non plus.
 	if ferrvault.IsRateLimited(probeErr) {
+		after := r.rateLimit().next(req.NamespacedName)
 		logger.Info("probe rate limited, condition left as-is",
-			"requeueAfter", rateLimitRequeue)
-		return ctrl.Result{RequeueAfter: rateLimitRequeue}, nil
+			"requeueAfter", after)
+		return ctrl.Result{RequeueAfter: after}, nil
 	}
+	r.rateLimit().forget(req.NamespacedName)
 
 	logger.Info("probe finished", "ready", status, "reason", reason)
 

internal/controller/ferrvaultsecret_controller.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"sort"
+	"sync"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
@@ -28,14 +29,6 @@ const (
 	fvAnnotationContentHash       = "ferrvault.com/content-hash"
 	fvAnnotationRestartedAt       = "ferrvault.com/restarted-at"
 	fvSecretFinalizer             = "ferrvault.com/secret-cleanup"
-
-	// How long to wait after a 429 before trying again.
-	//
-	// The API's bucket refills at one token per second, so a short wait is
-	// enough for a handful of resources to get through. Long enough not to
-	// hammer a server that just asked for room, short enough that a resource
-	// blocked by someone else's burst is current again within the minute.
-	rateLimitRequeue = 20 * time.Second
 )
 
 type FerrVaultSecretReconciler struct {
@@ -44,6 +37,15 @@ type FerrVaultSecretReconciler struct {
 	DefaultRefreshInterval time.Duration
 	ClientFactory          ClientFactory
 	Broker                 *TokenBroker
+	Heartbeat              *Heartbeat
+
+	backoffOnce sync.Once
+	backoff     *rateLimitBackoff
+}
+
+func (r *FerrVaultSecretReconciler) rateLimit() *rateLimitBackoff {
+	r.backoffOnce.Do(func() { r.backoff = newRateLimitBackoff() })
+	return r.backoff
 }
 
 // +kubebuilder:rbac:groups=ferrvault.com,resources=ferrvaultsecrets,verbs=get;list;watch;create;update;patch;delete
@@ -57,13 +59,15 @@ func (r *FerrVaultSecretReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 	begin := time.Now()
 	result := "failure"
 	defer func() {
+		r.Heartbeat.Beat(time.Now())
 		ObserveReconcile(begin, result)
 	}()
 
 	var cr fvv1alpha1.FerrVaultSecret
 	if err := r.Get(ctx, req.NamespacedName, &cr); err != nil {
 		if apierrors.IsNotFound(err) {
 			DeleteLastSyncTimestamp(req.Namespace, req.Name)
+			r.rateLimit().forget(req.NamespacedName)
 			result = "success"
 			return ctrl.Result{}, nil
 		}
@@ -125,18 +129,20 @@ func (r *FerrVaultSecretReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 			cr.Spec.Selector.Names,
 		)
 	}
+	// A 429 is a "call me back", not a failure. Requeue and leave the status
+	// untouched: the resource may well be up to date from an earlier successful
+	// pass, and rewriting Ready=False here is what made fourteen healthy
+	// secrets report as broken while their data was current.
+	if err != nil && ferrvault.IsRateLimited(err) {
+		after := r.rateLimit().next(req.NamespacedName)
+		logger.Info("rate limited by the FerrVault API, backing off",
+			"requeueAfter", after)
+		IncSyncError("RateLimited")
+		return ctrl.Result{RequeueAfter: after}, nil
+	}
+	r.rateLimit().forget(req.NamespacedName)
+
 	if err != nil {
-		// A 429 is a "call me back", not a failure. Requeue shortly and leave
-		// the status untouched: the resource may well be up to date from an
-		// earlier successful pass, and rewriting Ready=False here is what made
-		// fourteen healthy secrets report as broken while their data was
-		// current. Nothing is lost — the requeue re-reads within the minute.
-		if ferrvault.IsRateLimited(err) {
-			logger.Info("rate limited by the FerrVault API, retrying shortly",
-				"requeueAfter", rateLimitRequeue)
-			IncSyncError("RateLimited")
-			return ctrl.Result{RequeueAfter: rateLimitRequeue}, nil
-		}
 		if ferrvault.IsAuthError(err) {
 			return r.failReadyWithRequeue(ctx, &cr, "AuthFailed", err.Error(), 5*time.Minute)
 		}

internal/controller/heartbeat.go

@@ -0,0 +1,72 @@
+package controller
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"sync"
+	"time"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/healthz"
+
+	fvv1alpha1 "github.com/FerrLabs/FerrVault/api/ferrvault/v1alpha1"
+)
+
+type Heartbeat struct {
+	mu   sync.Mutex
+	last time.Time
+}
+
+func NewHeartbeat(now time.Time) *Heartbeat {
+	return &Heartbeat{last: now}
+}
+
+func (h *Heartbeat) Beat(now time.Time) {
+	if h == nil {
+		return
+	}
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	if now.After(h.last) {
+		h.last = now
+	}
+}
+
+func (h *Heartbeat) Idle(now time.Time) time.Duration {
+	if h == nil {
+		return 0
+	}
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	return now.Sub(h.last)
+}
+
+func StallChecker(c client.Client, hb *Heartbeat, threshold time.Duration) healthz.Checker {
+	return func(req *http.Request) error {
+		idle := hb.Idle(time.Now())
+		if idle <= threshold {
+			return nil
+		}
+		watching, err := hasWatchedResources(req.Context(), c)
+		if err != nil || !watching {
+			return nil
+		}
+		return fmt.Errorf("no reconcile completed for %s", idle.Truncate(time.Second))
+	}
+}
+
+func hasWatchedResources(ctx context.Context, c client.Client) (bool, error) {
+	var conns fvv1alpha1.FerrVaultConnectionList
+	if err := c.List(ctx, &conns, client.Limit(1)); err != nil {
+		return false, err
+	}
+	if len(conns.Items) > 0 {
+		return true, nil
+	}
+	var secrets fvv1alpha1.FerrVaultSecretList
+	if err := c.List(ctx, &secrets, client.Limit(1)); err != nil {
+		return false, err
+	}
+	return len(secrets.Items) > 0, nil
+}

internal/controller/heartbeat_test.go

@@ -0,0 +1,84 @@
+package controller
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"testing"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
+
+	fvv1alpha1 "github.com/FerrLabs/FerrVault/api/ferrvault/v1alpha1"
+)
+
+const stallThresholdForTest = 15 * time.Minute
+
+func connectionFixture() *fvv1alpha1.FerrVaultConnection {
+	return &fvv1alpha1.FerrVaultConnection{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "ferrvault-dev"},
+	}
+}
+
+func checkerFor(t *testing.T, hb *Heartbeat, objs ...client.Object) error {
+	t.Helper()
+	c := fake.NewClientBuilder().WithScheme(newTestScheme(t)).WithObjects(objs...).Build()
+	return StallChecker(c, hb, stallThresholdForTest)(&http.Request{})
+}
+
+func TestABeatingOperatorIsLive(t *testing.T) {
+	hb := NewHeartbeat(time.Now().Add(-stallThresholdForTest + time.Minute))
+	if err := checkerFor(t, hb, connectionFixture()); err != nil {
+		t.Fatalf("reported dead while still reconciling: %v", err)
+	}
+}
+
+func TestAStalledOperatorWithWorkIsNotLive(t *testing.T) {
+	hb := NewHeartbeat(time.Now().Add(-stallThresholdForTest - time.Minute))
+	if err := checkerFor(t, hb, connectionFixture()); err == nil {
+		t.Fatal("reported live after reconciling nothing for longer than the threshold")
+	}
+}
+
+func TestAnIdleClusterIsNotRestarted(t *testing.T) {
+	hb := NewHeartbeat(time.Now().Add(-24 * time.Hour))
+	if err := checkerFor(t, hb); err != nil {
+		t.Fatalf("reported dead with no FerrVault resources to reconcile: %v", err)
+	}
+}
+
+func TestAnUnreadableCacheIsNotRestarted(t *testing.T) {
+	c := fake.NewClientBuilder().
+		WithScheme(newTestScheme(t)).
+		WithInterceptorFuncs(interceptor.Funcs{
+			List: func(context.Context, client.WithWatch, client.ObjectList, ...client.ListOption) error {
+				return errors.New("cache not synced")
+			},
+		}).
+		Build()
+
+	hb := NewHeartbeat(time.Now().Add(-24 * time.Hour))
+	if err := StallChecker(c, hb, stallThresholdForTest)(&http.Request{}); err != nil {
+		t.Fatalf("reported dead because the cache was unreadable: %v", err)
+	}
+}
+
+func TestBeatNeverMovesBackwards(t *testing.T) {
+	now := time.Now()
+	hb := NewHeartbeat(now)
+	hb.Beat(now.Add(-time.Hour))
+	if idle := hb.Idle(now); idle != 0 {
+		t.Fatalf("an older beat moved the heartbeat back by %s", idle)
+	}
+}
+
+func TestANilHeartbeatIsInert(t *testing.T) {
+	var hb *Heartbeat
+	hb.Beat(time.Now())
+	if idle := hb.Idle(time.Now()); idle != 0 {
+		t.Fatalf("a nil heartbeat reported %s idle", idle)
+	}
+}