Merge pull request #153 from makkes/concurrent-reconciles

Make concurrent reconciliation configurable
fluxcd · Jun 24, 2021 · d41b3f1 · d41b3f1
2 parents 6185181 + 03043e5
commit d41b3f1
Show file tree

Hide file tree

Showing 4 changed files with 28 additions and 6 deletions.
diff --git a/controllers/imagepolicy_controller.go b/controllers/imagepolicy_controller.go
@@ -30,6 +30,7 @@ import (
 	"k8s.io/client-go/tools/reference"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
@@ -57,6 +58,10 @@ type ImagePolicyReconciler struct {
 	Database              DatabaseReader
 }
 
+type ImagePolicyReconcilerOptions struct {
+	MaxConcurrentReconciles int
+}
+
 // +kubebuilder:rbac:groups=image.toolkit.fluxcd.io,resources=imagepolicies,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=image.toolkit.fluxcd.io,resources=imagepolicies/status,verbs=get;update;patch
 // +kubebuilder:rbac:groups=image.toolkit.fluxcd.io,resources=imagerepositories,verbs=get;list;watch
@@ -191,7 +196,7 @@ func (r *ImagePolicyReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 	return ctrl.Result{}, err
 }
 
-func (r *ImagePolicyReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func (r *ImagePolicyReconciler) SetupWithManager(mgr ctrl.Manager, opts ImagePolicyReconcilerOptions) error {
 	// index the policies by which image repo they point at, so that
 	// it's easy to list those out when an image repo changes.
 	if err := mgr.GetFieldIndexer().IndexField(context.Background(), &imagev1.ImagePolicy{}, imageRepoKey, func(obj client.Object) []string {
@@ -207,6 +212,9 @@ func (r *ImagePolicyReconciler) SetupWithManager(mgr ctrl.Manager) error {
 			&source.Kind{Type: &imagev1.ImageRepository{}},
 			handler.EnqueueRequestsFromMapFunc(r.imagePoliciesForRepository),
 		).
+		WithOptions(controller.Options{
+			MaxConcurrentReconciles: opts.MaxConcurrentReconciles,
+		}).
 		Complete(r)
 }
 

diff --git a/controllers/imagerepository_controller.go b/controllers/imagerepository_controller.go
@@ -42,6 +42,7 @@ import (
 	"k8s.io/client-go/tools/reference"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 
 	"github.com/fluxcd/pkg/apis/meta"
@@ -75,6 +76,10 @@ type ImageRepositoryReconciler struct {
 	}
 }
 
+type ImageRepositoryReconcilerOptions struct {
+	MaxConcurrentReconciles int
+}
+
 type dockerConfig struct {
 	Auths map[string]authn.AuthConfig
 }
@@ -352,10 +357,13 @@ func (r *ImageRepositoryReconciler) shouldScan(repo imagev1.ImageRepository, now
 	return false, when, nil
 }
 
-func (r *ImageRepositoryReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func (r *ImageRepositoryReconciler) SetupWithManager(mgr ctrl.Manager, opts ImageRepositoryReconcilerOptions) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&imagev1.ImageRepository{}).
 		WithEventFilter(predicate.Or(predicate.GenerationChangedPredicate{}, predicates.ReconcileRequestedPredicate{})).
+		WithOptions(controller.Options{
+			MaxConcurrentReconciles: opts.MaxConcurrentReconciles,
+		}).
 		Complete(r)
 }
 

diff --git a/controllers/suite_test.go b/controllers/suite_test.go
@@ -105,14 +105,14 @@ var _ = BeforeSuite(func(done Done) {
 		Scheme:   scheme.Scheme,
 		Database: database.NewBadgerDatabase(badgerDB),
 	}
-	Expect(imageRepoReconciler.SetupWithManager(k8sMgr)).To(Succeed())
+	Expect(imageRepoReconciler.SetupWithManager(k8sMgr, ImageRepositoryReconcilerOptions{})).To(Succeed())
 
 	imagePolicyReconciler = &ImagePolicyReconciler{
 		Client:   k8sMgr.GetClient(),
 		Scheme:   scheme.Scheme,
 		Database: database.NewBadgerDatabase(badgerDB),
 	}
-	Expect(imagePolicyReconciler.SetupWithManager(k8sMgr)).To(Succeed())
+	Expect(imagePolicyReconciler.SetupWithManager(k8sMgr, ImagePolicyReconcilerOptions{})).To(Succeed())
 
 	// this must be started for the caches to be running, and thereby
 	// for the client to be usable.

diff --git a/main.go b/main.go
@@ -68,6 +68,7 @@ func main() {
 		watchAllNamespaces      bool
 		storagePath             string
 		storageValueLogFileSize int64
+		concurrent              int
 	)
 
 	flag.StringVar(&metricsAddr, "metrics-addr", ":8080", "The address the metric endpoint binds to.")
@@ -77,6 +78,7 @@ func main() {
 		"Watch for custom resources in all namespaces, if set to false it will only watch the runtime namespace.")
 	flag.StringVar(&storagePath, "storage-path", "/data", "Where to store the persistent database of image metadata")
 	flag.Int64Var(&storageValueLogFileSize, "storage-value-log-file-size", 1<<28, "Set the database's memory mapped value log file size in bytes. Effective memory usage is about two times this size.")
+	flag.IntVar(&concurrent, "concurrent", 4, "The number of concurrent resource reconciles.")
 	clientOptions.BindFlags(flag.CommandLine)
 	logOptions.BindFlags(flag.CommandLine)
 	leaderElectionOptions.BindFlags(flag.CommandLine)
@@ -142,7 +144,9 @@ func main() {
 		ExternalEventRecorder: eventRecorder,
 		MetricsRecorder:       metricsRecorder,
 		Database:              db,
-	}).SetupWithManager(mgr); err != nil {
+	}).SetupWithManager(mgr, controllers.ImageRepositoryReconcilerOptions{
+		MaxConcurrentReconciles: concurrent,
+	}); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", imagev1.ImageRepositoryKind)
 		os.Exit(1)
 	}
@@ -153,7 +157,9 @@ func main() {
 		ExternalEventRecorder: eventRecorder,
 		MetricsRecorder:       metricsRecorder,
 		Database:              db,
-	}).SetupWithManager(mgr); err != nil {
+	}).SetupWithManager(mgr, controllers.ImagePolicyReconcilerOptions{
+		MaxConcurrentReconciles: concurrent,
+	}); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", imagev1.ImagePolicyKind)
 		os.Exit(1)
 	}