[WIP] 🌱 Controller Lifecycle Management

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
@@ -84,6 +87,14 @@ type Informer interface {
 	AddIndexers(indexers toolscache.Indexers) error
 	//HasSynced return true if the informers underlying store has synced
 	HasSynced() bool
+
+	// RemoveEventHandler currently just decrements a the count of event handlers
+	// The goals it to have SharedInformer support RemoveEventHandler (and actually remove
+	// the handler instead of just decrementing a count).
+	RemoveEventHandler(id int) error
+
+	// CountEventHandlers returns the number of event handlers added to an informer.
+	CountEventHandlers() int
 }
 
 // Options are the optional arguments for creating a new InformersMap object
@@ -126,7 +137,7 @@ func defaultOpts(config *rest.Config, opts Options) (Options, error) {
 	// Construct a new Mapper if unset
 	if opts.Mapper == nil {
 		var err error
-		opts.Mapper, err = apiutil.NewDiscoveryRESTMapper(config)
+		opts.Mapper, err = apiutil.NewDynamicRESTMapper(config)
 		if err != nil {
 			log.WithName("setup").Error(err, "Failed to get API Group-Resources")
 			return opts, fmt.Errorf("could not create RESTMapper from config")

pkg/cache/informer_cache.go

@@ -22,6 +22,7 @@ import (
 	"reflect"
 	"strings"
 
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -159,6 +160,24 @@ func (ip *informerCache) GetInformer(ctx context.Context, obj client.Object) (In
 	return i.Informer, err
 }
 
+// GetInformerNonBlocking returns the informer for the obj without waiting for its cache to sync.
+func (ip *informerCache) GetInformerNonBlocking(obj runtime.Object) (Informer, error) {
+	gvk, err := apiutil.GVKForObject(obj, ip.Scheme)
+	if err != nil {
+		return nil, err
+	}
+
+	// Use a cancelled context to signal non-blocking
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	_, i, err := ip.InformersMap.Get(ctx, gvk, obj)
+	if err != nil && !apierrors.IsTimeout(err) {
+		return nil, err
+	}
+	return i.Informer, nil
+}
+
 // NeedLeaderElection implements the LeaderElectionRunnable interface
 // to indicate that this can be started without requiring the leader lock
 func (ip *informerCache) NeedLeaderElection() bool {
@@ -216,3 +235,13 @@ func indexByField(indexer Informer, field string, extractor client.IndexerFunc)
 
 	return indexer.AddIndexers(cache.Indexers{internal.FieldIndexName(field): indexFunc})
 }
+
+// Remove removes an informer specified by the obj argument from the cache and stops it if it existed.
+func (ip *informerCache) Remove(ctx context.Context, obj runtime.Object) error {
+	gvk, err := apiutil.GVKForObject(obj, ip.Scheme)
+	if err != nil {
+		return err
+	}
+
+	return ip.InformersMap.Remove(gvk, obj)
+}

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/internal/counting_informer.go

@@ -0,0 +1,86 @@
+package internal
+
+import (
+	"fmt"
+	"time"
+
+	"k8s.io/client-go/tools/cache"
+)
+
+// CountingInformer exposes a way to track the number of EventHandlers
+// registered on an Informer.
+type CountingInformer interface {
+	cache.SharedIndexInformer
+	CountEventHandlers() int
+	RemoveEventHandler(id int) error
+}
+
+// HandlerCountingInformer implements the CountingInformer.
+// It increments the count every time AddEventHandler is called,
+// and decrements the count every time RemoveEventHandler is called.
+//
+// It doesn't actually RemoveEventHandlers because that feature is not implemented
+// in client-go, but we're are naming it this way to suggest what the interface would look
+// like if/when it does get added to client-go.
+//
+// We can get rid of this if apimachinery adds the ability to retrieve this from the SharedIndexInformer
+// but until then, we have to track it ourselves
+type HandlerCountingInformer struct {
+	// Informer is the cached informer
+	informer cache.SharedIndexInformer
+
+	// count indicates the number of EventHandlers registered on the informer
+	count int
+}
+
+func (i *HandlerCountingInformer) RemoveEventHandler(id int) error {
+	i.count--
+	fmt.Printf("decrement, count is %+v\n", i.count)
+	return nil
+}
+
+func (i *HandlerCountingInformer) AddEventHandler(handler cache.ResourceEventHandler) {
+	i.count++
+	fmt.Printf("increment, count is %+v\n", i.count)
+	i.informer.AddEventHandler(handler)
+}
+
+func (i *HandlerCountingInformer) CountEventHandlers() int {
+	return i.count
+}
+
+func (i *HandlerCountingInformer) AddEventHandlerWithResyncPeriod(handler cache.ResourceEventHandler, resyncPeriod time.Duration) {
+	i.count++
+	i.informer.AddEventHandlerWithResyncPeriod(handler, resyncPeriod)
+}
+func (i *HandlerCountingInformer) AddIndexers(indexers cache.Indexers) error {
+	return i.informer.AddIndexers(indexers)
+}
+
+func (i *HandlerCountingInformer) HasSynced() bool {
+	return i.informer.HasSynced()
+}
+
+func (i *HandlerCountingInformer) GetStore() cache.Store {
+	return i.informer.GetStore()
+}
+
+func (i *HandlerCountingInformer) GetController() cache.Controller {
+	return i.informer.GetController()
+}
+
+func (i *HandlerCountingInformer) LastSyncResourceVersion() string {
+	return i.informer.LastSyncResourceVersion()
+}
+
+func (i *HandlerCountingInformer) SetWatchErrorHandler(handler cache.WatchErrorHandler) error {
+	return i.informer.SetWatchErrorHandler(handler)
+}
+
+func (i *HandlerCountingInformer) GetIndexer() cache.Indexer {
+	return i.informer.GetIndexer()
+}
+
+func (i *HandlerCountingInformer) Run(stopCh <-chan struct{}) {
+	i.informer.Run(stopCh)
+}

pkg/cache/internal/deleg_map.go

@@ -92,6 +92,20 @@ func (m *InformersMap) Get(ctx context.Context, gvk schema.GroupVersionKind, obj
 	return m.structured.Get(ctx, gvk, obj)
 }
 
+// Remove will remove an new Informer from the InformersMap and stop it if it exists.
+func (m *InformersMap) Remove(gvk schema.GroupVersionKind, obj runtime.Object) error {
+	_, isUnstructured := obj.(*unstructured.Unstructured)
+	_, isUnstructuredList := obj.(*unstructured.UnstructuredList)
+	isUnstructured = isUnstructured || isUnstructuredList
+
+	switch {
+	case isUnstructured:
+		return m.unstructured.Remove(gvk)
+	default:
+		return m.structured.Remove(gvk)
+	}
+}
+
 // newStructuredInformersMap creates a new InformersMap for structured objects.
 func newStructuredInformersMap(config *rest.Config, scheme *runtime.Scheme, mapper meta.RESTMapper, resync time.Duration, namespace string) *specificInformersMap {
 	return newSpecificInformersMap(config, scheme, mapper, resync, namespace, createStructuredListWatch)

pkg/cache/internal/informers_map.go

@@ -65,11 +65,14 @@ func newSpecificInformersMap(config *rest.Config,
 
 // MapEntry contains the cached data for an Informer
 type MapEntry struct {
-	// Informer is the cached informer
-	Informer cache.SharedIndexInformer
+	// Informer is a SharedIndexInformer with addition count and remove event handler functionality.
+	Informer CountingInformer
 
 	// CacheReader wraps Informer and implements the CacheReader interface for a single type
 	Reader CacheReader
+
+	// Stop can be used to stop this individual informer without stopping the entire specificInformersMap.
+	stop chan struct{}
 }
 
 // specificInformersMap create and caches Informers for (runtime.Object, schema.GroupVersionKind) pairs.
@@ -121,6 +124,17 @@ type specificInformersMap struct {
 	namespace string
 }
 
+// Start starts the informer managed by a MapEntry.
+// Blocks until the informer stops. The informer can be stopped
+// either individually (via the entry's stop channel) or globally
+// via the provided stop argument.
+func (e *MapEntry) Start(stop <-chan struct{}) {
+	// Stop on either the whole map stopping or just this informer being removed.
+	internalStop, cancel := anyOf(stop, e.stop)
+	defer cancel()
+	e.Informer.Run(internalStop)
+}
+
 // Start calls Run on each of the informers and sets started to true.  Blocks on the context.
 // It doesn't return start because it can't return an error, and it's not a runnable directly.
 func (ip *specificInformersMap) Start(ctx context.Context) {
@@ -132,8 +146,8 @@ func (ip *specificInformersMap) Start(ctx context.Context) {
 		ip.stop = ctx.Done()
 
 		// Start each informer
-		for _, informer := range ip.informersByGVK {
-			go informer.Informer.Run(ctx.Done())
+		for _, entry := range ip.informersByGVK {
+			go entry.Start(ctx.Done())
 		}
 
 		// Set started to true so we immediately start any informers added later.
@@ -183,8 +197,12 @@ func (ip *specificInformersMap) Get(ctx context.Context, gvk schema.GroupVersion
 
 	if started && !i.Informer.HasSynced() {
 		// Wait for it to sync before returning the Informer so that folks don't read from a stale cache.
-		if !cache.WaitForCacheSync(ctx.Done(), i.Informer.HasSynced) {
-			return started, nil, apierrors.NewTimeoutError(fmt.Sprintf("failed waiting for %T Informer to sync", obj), 0)
+		// Cancel for context, informer stopping, or entire map stopping.
+		syncStop, cancel := mergeChan(ctx.Done(), i.stop, ip.stop)
+		defer cancel()
+		if !cache.WaitForCacheSync(syncStop, i.Informer.HasSynced) {
+			// Return entry even on timeout - caller may have intended a non-blocking fetch.
+			return started, i, apierrors.NewTimeoutError(fmt.Sprintf("failed waiting for %T Informer to sync", obj), 0)
 		}
 	}
 
@@ -212,20 +230,47 @@ func (ip *specificInformersMap) addInformerToMap(gvk schema.GroupVersionKind, ob
 		cache.NamespaceIndex: cache.MetaNamespaceIndexFunc,
 	})
 	i := &MapEntry{
-		Informer: ni,
+		Informer: &HandlerCountingInformer{ni, 0},
 		Reader:   CacheReader{indexer: ni.GetIndexer(), groupVersionKind: gvk},
+		stop:     make(chan struct{}),
 	}
 	ip.informersByGVK[gvk] = i
 
 	// Start the Informer if need by
 	// TODO(seans): write thorough tests and document what happens here - can you add indexers?
 	// can you add eventhandlers?
 	if ip.started {
-		go i.Informer.Run(ip.stop)
+		go i.Start(ip.stop)
+		//go i.Start(StopOptions{
+		//	StopChannel: ip.stop,
+		//})
 	}
 	return i, ip.started, nil
 }
 
+// Remove removes an informer entry and stops it if it was running.
+func (ip *specificInformersMap) Remove(gvk schema.GroupVersionKind) error {
+	ip.mu.Lock()
+	defer ip.mu.Unlock()
+
+	entry, ok := ip.informersByGVK[gvk]
+	if !ok {
+		return nil
+	}
+
+	chInformer, ok := entry.Informer.(*HandlerCountingInformer)
+	if !ok {
+		return fmt.Errorf("entry informer is not a HandlerCountingInformer")
+	}
+	if chInformer.CountEventHandlers() != 0 {
+		return fmt.Errorf("attempting to remove informer with %d references", chInformer.CountEventHandlers())
+	}
+
+	close(entry.stop)
+	delete(ip.informersByGVK, gvk)
+	return nil
+}
+
 // newListWatch returns a new ListWatch object that can be used to create a SharedIndexInformer.
 func createStructuredListWatch(gvk schema.GroupVersionKind, ip *specificInformersMap) (*cache.ListWatch, error) {
 	// Kubernetes APIs work against Resources, not GroupVersionKinds.  Map the

pkg/cache/internal/internal_suite_test.go

@@ -0,0 +1,14 @@
+package internal
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"sigs.k8s.io/controller-runtime/pkg/envtest/printer"
+)
+
+func TestCacheInternal(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecsWithDefaultAndCustomReporters(t, "Cache Internal Suite", []Reporter{printer.NewlineReporter{}})
+}