Enable starting/stopping controllers and their caches

Author: kevindelgado

pkg/cache/cache.go

@@ -58,6 +58,9 @@ type Informers interface {
 	// of the underlying object.
 	GetInformerForKind(ctx context.Context, gvk schema.GroupVersionKind) (Informer, error)
 
+	// Remove the informer for the given object.
+	Remove(ctx context.Context, obj runtime.Object) error
+
 	// Start runs all the informers known to this cache until the context is closed.
 	// It blocks.
 	Start(ctx context.Context) error

pkg/cache/informer_cache.go

@@ -237,7 +237,7 @@ func indexByField(indexer Informer, field string, extractor client.IndexerFunc)
 }
 
 // Remove removes an informer specified by the obj argument from the cache and stops it if it existed.
-func (ip *informerCache) Remove(obj runtime.Object) error {
+func (ip *informerCache) Remove(ctx context.Context, obj runtime.Object) error {
 	gvk, err := apiutil.GVKForObject(obj, ip.Scheme)
 	if err != nil {
 		return err

pkg/cache/informertest/fake_cache.go

@@ -125,6 +125,11 @@ func (c *FakeInformers) Start(ctx context.Context) error {
 	return c.Error
 }
 
+// Remove implements Cache
+func (c *FakeInformers) Remove(ctx context.Context, obj runtime.Object) error {
+	return c.Error
+}
+
 // IndexField implements Cache
 func (c *FakeInformers) IndexField(ctx context.Context, obj client.Object, field string, extractValue client.IndexerFunc) error {
 	return nil

pkg/cache/multi_namespace_cache.go

@@ -94,6 +94,15 @@ func (c *multiNamespaceCache) GetInformerForKind(ctx context.Context, gvk schema
 	return &multiNamespaceInformer{namespaceToInformer: informers}, nil
 }
 
+func (c *multiNamespaceCache) Remove(ctx context.Context, obj runtime.Object) error {
+	for _, cache := range c.namespaceToCache {
+		if err := cache.Remove(ctx, obj); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (c *multiNamespaceCache) Start(ctx context.Context) error {
 	for ns, cache := range c.namespaceToCache {
 		go func(ns string, cache Cache) {

pkg/internal/controller/controller.go

@@ -79,6 +79,9 @@ type Controller struct {
 	// undergo a major refactoring and redesign to allow for context to not be stored in a struct.
 	ctx context.Context
 
+	// saveWatches indicates not to clear startWatches when the controller is stopped.
+	saveWatches bool
+
 	// startWatches maintains a list of sources, handlers, and predicates to start when the controller is started.
 	startWatches []watchDescription
 
@@ -180,9 +183,15 @@ func (c *Controller) Start(ctx context.Context) 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,
+		// We should not usually 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
+
+		// The exception to this is when the saveWatches is set to true,
+		// meaning the controller is intending to run Start() again.
+		// In this case it needs knowledge of the watches for when the controller is restarted.
+		if !c.saveWatches {
+			c.startWatches = nil
+		}
 
 		if c.JitterPeriod == 0 {
 			c.JitterPeriod = 1 * time.Second
@@ -302,3 +311,13 @@ func (c *Controller) InjectFunc(f inject.Func) error {
 func (c *Controller) updateMetrics(reconcileTime time.Duration) {
 	ctrlmetrics.ReconcileTime.WithLabelValues(c.Name).Observe(reconcileTime.Seconds())
 }
+
+// ResetStart sets Started to false to enable running Start on the controller again.
+func (c *Controller) ResetStart() {
+	c.Started = false
+}
+
+// SaveWatches indicates that watches should not be cleared when the controller is stopped.
+func (c *Controller) SaveWatches() {
+	c.saveWatches = true
+}