Merge pull request #941 from alvaroaleman/syncing-source

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

Author: k8s-ci-robot

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