✨ Controller: Let sources sync even if they use a different cluster

Currently, the controller gets a cache injected and waits for that cache
to sync before starting workers. This model works well for single
cluster usecases, but doesn't work at all for multi-cluster usecases, as
the source may be using a completely different cache.

This PR adds a new source.SyncingSource interface and makes the
controller wait for all sources that implement it. Apart from solving
the issue mentioned in the title, this also allows to implement custom
sources that need syncing.

Author: alvaroaleman

pkg/controller/controller.go

@@ -103,7 +103,6 @@ func NewUnmanaged(name string, mgr manager.Manager, options Options) (Controller
 	// Create controller with dependencies set
 	c := &controller.Controller{
 		Do:       options.Reconciler,
-		Cache:    mgr.GetCache(),
 		Scheme:   mgr.GetScheme(),
 		Client:   mgr.GetClient(),
 		Recorder: mgr.GetEventRecorderFor(name),

pkg/internal/controller/controller.go

@@ -26,7 +26,6 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
-	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/internal/controller/metrics"
@@ -60,9 +59,6 @@ type Controller struct {
 	// Scheme is injected by the controllerManager when controllerManager.Start is called
 	Scheme *runtime.Scheme
 
-	// informers are injected by the controllerManager when controllerManager.Start is called
-	Cache cache.Cache
-
 	// MakeQueue constructs the queue for this controller once the controller is ready to start.
 	// This exists because the standard Kubernetes workqueues start themselves immediately, which
 	// leads to goroutine leaks if something calls controller.New repeatedly.
@@ -81,10 +77,6 @@ type Controller struct {
 	// JitterPeriod allows tests to reduce the JitterPeriod so they complete faster
 	JitterPeriod time.Duration
 
-	// WaitForCacheSync allows tests to mock out the WaitForCacheSync function to return an error
-	// defaults to Cache.WaitForCacheSync
-	WaitForCacheSync func(stopCh <-chan struct{}) bool
-
 	// Started is true if the Controller has been Started
 	Started bool
 
@@ -165,16 +157,18 @@ func (c *Controller) Start(stop <-chan struct{}) error {
 		// Start the SharedIndexInformer factories to begin populating the SharedIndexInformer caches
 		log.Info("Starting Controller", "controller", c.Name)
 
-		// Wait for the caches to be synced before starting workers
-		if c.WaitForCacheSync == nil {
-			c.WaitForCacheSync = c.Cache.WaitForCacheSync
-		}
-		if ok := c.WaitForCacheSync(stop); !ok {
-			// This code is unreachable right now since WaitForCacheSync will never return an error
-			// Leaving it here because that could happen in the future
-			err := fmt.Errorf("failed to wait for %s caches to sync", c.Name)
-			log.Error(err, "Could not wait for Cache to sync", "controller", c.Name)
-			return err
+		for _, watch := range c.watches {
+			syncingSource, ok := watch.src.(source.SyncingSource)
+			if !ok {
+				continue
+			}
+			if err := syncingSource.WaitForSync(stop); err != nil {
+				// This code is unreachable in case of kube watches since WaitForCacheSync will never return an error
+				// Leaving it here because that could happen in the future
+				err := fmt.Errorf("failed to wait for %s caches to sync: %w", c.Name, err)
+				log.Error(err, "Could not wait for Cache to sync", "controller", c.Name)
+				return err
+			}
 		}
 
 		if c.JitterPeriod == 0 {

pkg/internal/controller/controller_test.go

@@ -66,7 +66,6 @@ var _ = Describe("controller", func() {
 			MaxConcurrentReconciles: 1,
 			Do:                      fakeReconcile,
 			MakeQueue:               func() workqueue.RateLimitingInterface { return queue },
-			Cache:                   informers,
 		}
 		Expect(ctrl.InjectFunc(func(interface{}) error { return nil })).To(Succeed())
 	})
@@ -89,7 +88,10 @@ var _ = Describe("controller", func() {
 
 	Describe("Start", func() {
 		It("should return an error if there is an error waiting for the informers", func(done Done) {
-			ctrl.WaitForCacheSync = func(<-chan struct{}) bool { return false }
+			f := false
+			ctrl.watches = []watchDescription{{
+				src: source.NewKindWithCache(&corev1.Pod{}, &informertest.FakeInformers{Synced: &f}),
+			}}
 			ctrl.Name = "foo"
 			err := ctrl.Start(stop)
 			Expect(err).To(HaveOccurred())
@@ -111,8 +113,9 @@ var _ = Describe("controller", func() {
 			Expect(err).NotTo(HaveOccurred())
 			_, err = c.GetInformer(context.TODO(), &appsv1.ReplicaSet{})
 			Expect(err).NotTo(HaveOccurred())
-			ctrl.Cache = c
-			ctrl.WaitForCacheSync = func(<-chan struct{}) bool { return true }
+			ctrl.watches = []watchDescription{{
+				src: source.NewKindWithCache(&appsv1.Deployment{}, &informertest.FakeInformers{}),
+			}}
 
 			Expect(ctrl.Start(stopped)).NotTo(HaveOccurred())
 
@@ -162,7 +165,7 @@ var _ = Describe("controller", func() {
 	Describe("Watch", func() {
 		It("should inject dependencies into the Source", func() {
 			src := &source.Kind{Type: &corev1.Pod{}}
-			Expect(src.InjectCache(ctrl.Cache)).To(Succeed())
+			Expect(src.InjectCache(informers)).To(Succeed())
 			evthdl := &handler.EnqueueRequestForObject{}
 			found := false
 			ctrl.SetFields = func(i interface{}) error {
@@ -178,7 +181,7 @@ var _ = Describe("controller", func() {
 
 		It("should return an error if there is an error injecting into the Source", func() {
 			src := &source.Kind{Type: &corev1.Pod{}}
-			Expect(src.InjectCache(ctrl.Cache)).To(Succeed())
+			Expect(src.InjectCache(informers)).To(Succeed())
 			evthdl := &handler.EnqueueRequestForObject{}
 			expected := fmt.Errorf("expect fail source")
 			ctrl.SetFields = func(i interface{}) error {
@@ -193,7 +196,7 @@ var _ = Describe("controller", func() {
 
 		It("should inject dependencies into the EventHandler", func() {
 			src := &source.Kind{Type: &corev1.Pod{}}
-			Expect(src.InjectCache(ctrl.Cache)).To(Succeed())
+			Expect(src.InjectCache(informers)).To(Succeed())
 			evthdl := &handler.EnqueueRequestForObject{}
 			found := false
 			ctrl.SetFields = func(i interface{}) error {
@@ -231,7 +234,7 @@ var _ = Describe("controller", func() {
 
 		It("should inject dependencies into all of the Predicates", func() {
 			src := &source.Kind{Type: &corev1.Pod{}}
-			Expect(src.InjectCache(ctrl.Cache)).To(Succeed())
+			Expect(src.InjectCache(informers)).To(Succeed())
 			evthdl := &handler.EnqueueRequestForObject{}
 			pr1 := &predicate.Funcs{}
 			pr2 := &predicate.Funcs{}
@@ -254,7 +257,7 @@ var _ = Describe("controller", func() {
 
 		It("should return an error if there is an error injecting into any of the Predicates", func() {
 			src := &source.Kind{Type: &corev1.Pod{}}
-			Expect(src.InjectCache(ctrl.Cache)).To(Succeed())
+			Expect(src.InjectCache(informers)).To(Succeed())
 			evthdl := &handler.EnqueueRequestForObject{}
 			pr1 := &predicate.Funcs{}
 			pr2 := &predicate.Funcs{}

pkg/source/source.go

@@ -18,6 +18,7 @@ package source
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 
@@ -56,10 +57,17 @@ type Source interface {
 	Start(handler.EventHandler, workqueue.RateLimitingInterface, ...predicate.Predicate) error
 }
 
+// SyncingSource is a source that needs syncing prior to being usable. The controller
+// will call its WaitForSync prior to starting workers.
+type SyncingSource interface {
+	Source
+	WaitForSync(stop <-chan struct{}) error
+}
+
 // NewKindWithCache creates a Source without InjectCache, so that it is assured that the given cache is used
 // and not overwritten. It can be used to watch objects in a different cluster by passing the cache
 // from that other cluster
-func NewKindWithCache(object runtime.Object, cache cache.Cache) Source {
+func NewKindWithCache(object runtime.Object, cache cache.Cache) SyncingSource {
 	return &kindWithCache{kind: Kind{Type: object, cache: cache}}
 }
 
@@ -72,6 +80,10 @@ func (ks *kindWithCache) Start(handler handler.EventHandler, queue workqueue.Rat
 	return ks.kind.Start(handler, queue, prct...)
 }
 
+func (ks *kindWithCache) WaitForSync(stop <-chan struct{}) error {
+	return ks.kind.WaitForSync(stop)
+}
+
 // Kind is used to provide a source of events originating inside the cluster from Watches (e.g. Pod Create)
 type Kind struct {
 	// Type is the type of object to watch.  e.g. &v1.Pod{}
@@ -81,7 +93,7 @@ type Kind struct {
 	cache cache.Cache
 }
 
-var _ Source = &Kind{}
+var _ SyncingSource = &Kind{}
 
 // Start is internal and should be called only by the Controller to register an EventHandler with the Informer
 // to enqueue reconcile.Requests.
@@ -118,6 +130,16 @@ func (ks *Kind) String() string {
 	return fmt.Sprintf("kind source: unknown GVK")
 }
 
+// WaitForSync implements SyncingSource to allow controllers to wait with starting
+// workers until the cache is synced.
+func (ks *Kind) WaitForSync(stop <-chan struct{}) error {
+	if !ks.cache.WaitForCacheSync(stop) {
+		// Would be great to return something more informative here
+		return errors.New("cache did not sync")
+	}
+	return nil
+}
+
 var _ inject.Cache = &Kind{}
 
 // InjectCache is internal should be called only by the Controller.  InjectCache is used to inject
@@ -283,6 +305,8 @@ func (is *Informer) String() string {
 	return fmt.Sprintf("informer source: %p", is.Informer)
 }
 
+var _ Source = Func(nil)
+
 // Func is a function that implements Source
 type Func func(handler.EventHandler, workqueue.RateLimitingInterface, ...predicate.Predicate) error
 

pkg/source/source_test.go

@@ -210,6 +210,18 @@ var _ = Describe("Source", func() {
 			close(done)
 		})
 
+		It("should return an error if syncing fails", func(done Done) {
+			instance := source.Kind{}
+			f := false
+			Expect(instance.InjectCache(&informertest.FakeInformers{Synced: &f})).To(Succeed())
+			err := instance.WaitForSync(nil)
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(Equal("cache did not sync"))
+
+			close(done)
+
+		})
+
 		Context("for a Kind not in the cache", func() {
 			It("should return an error when Start is called", func(done Done) {
 				ic.Error = fmt.Errorf("test error")
@@ -227,6 +239,26 @@ var _ = Describe("Source", func() {
 		})
 	})
 
+	Describe("KindWithCache", func() {
+		It("should not allow injecting a cache", func() {
+			instance := source.NewKindWithCache(nil, nil)
+			injected, err := inject.CacheInto(&informertest.FakeInformers{}, instance)
+			Expect(err).To(BeNil())
+			Expect(injected).To(BeFalse())
+		})
+
+		It("should return an error if syncing fails", func(done Done) {
+			f := false
+			instance := source.NewKindWithCache(nil, &informertest.FakeInformers{Synced: &f})
+			err := instance.WaitForSync(nil)
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(Equal("cache did not sync"))
+
+			close(done)
+
+		})
+	})
+
 	Describe("Func", func() {
 		It("should be called from Start", func(done Done) {
 			run := false