UPSTREAM: <carry>: support custom informer creation + indexers

Co-authored-by: Fabian von Feilitzsch <[email protected]>

Author: ncdc

pkg/cache/cache.go

@@ -26,6 +26,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/cache"
 	toolscache "k8s.io/client-go/tools/cache"
 	"sigs.k8s.io/controller-runtime/pkg/cache/internal"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -108,6 +109,14 @@ type Options struct {
 	// So that all informers will not send list requests simultaneously.
 	Resync *time.Duration
 
+	// NewInformerFunc is a function that is used to create SharedIndexInformers.
+	// Defaults to cache.NewSharedIndexInformer from client-go
+	NewInformerFunc client.NewInformerFunc
+
+	// Indexers is the indexers that the informers will be configured to use.
+	// Will always have the standard NamespaceIndex.
+	Indexers cache.Indexers
+
 	// Namespace restricts the cache's ListWatch to the desired namespace
 	// Default watches all namespaces
 	Namespace string
@@ -163,7 +172,7 @@ func New(config *rest.Config, opts Options) (Cache, error) {
 		return nil, err
 	}
 
-	im := internal.NewInformersMap(config, opts.Scheme, opts.Mapper, *opts.Resync, opts.Namespace, selectorsByGVK, disableDeepCopyByGVK, transformByGVK)
+	im := internal.NewInformersMap(config, opts.Scheme, opts.Mapper, *opts.Resync, opts.Namespace, selectorsByGVK, disableDeepCopyByGVK, transformByGVK, opts.NewInformerFunc, opts.Indexers)
 	return &informerCache{InformersMap: im}, nil
 }
 
@@ -216,6 +225,10 @@ func defaultOpts(config *rest.Config, opts Options) (Options, error) {
 	if opts.Resync == nil {
 		opts.Resync = &defaultResyncTime
 	}
+
+	if opts.NewInformerFunc == nil {
+		opts.NewInformerFunc = cache.NewSharedIndexInformer
+	}
 	return opts, nil
 }
 

pkg/cache/internal/deleg_map.go

