Merge pull request #1163 from vincepri/watches-controller-bug

🐛 Controller.Watch() should not store watches if already started

pkg/internal/controller/controller.go

@@ -74,8 +74,8 @@ type Controller struct {
 
 	// TODO(community): Consider initializing a logger with the Controller Name as the tag
 
-	// watches maintains a list of sources, handlers, and predicates to start when the controller is started.
-	watches []watchDescription
+	// startWatches maintains a list of sources, handlers, and predicates to start when the controller is started.
+	startWatches []watchDescription
 
 	// Log is used to log messages to users during reconciliation, or for example when a watch is started.
 	Log logr.Logger
@@ -113,13 +113,16 @@ func (c *Controller) Watch(src source.Source, evthdler handler.EventHandler, prc
 		}
 	}
 
-	c.watches = append(c.watches, watchDescription{src: src, handler: evthdler, predicates: prct})
-	if c.Started {
-		c.Log.Info("Starting EventSource", "source", src)
-		return src.Start(evthdler, c.Queue, prct...)
+	// Controller hasn't started yet, store the watches locally and return.
+	//
+	// These watches are going to be held on the controller struct until the manager or user calls Start(...).
+	if !c.Started {
+		c.startWatches = append(c.startWatches, watchDescription{src: src, handler: evthdler, predicates: prct})
+		return nil
 	}
 
-	return nil
+	c.Log.Info("Starting EventSource", "source", src)
+	return src.Start(evthdler, c.Queue, prct...)
 }
 
 // Start implements controller.Controller
@@ -143,7 +146,7 @@ func (c *Controller) Start(stop <-chan struct{}) error {
 		// NB(directxman12): launch the sources *before* trying to wait for the
 		// caches to sync so that they have a chance to register their intendeded
 		// caches.
-		for _, watch := range c.watches {
+		for _, watch := range c.startWatches {
 			c.Log.Info("Starting EventSource", "source", watch.src)
 			if err := watch.src.Start(watch.handler, c.Queue, watch.predicates...); err != nil {
 				return err
@@ -153,7 +156,7 @@ func (c *Controller) Start(stop <-chan struct{}) error {
 		// Start the SharedIndexInformer factories to begin populating the SharedIndexInformer caches
 		c.Log.Info("Starting Controller")
 
-		for _, watch := range c.watches {
+		for _, watch := range c.startWatches {
 			syncingSource, ok := watch.src.(source.SyncingSource)
 			if !ok {
 				continue
@@ -167,6 +170,12 @@ func (c *Controller) Start(stop <-chan struct{}) error {
 			}
 		}
 
+		// All the watches have been started, we can reset the local slice.
+		//
+		// We should never hold watches more than necessary, each watch source can hold a backing cache,
+		// which won't be garbage collected if we hold a reference to it.
+		c.startWatches = nil
+
 		if c.JitterPeriod == 0 {
 			c.JitterPeriod = 1 * time.Second
 		}

pkg/internal/controller/controller_test.go

@@ -91,7 +91,7 @@ var _ = Describe("controller", func() {
 	Describe("Start", func() {
 		It("should return an error if there is an error waiting for the informers", func(done Done) {
 			f := false
-			ctrl.watches = []watchDescription{{
+			ctrl.startWatches = []watchDescription{{
 				src: source.NewKindWithCache(&corev1.Pod{}, &informertest.FakeInformers{Synced: &f}),
 			}}
 			ctrl.Name = "foo"
@@ -115,7 +115,7 @@ var _ = Describe("controller", func() {
 			Expect(err).NotTo(HaveOccurred())
 			_, err = c.GetInformer(context.TODO(), &appsv1.ReplicaSet{})
 			Expect(err).NotTo(HaveOccurred())
-			ctrl.watches = []watchDescription{{
+			ctrl.startWatches = []watchDescription{{
 				src: source.NewKindWithCache(&appsv1.Deployment{}, &informertest.FakeInformers{}),
 			}}