added leader election for runnables

Author: GrigoriyMikhalkin

pkg/controller/controller.go

@@ -35,6 +35,13 @@ type Options struct {
 
 	// Reconciler reconciles an object
 	Reconciler reconcile.Reconciler
+
+	// LeaderElection determines whether or not to use leader election when starting the controller.
+	// Defaults to true
+	LeaderElection 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
@@ -89,6 +96,8 @@ func New(name string, mgr manager.Manager, options Options) (Controller, error)
 		Queue:                   workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), name),
 		MaxConcurrentReconciles: options.MaxConcurrentReconciles,
 		Name:                    name,
+		LeaderElection:          options.LeaderElection,
+		LeaderElectionID:        options.LeaderElectionID,
 	}
 
 	// Add the controller as a Manager components

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.
@@ -256,3 +264,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/leaderelection/leader_election.go

@@ -52,11 +52,6 @@ func NewResourceLock(config *rest.Config, recorderProvider recorder.Provider, op
 		return nil, nil
 	}
 
-	// Default the LeaderElectionID
-	if options.LeaderElectionID == "" {
-		options.LeaderElectionID = "controller-leader-election-helper"
-	}
-
 	// Default the namespace (if running in cluster)
 	if options.LeaderElectionNamespace == "" {
 		var err error

pkg/manager/internal.go

@@ -18,6 +18,7 @@ package manager
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net"
 	"net/http"
@@ -34,6 +35,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	logf "sigs.k8s.io/controller-runtime/pkg/internal/log"
+	crleaderelection "sigs.k8s.io/controller-runtime/pkg/leaderelection"
 	"sigs.k8s.io/controller-runtime/pkg/metrics"
 	"sigs.k8s.io/controller-runtime/pkg/recorder"
 	"sigs.k8s.io/controller-runtime/pkg/runtime/inject"
@@ -82,8 +84,16 @@ type controllerManager struct {
 	// (and EventHandlers, Sources and Predicates).
 	recorderProvider recorder.Provider
 
-	// resourceLock forms the basis for leader election
-	resourceLock resourcelock.Interface
+	// leaderElectionNamespace determines the namespace in which the leader
+	// election configmap will be created.
+	leaderElectionNamespace string
+
+	// leaderElectionID determines the name of the configmap that leader election
+	// will use for holding the leader lock.
+	leaderElectionID string
+
+	// resourceLocksMap maps leader election ID to resource lock
+	resourceLocksMap map[string]resourcelock.Interface
 
 	// mapper is used to map resources to kind, and map kind and version.
 	mapper meta.RESTMapper
@@ -123,6 +133,9 @@ type controllerManager struct {
 	// retryPeriod is the duration the LeaderElector clients should wait
 	// between tries of actions.
 	retryPeriod time.Duration
+
+	// Dependency injection for testing
+	newResourceLock func(config *rest.Config, recorderProvider recorder.Provider, options crleaderelection.Options) (resourcelock.Interface, error)
 }
 
 // Add sets dependencies on i, and adds it to the list of Runnables to start.
@@ -265,8 +278,18 @@ func (cm *controllerManager) Start(stop <-chan struct{}) error {
 
 	go cm.startNonLeaderElectionRunnables()
 
-	if cm.resourceLock != nil {
-		err := cm.startLeaderElection()
+	if resourceLock := cm.resourceLocksMap[cm.leaderElectionID]; resourceLock != nil {
+		err := cm.startLeaderElection(cm.leaderElectionID, leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(_ context.Context) {
+				cm.startLeaderElectionRunnables()
+			},
+			OnStoppedLeading: func() {
+				// Most implementations of leader election log.Fatal() here.
+				// Since Start is wrapped in log.Fatal when called, we can just return
+				// an error here which will cause the program to exit.
+				cm.errChan <- fmt.Errorf("leader election lost")
+			},
+		})
 		if err != nil {
 			return err
 		}
@@ -313,7 +336,46 @@ func (cm *controllerManager) startLeaderElectionRunnables() {
 		// Write any Start errors to a channel so we can return them
 		ctrl := c
 		go func() {
-			cm.errChan <- ctrl.Start(cm.internalStop)
+			if leRunnable, ok := ctrl.(LeaderElectionRunnable); ok {
+				leID := leRunnable.GetID()
+
+				// Check that leader election ID is defined
+				if leID == "" {
+					cm.errChan <- errors.New("LeaderElectionID must be configured")
+					return
+				}
+
+				// Check that leader election ID is unique
+				if _, exists := cm.resourceLocksMap[leID]; exists {
+					cm.errChan <- errors.New("LeaderElectionID must be unique")
+					return
+				}
+
+				// Create resource lock
+				resourceLock, err := cm.newResourceLock(cm.config, cm.recorderProvider, crleaderelection.Options{
+					LeaderElection:          true,
+					LeaderElectionID:        leID,
+					LeaderElectionNamespace: cm.leaderElectionNamespace,
+				})
+				if err != nil {
+					cm.errChan <- err
+				}
+
+				cm.resourceLocksMap[leID] = resourceLock
+				err = cm.startLeaderElection(leID, leaderelection.LeaderCallbacks{
+					OnStartedLeading: func(_ context.Context) {
+						cm.errChan <- ctrl.Start(cm.internalStop)
+					},
+					OnStoppedLeading: func() {
+						cm.errChan <- fmt.Errorf("runnable leader election lost")
+					},
+				})
+				if err != nil {
+					cm.errChan <- err
+				}
+			} else {
+				cm.errChan <- ctrl.Start(cm.internalStop)
+			}
 		}()
 	}
 }
@@ -339,23 +401,13 @@ func (cm *controllerManager) waitForCache() {
 	cm.started = true
 }
 
-func (cm *controllerManager) startLeaderElection() (err error) {
+func (cm *controllerManager) startLeaderElection(leaderElectionID string, callbacks leaderelection.LeaderCallbacks) (err error) {
 	l, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
-		Lock:          cm.resourceLock,
+		Lock:          cm.resourceLocksMap[leaderElectionID],
 		LeaseDuration: cm.leaseDuration,
 		RenewDeadline: cm.renewDeadline,
 		RetryPeriod:   cm.retryPeriod,
-		Callbacks: leaderelection.LeaderCallbacks{
-			OnStartedLeading: func(_ context.Context) {
-				cm.startLeaderElectionRunnables()
-			},
-			OnStoppedLeading: func() {
-				// Most implementations of leader election log.Fatal() here.
-				// Since Start is wrapped in log.Fatal when called, we can just return
-				// an error here which will cause the program to exit.
-				cm.errChan <- fmt.Errorf("leader election lost")
-			},
-		},
+		Callbacks:     callbacks,
 	})
 	if err != nil {
 		return err

pkg/manager/manager.go

@@ -190,6 +190,9 @@ type LeaderElectionRunnable interface {
 	// NeedLeaderElection returns true if the Runnable needs to be run in the leader election mode.
 	// e.g. controllers need to be run in leader election mode, while webhook server doesn't.
 	NeedLeaderElection() bool
+
+	// GetID returns leader election ID
+	GetID() string
 }
 
 // New returns a new Manager for creating Controllers.
@@ -241,6 +244,9 @@ func New(config *rest.Config, options Options) (Manager, error) {
 	if err != nil {
 		return nil, err
 	}
+	resourceLockMap := map[string]resourcelock.Interface{
+		options.LeaderElectionID: resourceLock,
+	}
 
 	// Create the mertics listener. This will throw an error if the metrics bind
 	// address is invalid or already in use.
@@ -252,24 +258,27 @@ func New(config *rest.Config, options Options) (Manager, error) {
 	stop := make(chan struct{})
 
 	return &controllerManager{
-		config:           config,
-		scheme:           options.Scheme,
-		errChan:          make(chan error),
-		cache:            cache,
-		fieldIndexes:     cache,
-		client:           writeObj,
-		apiReader:        apiReader,
-		recorderProvider: recorderProvider,
-		resourceLock:     resourceLock,
-		mapper:           mapper,
-		metricsListener:  metricsListener,
-		internalStop:     stop,
-		internalStopper:  stop,
-		port:             options.Port,
-		host:             options.Host,
-		leaseDuration:    *options.LeaseDuration,
-		renewDeadline:    *options.RenewDeadline,
-		retryPeriod:      *options.RetryPeriod,
+		config:                  config,
+		scheme:                  options.Scheme,
+		errChan:                 make(chan error),
+		cache:                   cache,
+		fieldIndexes:            cache,
+		client:                  writeObj,
+		apiReader:               apiReader,
+		recorderProvider:        recorderProvider,
+		resourceLocksMap:        resourceLockMap,
+		mapper:                  mapper,
+		metricsListener:         metricsListener,
+		internalStop:            stop,
+		internalStopper:         stop,
+		port:                    options.Port,
+		host:                    options.Host,
+		leaderElectionNamespace: options.LeaderElectionNamespace,
+		leaderElectionID:        options.LeaderElectionID,
+		leaseDuration:           *options.LeaseDuration,
+		renewDeadline:           *options.RenewDeadline,
+		retryPeriod:             *options.RetryPeriod,
+		newResourceLock:         options.newResourceLock,
 	}, nil
 }
 
@@ -317,6 +326,11 @@ func setOptionsDefaults(options Options) Options {
 		options.newRecorderProvider = internalrecorder.NewProvider
 	}
 
+	// Default the LeaderElectionID
+	if options.LeaderElectionID == "" {
+		options.LeaderElectionID = "controller-leader-election-helper"
+	}
+
 	// Allow newResourceLock to be mocked
 	if options.newResourceLock == nil {
 		options.newResourceLock = leaderelection.NewResourceLock

pkg/manager/manager_test.go

@@ -285,6 +285,34 @@ var _ = Describe("manger.Manager", func() {
 				<-c3
 			})
 
+			It("shouldn't allow empty LeaderElectionID for controller", func(done Done) {
+				m, err := New(cfg, options)
+				Expect(err).ToNot(HaveOccurred())
+
+				r := &leRunnable{
+					leaderElectionID: "",
+				}
+				Expect(m.Add(r)).To(Succeed())
+				Expect(m.Start(stop).Error()).To(ContainSubstring("LeaderElectionID must be configured"))
+
+				close(done)
+			})
+
+			It("shouldn't allow non unique LeaderElectionID for controller", func(done Done) {
+				if options.LeaderElectionID != "" {
+					m, err := New(cfg, options)
+					Expect(err).ToNot(HaveOccurred())
+
+					r := &leRunnable{
+						leaderElectionID: options.LeaderElectionID,
+					}
+					Expect(m.Add(r)).To(Succeed())
+					Expect(m.Start(stop).Error()).To(ContainSubstring("LeaderElectionID must be unique"))
+				}
+
+				close(done)
+			})
+
 			It("should return an error if any non-leaderelection Components fail to Start", func() {
 				// TODO(mengqiy): implement this after resolving https://github.com/kubernetes-sigs/controller-runtime/issues/429
 			})
@@ -711,3 +739,19 @@ func (i *injectable) InjectStopChannel(stop <-chan struct{}) error {
 func (i *injectable) Start(<-chan struct{}) error {
 	return nil
 }
+
+type leRunnable struct {
+	leaderElectionID string
+}
+
+func (*leRunnable) Start(<-chan struct{}) error {
+	return nil
+}
+
+func (*leRunnable) NeedLeaderElection() bool {
+	return true
+}
+
+func (le *leRunnable) GetID() string {
+	return le.leaderElectionID
+}

pkg/webhook/server.go

@@ -95,6 +95,10 @@ func (*Server) NeedLeaderElection() bool {
 	return false
 }
 
+func (*Server) GetID() string {
+	return ""
+}
+
 // Register marks the given webhook as being served at the given path.
 // It panics if two hooks are registered on the same path.
 func (s *Server) Register(path string, hook http.Handler) {