@@ -27,6 +27,8 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 // InformersMap create and caches Informers for (runtime.Object, schema.GroupVersionKind) pairs.
@@ -53,11 +55,13 @@ func NewInformersMap(config *rest.Config,
 	selectors SelectorsByGVK,
 	disableDeepCopy DisableDeepCopyByGVK,
 	transformers TransformFuncByObject,
+	newInformerFunc client.NewInformerFunc,
+	indexers cache.Indexers,
 ) *InformersMap {
 	return &InformersMap{
-		structured:   newStructuredInformersMap(config, scheme, mapper, resync, namespace, selectors, disableDeepCopy, transformers),
-		unstructured: newUnstructuredInformersMap(config, scheme, mapper, resync, namespace, selectors, disableDeepCopy, transformers),
-		metadata:     newMetadataInformersMap(config, scheme, mapper, resync, namespace, selectors, disableDeepCopy, transformers),
+		structured:   newStructuredInformersMap(config, scheme, mapper, resync, namespace, selectors, disableDeepCopy, transformers, newInformerFunc, indexers),
+		unstructured: newUnstructuredInformersMap(config, scheme, mapper, resync, namespace, selectors, disableDeepCopy, transformers, newInformerFunc, indexers),
+		metadata:     newMetadataInformersMap(config, scheme, mapper, resync, namespace, selectors, disableDeepCopy, transformers, newInformerFunc, indexers),
 
 		Scheme: scheme,
 	}
@@ -109,18 +113,18 @@ func (m *InformersMap) Get(ctx context.Context, gvk schema.GroupVersionKind, obj
 
 // newStructuredInformersMap creates a new InformersMap for structured objects.
 func newStructuredInformersMap(config *rest.Config, scheme *runtime.Scheme, mapper meta.RESTMapper, resync time.Duration,
-	namespace string, selectors SelectorsByGVK, disableDeepCopy DisableDeepCopyByGVK, transformers TransformFuncByObject) *specificInformersMap {
-	return newSpecificInformersMap(config, scheme, mapper, resync, namespace, selectors, disableDeepCopy, transformers, createStructuredListWatch)
+	namespace string, selectors SelectorsByGVK, disableDeepCopy DisableDeepCopyByGVK, transformers TransformFuncByObject, newInformerFunc client.NewInformerFunc, indexers cache.Indexers) *specificInformersMap {
+	return newSpecificInformersMap(config, scheme, mapper, resync, namespace, selectors, disableDeepCopy, transformers, createStructuredListWatch, newInformerFunc, indexers)
 }
 
 // newUnstructuredInformersMap creates a new InformersMap for unstructured objects.
 func newUnstructuredInformersMap(config *rest.Config, scheme *runtime.Scheme, mapper meta.RESTMapper, resync time.Duration,
-	namespace string, selectors SelectorsByGVK, disableDeepCopy DisableDeepCopyByGVK, transformers TransformFuncByObject) *specificInformersMap {
-	return newSpecificInformersMap(config, scheme, mapper, resync, namespace, selectors, disableDeepCopy, transformers, createUnstructuredListWatch)
+	namespace string, selectors SelectorsByGVK, disableDeepCopy DisableDeepCopyByGVK, transformers TransformFuncByObject, newInformerFunc client.NewInformerFunc, indexers cache.Indexers) *specificInformersMap {
+	return newSpecificInformersMap(config, scheme, mapper, resync, namespace, selectors, disableDeepCopy, transformers, createUnstructuredListWatch, newInformerFunc, indexers)
 }
 
 // newMetadataInformersMap creates a new InformersMap for metadata-only objects.
 func newMetadataInformersMap(config *rest.Config, scheme *runtime.Scheme, mapper meta.RESTMapper, resync time.Duration,
-	namespace string, selectors SelectorsByGVK, disableDeepCopy DisableDeepCopyByGVK, transformers TransformFuncByObject) *specificInformersMap {
-	return newSpecificInformersMap(config, scheme, mapper, resync, namespace, selectors, disableDeepCopy, transformers, createMetadataListWatch)
+	namespace string, selectors SelectorsByGVK, disableDeepCopy DisableDeepCopyByGVK, transformers TransformFuncByObject, newInformerFunc client.NewInformerFunc, indexers cache.Indexers) *specificInformersMap {
+	return newSpecificInformersMap(config, scheme, mapper, resync, namespace, selectors, disableDeepCopy, transformers, createMetadataListWatch, newInformerFunc, indexers)
 }

pkg/cache/internal/informers_map.go

@@ -35,6 +35,7 @@ import (
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
 
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
 )
 
@@ -56,21 +57,25 @@ func newSpecificInformersMap(config *rest.Config,
 	disableDeepCopy DisableDeepCopyByGVK,
 	transformers TransformFuncByObject,
 	createListWatcher createListWatcherFunc,
+	newInformerFunc client.NewInformerFunc,
+	indexers cache.Indexers,
 ) *specificInformersMap {
 	ip := &specificInformersMap{
-		config:            config,
-		Scheme:            scheme,
-		mapper:            mapper,
-		informersByGVK:    make(map[schema.GroupVersionKind]*MapEntry),
-		codecs:            serializer.NewCodecFactory(scheme),
-		paramCodec:        runtime.NewParameterCodec(scheme),
-		resync:            resync,
-		startWait:         make(chan struct{}),
-		createListWatcher: createListWatcher,
-		namespace:         namespace,
-		selectors:         selectors.forGVK,
-		disableDeepCopy:   disableDeepCopy,
-		transformers:      transformers,
+		config:             config,
+		Scheme:             scheme,
+		mapper:             mapper,
+		informersByGVK:     make(map[schema.GroupVersionKind]*MapEntry),
+		codecs:             serializer.NewCodecFactory(scheme),
+		paramCodec:         runtime.NewParameterCodec(scheme),
+		resync:             resync,
+		startWait:          make(chan struct{}),
+		createListWatcher:  createListWatcher,
+		namespace:          namespace,
+		selectors:          selectors.forGVK,
+		disableDeepCopy:    disableDeepCopy,
+		transformers:       transformers,
+		newInformerFunc:    newInformerFunc,
+		additionalIndexers: indexers,
 	}
 	return ip
 }
@@ -141,6 +146,15 @@ type specificInformersMap struct {
 
 	// transform funcs are applied to objects before they are committed to the cache
 	transformers TransformFuncByObject
+
+	// keyFunction is the cache.KeyFunc informers will be configured to use
+	keyFunction cache.KeyFunc
+
+	// additionalIndexers is the indexers that the informers will be configured to use.
+	// Will not allow overwriting the standard NamespaceIndex.
+	additionalIndexers cache.Indexers
+
+	newInformerFunc client.NewInformerFunc
 }
 
 // Start calls Run on each of the informers and sets started to true.  Blocks on the context.
@@ -230,9 +244,13 @@ func (ip *specificInformersMap) addInformerToMap(gvk schema.GroupVersionKind, ob
 	if err != nil {
 		return nil, false, err
 	}
-	ni := cache.NewSharedIndexInformer(lw, obj, resyncPeriod(ip.resync)(), cache.Indexers{
-		cache.NamespaceIndex: cache.MetaNamespaceIndexFunc,
-	})
+	indexers := cache.Indexers{}
+	for indexName, indexer := range ip.additionalIndexers {
+		indexers[indexName] = indexer
+	}
+	indexers[cache.NamespaceIndex] = cache.MetaNamespaceIndexFunc
+
+	ni := ip.newInformerFunc(lw, obj, resyncPeriod(ip.resync)(), indexers)
 
 	// Check to see if there is a transformer for this gvk
 	if err := ni.SetTransform(ip.transformers.Get(gvk)); err != nil {

pkg/client/interfaces.go

@@ -18,8 +18,10 @@ package client
 
 import (
 	"context"
+	"time"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/client-go/tools/cache"
 
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -43,6 +45,10 @@ type Patch interface {
 	Data(obj Object) ([]byte, error)
 }
 
+// NewInformerFunc describes a function that creates SharedIndexInformers.
+// Its signature matches cache.NewSharedIndexInformer from client-go
+type NewInformerFunc func(cache.ListerWatcher, runtime.Object, time.Duration, cache.Indexers) cache.SharedIndexInformer
+
 // TODO(directxman12): is there a sane way to deal with get/delete options?
 
 // Reader knows how to read and list Kubernetes objects.