feat: introduced custom rate limiter based on options pattern

On-behalf-of: SAP [email protected]

Author: OlegErshov

controller/lifecycle/controllerruntime/lifecycle.go

@@ -15,6 +15,7 @@ import (
 	"github.com/platform-mesh/golang-commons/controller/lifecycle"
 	"github.com/platform-mesh/golang-commons/controller/lifecycle/api"
 	"github.com/platform-mesh/golang-commons/controller/lifecycle/conditions"
+	"github.com/platform-mesh/golang-commons/controller/lifecycle/ratelimiter"
 	"github.com/platform-mesh/golang-commons/controller/lifecycle/runtimeobject"
 	"github.com/platform-mesh/golang-commons/controller/lifecycle/spread"
 	"github.com/platform-mesh/golang-commons/controller/lifecycle/subroutine"
@@ -29,6 +30,7 @@ type LifecycleManager struct {
 	spreader           *spread.Spreader
 	conditionsManager  *conditions.ConditionManager
 	prepareContextFunc api.PrepareContextFunc
+	rateLimiterConfig  ratelimiter.Config
 }
 
 func NewLifecycleManager(subroutines []subroutine.Subroutine, operatorName string, controllerName string, client client.Client, log *logger.Logger) *LifecycleManager {
@@ -83,10 +85,15 @@ func (l *LifecycleManager) SetupWithManagerBuilder(mgr ctrl.Manager, maxReconcil
 	}
 
 	eventPredicates = append([]predicate.Predicate{filter.DebugResourcesBehaviourPredicate(debugLabelValue)}, eventPredicates...)
+	opts := controller.Options{
+		MaxConcurrentReconciles: maxReconciles,
+		RateLimiter:             ratelimiter.NewStaticThenExponentialRateLimiter[reconcile.Request](l.rateLimiterConfig),
+	}
+
 	return ctrl.NewControllerManagedBy(mgr).
 		Named(reconcilerName).
 		For(instance).
-		WithOptions(controller.Options{MaxConcurrentReconciles: maxReconciles}).
+		WithOptions(opts).
 		WithEventFilter(predicate.And(eventPredicates...)), nil
 }
 func (l *LifecycleManager) SetupWithManager(mgr ctrl.Manager, maxReconciles int, reconcilerName string, instance runtimeobject.RuntimeObject, debugLabelValue string, r reconcile.Reconciler, log *logger.Logger, eventPredicates ...predicate.Predicate) error {
@@ -123,3 +130,9 @@ func (l *LifecycleManager) WithConditionManagement() *LifecycleManager {
 	l.conditionsManager = conditions.NewConditionManager()
 	return l
 }
+
+func (l *LifecycleManager) WithRateLimiter(opts ...ratelimiter.Option) *LifecycleManager {
+	cfg := ratelimiter.NewConfig(opts...)
+	l.rateLimiterConfig = cfg
+	return l
+}

controller/lifecycle/controllerruntime/lifecycle_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	goerrors "errors"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -13,6 +14,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 
+	"github.com/platform-mesh/golang-commons/controller/lifecycle/ratelimiter"
 	"github.com/platform-mesh/golang-commons/controller/lifecycle/runtimeobject"
 	"github.com/platform-mesh/golang-commons/controller/lifecycle/subroutine"
 	pmtesting "github.com/platform-mesh/golang-commons/controller/testSupport"
@@ -152,6 +154,27 @@ func TestLifecycle(t *testing.T) {
 		assert.True(t, true, l.ConditionsManager() != nil)
 	})
 
+	t.Run("WithRateLimiter", func(t *testing.T) {
+		fakeClient := pmtesting.CreateFakeClient(t, &pmtesting.TestApiObject{})
+		_, log := createLifecycleManager([]subroutine.Subroutine{}, fakeClient)
+
+		l := NewLifecycleManager([]subroutine.Subroutine{}, "test-operator", "test-controller", fakeClient, log.Logger)
+		expectedCfg := ratelimiter.Config{
+			StaticRequeueDelay:        5 * time.Second,
+			StaticWindow:              10 * time.Second,
+			ExponentialInitialBackoff: 5 * time.Second,
+			ExponentialMaxBackoff:     time.Minute,
+		}
+		l.WithRateLimiter(
+			ratelimiter.WithRequeueDelay(expectedCfg.StaticRequeueDelay),
+			ratelimiter.WithStaticWindow(expectedCfg.StaticWindow),
+			ratelimiter.WithExponentialInitialBackoff(expectedCfg.ExponentialInitialBackoff),
+			ratelimiter.WithExponentialMaxBackoff(expectedCfg.ExponentialMaxBackoff),
+		)
+
+		assert.Equal(t, expectedCfg, l.rateLimiterConfig)
+	})
+
 }
 
 type testReconciler struct {

controller/lifecycle/multicluster/lifecycle.go

@@ -17,6 +17,7 @@ import (
 	"github.com/platform-mesh/golang-commons/controller/lifecycle"
 	"github.com/platform-mesh/golang-commons/controller/lifecycle/api"
 	"github.com/platform-mesh/golang-commons/controller/lifecycle/conditions"
+	"github.com/platform-mesh/golang-commons/controller/lifecycle/ratelimiter"
 	"github.com/platform-mesh/golang-commons/controller/lifecycle/runtimeobject"
 	"github.com/platform-mesh/golang-commons/controller/lifecycle/spread"
 	"github.com/platform-mesh/golang-commons/controller/lifecycle/subroutine"
@@ -35,6 +36,7 @@ type LifecycleManager struct {
 	spreader           *spread.Spreader
 	conditionsManager  *conditions.ConditionManager
 	prepareContextFunc api.PrepareContextFunc
+	rateLimiterConfig  ratelimiter.Config
 }
 
 func NewLifecycleManager(subroutines []subroutine.Subroutine, operatorName string, controllerName string, mgr ClusterGetter, log *logger.Logger) *LifecycleManager {
@@ -95,7 +97,9 @@ func (l *LifecycleManager) SetupWithManagerBuilder(mgr mcmanager.Manager, maxRec
 	eventPredicates = append([]predicate.Predicate{filter.DebugResourcesBehaviourPredicate(debugLabelValue)}, eventPredicates...)
 	opts := controller.TypedOptions[mcreconcile.Request]{
 		MaxConcurrentReconciles: maxReconciles,
+		RateLimiter:             ratelimiter.NewStaticThenExponentialRateLimiter[mcreconcile.Request](l.rateLimiterConfig),
 	}
+
 	return mcbuilder.ControllerManagedBy(mgr).
 		Named(reconcilerName).
 		For(instance).
@@ -136,3 +140,9 @@ func (l *LifecycleManager) WithConditionManagement() api.Lifecycle {
 	l.conditionsManager = conditions.NewConditionManager()
 	return l
 }
+
+func (l *LifecycleManager) WithRateLimiter(opts ...ratelimiter.Option) *LifecycleManager {
+	cfg := ratelimiter.NewConfig(opts...)
+	l.rateLimiterConfig = cfg
+	return l
+}

controller/lifecycle/ratelimiter/config.go

@@ -0,0 +1,55 @@
+package ratelimiter
+
+import (
+	"time"
+)
+
+type Config struct {
+	StaticRequeueDelay        time.Duration
+	StaticWindow              time.Duration
+	ExponentialInitialBackoff time.Duration
+	ExponentialMaxBackoff     time.Duration
+}
+
+var defaultConfig = Config{
+	StaticRequeueDelay:        5 * time.Second,
+	StaticWindow:              1 * time.Minute,
+	ExponentialInitialBackoff: 5 * time.Second,
+	ExponentialMaxBackoff:     2 * time.Minute,
+}
+
+type Option func(*Config)
+
+func WithStaticWindow(d time.Duration) Option {
+	return func(c *Config) {
+		c.StaticWindow = d
+	}
+}
+
+func WithRequeueDelay(d time.Duration) Option {
+	return func(c *Config) {
+		c.StaticRequeueDelay = d
+	}
+}
+
+func WithExponentialInitialBackoff(d time.Duration) Option {
+	return func(c *Config) {
+		c.ExponentialInitialBackoff = d
+	}
+}
+
+func WithExponentialMaxBackoff(d time.Duration) Option {
+	return func(c *Config) {
+		c.ExponentialMaxBackoff = d
+	}
+}
+
+func NewConfig(options ...Option) Config {
+	cfg := defaultConfig
+
+	for _, option := range options {
+		option(&cfg)
+	}
+
+	return cfg
+}

controller/lifecycle/ratelimiter/static_exponential.go

@@ -0,0 +1,65 @@
+package ratelimiter
+
+import (
+	"sync"
+	"time"
+
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/utils/clock"
+)
+
+type StaticThenExponentialRateLimiter[T comparable] struct {
+	failuresLock sync.Mutex
+	firstAttempt map[T]time.Time
+
+	staticDelay  time.Duration
+	staticWindow time.Duration
+
+	exponential workqueue.TypedRateLimiter[T]
+	clock       clock.Clock
+}
+
+func NewStaticThenExponentialRateLimiter[T comparable](cfg Config) *StaticThenExponentialRateLimiter[T] {
+	return &StaticThenExponentialRateLimiter[T]{
+		staticDelay:  cfg.StaticRequeueDelay,
+		staticWindow: cfg.StaticWindow,
+		exponential: workqueue.NewTypedItemExponentialFailureRateLimiter[T](
+			cfg.ExponentialInitialBackoff,
+			cfg.ExponentialMaxBackoff,
+		),
+		firstAttempt: make(map[T]time.Time),
+		clock:        clock.RealClock{},
+	}
+}
+
+func (r *StaticThenExponentialRateLimiter[T]) When(item T) time.Duration {
+	r.failuresLock.Lock()
+	defer r.failuresLock.Unlock()
+
+	now := r.clock.Now()
+
+	first, exists := r.firstAttempt[item]
+	if !exists {
+		first = now
+		r.firstAttempt[item] = first
+	}
+
+	timeSinceFirst := now.Sub(first)
+	if timeSinceFirst <= r.staticWindow {
+		return r.staticDelay
+	}
+
+	return r.exponential.When(item)
+}
+
+func (r *StaticThenExponentialRateLimiter[T]) Forget(item T) {
+	r.failuresLock.Lock()
+	defer r.failuresLock.Unlock()
+
+	delete(r.firstAttempt, item)
+	r.exponential.Forget(item)
+}
+
+func (r *StaticThenExponentialRateLimiter[T]) NumRequeues(item T) int {
+	return r.exponential.NumRequeues(item)
+}

controller/lifecycle/ratelimiter/static_exponential_test.go

@@ -0,0 +1,33 @@
+package ratelimiter
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/types"
+	clocktesting "k8s.io/utils/clock/testing"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+)
+
+func TestStaticThenExponentialRateLimiter_Forget(t *testing.T) {
+	cfg := Config{
+		StaticRequeueDelay:        1 * time.Second,
+		StaticWindow:              5 * time.Second,
+		ExponentialInitialBackoff: 2 * time.Second,
+		ExponentialMaxBackoff:     1 * time.Minute,
+	}
+	limiter := NewStaticThenExponentialRateLimiter[reconcile.Request](cfg)
+	fakeClock := clocktesting.NewFakeClock(time.Now())
+	limiter.clock = fakeClock
+
+	item := reconcile.Request{NamespacedName: types.NamespacedName{Name: "name", Namespace: "namespace"}}
+	require.Equal(t, cfg.StaticRequeueDelay, limiter.When(item))
+	fakeClock.Step(10 * time.Second)
+	_ = limiter.When(item)
+
+	limiter.Forget(item)
+	require.Equal(t, 0, limiter.NumRequeues(item))
+	delay := limiter.When(item)
+	require.Equal(t, cfg.StaticRequeueDelay, delay)
+}