feat: dynamic ratelimiter for gracefuleviction

Signed-off-by: whosefriendA <[email protected]>

Author: whosefriendA

cmd/controller-manager/app/controllermanager.go

@@ -299,6 +299,7 @@ func startClusterController(ctx controllerscontext.Context) (enabled bool, err e
 			RateLimiterOptions:                 ctx.Opts.RateLimiterOptions,
 			EnableNoExecuteTaintEviction:       ctx.Opts.FailoverConfiguration.EnableNoExecuteTaintEviction,
 			NoExecuteTaintEvictionPurgeMode:    ctx.Opts.FailoverConfiguration.NoExecuteTaintEvictionPurgeMode,
+			EvictionQueueOptions:               ctx.Opts.EvictionQueueOptions,
 		}
 		if err := taintManager.SetupWithManager(mgr); err != nil {
 			return false, err
@@ -924,6 +925,7 @@ func setupControllers(ctx context.Context, mgr controllerruntime.Manager, opts *
 			HPAControllerConfiguration:        opts.HPAControllerConfiguration,
 			FederatedResourceQuotaOptions:     opts.FederatedResourceQuotaOptions,
 			FailoverConfiguration:             opts.FailoverOptions,
+			EvictionQueueOptions:              opts.EvictionQueueOptions,
 		},
 		Context:                     ctx,
 		DynamicClientSet:            dynamicClientSet,

cmd/controller-manager/app/options/options.go

@@ -18,6 +18,8 @@ package options
 
 import (
 	"fmt"
+	evictionQueueconfig "github.com/karmada-io/karmada/pkg/controllers/cluster/evictionqueue_config"
+
 	"regexp"
 	"strings"
 	"time"
@@ -148,6 +150,8 @@ type Options struct {
 	FederatedResourceQuotaOptions FederatedResourceQuotaOptions
 	// FailoverOptions holds the Failover configurations.
 	FailoverOptions FailoverOptions
+	// EvictionQueueOptions holds the GracefulEviction
+	EvictionQueueOptions evictionQueueconfig.EvictionQueueOptions
 }
 
 // NewOptions builds an empty options.
@@ -234,6 +238,7 @@ func (o *Options) AddFlags(flags *pflag.FlagSet, allControllers, disabledByDefau
 	o.HPAControllerConfiguration.AddFlags(flags)
 	o.FederatedResourceQuotaOptions.AddFlags(flags)
 	o.FailoverOptions.AddFlags(flags)
+	o.EvictionQueueOptions.AddFlags(flags)
 	features.FeatureGate.AddFlag(flags)
 }
 

pkg/controllers/cluster/dynamic_rate_limiter.go

@@ -0,0 +1,145 @@
+/*
+Copyright 2022 The Karmada Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cluster
+
+import (
+	"time"
+
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/klog/v2"
+
+	clusterv1alpha1 "github.com/karmada-io/karmada/pkg/apis/cluster/v1alpha1"
+	config "github.com/karmada-io/karmada/pkg/controllers/cluster/evictionqueue_config"
+	"github.com/karmada-io/karmada/pkg/metrics"
+	"github.com/karmada-io/karmada/pkg/sharedcli/ratelimiterflag"
+	"github.com/karmada-io/karmada/pkg/util"
+	"github.com/karmada-io/karmada/pkg/util/fedinformer/genericmanager"
+)
+
+// maxEvictionDelay is the maximum delay for eviction when the rate is 0
+const maxEvictionDelay = 1000 * time.Second
+
+// DynamicRateLimiter adjusts its rate based on the overall health of clusters.
+// It implements the workqueue.RateLimiter interface with dynamic behavior.
+type DynamicRateLimiter[T comparable] struct {
+	resourceEvictionRate          float32
+	secondaryResourceEvictionRate float32
+	unhealthyClusterThreshold     float32
+	largeClusterNumThreshold      int
+	informerManager               genericmanager.SingleClusterInformerManager
+}
+
+// NewDynamicRateLimiter creates a new DynamicRateLimiter with the given options.
+func NewDynamicRateLimiter[T comparable](informerManager genericmanager.SingleClusterInformerManager, opts config.EvictionQueueOptions) workqueue.TypedRateLimiter[T] {
+	return &DynamicRateLimiter[T]{
+		resourceEvictionRate:          opts.ResourceEvictionRate,
+		secondaryResourceEvictionRate: opts.SecondaryResourceEvictionRate,
+		unhealthyClusterThreshold:     opts.UnhealthyClusterThreshold,
+		largeClusterNumThreshold:      opts.LargeClusterNumThreshold,
+		informerManager:               informerManager,
+	}
+}
+
+// When determines how long to wait before processing an item.
+// Returns a longer delay when the system is unhealthy.
+func (d *DynamicRateLimiter[T]) When(item T) time.Duration {
+	currentRate := d.getCurrentRate()
+	if currentRate == 0 {
+		return maxEvictionDelay
+	}
+	return time.Duration(1 / currentRate * float32(time.Second))
+}
+
+// getCurrentRate calculates the appropriate rate based on cluster health:
+// - Normal rate when system is healthy
+// - Secondary rate when system is unhealthy but large-scale
+// - Zero (halt evictions) when system is unhealthy and small-scale
+func (d *DynamicRateLimiter[T]) getCurrentRate() float32 {
+	clusterGVR := clusterv1alpha1.SchemeGroupVersion.WithResource("clusters")
+
+	var lister = d.informerManager.Lister(clusterGVR)
+	if lister == nil {
+		klog.Errorf("Failed to get cluster lister, halting eviction for safety")
+		return 0
+	}
+
+	clusters, err := lister.List(labels.Everything())
+	if err != nil {
+		klog.Errorf("Failed to list clusters from informer cache: %v, halting eviction for safety", err)
+		return 0
+	}
+
+	totalClusters := len(clusters)
+	if totalClusters == 0 {
+		return d.resourceEvictionRate
+	}
+
+	unhealthyClusters := 0
+	for _, clusterObj := range clusters {
+		cluster, ok := clusterObj.(*clusterv1alpha1.Cluster)
+		if !ok {
+			continue
+		}
+		if !util.IsClusterReady(&cluster.Status) {
+			unhealthyClusters++
+		}
+	}
+
+	// Update metrics
+	failureRate := float32(unhealthyClusters) / float32(totalClusters)
+	metrics.RecordClusterHealthMetrics(unhealthyClusters, float64(failureRate))
+
+	// Determine rate based on health status
+	isUnhealthy := failureRate > d.unhealthyClusterThreshold
+	if !isUnhealthy {
+		return d.resourceEvictionRate
+	}
+
+	isLargeScale := totalClusters > d.largeClusterNumThreshold
+	if isLargeScale {
+		klog.V(2).Infof("System is unhealthy (failure rate: %.2f), downgrading eviction rate to secondary rate: %.2f/s",
+			failureRate, d.secondaryResourceEvictionRate)
+		return d.secondaryResourceEvictionRate
+	}
+
+	klog.V(2).Infof("System is unhealthy (failure rate: %.2f) and instance is small, halting eviction.", failureRate)
+	return 0
+}
+
+// Forget is a no-op as this rate limiter doesn't track individual items.
+func (d *DynamicRateLimiter[T]) Forget(item T) {
+	// No-op
+}
+
+// NumRequeues always returns 0 as this rate limiter doesn't track retries.
+func (d *DynamicRateLimiter[T]) NumRequeues(item T) int {
+	return 0
+}
+
+// NewGracefulEvictionRateLimiter creates a combined rate limiter for eviction.
+// It uses the maximum delay from both dynamic and default rate limiters to ensure
+// both cluster health and retry backoff are considered.
+func NewGracefulEvictionRateLimiter[T comparable](
+	informerManager genericmanager.SingleClusterInformerManager,
+	evictionOpts config.EvictionQueueOptions,
+	rateLimiterOpts ratelimiterflag.Options) workqueue.TypedRateLimiter[T] {
+
+	dynamicLimiter := NewDynamicRateLimiter[T](informerManager, evictionOpts)
+	defaultLimiter := ratelimiterflag.DefaultControllerRateLimiter[T](rateLimiterOpts)
+	return workqueue.NewTypedMaxOfRateLimiter[T](dynamicLimiter, defaultLimiter)
+}

pkg/controllers/cluster/eviction_worker.go

@@ -0,0 +1,198 @@
+/*
+Copyright 2022 The Karmada Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cluster
+
+import (
+	"context"
+	"time"
+
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/klog/v2"
+
+	config "github.com/karmada-io/karmada/pkg/controllers/cluster/evictionqueue_config"
+	"github.com/karmada-io/karmada/pkg/metrics"
+	"github.com/karmada-io/karmada/pkg/sharedcli/ratelimiterflag"
+	"github.com/karmada-io/karmada/pkg/util"
+	"github.com/karmada-io/karmada/pkg/util/fedinformer/genericmanager"
+)
+
+// EvictionWorker enhances AsyncWorker with dynamic rate limiting and metrics
+// for eviction operations. It provides a queue that adjusts its processing rate
+// based on cluster health status.
+type EvictionWorker interface {
+	util.AsyncWorker
+}
+
+type evictionWorker struct {
+	name             string
+	keyFunc          util.KeyFunc
+	reconcileFunc    util.ReconcileFunc
+	resourceKindFunc func(key interface{}) (clusterName, resourceKind string)
+	queue            workqueue.TypedRateLimitingInterface[any]
+}
+
+// EvictionWorkerOptions configures a new EvictionWorker instance.
+type EvictionWorkerOptions struct {
+	// Name is the queue's name used for metrics and logging
+	Name string
+
+	// KeyFunc generates keys from objects for queue operations
+	KeyFunc util.KeyFunc
+
+	// ReconcileFunc processes keys from the queue
+	ReconcileFunc util.ReconcileFunc
+
+	// ResourceKindFunc returns resource metadata for metrics collection
+	ResourceKindFunc func(key interface{}) (clusterName, resourceKind string)
+
+	// InformerManager provides cluster information for dynamic rate limiting
+	InformerManager genericmanager.SingleClusterInformerManager
+
+	// EvictionQueueOptions configures dynamic rate limiting behavior
+	EvictionQueueOptions config.EvictionQueueOptions
+
+	// RateLimiterOptions configures general rate limiter behavior
+	RateLimiterOptions ratelimiterflag.Options
+}
+
+// NewEvictionWorker creates a new EvictionWorker with dynamic rate limiting.
+func NewEvictionWorker(opts EvictionWorkerOptions) EvictionWorker {
+	rateLimiter := NewGracefulEvictionRateLimiter[interface{}](
+		opts.InformerManager,
+		opts.EvictionQueueOptions,
+		opts.RateLimiterOptions,
+	)
+
+	return &evictionWorker{
+		name:             opts.Name,
+		keyFunc:          opts.KeyFunc,
+		reconcileFunc:    opts.ReconcileFunc,
+		resourceKindFunc: opts.ResourceKindFunc,
+		queue: workqueue.NewRateLimitingQueueWithConfig(rateLimiter, workqueue.RateLimitingQueueConfig{
+			Name: opts.Name,
+		}),
+	}
+}
+
+// Enqueue converts an object to a key and adds it to the queue.
+func (w *evictionWorker) Enqueue(obj interface{}) {
+	key, err := w.keyFunc(obj)
+	if err != nil {
+		klog.Errorf("Failed to generate key for obj: %+v, err: %v", obj, err)
+		return
+	}
+
+	if key == nil {
+		return
+	}
+
+	w.Add(key)
+}
+
+// Add puts an item into the queue and updates metrics.
+func (w *evictionWorker) Add(item interface{}) {
+	if item == nil {
+		klog.Warningf("Ignore nil item from queue")
+		return
+	}
+
+	w.queue.Add(item)
+	metrics.RecordEvictionQueueMetrics(w.name, float64(w.queue.Len()))
+
+	// Update resource kind metrics if possible
+	if w.resourceKindFunc != nil {
+		clusterName, resourceKind := w.resourceKindFunc(item)
+		metrics.RecordEvictionKindMetrics(clusterName, resourceKind, true)
+	}
+}
+
+// AddAfter adds an item to the queue after a delay and updates metrics.
+func (w *evictionWorker) AddAfter(item interface{}, duration time.Duration) {
+	if item == nil {
+		klog.Warningf("Ignore nil item from queue")
+		return
+	}
+
+	w.queue.AddAfter(item, duration)
+	metrics.RecordEvictionQueueMetrics(w.name, float64(w.queue.Len()))
+
+	// Update resource kind metrics if possible
+	if w.resourceKindFunc != nil {
+		clusterName, resourceKind := w.resourceKindFunc(item)
+		metrics.RecordEvictionKindMetrics(clusterName, resourceKind, true)
+	}
+}
+
+// worker processes items from the queue until the context is canceled.
+func (w *evictionWorker) worker(ctx context.Context) {
+	for w.processNextWorkItem(ctx) {
+	}
+}
+
+// processNextWorkItem handles a single item from the queue with metrics tracking.
+// Returns false when the queue is shutting down, true otherwise.
+func (w *evictionWorker) processNextWorkItem(ctx context.Context) bool {
+	key, quit := w.queue.Get()
+	if quit {
+		return false
+	}
+	defer w.queue.Done(key)
+
+	// Update queue metrics
+	metrics.RecordEvictionQueueMetrics(w.name, float64(w.queue.Len()))
+
+	// Get resource metadata for metrics
+	var clusterName, resourceKind string
+	if w.resourceKindFunc != nil {
+		clusterName, resourceKind = w.resourceKindFunc(key)
+	}
+
+	// Process the item and measure latency
+	startTime := time.Now()
+	err := w.reconcileFunc(key)
+	metrics.RecordEvictionProcessingMetrics(w.name, err, startTime)
+
+	if err != nil {
+		// Requeue with rate limiting on error
+		w.queue.AddRateLimited(key)
+		// Item remains in queue, so don't decrease metrics count
+		return true
+	}
+
+	// Successfully processed
+	w.queue.Forget(key)
+
+	// Decrease resource kind count only after successful processing
+	metrics.RecordEvictionKindMetrics(clusterName, resourceKind, false)
+
+	return true
+}
+
+// Run starts worker goroutines and ensures cleanup when context is canceled.
+func (w *evictionWorker) Run(ctx context.Context, workerNumber int) {
+	klog.Infof("Starting %d workers for eviction worker %s", workerNumber, w.name)
+	for i := 0; i < workerNumber; i++ {
+		go w.worker(ctx)
+	}
+
+	// Clean up when context is canceled
+	go func() {
+		<-ctx.Done()
+		klog.Infof("Shutting down eviction worker %s", w.name)
+		w.queue.ShutDown()
+	}()
+}