added leader election for runnables

Author: GrigoriyMikhalkin

pkg/builder/controller_test.go

@@ -62,6 +62,11 @@ var _ = Describe("application", func() {
 			instance, err := ControllerManagedBy(m).
 				For(&appsv1.ReplicaSet{}).
 				Owns(&appsv1.ReplicaSet{}).
+				WithOptions(controller.Options{
+					LeaderElection: &controller.LeaderElectionOptions{
+						NeedLeaderElection: false,
+					},
+				}).
 				Build(noop)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(instance).NotTo(BeNil())
@@ -76,6 +81,11 @@ var _ = Describe("application", func() {
 			instance, err := ControllerManagedBy(m).
 				For(&fakeType{}).
 				Owns(&appsv1.ReplicaSet{}).
+				WithOptions(controller.Options{
+					LeaderElection: &controller.LeaderElectionOptions{
+						NeedLeaderElection: false,
+					},
+				}).
 				Build(noop)
 			Expect(err).To(MatchError(ContainSubstring("no kind is registered for the type builder.fakeType")))
 			Expect(instance).To(BeNil())
@@ -98,6 +108,11 @@ var _ = Describe("application", func() {
 			instance, err := ControllerManagedBy(m).
 				For(&appsv1.ReplicaSet{}).
 				Owns(&appsv1.ReplicaSet{}).
+				WithOptions(controller.Options{
+					LeaderElection: &controller.LeaderElectionOptions{
+						NeedLeaderElection: false,
+					},
+				}).
 				Build(noop)
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("expected error"))
@@ -121,7 +136,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{
+						NeedLeaderElection: false,
+					},
+				}).
 				Build(noop)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(instance).NotTo(BeNil())
@@ -164,6 +184,11 @@ var _ = Describe("application", func() {
 			ctrl1, err := ControllerManagedBy(m).
 				For(&TestDefaultValidator{}).
 				Owns(&appsv1.ReplicaSet{}).
+				WithOptions(controller.Options{
+					LeaderElection: &controller.LeaderElectionOptions{
+						NeedLeaderElection: false,
+					},
+				}).
 				Build(noop)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(ctrl1).NotTo(BeNil())
@@ -172,6 +197,11 @@ var _ = Describe("application", func() {
 			ctrl2, err := ControllerManagedBy(m).
 				For(&TestDefaultValidator{}).
 				Owns(&appsv1.ReplicaSet{}).
+				WithOptions(controller.Options{
+					LeaderElection: &controller.LeaderElectionOptions{
+						NeedLeaderElection: false,
+					},
+				}).
 				Build(noop)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(ctrl2).NotTo(BeNil())
@@ -185,7 +215,12 @@ var _ = Describe("application", func() {
 
 			bldr := ControllerManagedBy(m).
 				For(&appsv1.Deployment{}).
-				Owns(&appsv1.ReplicaSet{})
+				Owns(&appsv1.ReplicaSet{}).
+				WithOptions(controller.Options{
+					LeaderElection: &controller.LeaderElectionOptions{
+						NeedLeaderElection: false,
+					},
+				})
 			doReconcileTest("3", stop, bldr, m, false)
 			close(done)
 		}, 10)
@@ -196,6 +231,11 @@ var _ = Describe("application", func() {
 
 			bldr := ControllerManagedBy(m).
 				For(&appsv1.Deployment{}).
+				WithOptions(controller.Options{
+					LeaderElection: &controller.LeaderElectionOptions{
+						NeedLeaderElection: false,
+					},
+				}).
 				Watches( // Equivalent of Owns
 					&source.Kind{Type: &appsv1.ReplicaSet{}},
 					&handler.EnqueueRequestForOwner{OwnerType: &appsv1.Deployment{}, IsController: true})

pkg/cache/informer_cache.go

@@ -169,6 +169,12 @@ func (ip *informerCache) NeedLeaderElection() bool {
 	return false
 }
 
+// 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
 // value(s) from the given field.  This index can then be used by passing a field selector
 // to List. For one-to-one compatibility with "normal" field selectors, only return one value.

pkg/controller/controller.go

@@ -41,6 +41,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 {
+	// NeedLeaderElection determines whether or not to use leader election when starting the controller.
+	NeedLeaderElection bool
+
+	// 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 +95,13 @@ func New(name string, mgr manager.Manager, options Options) (Controller, error)
 		options.RateLimiter = workqueue.DefaultControllerRateLimiter()
 	}
 
+	if options.LeaderElection == nil {
+		options.LeaderElection = &LeaderElectionOptions{
+			NeedLeaderElection: true,
+			LeaderElectionID:   name,
+		}
+	}
+
 	// Inject dependencies into Reconciler
 	if err := mgr.SetFields(options.Reconciler); err != nil {
 		return nil, err
@@ -101,6 +120,8 @@ func New(name string, mgr manager.Manager, options Options) (Controller, error)
 		},
 		MaxConcurrentReconciles: options.MaxConcurrentReconciles,
 		Name:                    name,
+		LeaderElection:          options.LeaderElection.NeedLeaderElection,
+		LeaderElectionID:        options.LeaderElection.LeaderElectionID,
 	}
 
 	// Add the controller as a Manager components

pkg/controller/controller_integration_test.go

@@ -65,6 +65,9 @@ var _ = Describe("controller", func() {
 						reconciled <- request
 						return reconcile.Result{}, nil
 					}),
+				LeaderElection: &controller.LeaderElectionOptions{
+					NeedLeaderElection: false,
+				},
 			})
 			Expect(err).NotTo(HaveOccurred())
 

pkg/controller/controller_test.go

@@ -48,7 +48,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{
+					NeedLeaderElection: false,
+				},
+			})
 			Expect(c).To(BeNil())
 			Expect(err.Error()).To(ContainSubstring("must specify Name for Controller"))
 
@@ -70,7 +75,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{
+					NeedLeaderElection: false,
+				},
+			})
 			Expect(c).To(BeNil())
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("expected error"))
@@ -82,11 +92,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{
+					NeedLeaderElection: false,
+				},
+			})
 			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{
+					NeedLeaderElection: false,
+				},
+			})
 			Expect(err).NotTo(HaveOccurred())
 			Expect(c2).ToNot(BeNil())
 

pkg/internal/controller/controller.go

@@ -50,6 +50,14 @@ type Controller struct {
 	// MaxConcurrentReconciles is the maximum number of concurrent Reconciles which can be run. Defaults to 1.
 	MaxConcurrentReconciles int
 
+	// LeaderElection determines whether or not to use leader election when
+	// starting the controller.
+	LeaderElection bool
+
+	// 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 +304,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) NeedLeaderElection() bool {
+	return c.LeaderElection
+}
+
+func (c *Controller) GetID() string {
+	return c.LeaderElectionID
+}

pkg/internal/recorder/recorder_integration_test.go

@@ -61,6 +61,9 @@ var _ = Describe("recorder", func() {
 						recorder.Event(dp, corev1.EventTypeNormal, "test-reason", "test-msg")
 						return reconcile.Result{}, nil
 					}),
+				LeaderElection: &controller.LeaderElectionOptions{
+					NeedLeaderElection: false,
+				},
 			})
 			Expect(err).NotTo(HaveOccurred())