Merge pull request #3087 from tareksha/typed_waitforsync

[release-0.19] 🐛fix(controller): support WaitForSync in custom TypedSyncingSource

Author: k8s-ci-robot

pkg/internal/controller/controller.go

@@ -183,7 +183,7 @@ func (c *Controller[request]) Start(ctx context.Context) error {
 		c.LogConstructor(nil).Info("Starting Controller")
 
 		for _, watch := range c.startWatches {
-			syncingSource, ok := watch.(source.SyncingSource)
+			syncingSource, ok := watch.(source.TypedSyncingSource[request])
 			if !ok {
 				continue
 			}

pkg/internal/controller/controller_test.go

@@ -46,6 +46,10 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/source"
 )
 
+type TestRequest struct {
+	Key string
+}
+
 var _ = Describe("controller", func() {
 	var fakeReconcile *fakeReconciler
 	var ctrl *Controller[reconcile.Request]
@@ -323,6 +327,41 @@ var _ = Describe("controller", func() {
 			Expect(err.Error()).To(Equal("controller was started more than once. This is likely to be caused by being added to a manager multiple times"))
 		})
 
+		It("should check for correct TypedSyncingSource if custom types are used", func() {
+			queue := &controllertest.TypedQueue[TestRequest]{
+				TypedInterface: workqueue.NewTyped[TestRequest](),
+			}
+			ctrl := &Controller[TestRequest]{
+				NewQueue: func(string, workqueue.TypedRateLimiter[TestRequest]) workqueue.TypedRateLimitingInterface[TestRequest] {
+					return queue
+				},
+				LogConstructor: func(*TestRequest) logr.Logger {
+					return log.RuntimeLog.WithName("controller").WithName("test")
+				},
+			}
+			ctrl.CacheSyncTimeout = time.Second
+			src := &bisignallingSource[TestRequest]{
+				startCall: make(chan workqueue.TypedRateLimitingInterface[TestRequest]),
+				startDone: make(chan error, 1),
+				waitCall:  make(chan struct{}),
+				waitDone:  make(chan error, 1),
+			}
+			ctrl.startWatches = []source.TypedSource[TestRequest]{src}
+			ctrl.Name = "foo"
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			startCh := make(chan error)
+			go func() {
+				defer GinkgoRecover()
+				startCh <- ctrl.Start(ctx)
+			}()
+			Eventually(src.startCall).Should(Receive(Equal(queue)))
+			src.startDone <- nil
+			Eventually(src.waitCall).Should(BeClosed())
+			src.waitDone <- nil
+			cancel()
+			Eventually(startCh).Should(Receive(Succeed()))
+		})
 	})
 
 	Describe("Processing queue items from a Controller", func() {
@@ -875,3 +914,40 @@ func (c *cacheWithIndefinitelyBlockingGetInformer) GetInformer(ctx context.Conte
 	<-ctx.Done()
 	return nil, errors.New("GetInformer timed out")
 }
+
+type bisignallingSource[T comparable] struct {
+	// receives the queue that is passed to Start
+	startCall chan workqueue.TypedRateLimitingInterface[T]
+	// passes an error to return from Start
+	startDone chan error
+	// closed when WaitForSync is called
+	waitCall chan struct{}
+	// passes an error to return from WaitForSync
+	waitDone chan error
+}
+
+var _ source.TypedSyncingSource[int] = (*bisignallingSource[int])(nil)
+
+func (t *bisignallingSource[T]) Start(ctx context.Context, q workqueue.TypedRateLimitingInterface[T]) error {
+	select {
+	case t.startCall <- q:
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+	select {
+	case err := <-t.startDone:
+		return err
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+func (t *bisignallingSource[T]) WaitForSync(ctx context.Context) error {
+	close(t.waitCall)
+	select {
+	case err := <-t.waitDone:
+		return err
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}