karmada-io
/
karmada
mirror of https://github.com/karmada-io/karmada.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refactor ratelimit flags to make it easier to share with karmada-agent. 

Signed-off-by: RainbowMango <qdurenhongcai@gmail.com>
This commit is contained in:
RainbowMango 2022-03-15 20:10:34 +08:00
parent d8bcedc3c1
commit 11fd0f935c
12 changed files with 84 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
cmd/controller-manager/app/controllermanager.go
View File

@ -50,7 +50,6 @@ import (
"github.com/karmada-io/karmada/pkg/util/informermanager"
"github.com/karmada-io/karmada/pkg/util/objectwatcher"
"github.com/karmada-io/karmada/pkg/util/overridemanager"
"github.com/karmada-io/karmada/pkg/util/ratelimiter"
"github.com/karmada-io/karmada/pkg/version"
"github.com/karmada-io/karmada/pkg/version/sharedcommand"
)
@ -426,12 +425,7 @@ func setupControllers(mgr controllerruntime.Manager, opts *options.Options, stop
}
objectWatcher := objectwatcher.NewObjectWatcher(mgr.GetClient(), mgr.GetRESTMapper(), util.NewClusterDynamicClientSet, resourceInterpreter)
rateLimiterOptions := ratelimiter.Options{
BaseDelay: opts.RateLimiterBaseDelay,
MaxDelay: opts.RateLimiterMaxDelay,
QPS: opts.RateLimiterQPS,
BucketSize: opts.RateLimiterBucketSize,
}
resourceDetector := &detector.ResourceDetector{
DiscoveryClientSet: discoverClientSet,
Client: mgr.GetClient(),
@ -444,7 +438,7 @@ func setupControllers(mgr controllerruntime.Manager, opts *options.Options, stop
EventRecorder: mgr.GetEventRecorderFor("resource-detector"),
ConcurrentResourceTemplateSyncs: opts.ConcurrentResourceTemplateSyncs,
ConcurrentResourceBindingSyncs: opts.ConcurrentResourceBindingSyncs,
RateLimiterOptions: rateLimiterOptions,
RateLimiterOptions: opts.RateLimiterOpts,
}
if err := mgr.Add(resourceDetector); err != nil {
klog.Fatalf("Failed to setup resource detector: %v", err)
@ -480,7 +474,7 @@ func setupControllers(mgr controllerruntime.Manager, opts *options.Options, stop
ClusterAPIBurst: opts.ClusterAPIBurst,
SkippedPropagatingNamespaces: opts.SkippedPropagatingNamespaces,
ConcurrentWorkSyncs: opts.ConcurrentWorkSyncs,
RateLimiterOptions: rateLimiterOptions,
RateLimiterOptions: opts.RateLimiterOpts,
},
StopChan: stopChan,
DynamicClientSet: dynamicClientSet,

17
cmd/controller-manager/app/options/options.go
View File

@ -11,6 +11,7 @@ import (
componentbaseconfig "k8s.io/component-base/config"
"github.com/karmada-io/karmada/pkg/features"
"github.com/karmada-io/karmada/pkg/sharedcli/ratelimiterflag"
"github.com/karmada-io/karmada/pkg/util"
)
@ -97,14 +98,8 @@ type Options struct {
ConcurrentNamespaceSyncs int
// ConcurrentResourceTemplateSyncs is the number of resource templates that are allowed to sync concurrently.
ConcurrentResourceTemplateSyncs int
// RateLimiterBaseDelay is the base delay for ItemExponentialFailureRateLimiter.
RateLimiterBaseDelay time.Duration
// RateLimiterMaxDelay is the max delay for ItemExponentialFailureRateLimiter.
RateLimiterMaxDelay time.Duration
// RateLimiterQPS is the qps for BucketRateLimiter
RateLimiterQPS int
// RateLimiterBucketSize is the bucket size for BucketRateLimiter
RateLimiterBucketSize int
RateLimiterOpts ratelimiterflag.Options
}
// NewOptions builds an empty options.
@ -164,9 +159,7 @@ func (o *Options) AddFlags(flags *pflag.FlagSet, allControllers []string) {
flags.IntVar(&o.ConcurrentWorkSyncs, "concurrent-work-syncs", 5, "The number of Works that are allowed to sync concurrently.")
flags.IntVar(&o.ConcurrentNamespaceSyncs, "concurrent-namespace-syncs", 1, "The number of Namespaces that are allowed to sync concurrently.")
flags.IntVar(&o.ConcurrentResourceTemplateSyncs, "concurrent-resource-template-syncs", 5, "The number of resource templates that are allowed to sync concurrently.")
flags.DurationVar(&o.RateLimiterBaseDelay, "rate-limiter-base-delay", time.Millisecond*5, "The base delay for rate limiter.")
flags.DurationVar(&o.RateLimiterMaxDelay, "rate-limiter-max-delay", time.Second*1000, "The max delay for rate limiter.")
flags.IntVar(&o.RateLimiterQPS, "rate-limiter-qps", 10, "The qps for rate limier.")
flags.IntVar(&o.RateLimiterBucketSize, "rate-limiter-bucket-size", 100, "The bucket size for rate limier.")
o.RateLimiterOpts.AddFlags(flags)
features.FeatureGate.AddFlag(flags)
}

6
pkg/controllers/binding/binding_controller.go
View File

@ -29,11 +29,11 @@ import (
workv1alpha1 "github.com/karmada-io/karmada/pkg/apis/work/v1alpha1"
workv1alpha2 "github.com/karmada-io/karmada/pkg/apis/work/v1alpha2"
"github.com/karmada-io/karmada/pkg/resourceinterpreter"
"github.com/karmada-io/karmada/pkg/sharedcli/ratelimiterflag"
"github.com/karmada-io/karmada/pkg/util"
"github.com/karmada-io/karmada/pkg/util/helper"
"github.com/karmada-io/karmada/pkg/util/informermanager"
"github.com/karmada-io/karmada/pkg/util/overridemanager"
"github.com/karmada-io/karmada/pkg/util/ratelimiter"
"github.com/karmada-io/karmada/pkg/util/restmapper"
)
@ -49,7 +49,7 @@ type ResourceBindingController struct {
RESTMapper meta.RESTMapper
OverrideManager overridemanager.OverrideManager
ResourceInterpreter resourceinterpreter.ResourceInterpreter
RateLimiterOptions ratelimiter.Options
RateLimiterOptions ratelimiterflag.Options
}
// Reconcile performs a full reconciliation for the object referred to by the Request.
@ -221,7 +221,7 @@ func (c *ResourceBindingController) SetupWithManager(mgr controllerruntime.Manag
Watches(&source.Kind{Type: &policyv1alpha1.OverridePolicy{}}, handler.EnqueueRequestsFromMapFunc(c.newOverridePolicyFunc())).
Watches(&source.Kind{Type: &policyv1alpha1.ClusterOverridePolicy{}}, handler.EnqueueRequestsFromMapFunc(c.newOverridePolicyFunc())).
WithOptions(controller.Options{
RateLimiter: ratelimiter.DefaultControllerRateLimiter(c.RateLimiterOptions),
RateLimiter: ratelimiterflag.DefaultControllerRateLimiter(c.RateLimiterOptions),
}).
Complete(c)
}

6
pkg/controllers/binding/cluster_resource_binding_controller.go
View File

@ -25,11 +25,11 @@ import (
workv1alpha1 "github.com/karmada-io/karmada/pkg/apis/work/v1alpha1"
workv1alpha2 "github.com/karmada-io/karmada/pkg/apis/work/v1alpha2"
"github.com/karmada-io/karmada/pkg/resourceinterpreter"
"github.com/karmada-io/karmada/pkg/sharedcli/ratelimiterflag"
"github.com/karmada-io/karmada/pkg/util"
"github.com/karmada-io/karmada/pkg/util/helper"
"github.com/karmada-io/karmada/pkg/util/informermanager"
"github.com/karmada-io/karmada/pkg/util/overridemanager"
"github.com/karmada-io/karmada/pkg/util/ratelimiter"
)
// ClusterResourceBindingControllerName is the controller name that will be used when reporting events.
@ -44,7 +44,7 @@ type ClusterResourceBindingController struct {
RESTMapper meta.RESTMapper
OverrideManager overridemanager.OverrideManager
ResourceInterpreter resourceinterpreter.ResourceInterpreter
RateLimiterOptions ratelimiter.Options
RateLimiterOptions ratelimiterflag.Options
}
// Reconcile performs a full reconciliation for the object referred to by the Request.
@ -162,7 +162,7 @@ func (c *ClusterResourceBindingController) SetupWithManager(mgr controllerruntim
Watches(&source.Kind{Type: &policyv1alpha1.OverridePolicy{}}, handler.EnqueueRequestsFromMapFunc(c.newOverridePolicyFunc())).
Watches(&source.Kind{Type: &policyv1alpha1.ClusterOverridePolicy{}}, handler.EnqueueRequestsFromMapFunc(c.newOverridePolicyFunc())).
WithOptions(controller.Options{
RateLimiter: ratelimiter.DefaultControllerRateLimiter(c.RateLimiterOptions),
RateLimiter: ratelimiterflag.DefaultControllerRateLimiter(c.RateLimiterOptions),
}).
Complete(c)
}

5
pkg/controllers/context/context.go
View File

@ -8,10 +8,10 @@ import (
controllerruntime "sigs.k8s.io/controller-runtime"
"github.com/karmada-io/karmada/pkg/resourceinterpreter"
"github.com/karmada-io/karmada/pkg/sharedcli/ratelimiterflag"
"github.com/karmada-io/karmada/pkg/util/informermanager"
"github.com/karmada-io/karmada/pkg/util/objectwatcher"
"github.com/karmada-io/karmada/pkg/util/overridemanager"
"github.com/karmada-io/karmada/pkg/util/ratelimiter"
)
// Options defines all the parameters required by our controllers.
@ -50,7 +50,8 @@ type Options struct {
ClusterName string
// ConcurrentWorkSyncs is the number of Works that are allowed to sync concurrently.
ConcurrentWorkSyncs int
RateLimiterOptions ratelimiter.Options
// RateLimiterOptions contains the options for rate limiter.
RateLimiterOptions ratelimiterflag.Options
}
// Context defines the context object for controller.

6
pkg/controllers/execution/execution_controller.go
View File

@ -20,13 +20,13 @@ import (
"sigs.k8s.io/controller-runtime/pkg/predicate"
workv1alpha1 "github.com/karmada-io/karmada/pkg/apis/work/v1alpha1"
"github.com/karmada-io/karmada/pkg/sharedcli/ratelimiterflag"
"github.com/karmada-io/karmada/pkg/util"
"github.com/karmada-io/karmada/pkg/util/helper"
"github.com/karmada-io/karmada/pkg/util/informermanager"
"github.com/karmada-io/karmada/pkg/util/informermanager/keys"
"github.com/karmada-io/karmada/pkg/util/names"
"github.com/karmada-io/karmada/pkg/util/objectwatcher"
"github.com/karmada-io/karmada/pkg/util/ratelimiter"
)
const (
@ -43,7 +43,7 @@ type Controller struct {
PredicateFunc predicate.Predicate
InformerManager informermanager.MultiClusterInformerManager
ClusterClientSetFunc func(clusterName string, client client.Client) (*util.DynamicClusterClient, error)
RatelimiterOption ratelimiter.Options
RatelimiterOption ratelimiterflag.Options
}
// Reconcile performs a full reconciliation for the object referred to by the Request.
@ -104,7 +104,7 @@ func (c *Controller) SetupWithManager(mgr controllerruntime.Manager) error {
WithEventFilter(predicate.GenerationChangedPredicate{}).
WithEventFilter(c.PredicateFunc).
WithOptions(controller.Options{
RateLimiter: ratelimiter.DefaultControllerRateLimiter(c.RatelimiterOption),
RateLimiter: ratelimiterflag.DefaultControllerRateLimiter(c.RatelimiterOption),
}).
Complete(c)
}

6
pkg/controllers/status/cluster_status_controller.go
View File

@ -32,10 +32,10 @@ import (
"sigs.k8s.io/controller-runtime/pkg/predicate"
clusterv1alpha1 "github.com/karmada-io/karmada/pkg/apis/cluster/v1alpha1"
"github.com/karmada-io/karmada/pkg/sharedcli/ratelimiterflag"
"github.com/karmada-io/karmada/pkg/util"
"github.com/karmada-io/karmada/pkg/util/helper"
"github.com/karmada-io/karmada/pkg/util/informermanager"
"github.com/karmada-io/karmada/pkg/util/ratelimiter"
)
const (
@ -84,7 +84,7 @@ type ClusterStatusController struct {
ClusterLeaseControllers sync.Map
ClusterCacheSyncTimeout metav1.Duration
RateLimiterOptions ratelimiter.Options
RateLimiterOptions ratelimiterflag.Options
}
// Reconcile syncs status of the given member cluster.
@ -117,7 +117,7 @@ func (c *ClusterStatusController) Reconcile(ctx context.Context, req controllerr
// SetupWithManager creates a controller and register to controller manager.
func (c *ClusterStatusController) SetupWithManager(mgr controllerruntime.Manager) error {
return controllerruntime.NewControllerManagedBy(mgr).For(&clusterv1alpha1.Cluster{}).WithEventFilter(c.PredicateFunc).WithOptions(controller.Options{
RateLimiter: ratelimiter.DefaultControllerRateLimiter(c.RateLimiterOptions),
RateLimiter: ratelimiterflag.DefaultControllerRateLimiter(c.RateLimiterOptions),
}).Complete(c)
}

6
pkg/controllers/status/workstatus_controller.go
View File

@ -21,13 +21,13 @@ import (
clusterv1alpha1 "github.com/karmada-io/karmada/pkg/apis/cluster/v1alpha1"
workv1alpha1 "github.com/karmada-io/karmada/pkg/apis/work/v1alpha1"
"github.com/karmada-io/karmada/pkg/sharedcli/ratelimiterflag"
"github.com/karmada-io/karmada/pkg/util"
"github.com/karmada-io/karmada/pkg/util/helper"
"github.com/karmada-io/karmada/pkg/util/informermanager"
"github.com/karmada-io/karmada/pkg/util/informermanager/keys"
"github.com/karmada-io/karmada/pkg/util/names"
"github.com/karmada-io/karmada/pkg/util/objectwatcher"
"github.com/karmada-io/karmada/pkg/util/ratelimiter"
"github.com/karmada-io/karmada/pkg/util/restmapper"
)
@ -49,7 +49,7 @@ type WorkStatusController struct {
PredicateFunc predicate.Predicate
ClusterClientSetFunc func(clusterName string, client client.Client) (*util.DynamicClusterClient, error)
ClusterCacheSyncTimeout metav1.Duration
RateLimiterOptions ratelimiter.Options
RateLimiterOptions ratelimiterflag.Options
}
// Reconcile performs a full reconciliation for the object referred to by the Request.
@ -452,6 +452,6 @@ func (c *WorkStatusController) getSingleClusterManager(cluster *clusterv1alpha1.
// SetupWithManager creates a controller and register to controller manager.
func (c *WorkStatusController) SetupWithManager(mgr controllerruntime.Manager) error {
return controllerruntime.NewControllerManagedBy(mgr).For(&workv1alpha1.Work{}).WithEventFilter(c.PredicateFunc).WithOptions(controller.Options{
RateLimiter: ratelimiter.DefaultControllerRateLimiter(c.RateLimiterOptions),
RateLimiter: ratelimiterflag.DefaultControllerRateLimiter(c.RateLimiterOptions),
}).Complete(c)
}

4
pkg/detector/detector.go
View File

@ -30,13 +30,13 @@ import (
policyv1alpha1 "github.com/karmada-io/karmada/pkg/apis/policy/v1alpha1"
workv1alpha2 "github.com/karmada-io/karmada/pkg/apis/work/v1alpha2"
"github.com/karmada-io/karmada/pkg/resourceinterpreter"
"github.com/karmada-io/karmada/pkg/sharedcli/ratelimiterflag"
"github.com/karmada-io/karmada/pkg/util"
"github.com/karmada-io/karmada/pkg/util/helper"
"github.com/karmada-io/karmada/pkg/util/informermanager"
"github.com/karmada-io/karmada/pkg/util/informermanager/keys"
"github.com/karmada-io/karmada/pkg/util/lifted"
"github.com/karmada-io/karmada/pkg/util/names"
"github.com/karmada-io/karmada/pkg/util/ratelimiter"
"github.com/karmada-io/karmada/pkg/util/restmapper"
)
@ -81,7 +81,7 @@ type ResourceDetector struct {
// RateLimiterOptions is the configuration for rate limiter which may significantly influence the performance of
// the controller.
RateLimiterOptions ratelimiter.Options
RateLimiterOptions ratelimiterflag.Options
stopCh <-chan struct{}
}

53
pkg/sharedcli/ratelimiterflag/ratelimiterflag.go Normal file
View File

@ -0,0 +1,53 @@
package ratelimiterflag
import (
"time"
"github.com/spf13/pflag"
"golang.org/x/time/rate"
"k8s.io/client-go/util/workqueue"
)
// Options are options for rate limiter.
type Options struct {
// RateLimiterBaseDelay is the base delay for ItemExponentialFailureRateLimiter.
RateLimiterBaseDelay time.Duration
// RateLimiterMaxDelay is the max delay for ItemExponentialFailureRateLimiter.
RateLimiterMaxDelay time.Duration
// RateLimiterQPS is the qps for BucketRateLimiter
RateLimiterQPS int
// RateLimiterBucketSize is the bucket size for BucketRateLimiter
RateLimiterBucketSize int
}
// AddFlags adds flags to the specified FlagSet.
func (o *Options) AddFlags(fs *pflag.FlagSet) {
fs.DurationVar(&o.RateLimiterBaseDelay, "rate-limiter-base-delay", 5*time.Millisecond, "The base delay for rate limiter.")
fs.DurationVar(&o.RateLimiterMaxDelay, "rate-limiter-max-delay", 1000*time.Second, "The max delay for rate limiter.")
fs.IntVar(&o.RateLimiterQPS, "rate-limiter-qps", 10, "The QPS for rate limier.")
fs.IntVar(&o.RateLimiterBucketSize, "rate-limiter-bucket-size", 100, "The bucket size for rate limier.")
}
// DefaultControllerRateLimiter provide a default rate limiter for controller, and users can tune it by corresponding flags.
func DefaultControllerRateLimiter(opts Options) workqueue.RateLimiter {
// set defaults
if opts.RateLimiterBaseDelay <= 0 {
opts.RateLimiterBaseDelay = 5 * time.Millisecond
}
if opts.RateLimiterMaxDelay <= 0 {
opts.RateLimiterMaxDelay = 1000 * time.Second
}
if opts.RateLimiterQPS <= 0 {
opts.RateLimiterQPS = 10
}
if opts.RateLimiterBucketSize <= 0 {
opts.RateLimiterBucketSize = 100
}
return workqueue.NewMaxOfRateLimiter(
workqueue.NewItemExponentialFailureRateLimiter(opts.RateLimiterBaseDelay, opts.RateLimiterMaxDelay),
&workqueue.BucketRateLimiter{Limiter: rate.NewLimiter(rate.Limit(opts.RateLimiterQPS), opts.RateLimiterBucketSize)},
)
}

37
pkg/util/ratelimiter/rate_limiter.go
View File

@ -1,37 +0,0 @@
package ratelimiter
import (
"time"
"golang.org/x/time/rate"
"k8s.io/client-go/util/workqueue"
)
// Options are options for rate limiter.
type Options struct {
BaseDelay time.Duration
MaxDelay time.Duration
QPS int
BucketSize int
}
// DefaultControllerRateLimiter provide a default rate limiter for controller, and users can tune it by corresponding flags.
func DefaultControllerRateLimiter(opts Options) workqueue.RateLimiter {
// set defaults
if opts.BaseDelay <= 0 {
opts.BaseDelay = 5 * time.Millisecond
}
if opts.MaxDelay <= 0 {
opts.MaxDelay = 1000 * time.Second
}
if opts.QPS <= 0 {
opts.QPS = 10
}
if opts.BucketSize <= 0 {
opts.BucketSize = 100
}
return workqueue.NewMaxOfRateLimiter(
workqueue.NewItemExponentialFailureRateLimiter(opts.BaseDelay, opts.MaxDelay),
&workqueue.BucketRateLimiter{Limiter: rate.NewLimiter(rate.Limit(opts.QPS), opts.BucketSize)},
)
}

6
pkg/util/worker.go
View File

@ -7,7 +7,7 @@ import (
"k8s.io/client-go/util/workqueue"
"k8s.io/klog/v2"
"github.com/karmada-io/karmada/pkg/util/ratelimiter"
"github.com/karmada-io/karmada/pkg/sharedcli/ratelimiterflag"
)
const (
@ -64,7 +64,7 @@ type Options struct {
Name string
KeyFunc KeyFunc
ReconcileFunc ReconcileFunc
RateLimiterOptions ratelimiter.Options
RateLimiterOptions ratelimiterflag.Options
}
// NewAsyncWorker returns a asyncWorker which can process resource periodic.
@ -72,7 +72,7 @@ func NewAsyncWorker(opt Options) AsyncWorker {
return &asyncWorker{
keyFunc: opt.KeyFunc,
reconcileFunc: opt.ReconcileFunc,
queue: workqueue.NewNamedRateLimitingQueue(ratelimiter.DefaultControllerRateLimiter(opt.RateLimiterOptions), opt.Name),
queue: workqueue.NewNamedRateLimitingQueue(ratelimiterflag.DefaultControllerRateLimiter(opt.RateLimiterOptions), opt.Name),
}
}