Allow controllers to be started without adding them to the manager

As far as I can tell the main reason we add controllers to the manager is to
ensure that we've been elected leader before we start any controllers. The
downside of this design is that it's not possible to stop individual
controllers, or remove controllers from the manager.

I believe this commit is the minimum possible change necessary to allow
controllers to be started and stopped on-demand. It allows a controller to be
created, started, and stopped without ever being added to the manager. Any
controller that is started separately from the manager must handle its own
leader election.

kubernetes-sigs#730

Signed-off-by: Nic Cope <[email protected]>

Author: negz

pkg/controller/controller.go

@@ -67,6 +67,17 @@ type Controller interface {
 // New returns a new Controller registered with the Manager.  The Manager will ensure that shared Caches have
 // been synced before the Controller is Started.
 func New(name string, mgr manager.Manager, options Options) (Controller, error) {
+	c, err := Configure(name, mgr, options)
+	if err != nil {
+		return nil, err
+	}
+
+	// Add the controller as a Manager components
+	return c, mgr.Add(c)
+}
+
+// Configure a new controller without starting it or adding it to the manager.
+func Configure(name string, mgr manager.Manager, options Options) (Controller, error) {
 	if options.Reconciler == nil {
 		return nil, fmt.Errorf("must specify Reconciler")
 	}
@@ -103,6 +114,5 @@ func New(name string, mgr manager.Manager, options Options) (Controller, error)
 		Name:                    name,
 	}
 
-	// Add the controller as a Manager components
-	return c, mgr.Add(c)
+	return c, nil
 }

pkg/manager/internal.go

@@ -134,6 +134,8 @@ type controllerManager struct {
 	// It and `internalStop` should point to the same channel.
 	internalStopper chan<- struct{}
 
+	elected chan struct{}
+
 	startCache func(stop <-chan struct{}) error
 
 	// port is the port that the webhook server serves at.
@@ -457,6 +459,8 @@ func (cm *controllerManager) Start(stop <-chan struct{}) error {
 			return err
 		}
 	} else {
+		// Treat not having an election the same as being elected.
+		close(cm.elected)
 		go cm.startLeaderElectionRunnables()
 	}
 
@@ -545,6 +549,7 @@ func (cm *controllerManager) startLeaderElection() (err error) {
 		RetryPeriod:   cm.retryPeriod,
 		Callbacks: leaderelection.LeaderCallbacks{
 			OnStartedLeading: func(_ context.Context) {
+				close(cm.elected)
 				cm.startLeaderElectionRunnables()
 			},
 			OnStoppedLeading: func() {
@@ -572,3 +577,7 @@ func (cm *controllerManager) startLeaderElection() (err error) {
 	go l.Run(ctx)
 	return nil
 }
+
+func (cm *controllerManager) Elected() <-chan struct{} {
+	return cm.elected
+}

pkg/manager/manager.go

@@ -51,6 +51,10 @@ type Manager interface {
 	// non-leaderelection mode (always running) or leader election mode (managed by leader election if enabled).
 	Add(Runnable) error
 
+	// Elected is closed when this manager is elected the leader, or when no
+	// election is configured.
+	Elected() <-chan struct{}
+
 	// SetFields will set any dependencies on an object for which the object has implemented the inject
 	// interface - e.g. inject.Client.
 	SetFields(interface{}) error
@@ -325,6 +329,7 @@ func New(config *rest.Config, options Options) (Manager, error) {
 		metricsExtraHandlers:  metricsExtraHandlers,
 		internalStop:          stop,
 		internalStopper:       stop,
+		elected:               make(chan struct{}),
 		port:                  options.Port,
 		host:                  options.Host,
 		certDir:               options.CertDir,