Add support for additional default indexers in cache

On-behalf-of: SAP [email protected]
Co-authored-by: Vince Prignano <[email protected]>
Co-authored-by: Dr. Stefan Schimanski <[email protected]>
Signed-off-by: Marvin Beckers <[email protected]>

Author: 3 people

pkg/cache/cache.go

@@ -207,6 +207,10 @@ type Options struct {
 	// to reduce the caches memory usage.
 	DefaultTransform toolscache.TransformFunc
 
+	// AdditionalDefaultIndexes are indexes that are added to every informer
+	// beyond the namespace-name and namespace ones.
+	AdditionalDefaultIndexes client.Indexers
+
 	// DefaultWatchErrorHandler will be used to the WatchErrorHandler which is called
 	// whenever ListAndWatch drops the connection with an error.
 	//
@@ -239,8 +243,10 @@ type Options struct {
 	// If unset, this will fall through to the Default* settings.
 	ByObject map[client.Object]ByObject
 
-	// newInformer allows overriding of NewSharedIndexInformer for testing.
-	newInformer *func(toolscache.ListerWatcher, runtime.Object, time.Duration, toolscache.Indexers) toolscache.SharedIndexInformer
+	// NewInformer allows to override NewSharedIndexInformer.
+	// NOTE: LOW LEVEL PRIMITIVE!
+	// Only use a custom informer if you know what you are doing.
+	NewInformer func(toolscache.ListerWatcher, runtime.Object, time.Duration, toolscache.Indexers) toolscache.SharedIndexInformer
 }
 
 // ByObject offers more fine-grained control over the cache's ListWatch by object.
@@ -428,11 +434,12 @@ func newCache(restConfig *rest.Config, opts Options) newCacheFunc {
 					Label: config.LabelSelector,
 					Field: config.FieldSelector,
 				},
-				Transform:             config.Transform,
-				WatchErrorHandler:     opts.DefaultWatchErrorHandler,
-				UnsafeDisableDeepCopy: ptr.Deref(config.UnsafeDisableDeepCopy, false),
-				EnableWatchBookmarks:  ptr.Deref(config.EnableWatchBookmarks, true),
-				NewInformer:           opts.newInformer,
+				Transform:                config.Transform,
+				WatchErrorHandler:        opts.DefaultWatchErrorHandler,
+				UnsafeDisableDeepCopy:    ptr.Deref(config.UnsafeDisableDeepCopy, false),
+				EnableWatchBookmarks:     ptr.Deref(config.EnableWatchBookmarks, true),
+				NewInformer:              opts.NewInformer,
+				AdditionalDefaultIndexes: opts.AdditionalDefaultIndexes,
 			}),
 			readerFailOnMissingInformer: opts.ReaderFailOnMissingInformer,
 		}

pkg/cache/cache_test.go

