HA components

Author: Mengqi Yu

pkg/manager/internal.go

@@ -50,8 +50,11 @@ type controllerManager struct {
 	// to scheme.scheme.
 	scheme *runtime.Scheme
 
-	// runnables is the set of Controllers that the controllerManager injects deps into and Starts.
+	// runnables is the set of Controllers and (or) webhook server that the controllerManager injects deps into and Starts.
 	runnables []Runnable
+	// haRunnables is the set of webhook servers that the controllerManager injects deps into and Starts.
+	// These Runnables are in HA mode (not blocked by lead election)
+	haRunnables []Runnable
 
 	cache cache.Cache
 
@@ -104,7 +107,7 @@ type controllerManager struct {
 }
 
 // Add sets dependencies on i, and adds it to the list of runnables to start.
-func (cm *controllerManager) Add(r Runnable) error {
+func (cm *controllerManager) add(r Runnable, leaderElection bool) error {
 	cm.mu.Lock()
 	defer cm.mu.Unlock()
 
@@ -114,7 +117,11 @@ func (cm *controllerManager) Add(r Runnable) error {
 	}
 
 	// Add the runnable to the list
-	cm.runnables = append(cm.runnables, r)
+	if leaderElection {
+		cm.runnables = append(cm.runnables, r)
+	} else {
+		cm.haRunnables = append(cm.haRunnables, r)
+	}
 	if cm.started {
 		// If already started, start the controller
 		go func() {
@@ -125,6 +132,16 @@ func (cm *controllerManager) Add(r Runnable) error {
 	return nil
 }
 
+// Add sets dependencies on i, and adds it to the list of runnables to start.
+func (cm *controllerManager) Add(r Runnable) error {
+	return cm.add(r, true)
+}
+
+// AddHA sets dependencies on i, and adds it to the list of runnables to start in HA mode.
+func (cm *controllerManager) AddHA(r Runnable) error {
+	return cm.add(r, false)
+}
+
 func (cm *controllerManager) SetFields(i interface{}) error {
 	if _, err := inject.ConfigInto(cm.config, i); err != nil {
 		return err
@@ -235,13 +252,15 @@ func (cm *controllerManager) Start(stop <-chan struct{}) error {
 		go cm.serveMetrics(cm.internalStop)
 	}
 
+	go cm.start()
+
 	if cm.resourceLock != nil {
 		err := cm.startLeaderElection()
 		if err != nil {
 			return err
 		}
 	} else {
-		go cm.start()
+		go cm.startLeaderElected()
 	}
 
 	select {
@@ -272,6 +291,24 @@ func (cm *controllerManager) start() {
 	// TODO(community): Check the return value and write a test
 	cm.cache.WaitForCacheSync(cm.internalStop)
 
+	// Start the HA runnables after the cache has synced
+	for _, c := range cm.haRunnables {
+		// 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
+		go func() {
+			cm.errChan <- ctrl.Start(cm.internalStop)
+		}()
+	}
+
+	cm.started = true
+}
+
+func (cm *controllerManager) startLeaderElected() {
+	// Wait for the caches to sync.
+	// TODO(community): Check the return value and write a test
+	cm.cache.WaitForCacheSync(cm.internalStop)
+
 	// Start the runnables after the cache has synced
 	for _, c := range cm.runnables {
 		// Controllers block, but we want to return an error if any have an error starting.
@@ -295,7 +332,7 @@ func (cm *controllerManager) startLeaderElection() (err error) {
 		RetryPeriod:   2 * time.Second,
 		Callbacks: leaderelection.LeaderCallbacks{
 			OnStartedLeading: func(_ context.Context) {
-				cm.start()
+				cm.startLeaderElected()
 			},
 			OnStoppedLeading: func() {
 				// Most implementations of leader election log.Fatal() here.

pkg/manager/manager.go

@@ -42,11 +42,18 @@ import (
 // Manager initializes shared dependencies such as Caches and Clients, and provides them to Runnables.
 // A Manager is required to create Controllers.
 type Manager interface {
-	// Add will set reqeusted dependencies on the component, and cause the component to be
+	// Add will set requested dependencies on the component, and cause the component to be
 	// started when Start is called.  Add will inject any dependencies for which the argument
-	// implements the inject interface - e.g. inject.Client
+	// implements the inject interface - e.g. inject.Client. Runnable added by calling `Add`
+	// will be managed by leader election if the leader election is enabled.
 	Add(Runnable) error
 
+	// AddHA will set reqeusted dependencies on the component, and cause the component to be
+	// started when Start is called.  Add will inject any dependencies for which the argument
+	// implements the inject interface - e.g. inject.Client. Runnable added by calling `AddHA`
+	// are NOT managed by leader election and all replicas will be running.
+	AddHA(Runnable) error
+
 	// SetFields will set any dependencies on an object for which the object has implemented the inject
 	// interface - e.g. inject.Client.
 	SetFields(interface{}) error

pkg/manager/manager_test.go

@@ -218,12 +218,28 @@ var _ = Describe("manger.Manager", func() {
 					return nil
 				}))
 
+				c3 := make(chan struct{})
+				m.AddHA(RunnableFunc(func(s <-chan struct{}) error {
+					defer close(c3)
+					defer GinkgoRecover()
+					return nil
+				}))
+
+				c4 := make(chan struct{})
+				m.Add(RunnableFunc(func(s <-chan struct{}) error {
+					defer close(c4)
+					defer GinkgoRecover()
+					return nil
+				}))
+
 				go func() {
 					defer GinkgoRecover()
 					Expect(m.Start(stop)).NotTo(HaveOccurred())
 				}()
 				<-c1
 				<-c2
+				<-c3
+				<-c4
 
 				close(done)
 			})
@@ -251,7 +267,7 @@ var _ = Describe("manger.Manager", func() {
 				close(done)
 			})
 
-			It("should return an error if any Components fail to Start", func(done Done) {
+			It("should return an error if any Components that managed by leader election fail to Start", func(done Done) {
 				m, err := New(cfg, options)
 				Expect(err).NotTo(HaveOccurred())
 				c1 := make(chan struct{})
@@ -284,6 +300,40 @@ var _ = Describe("manger.Manager", func() {
 				<-c2
 				<-c3
 			})
+
+			It("should return an error if any Components fail to Start", func(done Done) {
+				m, err := New(cfg, options)
+				Expect(err).NotTo(HaveOccurred())
+				c1 := make(chan struct{})
+				m.AddHA(RunnableFunc(func(s <-chan struct{}) error {
+					defer GinkgoRecover()
+					defer close(c1)
+					return nil
+				}))
+
+				c2 := make(chan struct{})
+				m.AddHA(RunnableFunc(func(s <-chan struct{}) error {
+					defer GinkgoRecover()
+					defer close(c2)
+					return fmt.Errorf("expected error")
+				}))
+
+				c3 := make(chan struct{})
+				m.Add(RunnableFunc(func(s <-chan struct{}) error {
+					defer GinkgoRecover()
+					defer close(c3)
+					return nil
+				}))
+
+				go func() {
+					defer GinkgoRecover()
+					Expect(m.Start(stop)).NotTo(HaveOccurred())
+					close(done)
+				}()
+				<-c1
+				<-c2
+				<-c3
+			})
 		}
 
 		Context("with defaults", func() {