✨ Allow configuring more granular cache filtering

This change implements most of the [cache options design][0], in
particular it is now possible to:
* Configure Namespaces per type
* Configure filtering per namespace

What is still missing is to allow configuring namespace-level default
settings that will be used for all namespaces not explicitly configured.
There is nothing in the way of doing that as a follow-up.

The implementation slightly derivates from the design document in that
the filter settings in `ByGVK` do not use the `Config` struct but
instead are directly embedded. This allows us to not break compatibility
and is more ergonomic to use.

The implementation is based on a new and internal per-type delegating
"meta cache".

[0]: https://github.com/kubernetes-sigs/controller-runtime/blob/main/designs/cache_options.md

Author: alvaroaleman

go.mod

@@ -15,6 +15,7 @@ require (
 	github.com/prometheus/client_model v0.4.0
 	go.uber.org/goleak v1.2.1
 	go.uber.org/zap v1.24.0
+	golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e
 	golang.org/x/sys v0.10.0
 	gomodules.xyz/jsonpatch/v2 v2.3.0
 	k8s.io/api v0.28.0-alpha.3

go.sum

@@ -128,6 +128,8 @@ go.uber.org/zap v1.24.0/go.mod h1:2kMP+WWQ8aoFoedH3T2sq6iJ2yDWpHbP0f6MQbS9Gkg=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e h1:+WEEuIdZHnUeJJmEUjyYC2gfUMj69yZXw17EnHg/otA=
+golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e/go.mod h1:Kr81I6Kryrl9sr8s2FK3vxD90NdsKWRuOIl2O4CvYbA=
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=

pkg/cache/cache.go

@@ -22,8 +22,8 @@ import (
 	"net/http"
 	"time"
 
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -144,9 +144,13 @@ type Options struct {
 	// instead of `reconcile.Result{}`.
 	SyncPeriod *time.Duration
 
-	// Namespaces restricts the cache's ListWatch to the desired namespaces
-	// Per default ListWatch watches all namespaces
-	Namespaces []string
+	// DefaultNamespaces maps namespace names to cache settings. If set, only
+	// the namespaces in here will be watched and it will by used to default
+	// ByObject.Namespaces for all objects if that is nil.
+	//
+	// The options in the Config that are nil will be defaulted from
+	// the respective Default* settings.
+	DefaultNamespaces map[string]Config
 
 	// DefaultLabelSelector will be used as a label selector for all object types
 	// unless they have a more specific selector set in ByObject.
@@ -160,20 +164,40 @@ type Options struct {
 	// unless they have a more specific transform set in ByObject.
 	DefaultTransform toolscache.TransformFunc
 
-	// UnsafeDisableDeepCopy indicates not to deep copy objects during get or
-	// list objects for EVERY object.
+	// DefaultUnsafeDisableDeepCopy is the default for UnsafeDisableDeepCopy
+	// for everything that doesn't specify this.
+	//
 	// Be very careful with this, when enabled you must DeepCopy any object before mutating it,
 	// otherwise you will mutate the object in the cache.
 	//
 	// This is a global setting for all objects, and can be overridden by the ByObject setting.
-	UnsafeDisableDeepCopy *bool
+	DefaultUnsafeDisableDeepCopy *bool
 
 	// ByObject restricts the cache's ListWatch to the desired fields per GVK at the specified object.
+	// object, this will fall through to Default* settings.
 	ByObject map[client.Object]ByObject
 }
 
 // ByObject offers more fine-grained control over the cache's ListWatch by object.
 type ByObject struct {
+	// Namespaces maps a namespace name to cache setting. If set, only the
+	// namespaces in this map will be cached.
+	//
+	// Settings in the map value that are unset because either the value as a
+	// whole is nil or because the specific setting is nil will be defaulted.
+	// Use an empty value for the specific setting to prevent that.
+	//
+	// It is possible to have specific Config for just some namespaces
+	// but cache all namespaces by using the AllNamespaces const as the map key.
+	// This wil then include all namespaces that do not have a more specific
+	// setting.
+	//
+	// A nil map allows to default this to the cache's DefaultNamespaces setting.
+	// An empty map prevents this.
+	//
+	// This must be unset for cluster-scoped objects.
+	Namespaces map[string]Config
+
 	// Label represents a label selector for the object.
 	Label labels.Selector
 
@@ -194,48 +218,111 @@ type ByObject struct {
 	UnsafeDisableDeepCopy *bool
 }
 
+// Config describes all potential options for a given watch.
+type Config struct {
+	// LabelSelector specifies a label selector. A nil value allows to
+	// default this.
+	LabelSelector labels.Selector
+
+	// FieldSelector specifics a field selector. A nil value allows to
+	// default this.
+	FieldSelector fields.Selector
+
+	// Transform specifies a transform func. A nil value allows to default
+	// this.
+	Transform toolscache.TransformFunc
+
+	// UnsafeDisableDeepCopy specifies if List and Get requests against the
+	// cache should not DeepCopy. A nil value allows to default this.
+	UnsafeDisableDeepCopy *bool
+}
+
 // NewCacheFunc - Function for creating a new cache from the options and a rest config.
 type NewCacheFunc func(config *rest.Config, opts Options) (Cache, error)
 
 // New initializes and returns a new Cache.
-func New(config *rest.Config, opts Options) (Cache, error) {
-	if len(opts.Namespaces) == 0 {
-		opts.Namespaces = []string{metav1.NamespaceAll}
+func New(cfg *rest.Config, opts Options) (Cache, error) {
+	opts, err := defaultOpts(cfg, opts)
+	if err != nil {
+		return nil, err
 	}
-	if len(opts.Namespaces) > 1 {
-		return newMultiNamespaceCache(config, opts)
+
+	newCache := newCache(cfg, opts)
+
+	var defaultCache Cache
+	if len(opts.DefaultNamespaces) > 0 {
+		defaultConfig := optionDefaultsToConfig(&opts)
+		defaultCache = newMultiNamespaceCache(newCache, opts.Scheme, opts.Mapper, opts.DefaultNamespaces, &defaultConfig)
+	} else {
+		defaultCache = newCache(optionDefaultsToConfig(&opts), corev1.NamespaceAll)
 	}
 
-	opts, err := defaultOpts(config, opts)
-	if err != nil {
-		return nil, err
+	if len(opts.ByObject) == 0 {
+		return defaultCache, nil
 	}
 
-	byGVK, err := convertToInformerOptsByGVK(opts.ByObject, opts.Scheme)
-	if err != nil {
-		return nil, err
+	delegating := &delegatingByTypeCache{
+		scheme:       opts.Scheme,
+		caches:       make(map[schema.GroupVersionKind]Cache, len(opts.ByObject)),
+		defaultCache: defaultCache,
 	}
-	// Set the default selector and transform.
-	byGVK[schema.GroupVersionKind{}] = internal.InformersOptsByGVK{
-		Selector: internal.Selector{
-			Label: opts.DefaultLabelSelector,
-			Field: opts.DefaultFieldSelector,
-		},
+
+	for tp, config := range opts.ByObject {
+		gvk, err := apiutil.GVKForObject(tp, opts.Scheme)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get GVK for type %T: %w", tp, err)
+		}
+		var cache Cache
+		if len(config.Namespaces) > 0 {
+			cache = newMultiNamespaceCache(newCache, opts.Scheme, opts.Mapper, config.Namespaces, nil)
+		} else {
+			cache = newCache(byObjectToConfig(config), corev1.NamespaceAll)
+		}
+		delegating.caches[gvk] = cache
+	}
+
+	return delegating, nil
+}
+
+func optionDefaultsToConfig(opts *Options) Config {
+	return Config{
+		LabelSelector:         opts.DefaultLabelSelector,
+		FieldSelector:         opts.DefaultFieldSelector,
 		Transform:             opts.DefaultTransform,
-		UnsafeDisableDeepCopy: opts.UnsafeDisableDeepCopy,
+		UnsafeDisableDeepCopy: opts.DefaultUnsafeDisableDeepCopy,
+	}
+}
+
+func byObjectToConfig(byObject ByObject) Config {
+	return Config{
+		LabelSelector:         byObject.Label,
+		FieldSelector:         byObject.Field,
+		Transform:             byObject.Transform,
+		UnsafeDisableDeepCopy: byObject.UnsafeDisableDeepCopy,
 	}
+}
 
-	return &informerCache{
-		scheme: opts.Scheme,
-		Informers: internal.NewInformers(config, &internal.InformersOpts{
-			HTTPClient:   opts.HTTPClient,
-			Scheme:       opts.Scheme,
-			Mapper:       opts.Mapper,
-			ResyncPeriod: *opts.SyncPeriod,
-			Namespace:    opts.Namespaces[0],
-			ByGVK:        byGVK,
-		}),
-	}, nil
+type newCacheFunc func(config Config, namespace string) Cache
+
+func newCache(cfg *rest.Config, opts Options) newCacheFunc {
+	return func(config Config, namespace string) Cache {
+		return &informerCache{
+			scheme: opts.Scheme,
+			Informers: internal.NewInformers(cfg, &internal.InformersOpts{
+				HTTPClient:   opts.HTTPClient,
+				Scheme:       opts.Scheme,
+				Mapper:       opts.Mapper,
+				ResyncPeriod: *opts.SyncPeriod,
+				Namespace:    namespace,
+				Selector: internal.Selector{
+					Label: config.LabelSelector,
+					Field: config.FieldSelector,
+				},
+				Transform:             config.Transform,
+				UnsafeDisableDeepCopy: config.UnsafeDisableDeepCopy,
+			}),
+		}
+	}
 }
 
 func defaultOpts(config *rest.Config, opts Options) (Options, error) {
@@ -262,31 +349,70 @@ func defaultOpts(config *rest.Config, opts Options) (Options, error) {
 		}
 	}
 
+	for namespace, cfg := range opts.DefaultNamespaces {
+		cfg = defaultConfig(cfg, optionDefaultsToConfig(&opts))
+		opts.DefaultNamespaces[namespace] = cfg
+	}
+
+	for tp, cfg := range opts.ByObject {
+		isNamespaced, err := apiutil.IsObjectNamespaced(tp, opts.Scheme, opts.Mapper)
+		if err != nil {
+			return opts, fmt.Errorf("failed to determine if %T is namespaced: %w", tp, err)
+		}
+		if !isNamespaced && cfg.Namespaces != nil {
+			return opts, fmt.Errorf("type %T is not namespaced, but its ByObject.Namespaces setting is not nil", tp)
+		}
+
+		// Default the namespace-level configs first, because they need to use the undefaulted type-level config.
+		for namespace, config := range cfg.Namespaces {
+			// 1. Default from the undefaulted type-level config
+			config = defaultConfig(config, byObjectToConfig(cfg))
+
+			// 2. Default from the namespace-level config. This was defaulted from the global default config earlier, but
+			//    might not have an entry for the current namespace.
+			if defaultNamespaceSettings, hasDefaultNamespace := opts.DefaultNamespaces[namespace]; hasDefaultNamespace {
+				config = defaultConfig(config, defaultNamespaceSettings)
+			}
+
+			// 3. Default from the global defaults
+			config = defaultConfig(config, optionDefaultsToConfig(&opts))
+
+			cfg.Namespaces[namespace] = config
+		}
+
+		defaultedConfig := defaultConfig(byObjectToConfig(cfg), optionDefaultsToConfig(&opts))
+		cfg.Label = defaultedConfig.LabelSelector
+		cfg.Field = defaultedConfig.FieldSelector
+		cfg.Transform = defaultedConfig.Transform
+		cfg.UnsafeDisableDeepCopy = defaultedConfig.UnsafeDisableDeepCopy
+
+		if cfg.Namespaces == nil {
+			cfg.Namespaces = opts.DefaultNamespaces
+		}
+
+		opts.ByObject[tp] = cfg
+	}
+
 	// Default the resync period to 10 hours if unset
 	if opts.SyncPeriod == nil {
 		opts.SyncPeriod = &defaultSyncPeriod
 	}
 	return opts, nil
 }
 
-func convertToInformerOptsByGVK(in map[client.Object]ByObject, scheme *runtime.Scheme) (map[schema.GroupVersionKind]internal.InformersOptsByGVK, error) {
-	out := map[schema.GroupVersionKind]internal.InformersOptsByGVK{}
-	for object, byObject := range in {
-		gvk, err := apiutil.GVKForObject(object, scheme)
-		if err != nil {
-			return nil, err
-		}
-		if _, ok := out[gvk]; ok {
-			return nil, fmt.Errorf("duplicate cache options for GVK %v, cache.Options.ByObject has multiple types with the same GroupVersionKind", gvk)
-		}
-		out[gvk] = internal.InformersOptsByGVK{
-			Selector: internal.Selector{
-				Field: byObject.Field,
-				Label: byObject.Label,
-			},
-			Transform:             byObject.Transform,
-			UnsafeDisableDeepCopy: byObject.UnsafeDisableDeepCopy,
-		}
+func defaultConfig(toDefault, defaultFrom Config) Config {
+	if toDefault.LabelSelector == nil {
+		toDefault.LabelSelector = defaultFrom.LabelSelector
+	}
+	if toDefault.FieldSelector == nil {
+		toDefault.FieldSelector = defaultFrom.FieldSelector
 	}
-	return out, nil
+	if toDefault.Transform == nil {
+		toDefault.Transform = defaultFrom.Transform
+	}
+	if toDefault.UnsafeDisableDeepCopy == nil {
+		toDefault.UnsafeDisableDeepCopy = defaultFrom.UnsafeDisableDeepCopy
+	}
+
+	return toDefault
 }