Implement a per-item rate-limiter

.vscode/launch.json

@@ -14,7 +14,7 @@
                 "--log-level=3",
                 "--v=5",
                 "--enable-operator-policy=true",
-                "--evaluation-backoff=1",
+                "--evaluation-backoff=2",
             ],
             "env": {
                 "WATCH_NAMESPACE": "managed",
@@ -35,7 +35,7 @@
                 "--v=5",
                 "--enable-operator-policy=true",
                 "--target-kubeconfig-path=${workspaceFolder}/kubeconfig_managed2",
-                "--evaluation-backoff=1",
+                "--evaluation-backoff=2",
             ],
             "env": {
                 "WATCH_NAMESPACE": "managed",

controllers/configurationpolicy_controller.go

@@ -8,7 +8,6 @@ import (
 	"encoding/base64"
 	"errors"
 	"fmt"
-	"math"
 	"reflect"
 	"regexp"
 	"sort"
@@ -106,6 +105,7 @@ func (r *ConfigurationPolicyReconciler) SetupWithManager(
 		Named(ControllerName).
 		WithOptions(controller.Options{
 			MaxConcurrentReconciles: int(evaluationConcurrency),
+			RateLimiter:             newPolicyRateLimiter(r.EvalBackoffSeconds),
 		}).
 		For(&policyv1.ConfigurationPolicy{}, builder.WithPredicates(
 			predicate.Funcs{
@@ -449,13 +449,6 @@ func (r *ConfigurationPolicyReconciler) shouldEvaluatePolicy(
 		}
 	}
 
-	lastEvaluated, err := time.Parse(time.RFC3339, policy.Status.LastEvaluated)
-	if err != nil {
-		log.Error(err, "The policy has an invalid status.lastEvaluated value. Will evaluate it now.")
-
-		return true, 0
-	}
-
 	usesSelector := policy.Spec.NamespaceSelector.LabelSelector != nil ||
 		len(policy.Spec.NamespaceSelector.Include) != 0
 
@@ -488,20 +481,6 @@ func (r *ConfigurationPolicyReconciler) shouldEvaluatePolicy(
 		return false, 0
 
 	case errors.Is(getIntervalErr, policyv1.ErrIsWatch):
-		minNextEval := lastEvaluated.Add(time.Second * time.Duration(r.EvalBackoffSeconds))
-		durationLeft := minNextEval.Sub(now)
-
-		if durationLeft > 0 {
-			log.V(1).Info(
-				"The policy evaluation is configured for a watch event but rescheduling the evaluation due to the "+
-					"configured evaluation backoff",
-				"evaluationBackoffSeconds", r.EvalBackoffSeconds,
-				"remainingSeconds", math.Round(durationLeft.Seconds()),
-			)
-
-			return false, durationLeft
-		}
-
 		log.V(1).Info("The policy evaluation is configured for a watch event. Will evaluate now.")
 
 		return true, 0
@@ -516,6 +495,16 @@ func (r *ConfigurationPolicyReconciler) shouldEvaluatePolicy(
 		return true, 0
 	}
 
+	// At this point, we have a valid evaluation interval, we can now determine
+	// how long we need to wait (if at all).
+
+	lastEvaluated, err := time.Parse(time.RFC3339, policy.Status.LastEvaluated)
+	if err != nil {
+		log.Error(err, "The policy has an invalid status.lastEvaluated value. Will evaluate it now.")
+
+		return true, 0
+	}
+
 	nextEvaluation := lastEvaluated.Add(interval)
 	durationLeft := nextEvaluation.Sub(now)
 

controllers/operatorpolicy_controller.go

@@ -38,6 +38,7 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
@@ -128,6 +129,9 @@ func (r *OperatorPolicyReconciler) SetupWithManager(
 ) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		Named(OperatorControllerName).
+		WithOptions(controller.Options{
+			RateLimiter: newPolicyRateLimiter(2),
+		}).
 		For(&policyv1beta1.OperatorPolicy{}, builder.WithPredicates(predicate.Funcs{
 			// Skip most pure status/metadata updates
 			UpdateFunc: func(e event.UpdateEvent) bool {

controllers/ratelimit.go

@@ -0,0 +1,64 @@
+// Copyright (c) 2025 Red Hat, Inc.
+// Copyright Contributors to the Open Cluster Management project
+
+package controllers
+
+import (
+	"sync"
+	"time"
+
+	"golang.org/x/time/rate"
+	"k8s.io/client-go/util/workqueue"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+)
+
+func newPolicyRateLimiter(minimumSecondsPerItem uint32) workqueue.TypedRateLimiter[reconcile.Request] {
+	return workqueue.NewTypedMaxOfRateLimiter(
+		// Based on the one in [email protected] DefaultTypedControllerRateLimiter
+		workqueue.NewTypedItemExponentialFailureRateLimiter[reconcile.Request](
+			500*time.Millisecond, // base delay: 0.5 seconds (5ms in client-go's default)
+			10*time.Minute,       // max delay: 10 minutes (16m40s in client-go's default)
+		),
+		// This is an overall (not per-item) limiter with 10 qps, 100 bucket size.
+		// This is identical to the one in [email protected] DefaultTypedControllerRateLimiter
+		&workqueue.TypedBucketRateLimiter[reconcile.Request]{
+			Limiter: rate.NewLimiter(rate.Limit(10), 100),
+		},
+		// This limits each item individually, so each has a minimum interval between reconciles.
+		&PerItemRateLimiter[reconcile.Request]{
+			limiters: map[reconcile.Request]*rate.Limiter{},
+			rate:     rate.Every(time.Second * time.Duration(minimumSecondsPerItem)),
+			burst:    1,
+		},
+	)
+}
+
+type PerItemRateLimiter[T comparable] struct {
+	lock     sync.Mutex
+	limiters map[T]*rate.Limiter
+	rate     rate.Limit
+	burst    int
+}
+
+// Forget is a no-op for a PerItemRateLimiter. RateLimiters in client-go only limit retries on
+// failures, but this limiter applies to *all* requests.
+func (r *PerItemRateLimiter[T]) Forget(item T) {
+}
+
+// NumRequeues always returns 0 for a PerItemRateLimiter.
+func (r *PerItemRateLimiter[T]) NumRequeues(item T) int {
+	return 0
+}
+
+func (r *PerItemRateLimiter[T]) When(item T) time.Duration {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	limiter, ok := r.limiters[item]
+	if !ok {
+		limiter = rate.NewLimiter(r.rate, r.burst)
+		r.limiters[item] = limiter
+	}
+
+	return limiter.Reserve().Delay()
+}

go.mod

@@ -21,6 +21,7 @@ require (
 	github.com/stretchr/testify v1.10.0
 	go.uber.org/zap v1.27.0
 	golang.org/x/mod v0.24.0
+	golang.org/x/time v0.11.0
 	gopkg.in/yaml.v3 v3.0.1
 	k8s.io/api v0.31.9
 	k8s.io/apiextensions-apiserver v0.31.9
@@ -101,7 +102,6 @@ require (
 	golang.org/x/sys v0.33.0 // indirect
 	golang.org/x/term v0.32.0 // indirect
 	golang.org/x/text v0.25.0 // indirect
-	golang.org/x/time v0.11.0 // indirect
 	golang.org/x/tools v0.33.0 // indirect
 	gomodules.xyz/jsonpatch/v2 v2.5.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240528184218-531527333157 // indirect

main.go

@@ -806,7 +806,7 @@ func parseOpts(flags *pflag.FlagSet, args []string) *ctrlOpts {
 	flags.Uint32Var(
 		&opts.evalBackoffSeconds,
 		"evaluation-backoff",
-		10,
+		5,
 		"The number of seconds before a policy is eligible for reevaluation in watch mode (throttles frequently "+
 			"evaluated policies)",
 	)

main_test.go

@@ -21,8 +21,8 @@ func TestRunMain(t *testing.T) {
 		"--leader-elect=false",
 		fmt.Sprintf("--target-kubeconfig-path=%s", os.Getenv("TARGET_KUBECONFIG_PATH")),
 		"--log-level=1",
-		// Speed up the tests by not throttling the policy evaluations
-		"--evaluation-backoff=1",
+		// Speed up the tests by not throttling the policy evaluations very much
+		"--evaluation-backoff=2",
 		"--enable-operator-policy=true",
 	)