diff --git a/charts/ferrvault-operator/README.md b/charts/ferrvault-operator/README.md index 70e8a30..3101791 100644 --- a/charts/ferrvault-operator/README.md +++ b/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` | | diff --git a/charts/ferrvault-operator/templates/deployment.yaml b/charts/ferrvault-operator/templates/deployment.yaml index d6d528e..c780426 100644 --- a/charts/ferrvault-operator/templates/deployment.yaml +++ b/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 }} diff --git a/charts/ferrvault-operator/values.yaml b/charts/ferrvault-operator/values.yaml index 53e96e5..0a24129 100644 --- a/charts/ferrvault-operator/values.yaml +++ b/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. diff --git a/cmd/main.go b/cmd/main.go index eee3cb7..c9f7eb3 100644 --- a/cmd/main.go +++ b/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") diff --git a/internal/controller/ferrvaultconnection_controller.go b/internal/controller/ferrvaultconnection_controller.go index 2f0026e..ac22e49 100644 --- a/internal/controller/ferrvaultconnection_controller.go +++ b/internal/controller/ferrvaultconnection_controller.go @@ -3,6 +3,7 @@ package controller import ( "context" "fmt" + "sync" "time" corev1 "k8s.io/api/core/v1" @@ -35,8 +36,17 @@ 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 @@ -44,11 +54,13 @@ type FerrVaultConnectionReconciler struct { 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) diff --git a/internal/controller/ferrvaultsecret_controller.go b/internal/controller/ferrvaultsecret_controller.go index 3b70347..8d7c558 100644 --- a/internal/controller/ferrvaultsecret_controller.go +++ b/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,6 +59,7 @@ 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) }() @@ -64,6 +67,7 @@ func (r *FerrVaultSecretReconciler) Reconcile(ctx context.Context, req ctrl.Requ 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) } diff --git a/internal/controller/heartbeat.go b/internal/controller/heartbeat.go new file mode 100644 index 0000000..80c5a90 --- /dev/null +++ b/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 +} diff --git a/internal/controller/heartbeat_test.go b/internal/controller/heartbeat_test.go new file mode 100644 index 0000000..f091870 --- /dev/null +++ b/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) + } +} diff --git a/internal/controller/rate_limit_backoff.go b/internal/controller/rate_limit_backoff.go new file mode 100644 index 0000000..83e9554 --- /dev/null +++ b/internal/controller/rate_limit_backoff.go @@ -0,0 +1,65 @@ +package controller + +import ( + "math/rand" + "sync" + "time" + + "k8s.io/apimachinery/pkg/types" +) + +const ( + rateLimitBackoffBase = 5 * time.Second + rateLimitBackoffMax = 5 * time.Minute + rateLimitBackoffShift = 8 +) + +type rateLimitBackoff struct { + mu sync.Mutex + attempts map[types.NamespacedName]int + rand *rand.Rand +} + +func newRateLimitBackoff() *rateLimitBackoff { + return &rateLimitBackoff{ + attempts: make(map[types.NamespacedName]int), + rand: rand.New(rand.NewSource(time.Now().UnixNano())), + } +} + +func (b *rateLimitBackoff) next(key types.NamespacedName) time.Duration { + b.mu.Lock() + defer b.mu.Unlock() + + attempt := b.attempts[key] + b.attempts[key] = attempt + 1 + return b.jitter(backoffFor(attempt)) +} + +func (b *rateLimitBackoff) forget(key types.NamespacedName) { + b.mu.Lock() + defer b.mu.Unlock() + delete(b.attempts, key) +} + +func (b *rateLimitBackoff) jitter(d time.Duration) time.Duration { + half := d / 2 + if half <= 0 { + return d + } + return half + time.Duration(b.rand.Int63n(int64(half))) +} + +func backoffFor(attempt int) time.Duration { + if attempt < 0 { + attempt = 0 + } + if attempt > rateLimitBackoffShift { + attempt = rateLimitBackoffShift + } + d := rateLimitBackoffBase << uint(attempt) + if d > rateLimitBackoffMax { + return rateLimitBackoffMax + } + return d +} diff --git a/internal/controller/rate_limit_backoff_test.go b/internal/controller/rate_limit_backoff_test.go new file mode 100644 index 0000000..fb38c0f --- /dev/null +++ b/internal/controller/rate_limit_backoff_test.go @@ -0,0 +1,89 @@ +package controller + +import ( + "testing" + "time" + + "k8s.io/apimachinery/pkg/types" +) + +func TestBackoffGrowsAndCaps(t *testing.T) { + cases := []struct { + attempt int + want time.Duration + }{ + {attempt: 0, want: rateLimitBackoffBase}, + {attempt: 1, want: 10 * time.Second}, + {attempt: 2, want: 20 * time.Second}, + {attempt: 5, want: 160 * time.Second}, + {attempt: 6, want: rateLimitBackoffMax}, + {attempt: 99, want: rateLimitBackoffMax}, + {attempt: -1, want: rateLimitBackoffBase}, + } + for _, tc := range cases { + if got := backoffFor(tc.attempt); got != tc.want { + t.Errorf("backoffFor(%d) = %s, want %s", tc.attempt, got, tc.want) + } + } +} + +func TestConsecutiveRateLimitsOutrunAFixedWait(t *testing.T) { + const ( + resources = 29 + refillPerS = 1.0 + fixedWait = 20 * time.Second + ) + + offered := func(wait time.Duration) float64 { + return resources / wait.Seconds() + } + + if offered(fixedWait) <= refillPerS { + t.Fatalf("the fixed wait this replaces was not actually saturating, "+ + "offered %.2f/s against %.2f/s refill", offered(fixedWait), refillPerS) + } + + b := newRateLimitBackoff() + key := types.NamespacedName{Namespace: "default", Name: "app-secrets"} + var last time.Duration + for round := 0; round < 8; round++ { + last = b.next(key) + if offered(last) <= refillPerS { + return + } + } + t.Fatalf("still offering %.2f/s after eight rounds, last wait %s", + offered(last), last) +} + +func TestJitterStaysWithinHalfTheDelay(t *testing.T) { + b := newRateLimitBackoff() + key := types.NamespacedName{Namespace: "default", Name: "app-secrets"} + for i := 0; i < 200; i++ { + b.forget(key) + got := b.next(key) + if got < rateLimitBackoffBase/2 || got >= rateLimitBackoffBase { + t.Fatalf("jittered base = %s, want within [%s, %s)", + got, rateLimitBackoffBase/2, rateLimitBackoffBase) + } + } +} + +func TestForgetResetsAndKeysAreIndependent(t *testing.T) { + b := newRateLimitBackoff() + one := types.NamespacedName{Namespace: "default", Name: "one"} + two := types.NamespacedName{Namespace: "default", Name: "two"} + + for i := 0; i < 4; i++ { + b.next(one) + } + + if got := b.next(two); got >= rateLimitBackoffBase { + t.Errorf("a second key started at %s, want a base-length wait", got) + } + + b.forget(one) + if got := b.next(one); got >= rateLimitBackoffBase { + t.Errorf("after forget the wait was %s, want a base-length wait", got) + } +}