✨ Implement graceful shutdown

Co-authored-by: david.benque <[email protected]>

Author: alvaroaleman

pkg/manager/internal.go

@@ -18,9 +18,11 @@ package manager
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net"
 	"net/http"
+	"os"
 	"sync"
 	"time"
 
@@ -44,9 +46,10 @@ import (
 
 const (
 	// Values taken from: https://github.com/kubernetes/apiserver/blob/master/pkg/apis/config/v1alpha1/defaults.go
-	defaultLeaseDuration = 15 * time.Second
-	defaultRenewDeadline = 10 * time.Second
-	defaultRetryPeriod   = 2 * time.Second
+	defaultLeaseDuration          = 15 * time.Second
+	defaultRenewDeadline          = 10 * time.Second
+	defaultRetryPeriod            = 2 * time.Second
+	defaultGracefulShutdownPeriod = 30 * time.Second
 
 	defaultReadinessEndpoint = "/readyz"
 	defaultLivenessEndpoint  = "/healthz"
@@ -118,11 +121,7 @@ type controllerManager struct {
 	started        bool
 	startedLeader  bool
 	healthzStarted bool
-
-	// NB(directxman12): we don't just use an error channel here to avoid the situation where the
-	// error channel is too small and we end up blocking some goroutines waiting to report their errors.
-	// errSignal lets us track when we should stop because an error occurred
-	errSignal *errSignaler
+	errChan        chan error
 
 	// internalStop is the stop channel *actually* used by everything involved
 	// with the manager as a stop channel, so that we can pass a stop channel
@@ -134,6 +133,9 @@ type controllerManager struct {
 	// It and `internalStop` should point to the same channel.
 	internalStopper chan<- struct{}
 
+	// stop procedure engaged. In other words, we should not add anything else to the manager
+	stopProcedureEngaged bool
+
 	// elected is closed when this manager becomes the leader of a group of
 	// managers, either because it won a leader election or because no leader
 	// election was configured.
@@ -161,57 +163,27 @@ type controllerManager struct {
 	// retryPeriod is the duration the LeaderElector clients should wait
 	// between tries of actions.
 	retryPeriod time.Duration
-}
 
-type errSignaler struct {
-	// errSignal indicates that an error occurred, when closed.  It shouldn't
-	// be written to.
-	errSignal chan struct{}
+	// waitForRunnable is holding the number of runnables currently running so that
+	// we can wait for them to exit before quitting the manager
+	waitForRunnable sync.WaitGroup
 
-	// err is the received error
-	err error
+	// gracefulShutdownTimeout is the duration given to runnable to stop
+	// before the manager actually returns on stop.
+	gracefulShutdownTimeout time.Duration
 
-	mu sync.Mutex
-}
-
-func (r *errSignaler) SignalError(err error) {
-	r.mu.Lock()
-	defer r.mu.Unlock()
-
-	if err == nil {
-		// non-error, ignore
-		log.Error(nil, "SignalError called without an (with a nil) error, which should never happen, ignoring")
-		return
-	}
-
-	if r.err != nil {
-		// we already have an error, don't try again
-		return
-	}
-
-	// save the error and report it
-	r.err = err
-	close(r.errSignal)
-}
-
-func (r *errSignaler) Error() error {
-	r.mu.Lock()
-	defer r.mu.Unlock()
-
-	return r.err
-}
-
-func (r *errSignaler) GotError() chan struct{} {
-	r.mu.Lock()
-	defer r.mu.Unlock()
-
-	return r.errSignal
+	// onStoppedLeading is callled when the leader election lease is lost.
+	// It can be overridden for tests.
+	onStoppedLeading func()
 }
 
 // Add sets dependencies on i, and adds it to the list of Runnables to start.
 func (cm *controllerManager) Add(r Runnable) error {
 	cm.mu.Lock()
 	defer cm.mu.Unlock()
+	if cm.stopProcedureEngaged {
+		return errors.New("can't accept new runnable as stop procedure is already engaged")
+	}
 
 	// Set dependencies on the object
 	if err := cm.SetFields(r); err != nil {
@@ -231,11 +203,7 @@ func (cm *controllerManager) Add(r Runnable) error {
 
 	if shouldStart {
 		// If already started, start the controller
-		go func() {
-			if err := r.Start(cm.internalStop); err != nil {
-				cm.errSignal.SignalError(err)
-			}
-		}()
+		cm.startRunnable(r)
 	}
 
 	return nil
@@ -293,6 +261,10 @@ func (cm *controllerManager) AddHealthzCheck(name string, check healthz.Checker)
 	cm.mu.Lock()
 	defer cm.mu.Unlock()
 
+	if cm.stopProcedureEngaged {
+		return errors.New("can't accept new healthCheck as stop procedure is already engaged")
+	}
+
 	if cm.healthzStarted {
 		return fmt.Errorf("unable to add new checker because healthz endpoint has already been created")
 	}
@@ -310,6 +282,10 @@ func (cm *controllerManager) AddReadyzCheck(name string, check healthz.Checker)
 	cm.mu.Lock()
 	defer cm.mu.Unlock()
 
+	if cm.stopProcedureEngaged {
+		return errors.New("can't accept new ready check as stop procedure is already engaged")
+	}
+
 	if cm.healthzStarted {
 		return fmt.Errorf("unable to add new checker because readyz endpoint has already been created")
 	}
@@ -389,17 +365,18 @@ func (cm *controllerManager) serveMetrics(stop <-chan struct{}) {
 		Handler: mux,
 	}
 	// Run the server
-	go func() {
+	cm.startRunnable(RunnableFunc(func(stop <-chan struct{}) error {
 		log.Info("starting metrics server", "path", defaultMetricsEndpoint)
 		if err := server.Serve(cm.metricsListener); err != nil && err != http.ErrServerClosed {
-			cm.errSignal.SignalError(err)
+			return err
 		}
-	}()
+		return nil
+	}))
 
 	// Shutdown the server when stop is closed
 	<-stop
 	if err := server.Shutdown(context.Background()); err != nil {
-		cm.errSignal.SignalError(err)
+		cm.errChan <- err
 	}
 }
 
@@ -420,27 +397,39 @@ func (cm *controllerManager) serveHealthProbes(stop <-chan struct{}) {
 		Handler: mux,
 	}
 	// Run server
-	go func() {
+	cm.startRunnable(RunnableFunc(func(stop <-chan struct{}) error {
 		if err := server.Serve(cm.healthProbeListener); err != nil && err != http.ErrServerClosed {
-			cm.errSignal.SignalError(err)
+			return err
 		}
-	}()
+		return nil
+	}))
 	cm.healthzStarted = true
 	cm.mu.Unlock()
 
 	// Shutdown the server when stop is closed
 	<-stop
 	if err := server.Shutdown(context.Background()); err != nil {
-		cm.errSignal.SignalError(err)
+		cm.errChan <- err
 	}
 }
 
-func (cm *controllerManager) Start(stop <-chan struct{}) error {
-	// join the passed-in stop channel as an upstream feeding into cm.internalStopper
-	defer close(cm.internalStopper)
+func (cm *controllerManager) Start(stop <-chan struct{}) (err error) {
+	// This chan indicates that stop is complete, in other words all runnables have returned or timeout on stop request
+	stopComplete := make(chan struct{})
+	defer close(stopComplete)
+	defer func() {
+		stopErr := cm.engageStopProcedure(stopComplete)
+		if stopErr != nil {
+			if err != nil {
+				err = fmt.Errorf("%w, afterwards waiting for graceful shtudown failed with err %v", err, stopErr)
+			} else {
+				err = stopErr
+			}
+		}
+	}()
 
 	// initialize this here so that we reset the signal channel state on every start
-	cm.errSignal = &errSignaler{errSignal: make(chan struct{})}
+	cm.errChan = make(chan error)
 
 	// Metrics should be served whether the controller is leader or not.
 	// (If we don't serve metrics for non-leaders, prometheus will still scrape
@@ -471,9 +460,51 @@ func (cm *controllerManager) Start(stop <-chan struct{}) error {
 	case <-stop:
 		// We are done
 		return nil
-	case <-cm.errSignal.GotError():
-		// Error starting a controller
-		return cm.errSignal.Error()
+	case err := <-cm.errChan:
+		// Error starting or running a runnable
+		return err
+	}
+}
+
+// engageStopProcedure signals all runnables to stop, reads potential errors
+// from the errChan and waits for them to end. It must not be called more than once.
+func (cm *controllerManager) engageStopProcedure(stopComplete chan struct{}) error {
+	close(cm.internalStopper)
+	cm.mu.Lock()
+	defer cm.mu.Unlock()
+	cm.stopProcedureEngaged = true
+	go func() {
+		for {
+			select {
+			case err, ok := <-cm.errChan:
+				if ok {
+					log.Error(err, "error received after stop sequence was engaged")
+				}
+			case <-stopComplete:
+				return
+			}
+		}
+	}()
+	return cm.waitForRunnableToEnd()
+}
+
+// waitForRunnableToEnd blocks until all runnables ended or the
+// tearDownTimeout was reached. In the latter case, an error is returned.
+func (cm *controllerManager) waitForRunnableToEnd() error {
+	gracefulShutdownTimer := time.NewTimer(cm.gracefulShutdownTimeout)
+	defer gracefulShutdownTimer.Stop()
+	allStopped := make(chan struct{})
+
+	go func() {
+		cm.waitForRunnable.Wait()
+		close(allStopped)
+	}()
+
+	select {
+	case <-allStopped:
+		return nil
+	case <-gracefulShutdownTimer.C:
+		return fmt.Errorf("not all runnables have stopped within the grace period of %s", cm.gracefulShutdownTimeout.String())
 	}
 }
 
@@ -487,15 +518,7 @@ func (cm *controllerManager) startNonLeaderElectionRunnables() {
 	for _, c := range cm.nonLeaderElectionRunnables {
 		// 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() {
-			if err := ctrl.Start(cm.internalStop); err != nil {
-				cm.errSignal.SignalError(err)
-			}
-			// we use %T here because we don't have a good stand-in for "name",
-			// and the full runnable might not serialize (mutexes, etc)
-			log.V(1).Info("non-leader-election runnable finished", "runnable type", fmt.Sprintf("%T", ctrl))
-		}()
+		cm.startRunnable(c)
 	}
 }
 
@@ -509,15 +532,7 @@ func (cm *controllerManager) startLeaderElectionRunnables() {
 	for _, c := 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
-		go func() {
-			if err := ctrl.Start(cm.internalStop); err != nil {
-				cm.errSignal.SignalError(err)
-			}
-			// we use %T here because we don't have a good stand-in for "name",
-			// and the full runnable might not serialize (mutexes, etc)
-			log.V(1).Info("leader-election runnable finished", "runnable type", fmt.Sprintf("%T", ctrl))
-		}()
+		cm.startRunnable(c)
 	}
 
 	cm.startedLeader = true
@@ -532,19 +547,30 @@ func (cm *controllerManager) waitForCache() {
 	if cm.startCache == nil {
 		cm.startCache = cm.cache.Start
 	}
-	go func() {
-		if err := cm.startCache(cm.internalStop); err != nil {
-			cm.errSignal.SignalError(err)
-		}
-	}()
+	cm.startRunnable(RunnableFunc(func(stop <-chan struct{}) error {
+		return cm.startCache(stop)
+	}))
 
 	// Wait for the caches to sync.
 	// TODO(community): Check the return value and write a test
 	cm.cache.WaitForCacheSync(cm.internalStop)
+	// TODO: This should be the return value of cm.cache.WaitForCacheSync but we abuse
+	// cm.started as check if we already started the cache so it must always become true.
+	// Making sure that the cache doesn't get started twice is needed to not get a "close
+	// of closed channel" panic
 	cm.started = true
 }
 
 func (cm *controllerManager) startLeaderElection() (err error) {
+	if cm.onStoppedLeading == nil {
+		cm.onStoppedLeading = func() {
+			// We have to exit here, otherwise the graceful shutdown or anything
+			// else that keeps the binary running might allow controllers that
+			// need leader election to run after we lost the lease.
+			log.Info("leader election lost, exiting")
+			os.Exit(1)
+		}
+	}
 	l, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
 		Lock:          cm.resourceLock,
 		LeaseDuration: cm.leaseDuration,
@@ -555,12 +581,7 @@ func (cm *controllerManager) startLeaderElection() (err error) {
 				close(cm.elected)
 				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.errSignal.SignalError(fmt.Errorf("leader election lost"))
-			},
+			OnStoppedLeading: cm.onStoppedLeading,
 		},
 	})
 	if err != nil {
@@ -584,3 +605,13 @@ func (cm *controllerManager) startLeaderElection() (err error) {
 func (cm *controllerManager) Elected() <-chan struct{} {
 	return cm.elected
 }
+
+func (cm *controllerManager) startRunnable(r Runnable) {
+	cm.waitForRunnable.Add(1)
+	go func() {
+		defer cm.waitForRunnable.Done()
+		if err := r.Start(cm.internalStop); err != nil {
+			cm.errChan <- err
+		}
+	}()
+}