✨ added per controller leader election

pkg/builder/controller_test.go

@@ -33,6 +33,7 @@ import (
 	"k8s.io/client-go/util/workqueue"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/leaderelection"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/scheme"
@@ -62,6 +63,11 @@ var _ = Describe("application", func() {
 			instance, err := ControllerManagedBy(m).
 				For(&appsv1.ReplicaSet{}).
 				Owns(&appsv1.ReplicaSet{}).
+				WithOptions(controller.Options{
+					LeaderElection: &controller.LeaderElectionOptions{
+						LeaderElectionMode: leaderelection.NonLeaderElectionMode,
+					},
+				}).
 				Build(noop)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(instance).NotTo(BeNil())
@@ -76,6 +82,11 @@ var _ = Describe("application", func() {
 			instance, err := ControllerManagedBy(m).
 				For(&fakeType{}).
 				Owns(&appsv1.ReplicaSet{}).
+				WithOptions(controller.Options{
+					LeaderElection: &controller.LeaderElectionOptions{
+						LeaderElectionMode: leaderelection.NonLeaderElectionMode,
+					},
+				}).
 				Build(noop)
 			Expect(err).To(MatchError(ContainSubstring("no kind is registered for the type builder.fakeType")))
 			Expect(instance).To(BeNil())
@@ -98,6 +109,11 @@ var _ = Describe("application", func() {
 			instance, err := ControllerManagedBy(m).
 				For(&appsv1.ReplicaSet{}).
 				Owns(&appsv1.ReplicaSet{}).
+				WithOptions(controller.Options{
+					LeaderElection: &controller.LeaderElectionOptions{
+						LeaderElectionMode: leaderelection.NonLeaderElectionMode,
+					},
+				}).
 				Build(noop)
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("expected error"))
@@ -121,7 +137,12 @@ var _ = Describe("application", func() {
 			instance, err := ControllerManagedBy(m).
 				For(&appsv1.ReplicaSet{}).
 				Owns(&appsv1.ReplicaSet{}).
-				WithOptions(controller.Options{MaxConcurrentReconciles: maxConcurrentReconciles}).
+				WithOptions(controller.Options{
+					MaxConcurrentReconciles: maxConcurrentReconciles,
+					LeaderElection: &controller.LeaderElectionOptions{
+						LeaderElectionMode: leaderelection.NonLeaderElectionMode,
+					},
+				}).
 				Build(noop)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(instance).NotTo(BeNil())
@@ -164,6 +185,11 @@ var _ = Describe("application", func() {
 			ctrl1, err := ControllerManagedBy(m).
 				For(&TestDefaultValidator{}).
 				Owns(&appsv1.ReplicaSet{}).
+				WithOptions(controller.Options{
+					LeaderElection: &controller.LeaderElectionOptions{
+						LeaderElectionMode: leaderelection.NonLeaderElectionMode,
+					},
+				}).
 				Build(noop)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(ctrl1).NotTo(BeNil())
@@ -172,6 +198,11 @@ var _ = Describe("application", func() {
 			ctrl2, err := ControllerManagedBy(m).
 				For(&TestDefaultValidator{}).
 				Owns(&appsv1.ReplicaSet{}).
+				WithOptions(controller.Options{
+					LeaderElection: &controller.LeaderElectionOptions{
+						LeaderElectionMode: leaderelection.NonLeaderElectionMode,
+					},
+				}).
 				Build(noop)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(ctrl2).NotTo(BeNil())
@@ -185,7 +216,12 @@ var _ = Describe("application", func() {
 
 			bldr := ControllerManagedBy(m).
 				For(&appsv1.Deployment{}).
-				Owns(&appsv1.ReplicaSet{})
+				Owns(&appsv1.ReplicaSet{}).
+				WithOptions(controller.Options{
+					LeaderElection: &controller.LeaderElectionOptions{
+						LeaderElectionMode: leaderelection.NonLeaderElectionMode,
+					},
+				})
 			doReconcileTest("3", stop, bldr, m, false)
 			close(done)
 		}, 10)
@@ -196,6 +232,11 @@ var _ = Describe("application", func() {
 
 			bldr := ControllerManagedBy(m).
 				For(&appsv1.Deployment{}).
+				WithOptions(controller.Options{
+					LeaderElection: &controller.LeaderElectionOptions{
+						LeaderElectionMode: leaderelection.NonLeaderElectionMode,
+					},
+				}).
 				Watches( // Equivalent of Owns
 					&source.Kind{Type: &appsv1.ReplicaSet{}},
 					&handler.EnqueueRequestForOwner{OwnerType: &appsv1.Deployment{}, IsController: true})

pkg/cache/informer_cache.go

@@ -30,6 +30,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/cache/internal"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
+	"sigs.k8s.io/controller-runtime/pkg/leaderelection"
 )
 
 var (
@@ -163,10 +164,16 @@ func (ip *informerCache) GetInformer(obj runtime.Object) (Informer, error) {
 	return i.Informer, err
 }
 
-// NeedLeaderElection implements the LeaderElectionRunnable interface
+// GetLeaderElectionMode implements the LeaderElectionRunnable interface
 // to indicate that this can be started without requiring the leader lock
-func (ip *informerCache) NeedLeaderElection() bool {
-	return false
+func (ip *informerCache) GetLeaderElectionMode() leaderelection.Mode {
+	return leaderelection.NonLeaderElectionMode
+}
+
+// GetID implements the LeaderElectionRunnable interface.
+// It's dummy method that always returns empty string as informerCache doesn't need leader election.
+func (ip *informerCache) GetID() string {
+	return ""
 }
 
 // IndexField adds an indexer to the underlying cache, using extraction function to get

pkg/cache/informer_cache_test.go

@@ -8,6 +8,7 @@ import (
 
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
+	"sigs.k8s.io/controller-runtime/pkg/leaderelection"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 )
 
@@ -23,6 +24,6 @@ var _ = Describe("informerCache", func() {
 
 		leaderElectionRunnable, ok := c.(manager.LeaderElectionRunnable)
 		Expect(ok).To(BeTrue())
-		Expect(leaderElectionRunnable.NeedLeaderElection()).To(BeFalse())
+		Expect(leaderElectionRunnable.GetLeaderElectionMode()).To(Equal(leaderelection.NonLeaderElectionMode))
 	})
 })

pkg/controller/controller.go

@@ -22,6 +22,7 @@ import (
 	"k8s.io/client-go/util/workqueue"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/internal/controller"
+	"sigs.k8s.io/controller-runtime/pkg/leaderelection"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/ratelimiter"
@@ -41,6 +42,18 @@ type Options struct {
 	// Defaults to MaxOfRateLimiter which has both overall and per-item rate limiting.
 	// The overall is a token bucket and the per-item is exponential.
 	RateLimiter ratelimiter.RateLimiter
+
+	// Leader election is by default
+	LeaderElection *LeaderElectionOptions
+}
+
+// Leader Election options
+type LeaderElectionOptions struct {
+	// LeaderElectionMode determines what leader election mode to use when starting the controller.
+	LeaderElectionMode leaderelection.Mode
+
+	// LeaderElectionID determines the name of the configmap that leader election will use for holding the leader lock.
+	LeaderElectionID string
 }
 
 // Controller implements a Kubernetes API.  A Controller manages a work queue fed reconcile.Requests
@@ -83,6 +96,14 @@ func New(name string, mgr manager.Manager, options Options) (Controller, error)
 		options.RateLimiter = workqueue.DefaultControllerRateLimiter()
 	}
 
+	if options.LeaderElection == nil {
+		// Defaulting to per-manager leader election mode for backwards compatibility
+		options.LeaderElection = &LeaderElectionOptions{
+			LeaderElectionMode: leaderelection.PerManagerLeaderElectionMode,
+			LeaderElectionID:   name,
+		}
+	}
+
 	// Inject dependencies into Reconciler
 	if err := mgr.SetFields(options.Reconciler); err != nil {
 		return nil, err
@@ -101,6 +122,8 @@ func New(name string, mgr manager.Manager, options Options) (Controller, error)
 		},
 		MaxConcurrentReconciles: options.MaxConcurrentReconciles,
 		Name:                    name,
+		LeaderElectionMode:      options.LeaderElection.LeaderElectionMode,
+		LeaderElectionID:        options.LeaderElection.LeaderElectionID,
 	}
 
 	// Add the controller as a Manager components

pkg/controller/controller_integration_test.go

@@ -28,6 +28,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllertest"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/leaderelection"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
@@ -65,6 +66,9 @@ var _ = Describe("controller", func() {
 						reconciled <- request
 						return reconcile.Result{}, nil
 					}),
+				LeaderElection: &controller.LeaderElectionOptions{
+					LeaderElectionMode: leaderelection.NonLeaderElectionMode,
+				},
 			})
 			Expect(err).NotTo(HaveOccurred())
 

pkg/controller/controller_test.go

@@ -25,6 +25,7 @@ import (
 
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/leaderelection"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/runtime/inject"
@@ -48,7 +49,12 @@ var _ = Describe("controller.Controller", func() {
 		It("should return an error if Name is not Specified", func(done Done) {
 			m, err := manager.New(cfg, manager.Options{})
 			Expect(err).NotTo(HaveOccurred())
-			c, err := controller.New("", m, controller.Options{Reconciler: rec})
+			c, err := controller.New("", m, controller.Options{
+				Reconciler: rec,
+				LeaderElection: &controller.LeaderElectionOptions{
+					LeaderElectionMode: leaderelection.NonLeaderElectionMode,
+				},
+			})
 			Expect(c).To(BeNil())
 			Expect(err.Error()).To(ContainSubstring("must specify Name for Controller"))
 
@@ -70,7 +76,12 @@ var _ = Describe("controller.Controller", func() {
 			m, err := manager.New(cfg, manager.Options{})
 			Expect(err).NotTo(HaveOccurred())
 
-			c, err := controller.New("foo", m, controller.Options{Reconciler: &failRec{}})
+			c, err := controller.New("foo", m, controller.Options{
+				Reconciler: &failRec{},
+				LeaderElection: &controller.LeaderElectionOptions{
+					LeaderElectionMode: leaderelection.NonLeaderElectionMode,
+				},
+			})
 			Expect(c).To(BeNil())
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("expected error"))
@@ -82,11 +93,21 @@ var _ = Describe("controller.Controller", func() {
 			m, err := manager.New(cfg, manager.Options{})
 			Expect(err).NotTo(HaveOccurred())
 
-			c1, err := controller.New("c1", m, controller.Options{Reconciler: rec})
+			c1, err := controller.New("c1", m, controller.Options{
+				Reconciler: rec,
+				LeaderElection: &controller.LeaderElectionOptions{
+					LeaderElectionMode: leaderelection.NonLeaderElectionMode,
+				},
+			})
 			Expect(err).NotTo(HaveOccurred())
 			Expect(c1).ToNot(BeNil())
 
-			c2, err := controller.New("c2", m, controller.Options{Reconciler: rec})
+			c2, err := controller.New("c2", m, controller.Options{
+				Reconciler: rec,
+				LeaderElection: &controller.LeaderElectionOptions{
+					LeaderElectionMode: leaderelection.NonLeaderElectionMode,
+				},
+			})
 			Expect(err).NotTo(HaveOccurred())
 			Expect(c2).ToNot(BeNil())
 

pkg/internal/controller/controller.go

@@ -32,6 +32,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/internal/controller/metrics"
 	logf "sigs.k8s.io/controller-runtime/pkg/internal/log"
+	"sigs.k8s.io/controller-runtime/pkg/leaderelection"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/runtime/inject"
@@ -50,6 +51,14 @@ type Controller struct {
 	// MaxConcurrentReconciles is the maximum number of concurrent Reconciles which can be run. Defaults to 1.
 	MaxConcurrentReconciles int
 
+	// LeaderElectionMode determines which leader election mode to use when
+	// starting the controller.
+	LeaderElectionMode leaderelection.Mode
+
+	// LeaderElectionID determines the name of the configmap that leader election
+	// will use for holding the leader lock.
+	LeaderElectionID string
+
 	// Reconciler is a function that can be called at any time with the Name / Namespace of an object and
 	// ensures that the state of the system matches the state specified in the object.
 	// Defaults to the DefaultReconcileFunc.
@@ -296,3 +305,11 @@ func (c *Controller) InjectFunc(f inject.Func) error {
 func (c *Controller) updateMetrics(reconcileTime time.Duration) {
 	ctrlmetrics.ReconcileTime.WithLabelValues(c.Name).Observe(reconcileTime.Seconds())
 }
+
+func (c *Controller) GetLeaderElectionMode() leaderelection.Mode {
+	return c.LeaderElectionMode
+}
+
+func (c *Controller) GetID() string {
+	return c.LeaderElectionID
+}

pkg/internal/recorder/recorder_integration_test.go

@@ -25,6 +25,7 @@ import (
 	ref "k8s.io/client-go/tools/reference"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/leaderelection"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
@@ -61,6 +62,9 @@ var _ = Describe("recorder", func() {
 						recorder.Event(dp, corev1.EventTypeNormal, "test-reason", "test-msg")
 						return reconcile.Result{}, nil
 					}),
+				LeaderElection: &controller.LeaderElectionOptions{
+					LeaderElectionMode: leaderelection.NonLeaderElectionMode,
+				},
 			})
 			Expect(err).NotTo(HaveOccurred())
 

pkg/leaderelection/leader_election.go

@@ -29,7 +29,20 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/recorder"
 )
 
-const inClusterNamespacePath = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"
+type Mode uint8
+
+const (
+	inClusterNamespacePath = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"
+
+	// NonLeaderElectionMode mode for Runnables that don't need leader election
+	NonLeaderElectionMode Mode = 0
+
+	// PerManagerLeaderElectionMode mode for Runnables that need per-manager leader election mode
+	PerManagerLeaderElectionMode Mode = 1
+
+	// PerControllerGroupLeaderElectionMode mode for Runnables that need per-controller leader election mode
+	PerControllerGroupLeaderElectionMode Mode = 2
+)
 
 // Options provides the required configuration to create a new resource lock
 type Options struct {