WIP adds Context support to manager in place of stop channels.

example/main.go

@@ -17,6 +17,7 @@ limitations under the License.
 package main
 
 import (
+	"context"
 	"flag"
 	"os"
 
@@ -128,7 +129,13 @@ func main() {
 		os.Exit(1)
 	}
 
-	if err := mgr.Start(signals.SetupSignalHandler()); err != nil {
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		<-signals.SetupSignalHandler()
+		cancel()
+	}()
+
+	if err := mgr.Start(ctx); err != nil {
 		entryLog.Error(err, "unable to run manager")
 		os.Exit(1)
 	}

pkg/builder/build_test.go

@@ -37,9 +37,11 @@ import (
 
 var _ = Describe("application", func() {
 	var stop chan struct{}
+	var ctx context.Context
 
 	BeforeEach(func() {
 		stop = make(chan struct{})
+		ctx = context.Background()
 		getConfig = func() (*rest.Config, error) { return cfg, nil }
 		newController = controller.New
 		newManager = manager.New
@@ -138,7 +140,7 @@ var _ = Describe("application", func() {
 			By("Starting the application")
 			go func() {
 				defer GinkgoRecover()
-				Expect(instance.Start(stop)).NotTo(HaveOccurred())
+				Expect(instance.Start(ctx)).NotTo(HaveOccurred())
 				By("Stopping the application")
 			}()
 

pkg/builder/example_test.go

@@ -27,7 +27,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
-	"sigs.k8s.io/controller-runtime/pkg/runtime/signals"
 )
 
 // NB: don't call SetLogger in init(), or else you'll mess up logging in the main suite.
@@ -49,7 +48,7 @@ func ExampleBuilder() {
 		os.Exit(1)
 	}
 
-	if err := rs.Start(signals.SetupSignalHandler()); err != nil {
+	if err := rs.Start(context.TODO()); err != nil {
 		log.Error(err, "Unable to run controller")
 		os.Exit(1)
 	}

pkg/controller/controller_integration_test.go

@@ -17,6 +17,8 @@ limitations under the License.
 package controller_test
 
 import (
+	"context"
+
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -35,9 +37,11 @@ import (
 var _ = Describe("controller", func() {
 	var reconciled chan reconcile.Request
 	var stop chan struct{}
+	var ctx context.Context
 
 	BeforeEach(func() {
 		stop = make(chan struct{})
+		ctx = context.Background()
 		reconciled = make(chan reconcile.Request)
 		Expect(cfg).NotTo(BeNil())
 	})
@@ -76,7 +80,7 @@ var _ = Describe("controller", func() {
 			By("Starting the Manager")
 			go func() {
 				defer GinkgoRecover()
-				Expect(cm.Start(stop)).NotTo(HaveOccurred())
+				Expect(cm.Start(ctx)).NotTo(HaveOccurred())
 			}()
 
 			deployment := &appsv1.Deployment{

pkg/controller/example_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package controller_test
 
 import (
+	"context"
 	"os"
 
 	"k8s.io/api/core/v1"
@@ -25,7 +26,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
-	"sigs.k8s.io/controller-runtime/pkg/runtime/signals"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 )
 
@@ -75,5 +75,5 @@ func ExampleController() {
 	}
 
 	// Start the Controller through the manager.
-	mgr.Start(signals.SetupSignalHandler())
+	mgr.Start(context.TODO())
 }

pkg/internal/recorder/recorder_integration_test.go

@@ -17,6 +17,8 @@ limitations under the License.
 package recorder_test
 
 import (
+	"context"
+
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -35,9 +37,11 @@ import (
 
 var _ = Describe("recorder", func() {
 	var stop chan struct{}
+	var ctx context.Context
 
 	BeforeEach(func() {
 		stop = make(chan struct{})
+		ctx = context.Background()
 		Expect(cfg).NotTo(BeNil())
 	})
 
@@ -71,7 +75,7 @@ var _ = Describe("recorder", func() {
 			By("Starting the Manager")
 			go func() {
 				defer GinkgoRecover()
-				Expect(cm.Start(stop)).NotTo(HaveOccurred())
+				Expect(cm.Start(ctx)).NotTo(HaveOccurred())
 			}()
 
 			deployment := &appsv1.Deployment{

pkg/manager/example_test.go

@@ -17,12 +17,12 @@ limitations under the License.
 package manager_test
 
 import (
+	"context"
 	"os"
 
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
-	"sigs.k8s.io/controller-runtime/pkg/runtime/signals"
 )
 
 var (
@@ -61,7 +61,7 @@ func ExampleManager_Add() {
 
 // This example starts a Manager that has had Runnables added.
 func ExampleManager_Start() {
-	err := mgr.Start(signals.SetupSignalHandler())
+	err := mgr.Start(context.TODO())
 	if err != nil {
 		log.Error(err, "unable start the manager")
 		os.Exit(1)

pkg/manager/internal.go

@@ -17,6 +17,7 @@ limitations under the License.
 package manager
 
 import (
+	"context"
 	"fmt"
 	"sync"
 	"time"
@@ -73,7 +74,7 @@ type controllerManager struct {
 	mu      sync.Mutex
 	started bool
 	errChan chan error
-	stop    <-chan struct{}
+	stop    chan struct{}
 
 	startCache func(stop <-chan struct{}) error
 }
@@ -157,11 +158,15 @@ func (cm *controllerManager) GetRESTMapper() meta.RESTMapper {
 	return cm.mapper
 }
 
-func (cm *controllerManager) Start(stop <-chan struct{}) error {
+func (cm *controllerManager) Start(ctx context.Context) error {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	defer close(cm.stop)
+
 	if cm.resourceLock == nil {
-		go cm.start(stop)
+		go cm.start(ctx)
 		select {
-		case <-stop:
+		case <-ctx.Done():
 			// we are done
 			return nil
 		case err := <-cm.errChan:
@@ -178,7 +183,16 @@ func (cm *controllerManager) Start(stop <-chan struct{}) error {
 		RenewDeadline: 10 * time.Second,
 		RetryPeriod:   2 * time.Second,
 		Callbacks: leaderelection.LeaderCallbacks{
-			OnStartedLeading: cm.start,
+			// Ignore the passed-in stop channel from leaderelection. The next
+			// thing it does anyway after closing its stop channel is call
+			// OnStoppedLeading.
+			// With the 1.12 release, this will accept a Context, which is a
+			// descendant of the same context that will be passed into
+			// `l.Run(ctx)` below. Then it will be safe to pass the context
+			// through to cm.start here.
+			OnStartedLeading: func(_ <-chan struct{}) {
+				cm.start(ctx)
+			},
 			OnStoppedLeading: func() {
 				// Most implementations of leader election log.Fatal() here.
 				// Since Start is wrapped in log.Fatal when called, we can just return
@@ -194,7 +208,7 @@ func (cm *controllerManager) Start(stop <-chan struct{}) error {
 	go l.Run()
 
 	select {
-	case <-stop:
+	case <-ctx.Done():
 		// We are done
 		return nil
 	case err := <-cm.errChan:
@@ -203,43 +217,37 @@ func (cm *controllerManager) Start(stop <-chan struct{}) error {
 	}
 }
 
-func (cm *controllerManager) start(stop <-chan struct{}) {
+func (cm *controllerManager) start(ctx context.Context) {
 	func() {
 		cm.mu.Lock()
 		defer cm.mu.Unlock()
 
-		cm.stop = stop
-
 		// Start the Cache. Allow the function to start the cache to be mocked out for testing
 		if cm.startCache == nil {
 			cm.startCache = cm.cache.Start
 		}
 		go func() {
-			if err := cm.startCache(stop); err != nil {
+			if err := cm.startCache(ctx.Done()); err != nil {
 				cm.errChan <- err
 			}
 		}()
 
 		// Wait for the caches to sync.
 		// TODO(community): Check the return value and write a test
-		cm.cache.WaitForCacheSync(stop)
+		cm.cache.WaitForCacheSync(ctx.Done())
 
 		// 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.
 			// Write any Start errors to a channel so we can return them
 			ctrl := c
 			go func() {
-				cm.errChan <- ctrl.Start(stop)
+				// Because a controller can be added and started at any time, we always
+				// give them cm.stop.
+				cm.errChan <- ctrl.Start(cm.stop)
 			}()
 		}
 
 		cm.started = true
 	}()
-
-	select {
-	case <-stop:
-		// We are done
-		return
-	}
 }

pkg/manager/manager.go

@@ -17,6 +17,7 @@ limitations under the License.
 package manager
 
 import (
+	"context"
 	"fmt"
 	"time"
 
@@ -52,7 +53,7 @@ type Manager interface {
 
 	// Start starts all registered Controllers and blocks until the Stop channel is closed.
 	// Returns an error if there is an error starting any controller.
-	Start(<-chan struct{}) error
+	Start(context.Context) error
 
 	// GetConfig returns an initialized Config
 	GetConfig() *rest.Config
@@ -204,6 +205,7 @@ func New(config *rest.Config, options Options) (Manager, error) {
 		recorderProvider: recorderProvider,
 		resourceLock:     resourceLock,
 		mapper:           mapper,
+		stop:             make(chan struct{}),
 	}, nil
 }
 

pkg/manager/manager_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package manager
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/go-logr/logr"
@@ -38,9 +39,11 @@ import (
 
 var _ = Describe("manger.Manager", func() {
 	var stop chan struct{}
+	var ctx context.Context
 
 	BeforeEach(func() {
 		stop = make(chan struct{})
+		ctx = context.Background()
 	})
 
 	AfterEach(func() {
@@ -146,7 +149,7 @@ var _ = Describe("manger.Manager", func() {
 
 				go func() {
 					defer GinkgoRecover()
-					Expect(m.Start(stop)).NotTo(HaveOccurred())
+					Expect(m.Start(ctx)).NotTo(HaveOccurred())
 				}()
 				<-c1
 				<-c2
@@ -157,9 +160,9 @@ var _ = Describe("manger.Manager", func() {
 			It("should stop when stop is called", func(done Done) {
 				m, err := New(cfg, options)
 				Expect(err).NotTo(HaveOccurred())
-				s := make(chan struct{})
-				close(s)
-				Expect(m.Start(s)).NotTo(HaveOccurred())
+				ctx, cancel := context.WithCancel(ctx)
+				cancel()
+				Expect(m.Start(ctx)).NotTo(HaveOccurred())
 
 				close(done)
 			})
@@ -172,7 +175,7 @@ var _ = Describe("manger.Manager", func() {
 				mgr.startCache = func(stop <-chan struct{}) error {
 					return fmt.Errorf("expected error")
 				}
-				Expect(m.Start(stop).Error()).To(ContainSubstring("expected error"))
+				Expect(m.Start(ctx).Error()).To(ContainSubstring("expected error"))
 
 				close(done)
 			})
@@ -203,7 +206,7 @@ var _ = Describe("manger.Manager", func() {
 
 				go func() {
 					defer GinkgoRecover()
-					Expect(m.Start(stop)).NotTo(HaveOccurred())
+					Expect(m.Start(ctx)).NotTo(HaveOccurred())
 					close(done)
 				}()
 				<-c1
@@ -244,7 +247,7 @@ var _ = Describe("manger.Manager", func() {
 
 				go func() {
 					defer GinkgoRecover()
-					Expect(m.Start(stop)).NotTo(HaveOccurred())
+					Expect(m.Start(ctx)).NotTo(HaveOccurred())
 				}()
 
 				// Wait for the Manager to start
@@ -271,7 +274,7 @@ var _ = Describe("manger.Manager", func() {
 
 			go func() {
 				defer GinkgoRecover()
-				Expect(m.Start(stop)).NotTo(HaveOccurred())
+				Expect(m.Start(ctx)).NotTo(HaveOccurred())
 			}()
 
 			// Wait for the Manager to start
@@ -327,8 +330,7 @@ var _ = Describe("manger.Manager", func() {
 				},
 				stop: func(stop <-chan struct{}) error {
 					defer GinkgoRecover()
-					// Manager stop chan has not been initialized.
-					Expect(stop).To(BeNil())
+					Expect(stop).NotTo(BeNil())
 					return nil
 				},
 				f: func(f inject.Func) error {