Initialize queue in warmup with test.

Author: godwinpang

pkg/internal/controller/controller.go

@@ -128,10 +128,10 @@ type Controller[request comparable] struct {
 	// startWatches maintains a list of sources, handlers, and predicates to start when the controller is started.
 	startWatches []source.TypedSource[request]
 
-	// startedEventSources is used to track if the event sources have been started.
+	// startedEventSourcesAndQueue is used to track if the event sources have been started.
 	// It ensures that we append sources to c.startWatches only until we call Start() / Warmup()
-	// It is true if startEventSourcesLocked has been called at least once.
-	startedEventSources bool
+	// It is true if startEventSourcesAndQueueLocked has been called at least once.
+	startedEventSourcesAndQueue bool
 
 	// didStartEventSourcesOnce is used to ensure that the event sources are only started once.
 	didStartEventSourcesOnce sync.Once
@@ -208,7 +208,7 @@ func (c *Controller[request]) Watch(src source.TypedSource[request]) error {
 
 	// Sources weren't started yet, store the watches locally and return.
 	// These sources are going to be held until either Warmup() or Start(...) is called.
-	if !c.startedEventSources {
+	if !c.startedEventSourcesAndQueue {
 		c.startWatches = append(c.startWatches, src)
 		return nil
 	}
@@ -237,7 +237,7 @@ func (c *Controller[request]) Warmup(ctx context.Context) error {
 	// Set the ctx so later calls to watch use this internal context
 	c.ctx = ctx
 
-	return c.startEventSourcesLocked(ctx)
+	return c.startEventSourcesAndQueueLocked(ctx)
 }
 
 // Start implements controller.Controller.
@@ -254,17 +254,6 @@ func (c *Controller[request]) Start(ctx context.Context) error {
 	// Set the internal context.
 	c.ctx = ctx
 
-	queue := c.NewQueue(c.Name, c.RateLimiter)
-	if priorityQueue, isPriorityQueue := queue.(priorityqueue.PriorityQueue[request]); isPriorityQueue {
-		c.Queue = priorityQueue
-	} else {
-		c.Queue = &priorityQueueWrapper[request]{TypedRateLimitingInterface: queue}
-	}
-	go func() {
-		<-ctx.Done()
-		c.Queue.ShutDown()
-	}()
-
 	wg := &sync.WaitGroup{}
 	err := func() error {
 		defer c.mu.Unlock()
@@ -275,7 +264,7 @@ func (c *Controller[request]) Start(ctx context.Context) error {
 		// NB(directxman12): launch the sources *before* trying to wait for the
 		// caches to sync so that they have a chance to register their intended
 		// caches.
-		if err := c.startEventSourcesLocked(ctx); err != nil {
+		if err := c.startEventSourcesAndQueueLocked(ctx); err != nil {
 			return err
 		}
 
@@ -308,12 +297,23 @@ func (c *Controller[request]) Start(ctx context.Context) error {
 	return nil
 }
 
-// startEventSourcesLocked launches all the sources registered with this controller and waits
+// startEventSourcesAndQueueLocked launches all the sources registered with this controller and waits
 // for them to sync. It returns an error if any of the sources fail to start or sync.
-func (c *Controller[request]) startEventSourcesLocked(ctx context.Context) error {
+func (c *Controller[request]) startEventSourcesAndQueueLocked(ctx context.Context) error {
 	var retErr error
 
 	c.didStartEventSourcesOnce.Do(func() {
+        queue := c.NewQueue(c.Name, c.RateLimiter)
+        if priorityQueue, isPriorityQueue := queue.(priorityqueue.PriorityQueue[request]); isPriorityQueue {
+            c.Queue = priorityQueue
+        } else {
+            c.Queue = &priorityQueueWrapper[request]{TypedRateLimitingInterface: queue}
+        }
+        go func() {
+            <-ctx.Done()
+            c.Queue.ShutDown()
+        }()
+
 		errGroup := &errgroup.Group{}
 		for _, watch := range c.startWatches {
 			log := c.LogConstructor(nil)
@@ -381,7 +381,7 @@ func (c *Controller[request]) startEventSourcesLocked(ctx context.Context) error
 
 		// Mark event sources as started after resetting the startWatches slice to no-op a Watch()
 		// call after event sources have been started.
-		c.startedEventSources = true
+		c.startedEventSourcesAndQueue = true
 	})
 
 	return retErr

pkg/internal/controller/controller_test.go

@@ -387,16 +387,26 @@ var _ = Describe("controller", func() {
 		})
 	})
 
-	Describe("startEventSourcesLocked", func() {
+	Describe("startEventSourcesAndQueueLocked", func() {
 		It("should return nil when no sources are provided", func() {
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
 
 			ctrl.startWatches = []source.TypedSource[reconcile.Request]{}
-			err := ctrl.startEventSourcesLocked(ctx)
+			err := ctrl.startEventSourcesAndQueueLocked(ctx)
 			Expect(err).NotTo(HaveOccurred())
 		})
 
+		It("should initialize controller queue when called", func() {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			ctrl.startWatches = []source.TypedSource[reconcile.Request]{}
+			err := ctrl.startEventSourcesAndQueueLocked(ctx)
+			Expect(err).NotTo(HaveOccurred())
+            Expect(ctrl.Queue).NotTo(BeNil())
+		})
+
 		It("should return an error if a source fails to start", func() {
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
@@ -410,7 +420,7 @@ var _ = Describe("controller", func() {
 			// Set a sufficiently long timeout to avoid timeouts interfering with the error being returned
 			ctrl.CacheSyncTimeout = 5 * time.Second
 			ctrl.startWatches = []source.TypedSource[reconcile.Request]{src}
-			err := ctrl.startEventSourcesLocked(ctx)
+			err := ctrl.startEventSourcesAndQueueLocked(ctx)
 			Expect(err).To(Equal(expectedErr))
 		})
 
@@ -424,7 +434,7 @@ var _ = Describe("controller", func() {
 			ctrl.Name = "test-controller"
 			ctrl.CacheSyncTimeout = 5 * time.Second
 
-			err := ctrl.startEventSourcesLocked(ctx)
+			err := ctrl.startEventSourcesAndQueueLocked(ctx)
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("failed to wait for test-controller caches to sync"))
 		})
@@ -440,7 +450,7 @@ var _ = Describe("controller", func() {
 			ctrl.Name = "test-controller"
 			ctrl.CacheSyncTimeout = 5 * time.Second
 
-			err := ctrl.startEventSourcesLocked(ctx)
+			err := ctrl.startEventSourcesAndQueueLocked(ctx)
 			Expect(err).NotTo(HaveOccurred())
 		})
 
@@ -464,7 +474,7 @@ var _ = Describe("controller", func() {
 			startErrCh := make(chan error)
 			go func() {
 				defer GinkgoRecover()
-				startErrCh <- ctrl.startEventSourcesLocked(sourceCtx)
+				startErrCh <- ctrl.startEventSourcesAndQueueLocked(sourceCtx)
 			}()
 
 			// Allow source to start successfully
@@ -499,7 +509,7 @@ var _ = Describe("controller", func() {
 
 			ctrl.startWatches = []source.TypedSource[reconcile.Request]{blockingSrc}
 
-			err := ctrl.startEventSourcesLocked(ctx)
+			err := ctrl.startEventSourcesAndQueueLocked(ctx)
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("timed out waiting for source"))
 		})
@@ -518,13 +528,13 @@ var _ = Describe("controller", func() {
 
 			ctrl.startWatches = []source.TypedSource[reconcile.Request]{src}
 
-			By("Calling startEventSourcesLocked multiple times in parallel")
+			By("Calling startEventSourcesAndQueueLocked multiple times in parallel")
 			var wg sync.WaitGroup
 			for i := 1; i <= 5; i++ {
 				wg.Add(1)
 				go func() {
 					defer wg.Done()
-					err := ctrl.startEventSourcesLocked(ctx)
+					err := ctrl.startEventSourcesAndQueueLocked(ctx)
 					// All calls should return the same nil error
 					Expect(err).NotTo(HaveOccurred())
 				}()
@@ -534,12 +544,12 @@ var _ = Describe("controller", func() {
 			Expect(startCount.Load()).To(Equal(int32(1)), "Source should only be started once even when called multiple times")
 		})
 
-		It("should block subsequent calls from returning until the first call to startEventSourcesLocked has returned", func() {
+		It("should block subsequent calls from returning until the first call to startEventSourcesAndQueueLocked has returned", func() {
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
 			ctrl.CacheSyncTimeout = 5 * time.Second
 
-			// finishSourceChan is closed to unblock startEventSourcesLocked from returning
+			// finishSourceChan is closed to unblock startEventSourcesAndQueueLocked from returning
 			finishSourceChan := make(chan struct{})
 
 			src := source.Func(func(ctx context.Context, _ workqueue.TypedRateLimitingInterface[reconcile.Request]) error {
@@ -548,35 +558,35 @@ var _ = Describe("controller", func() {
 			})
 			ctrl.startWatches = []source.TypedSource[reconcile.Request]{src}
 
-			By("Calling startEventSourcesLocked asynchronously")
+			By("Calling startEventSourcesAndQueueLocked asynchronously")
 			wg := sync.WaitGroup{}
 			go func() {
 				defer GinkgoRecover()
 				defer wg.Done()
 
 				wg.Add(1)
-				Expect(ctrl.startEventSourcesLocked(ctx)).To(Succeed())
+				Expect(ctrl.startEventSourcesAndQueueLocked(ctx)).To(Succeed())
 			}()
 
-			By("Calling startEventSourcesLocked again")
+			By("Calling startEventSourcesAndQueueLocked again")
 			var didSubsequentCallComplete atomic.Bool
 			go func() {
 				defer GinkgoRecover()
 				defer wg.Done()
 
 				wg.Add(1)
-				Expect(ctrl.startEventSourcesLocked(ctx)).To(Succeed())
+				Expect(ctrl.startEventSourcesAndQueueLocked(ctx)).To(Succeed())
 				didSubsequentCallComplete.Store(true)
 			}()
 
-			// Assert that second call to startEventSourcesLocked is blocked while source has not finished
+			// Assert that second call to startEventSourcesAndQueueLocked is blocked while source has not finished
 			Consistently(didSubsequentCallComplete.Load).Should(BeFalse())
 
 			By("Finishing source start + sync")
 			finishSourceChan <- struct{}{}
 
-			// Assert that second call to startEventSourcesLocked is now complete
-			Eventually(didSubsequentCallComplete.Load).Should(BeTrue(), "startEventSourcesLocked should complete after source is started and synced")
+			// Assert that second call to startEventSourcesAndQueueLocked is now complete
+			Eventually(didSubsequentCallComplete.Load).Should(BeTrue(), "startEventSourcesAndQueueLocked should complete after source is started and synced")
 			wg.Wait()
 		})
 
@@ -592,7 +602,7 @@ var _ = Describe("controller", func() {
 
 			ctrl.startWatches = []source.TypedSource[reconcile.Request]{src}
 
-			err := ctrl.startEventSourcesLocked(ctx)
+			err := ctrl.startEventSourcesAndQueueLocked(ctx)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(ctrl.startWatches).To(BeNil(), "startWatches should be reset to nil after returning")
 		})
@@ -1150,6 +1160,25 @@ var _ = Describe("controller", func() {
 			Expect(err.Error()).To(ContainSubstring("sync error"))
 		})
 
+		It("should call Start on sources with the appropriate non-nil queue", func() {
+			ctrl.CacheSyncTimeout = 10 * time.Second
+			started := false
+			ctx, cancel := context.WithCancel(context.Background())
+			src := source.Func(func(ctx context.Context, q workqueue.TypedRateLimitingInterface[reconcile.Request]) error {
+				defer GinkgoRecover()
+                Expect(q).ToNot(BeNil())
+				Expect(q).To(Equal(ctrl.Queue))
+
+				started = true
+				cancel() // Cancel the context so ctrl.Start() doesn't block forever
+				return nil
+			})
+			Expect(ctrl.Watch(src)).To(Succeed())
+			Expect(ctrl.Warmup(ctx)).To(Succeed())
+            Expect(ctrl.Queue).ToNot(BeNil())
+			Expect(started).To(BeTrue())
+		})
+
 		It("should return true if context is cancelled while waiting for source to start", func() {
 			// Setup controller with sources that complete with error
 			ctx, cancel := context.WithCancel(context.Background())