@@ -544,14 +544,9 @@ func NonBlockingGetTest(createCacheFunc func(config *rest.Config, opts cache.Opt
 			Expect(err).NotTo(HaveOccurred())
 
 			By("creating the informer cache")
-			v := reflect.ValueOf(&opts).Elem()
-			newInformerField := v.FieldByName("newInformer")
-			newFakeInformer := func(_ kcache.ListerWatcher, _ runtime.Object, _ time.Duration, _ kcache.Indexers) kcache.SharedIndexInformer {
+			opts.NewInformer = func(_ kcache.ListerWatcher, _ runtime.Object, _ time.Duration, _ kcache.Indexers) kcache.SharedIndexInformer {
 				return &controllertest.FakeInformer{Synced: false}
 			}
-			reflect.NewAt(newInformerField.Type(), newInformerField.Addr().UnsafePointer()).
-				Elem().
-				Set(reflect.ValueOf(&newFakeInformer))
 			informerCache, err = createCacheFunc(cfg, opts)
 			Expect(err).NotTo(HaveOccurred())
 			By("running the cache and waiting for it to sync")

pkg/cache/informer_cache.go

@@ -21,7 +21,6 @@ import (
 	"fmt"
 	"strings"
 
-	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -221,40 +220,5 @@ func (ic *informerCache) IndexField(ctx context.Context, obj client.Object, fiel
 }
 
 func indexByField(informer Informer, field string, extractValue client.IndexerFunc) error {
-	indexFunc := func(objRaw interface{}) ([]string, error) {
-		// TODO(directxman12): check if this is the correct type?
-		obj, isObj := objRaw.(client.Object)
-		if !isObj {
-			return nil, fmt.Errorf("object of type %T is not an Object", objRaw)
-		}
-		meta, err := apimeta.Accessor(obj)
-		if err != nil {
-			return nil, err
-		}
-		ns := meta.GetNamespace()
-
-		rawVals := extractValue(obj)
-		var vals []string
-		if ns == "" {
-			// if we're not doubling the keys for the namespaced case, just create a new slice with same length
-			vals = make([]string, len(rawVals))
-		} else {
-			// if we need to add non-namespaced versions too, double the length
-			vals = make([]string, len(rawVals)*2)
-		}
-		for i, rawVal := range rawVals {
-			// save a namespaced variant, so that we can ask
-			// "what are all the object matching a given index *in a given namespace*"
-			vals[i] = internal.KeyToNamespacedKey(ns, rawVal)
-			if ns != "" {
-				// if we have a namespace, also inject a special index key for listing
-				// regardless of the object namespace
-				vals[i+len(rawVals)] = internal.KeyToNamespacedKey("", rawVal)
-			}
-		}
-
-		return vals, nil
-	}
-
-	return informer.AddIndexers(cache.Indexers{internal.FieldIndexName(field): indexFunc})
+	return informer.AddIndexers(cache.Indexers{internal.FieldIndexName(field): internal.IndexFunc(extractValue)})
 }

pkg/cache/internal/informers.go

@@ -36,30 +36,32 @@ import (
 	"k8s.io/client-go/metadata"
 	"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"
 	"sigs.k8s.io/controller-runtime/pkg/internal/syncs"
 )
 
 // InformersOpts configures an InformerMap.
 type InformersOpts struct {
-	HTTPClient            *http.Client
-	Scheme                *runtime.Scheme
-	Mapper                meta.RESTMapper
-	ResyncPeriod          time.Duration
-	Namespace             string
-	NewInformer           *func(cache.ListerWatcher, runtime.Object, time.Duration, cache.Indexers) cache.SharedIndexInformer
-	Selector              Selector
-	Transform             cache.TransformFunc
-	UnsafeDisableDeepCopy bool
-	EnableWatchBookmarks  bool
-	WatchErrorHandler     cache.WatchErrorHandler
+	HTTPClient               *http.Client
+	Scheme                   *runtime.Scheme
+	Mapper                   meta.RESTMapper
+	ResyncPeriod             time.Duration
+	Namespace                string
+	NewInformer              func(cache.ListerWatcher, runtime.Object, time.Duration, cache.Indexers) cache.SharedIndexInformer
+	AdditionalDefaultIndexes client.Indexers
+	Selector                 Selector
+	Transform                cache.TransformFunc
+	UnsafeDisableDeepCopy    bool
+	EnableWatchBookmarks     bool
+	WatchErrorHandler        cache.WatchErrorHandler
 }
 
 // NewInformers creates a new InformersMap that can create informers under the hood.
 func NewInformers(config *rest.Config, options *InformersOpts) *Informers {
 	newInformer := cache.NewSharedIndexInformer
 	if options.NewInformer != nil {
-		newInformer = *options.NewInformer
+		newInformer = options.NewInformer
 	}
 	return &Informers{
 		config:     config,
@@ -71,17 +73,18 @@ func NewInformers(config *rest.Config, options *InformersOpts) *Informers {
 			Unstructured: make(map[schema.GroupVersionKind]*Cache),
 			Metadata:     make(map[schema.GroupVersionKind]*Cache),
 		},
-		codecs:                serializer.NewCodecFactory(options.Scheme),
-		paramCodec:            runtime.NewParameterCodec(options.Scheme),
-		resync:                options.ResyncPeriod,
-		startWait:             make(chan struct{}),
-		namespace:             options.Namespace,
-		selector:              options.Selector,
-		transform:             options.Transform,
-		unsafeDisableDeepCopy: options.UnsafeDisableDeepCopy,
-		enableWatchBookmarks:  options.EnableWatchBookmarks,
-		newInformer:           newInformer,
-		watchErrorHandler:     options.WatchErrorHandler,
+		codecs:                   serializer.NewCodecFactory(options.Scheme),
+		paramCodec:               runtime.NewParameterCodec(options.Scheme),
+		resync:                   options.ResyncPeriod,
+		startWait:                make(chan struct{}),
+		namespace:                options.Namespace,
+		selector:                 options.Selector,
+		transform:                options.Transform,
+		additionalDefaultIndexes: options.AdditionalDefaultIndexes,
+		unsafeDisableDeepCopy:    options.UnsafeDisableDeepCopy,
+		enableWatchBookmarks:     options.EnableWatchBookmarks,
+		newInformer:              newInformer,
+		watchErrorHandler:        options.WatchErrorHandler,
 	}
 }
 
@@ -173,10 +176,11 @@ type Informers struct {
 	// default or empty string means all namespaces
 	namespace string
 
-	selector              Selector
-	transform             cache.TransformFunc
-	unsafeDisableDeepCopy bool
-	enableWatchBookmarks  bool
+	selector                 Selector
+	transform                cache.TransformFunc
+	additionalDefaultIndexes client.Indexers
+	unsafeDisableDeepCopy    bool
+	enableWatchBookmarks     bool
 
 	// NewInformer allows overriding of the shared index informer constructor for testing.
 	newInformer func(cache.ListerWatcher, runtime.Object, time.Duration, cache.Indexers) cache.SharedIndexInformer
@@ -358,6 +362,13 @@ func (ip *Informers) addInformerToMap(gvk schema.GroupVersionKind, obj runtime.O
 	if err != nil {
 		return nil, false, err
 	}
+
+	indexers := make(cache.Indexers, len(ip.additionalDefaultIndexes)+1)
+	for k, fn := range ip.additionalDefaultIndexes {
+		indexers[FieldIndexName(k)] = IndexFunc(fn)
+	}
+
+	indexers[cache.NamespaceIndex] = cache.MetaNamespaceIndexFunc
 	sharedIndexInformer := ip.newInformer(&cache.ListWatch{
 		ListFunc: func(opts metav1.ListOptions) (runtime.Object, error) {
 			ip.selector.ApplyToList(&opts)
@@ -370,10 +381,7 @@ func (ip *Informers) addInformerToMap(gvk schema.GroupVersionKind, obj runtime.O
 			ip.selector.ApplyToList(&opts)
 			return listWatcher.WatchFunc(opts)
 		},
-	}, obj, calculateResyncPeriod(ip.resync), cache.Indexers{
-		cache.NamespaceIndex: cache.MetaNamespaceIndexFunc,
-	})
-
+	}, obj, calculateResyncPeriod(ip.resync), indexers)
 	// Set WatchErrorHandler on SharedIndexInformer if set
 	if ip.watchErrorHandler != nil {
 		if err := sharedIndexInformer.SetWatchErrorHandler(ip.watchErrorHandler); err != nil {
@@ -608,3 +616,43 @@ func restrictNamespaceBySelector(namespaceOpt string, s Selector) string {
 	}
 	return ""
 }
+
+// IndexFunc constructs a low-level cache.IndexFunc from a client.IndexerFunc.
+// Returned keys in the former are namespaced and non-namespaced variants of the
+// latter.
+func IndexFunc(extractValue client.IndexerFunc) cache.IndexFunc {
+	return func(objRaw interface{}) ([]string, error) {
+		// TODO(directxman12): check if this is the correct type?
+		obj, isObj := objRaw.(client.Object)
+		if !isObj {
+			return nil, fmt.Errorf("object of type %T is not an Object", objRaw)
+		}
+		meta, err := meta.Accessor(obj)
+		if err != nil {
+			return nil, err
+		}
+		ns := meta.GetNamespace()
+
+		rawVals := extractValue(obj)
+		var vals []string
+		if ns == "" {
+			// if we're not doubling the keys for the namespaced case, just create a new slice with same length
+			vals = make([]string, len(rawVals))
+		} else {
+			// if we need to add non-namespaced versions too, double the length
+			vals = make([]string, len(rawVals)*2)
+		}
+		for i, rawVal := range rawVals {
+			// save a namespaced variant, so that we can ask
+			// "what are all the object matching a given index *in a given namespace*"
+			vals[i] = KeyToNamespacedKey(ns, rawVal)
+			if ns != "" {
+				// if we have a namespace, also inject a special index key for listing
+				// regardless of the object namespace
+				vals[i+len(rawVals)] = KeyToNamespacedKey("", rawVal)
+			}
+		}
+
+		return vals, nil
+	}
+}

pkg/client/interfaces.go

@@ -190,6 +190,9 @@ type WithWatch interface {
 // namespaced and non-spaced variants, so keys do not need to include namespace.
 type IndexerFunc func(Object) []string
 
+// Indexers is a map of field name to IndexerFunc.
+type Indexers map[string]IndexerFunc
+
 // FieldIndexer knows how to index over a particular "field" such that it
 // can later be used by a field selector.
 type FieldIndexer interface {