Skip to content

Add configurable concurrent reconciliation to controllers #312

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Jun 7, 2025
Merged

Add configurable concurrent reconciliation to controllers #312

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion controller/job.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
utilerrors "k8s.io/apimachinery/pkg/util/errors"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/client"
ctrlcontroller "sigs.k8s.io/controller-runtime/pkg/controller"
"sigs.k8s.io/controller-runtime/pkg/handler"

"github.com/tinkerbell/rufio/api/v1alpha1"
Expand Down Expand Up @@ -221,7 +222,7 @@
}

// SetupWithManager sets up the controller with the Manager.
func (r *JobReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager) error {
func (r *JobReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, opts ctrlcontroller.Options) error {

Check warning on line 225 in controller/job.go

View check run for this annotation

Codecov / codecov/patch

controller/job.go#L225 

Added line #L225 was not covered by tests
if err := mgr.GetFieldIndexer().IndexField(
ctx,
&v1alpha1.Task{},
Expand All @@ -233,6 +234,7 @@

return ctrl.NewControllerManagedBy(mgr).
For(&v1alpha1.Job{}).
WithOptions(opts).

Check warning on line 237 in controller/job.go

View check run for this annotation

Codecov / codecov/patch

controller/job.go#L237 

Added line #L237 was not covered by tests
Watches(
&v1alpha1.Task{},
handler.EnqueueRequestForOwner(
Expand Down
5 changes: 4 additions & 1 deletion controller/machine.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,8 @@
"k8s.io/apimachinery/pkg/types"
utilerrors "k8s.io/apimachinery/pkg/util/errors"
"k8s.io/client-go/tools/record"
ctrlcontroller "sigs.k8s.io/controller-runtime/pkg/controller"

ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/client"

Expand Down Expand Up @@ -210,8 +212,9 @@
}

// SetupWithManager sets up the controller with the Manager.
func (r *MachineReconciler) SetupWithManager(mgr ctrl.Manager) error {
func (r *MachineReconciler) SetupWithManager(mgr ctrl.Manager, opts ctrlcontroller.Options) error {

Check warning on line 215 in controller/machine.go

View check run for this annotation

Codecov / codecov/patch

controller/machine.go#L215 

Added line #L215 was not covered by tests
return ctrl.NewControllerManagedBy(mgr).
For(&v1alpha1.Machine{}).
WithOptions(opts).

Check warning on line 218 in controller/machine.go

View check run for this annotation

Codecov / codecov/patch

controller/machine.go#L218 

Added line #L218 was not covered by tests
Complete(r)
}
4 changes: 3 additions & 1 deletion controller/task.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
utilerrors "k8s.io/apimachinery/pkg/util/errors"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/client"
ctrlcontroller "sigs.k8s.io/controller-runtime/pkg/controller"

"github.com/tinkerbell/rufio/api/v1alpha1"
)
Expand Down Expand Up @@ -274,8 +275,9 @@
}

// SetupWithManager sets up the controller with the Manager.
func (r *TaskReconciler) SetupWithManager(mgr ctrl.Manager) error {
func (r *TaskReconciler) SetupWithManager(mgr ctrl.Manager, opts ctrlcontroller.Options) error {

Check warning on line 278 in controller/task.go

View check run for this annotation

Codecov / codecov/patch

controller/task.go#L278 

Added line #L278 was not covered by tests
return ctrl.NewControllerManagedBy(mgr).
For(&v1alpha1.Task{}).
WithOptions(opts).

Check warning on line 281 in controller/task.go

View check run for this annotation

Codecov / codecov/patch

controller/task.go#L281 

Added line #L281 was not covered by tests
Complete(r)
}
18 changes: 10 additions & 8 deletions main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,16 +33,16 @@
"github.com/peterbourgon/ff/v3"
"github.com/peterbourgon/ff/v3/ffcli"
"github.com/rs/zerolog"
"github.com/tinkerbell/rufio/api/v1alpha1"
"github.com/tinkerbell/rufio/controller"
"k8s.io/apimachinery/pkg/runtime"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
clientgoscheme "k8s.io/client-go/kubernetes/scheme"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/cache"
ctrlcontroller "sigs.k8s.io/controller-runtime/pkg/controller"
"sigs.k8s.io/controller-runtime/pkg/healthz"
metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"

"github.com/tinkerbell/rufio/api/v1alpha1"
"github.com/tinkerbell/rufio/controller"
//+kubebuilder:scaffold:imports
)

Expand Down Expand Up @@ -86,6 +86,7 @@
var kubeconfig string
var kubeNamespace string
var bmcConnectTimeout time.Duration
var maxConcurrentReconciles int

Check warning on line 89 in main.go

View check run for this annotation

Codecov / codecov/patch

main.go#L89 

Added line #L89 was not covered by tests
fs := flag.NewFlagSet(appName, flag.ExitOnError)
fs.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "The address the metric endpoint binds to.")
fs.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.")
Expand All @@ -96,6 +97,7 @@
fs.StringVar(&kubeconfig, "kubeconfig", "", "Absolute path to the kubeconfig file.")
fs.StringVar(&kubeNamespace, "kube-namespace", "", "Namespace that the controller watches to reconcile objects.")
fs.DurationVar(&bmcConnectTimeout, "bmc-connect-timeout", 60*time.Second, "Timeout for establishing a connection to BMCs.")
fs.IntVar(&maxConcurrentReconciles, "max-concurrent-reconciles", 1, "Maximum number of concurrent reconciles per controller.")

Check warning on line 100 in main.go

View check run for this annotation

Codecov / codecov/patch

main.go#L100 

Added line #L100 was not covered by tests
cli := &ffcli.Command{
Name: appName,
FlagSet: fs,
Expand Down Expand Up @@ -144,7 +146,7 @@
bmcClientFactory := controller.NewClientFunc(bmcConnectTimeout)

// Setup controller reconcilers
setupReconcilers(ctx, mgr, bmcClientFactory)
setupReconcilers(ctx, mgr, bmcClientFactory, maxConcurrentReconciles)

Check warning on line 149 in main.go

View check run for this annotation

Codecov / codecov/patch

main.go#L149 

Added line #L149 was not covered by tests

//+kubebuilder:scaffold:builder

Expand Down Expand Up @@ -175,20 +177,20 @@
}

// setupReconcilers initializes the controllers with the Manager.
func setupReconcilers(ctx context.Context, mgr ctrl.Manager, bmcClientFactory controller.ClientFunc) {
func setupReconcilers(ctx context.Context, mgr ctrl.Manager, bmcClientFactory controller.ClientFunc, maxConcurrentReconciles int) {

Check warning on line 180 in main.go

View check run for this annotation

Codecov / codecov/patch

main.go#L180 

Added line #L180 was not covered by tests
err := (controller.NewMachineReconciler(
mgr.GetClient(),
mgr.GetEventRecorderFor("machine-controller"),
bmcClientFactory,
)).SetupWithManager(mgr)
)).SetupWithManager(mgr, ctrlcontroller.Options{MaxConcurrentReconciles: maxConcurrentReconciles})

Check warning on line 185 in main.go

View check run for this annotation

Codecov / codecov/patch

main.go#L185 

Added line #L185 was not covered by tests
if err != nil {
setupLog.Error(err, "unable to create controller", "controller", "Machine")
os.Exit(1)
}

err = (controller.NewJobReconciler(
mgr.GetClient(),
)).SetupWithManager(ctx, mgr)
)).SetupWithManager(ctx, mgr, ctrlcontroller.Options{MaxConcurrentReconciles: maxConcurrentReconciles})

Check warning on line 193 in main.go

View check run for this annotation

Codecov / codecov/patch

main.go#L193 

Added line #L193 was not covered by tests
if err != nil {
setupLog.Error(err, "unable to create controller", "controller", "Job")
os.Exit(1)
Expand All @@ -197,7 +199,7 @@
err = (controller.NewTaskReconciler(
mgr.GetClient(),
bmcClientFactory,
)).SetupWithManager(mgr)
)).SetupWithManager(mgr, ctrlcontroller.Options{MaxConcurrentReconciles: maxConcurrentReconciles})

Check warning on line 202 in main.go

View check run for this annotation

Codecov / codecov/patch

main.go#L202 

Added line #L202 was not covered by tests
if err != nil {
setupLog.Error(err, "unable to create controller", "controller", "Task")
os.Exit(1)
Expand Down