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"
@@ -58,9 +60,9 @@ type controllerManager struct {
 	// to scheme.scheme.
 	scheme *runtime.Scheme
 
-	// leaderElectionRunnables is the set of Controllers that the controllerManager injects deps into and Starts.
+	// leaderElectionRunnables is the map that groups runnables that use same leader election ID.
 	// These Runnables are managed by lead election.
-	leaderElectionRunnables []Runnable
+	leaderElectionRunnables map[string][]Runnable
 	// nonLeaderElectionRunnables is the set of webhook servers that the controllerManager injects deps into and Starts.
 	// These Runnables will not be blocked by lead election.
 	nonLeaderElectionRunnables []Runnable
@@ -82,8 +84,9 @@ 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 configmaps will be created.
+	leaderElectionNamespace string
 
 	// mapper is used to map resources to kind, and map kind and version.
 	mapper meta.RESTMapper
@@ -123,6 +126,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.
@@ -135,11 +141,27 @@ func (cm *controllerManager) Add(r Runnable) error {
 		return err
 	}
 
+	if cm.leaderElectionRunnables == nil {
+		cm.leaderElectionRunnables = make(map[string][]Runnable)
+	}
+
 	// Add the runnable to the leader election or the non-leaderelection list
-	if leRunnable, ok := r.(LeaderElectionRunnable); ok && !leRunnable.NeedLeaderElection() {
-		cm.nonLeaderElectionRunnables = append(cm.nonLeaderElectionRunnables, r)
+	if leRunnable, ok := r.(LeaderElectionRunnable); ok && leRunnable.NeedLeaderElection() {
+		runnables := []Runnable{r}
+		leID := leRunnable.GetID()
+
+		// Check that leader election ID is defined
+		if leID == "" {
+			return errors.New("LeaderElectionID must be configured")
+		}
+
+		if rs, exists := cm.leaderElectionRunnables[leID]; exists {
+			runnables = append(runnables, rs...)
+		}
+
+		cm.leaderElectionRunnables[leID] = runnables
 	} else {
-		cm.leaderElectionRunnables = append(cm.leaderElectionRunnables, r)
+		cm.nonLeaderElectionRunnables = append(cm.nonLeaderElectionRunnables, r)
 	}
 
 	if cm.started {
@@ -265,14 +287,7 @@ func (cm *controllerManager) Start(stop <-chan struct{}) error {
 
 	go cm.startNonLeaderElectionRunnables()
 
-	if cm.resourceLock != nil {
-		err := cm.startLeaderElection()
-		if err != nil {
-			return err
-		}
-	} else {
-		go cm.startLeaderElectionRunnables()
-	}
+	go cm.startLeaderElectionRunnables()
 
 	select {
 	case <-stop:
@@ -308,12 +323,38 @@ func (cm *controllerManager) startLeaderElectionRunnables() {
 	cm.waitForCache()
 
 	// Start the leader election Runnables after the cache has synced
-	for _, c := range cm.leaderElectionRunnables {
+	for leID, rs := range cm.leaderElectionRunnables {
 		// Controllers block, but we want to return an error if any have an error starting.
 		// Write any Start errors to a channel so we can return them
-		ctrl := c
+		leaderElectionID := leID
+		runnables := rs
 		go func() {
-			cm.errChan <- ctrl.Start(cm.internalStop)
+			// Create resource lock
+			resourceLock, err := cm.newResourceLock(cm.config, cm.recorderProvider, crleaderelection.Options{
+				LeaderElection:          true,
+				LeaderElectionID:        leaderElectionID,
+				LeaderElectionNamespace: cm.leaderElectionNamespace,
+			})
+			if err != nil {
+				cm.errChan <- err
+			}
+
+			err = cm.startLeaderElection(resourceLock, leaderelection.LeaderCallbacks{
+				OnStartedLeading: func(_ context.Context) {
+					for _, r := range runnables {
+						runnable := r
+						go func() {
+							cm.errChan <- runnable.Start(cm.internalStop)
+						}()
+					}
+				},
+				OnStoppedLeading: func() {
+					cm.errChan <- fmt.Errorf("runnable leader election lost")
+				},
+			})
+			if err != nil {
+				cm.errChan <- err
+			}
 		}()
 	}
 }
@@ -339,23 +380,13 @@ func (cm *controllerManager) waitForCache() {
 	cm.started = true
 }
 
-func (cm *controllerManager) startLeaderElection() (err error) {
+func (cm *controllerManager) startLeaderElection(resourceLock resourcelock.Interface, callbacks leaderelection.LeaderCallbacks) (err error) {
 	l, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
-		Lock:          cm.resourceLock,
+		Lock:          resourceLock,
 		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

@@ -103,18 +103,10 @@ type Options struct {
 	// value only if you know what you are doing. Defaults to 10 hours if unset.
 	SyncPeriod *time.Duration
 
-	// LeaderElection determines whether or not to use leader election when
-	// starting the manager.
-	LeaderElection bool
-
 	// LeaderElectionNamespace determines the namespace in which the leader
-	// election configmap will be created.
+	// election configmaps will be created.
 	LeaderElectionNamespace string
 
-	// LeaderElectionID determines the name of the configmap that leader election
-	// will use for holding the leader lock.
-	LeaderElectionID string
-
 	// LeaseDuration is the duration that non-leader candidates will
 	// wait to force acquire leadership. This is measured against time of
 	// last observed ack. Default is 15 seconds.
@@ -190,6 +182,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.
@@ -232,16 +227,6 @@ func New(config *rest.Config, options Options) (Manager, error) {
 		return nil, err
 	}
 
-	// Create the resource lock to enable leader election)
-	resourceLock, err := options.newResourceLock(config, recorderProvider, leaderelection.Options{
-		LeaderElection:          options.LeaderElection,
-		LeaderElectionID:        options.LeaderElectionID,
-		LeaderElectionNamespace: options.LeaderElectionNamespace,
-	})
-	if err != nil {
-		return nil, err
-	}
-
 	// Create the mertics listener. This will throw an error if the metrics bind
 	// address is invalid or already in use.
 	metricsListener, err := options.newMetricsListener(options.MetricsBindAddress)
@@ -252,24 +237,25 @@ 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,
+		mapper:                  mapper,
+		metricsListener:         metricsListener,
+		internalStop:            stop,
+		internalStopper:         stop,
+		port:                    options.Port,
+		host:                    options.Host,
+		leaderElectionNamespace: options.LeaderElectionNamespace,
+		leaseDuration:           *options.LeaseDuration,
+		renewDeadline:           *options.RenewDeadline,
+		retryPeriod:             *options.RetryPeriod,
+		newResourceLock:         options.newResourceLock,
 	}, nil
 }