Add option to disable workqueue bucket rate limiter (#4451)

This commit is contained in:
Junya Okabe
2026-04-23 06:26:39 +09:00
committed by GitHub
parent 012f1a5b23
commit a401686bd5
7 changed files with 96 additions and 37 deletions

View File

@@ -692,7 +692,7 @@ func (r *AutoscalingListenerReconciler) publishRunningListener(autoscalingListen
}
// SetupWithManager sets up the controller with the Manager.
func (r *AutoscalingListenerReconciler) SetupWithManager(mgr ctrl.Manager) error {
func (r *AutoscalingListenerReconciler) SetupWithManager(mgr ctrl.Manager, opts ...Option) error {
labelBasedWatchFunc := func(_ context.Context, obj client.Object) []reconcile.Request {
var requests []reconcile.Request
labels := obj.GetLabels()
@@ -716,14 +716,16 @@ func (r *AutoscalingListenerReconciler) SetupWithManager(mgr ctrl.Manager) error
return requests
}
return ctrl.NewControllerManagedBy(mgr).
For(&v1alpha1.AutoscalingListener{}).
Owns(&corev1.Pod{}).
Owns(&corev1.ServiceAccount{}).
Watches(&rbacv1.Role{}, handler.EnqueueRequestsFromMapFunc(labelBasedWatchFunc)).
Watches(&rbacv1.RoleBinding{}, handler.EnqueueRequestsFromMapFunc(labelBasedWatchFunc)).
WithEventFilter(predicate.ResourceVersionChangedPredicate{}).
Complete(r)
return builderWithOptions(
ctrl.NewControllerManagedBy(mgr).
For(&v1alpha1.AutoscalingListener{}).
Owns(&corev1.Pod{}).
Owns(&corev1.ServiceAccount{}).
Watches(&rbacv1.Role{}, handler.EnqueueRequestsFromMapFunc(labelBasedWatchFunc)).
Watches(&rbacv1.RoleBinding{}, handler.EnqueueRequestsFromMapFunc(labelBasedWatchFunc)).
WithEventFilter(predicate.ResourceVersionChangedPredicate{}),
opts,
).Complete(r)
}
func listenerContainerStatus(pod *corev1.Pod) *corev1.ContainerStatus {

View File

@@ -762,25 +762,27 @@ func (r *AutoscalingRunnerSetReconciler) listEphemeralRunnerSets(ctx context.Con
}
// SetupWithManager sets up the controller with the Manager.
func (r *AutoscalingRunnerSetReconciler) SetupWithManager(mgr ctrl.Manager) error {
return ctrl.NewControllerManagedBy(mgr).
For(&v1alpha1.AutoscalingRunnerSet{}).
Owns(&v1alpha1.EphemeralRunnerSet{}).
Watches(&v1alpha1.AutoscalingListener{}, handler.EnqueueRequestsFromMapFunc(
func(_ context.Context, o client.Object) []reconcile.Request {
autoscalingListener := o.(*v1alpha1.AutoscalingListener)
return []reconcile.Request{
{
NamespacedName: types.NamespacedName{
Namespace: autoscalingListener.Spec.AutoscalingRunnerSetNamespace,
Name: autoscalingListener.Spec.AutoscalingRunnerSetName,
func (r *AutoscalingRunnerSetReconciler) SetupWithManager(mgr ctrl.Manager, opts ...Option) error {
return builderWithOptions(
ctrl.NewControllerManagedBy(mgr).
For(&v1alpha1.AutoscalingRunnerSet{}).
Owns(&v1alpha1.EphemeralRunnerSet{}).
Watches(&v1alpha1.AutoscalingListener{}, handler.EnqueueRequestsFromMapFunc(
func(_ context.Context, o client.Object) []reconcile.Request {
autoscalingListener := o.(*v1alpha1.AutoscalingListener)
return []reconcile.Request{
{
NamespacedName: types.NamespacedName{
Namespace: autoscalingListener.Spec.AutoscalingRunnerSetNamespace,
Name: autoscalingListener.Spec.AutoscalingRunnerSetName,
},
},
},
}
},
)).
WithEventFilter(predicate.ResourceVersionChangedPredicate{}).
Complete(r)
}
},
)).
WithEventFilter(predicate.ResourceVersionChangedPredicate{}),
opts,
).Complete(r)
}
type autoscalingRunnerSetFinalizerDependencyCleaner struct {

View File

@@ -522,12 +522,14 @@ func (r *EphemeralRunnerSetReconciler) deleteEphemeralRunnerWithActionsClient(ct
}
// SetupWithManager sets up the controller with the Manager.
func (r *EphemeralRunnerSetReconciler) SetupWithManager(mgr ctrl.Manager) error {
return ctrl.NewControllerManagedBy(mgr).
For(&v1alpha1.EphemeralRunnerSet{}).
Owns(&v1alpha1.EphemeralRunner{}).
WithEventFilter(predicate.ResourceVersionChangedPredicate{}).
Complete(r)
func (r *EphemeralRunnerSetReconciler) SetupWithManager(mgr ctrl.Manager, opts ...Option) error {
return builderWithOptions(
ctrl.NewControllerManagedBy(mgr).
For(&v1alpha1.EphemeralRunnerSet{}).
Owns(&v1alpha1.EphemeralRunner{}).
WithEventFilter(predicate.ResourceVersionChangedPredicate{}),
opts,
).Complete(r)
}
type ephemeralRunnerStepper struct {

View File

@@ -1,8 +1,10 @@
package actionsgithubcom
import (
"k8s.io/client-go/util/workqueue"
"sigs.k8s.io/controller-runtime/pkg/builder"
"sigs.k8s.io/controller-runtime/pkg/controller"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
)
// Options is the optional configuration for the controllers, which can be
@@ -37,6 +39,25 @@ func WithMaxConcurrentReconciles(n int) Option {
}
}
// WithTypedRateLimiter sets the rate limiter for the controller's workqueue.
//
// By default, the controller-runtime uses
// workqueue.DefaultTypedControllerRateLimiter[reconcile.Request], which combines
// an exponential backoff per-item limiter with a token bucket overall limiter
// (10 QPS, 100 bucket size). In large-scale environments with many runner
// scale sets, the token bucket limiter can become a bottleneck for
// reconciliation throughput.
//
// Use this option to override the default rate limiter, for example, to use
// workqueue.DefaultTypedItemBasedRateLimiter[reconcile.Request], which removes
// the overall token bucket constraint while keeping the per-item exponential
// backoff.
func WithTypedRateLimiter(rateLimiter workqueue.TypedRateLimiter[reconcile.Request]) Option {
return func(b *controller.Options) {
b.RateLimiter = rateLimiter
}
}
// builderWithOptions applies the given options to the provided builder, if any.
// This is a helper function to avoid the need to import the controller-runtime package in every reconciler source file
// and the command package that creates the controller.