⚠️ Add ability for the delegating client to avoid caching objects

A long standing issue in controller runtime, the delegating client is
the default client that the manager creates. Whenever a user calls
Get/List on a typed object, the internal cache spins up an informer and
start watching *all* objects for that group-version-kind.

This change introduces the ability for the delegating client to take
a list of objects that should always hit the live api-server, and bypass
the cache.

We also offer the ability to build a manager.NewClientFunc with a new
builder that exposes the ability to mark which objects we want to be
uncached.

Signed-off-by: Vince Prignano <vincepri@vmware.com>

pkg/client/client_test.go

@@ -3065,10 +3065,11 @@ var _ = Describe("DelegatingClient", func() {
 			cachedReader := &fakeReader{}
 			cl, err := client.New(cfg, client.Options{})
 			Expect(err).NotTo(HaveOccurred())
-			dReader := client.NewDelegatingClient(client.NewDelegatingClientInput{
+			dReader, err := client.NewDelegatingClient(client.NewDelegatingClientInput{
 				CacheReader: cachedReader,
 				Client:      cl,
 			})
+			Expect(err).NotTo(HaveOccurred())
 			var actual appsv1.Deployment
 			key := client.ObjectKey{Namespace: "ns", Name: "name"}
 			Expect(dReader.Get(context.TODO(), key, &actual)).To(Succeed())
@@ -3079,10 +3080,11 @@ var _ = Describe("DelegatingClient", func() {
 			cachedReader := &fakeReader{}
 			cl, err := client.New(cfg, client.Options{})
 			Expect(err).NotTo(HaveOccurred())
-			dReader := client.NewDelegatingClient(client.NewDelegatingClientInput{
+			dReader, err := client.NewDelegatingClient(client.NewDelegatingClientInput{
 				CacheReader: cachedReader,
 				Client:      cl,
 			})
+			Expect(err).NotTo(HaveOccurred())
 			dep := &appsv1.Deployment{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:   "deployment1",
@@ -3123,10 +3125,11 @@ var _ = Describe("DelegatingClient", func() {
 			cachedReader := &fakeReader{}
 			cl, err := client.New(cfg, client.Options{})
 			Expect(err).NotTo(HaveOccurred())
-			dReader := client.NewDelegatingClient(client.NewDelegatingClientInput{
+			dReader, err := client.NewDelegatingClient(client.NewDelegatingClientInput{
 				CacheReader: cachedReader,
 				Client:      cl,
 			})
+			Expect(err).NotTo(HaveOccurred())
 			var actual appsv1.DeploymentList
 			Expect(dReader.List(context.Background(), &actual)).To(Succeed())
 			Expect(1).To(Equal(cachedReader.Called))
@@ -3136,10 +3139,11 @@ var _ = Describe("DelegatingClient", func() {
 			cachedReader := &fakeReader{}
 			cl, err := client.New(cfg, client.Options{})
 			Expect(err).NotTo(HaveOccurred())
-			dReader := client.NewDelegatingClient(client.NewDelegatingClientInput{
+			dReader, err := client.NewDelegatingClient(client.NewDelegatingClientInput{
 				CacheReader: cachedReader,
 				Client:      cl,
 			})
+			Expect(err).NotTo(HaveOccurred())
 
 			actual := &unstructured.UnstructuredList{}
 			actual.SetGroupVersionKind(schema.GroupVersionKind{

pkg/client/split.go

@@ -22,30 +22,44 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
 )
 
 // NewDelegatingClientInput encapsulates the input parameters to create a new delegating client.
 type NewDelegatingClientInput struct {
-	CacheReader Reader
-	Client      Client
+	CacheReader     Reader
+	Client          Client
+	UncachedObjects []runtime.Object
 }
 
 // NewDelegatingClient creates a new delegating client.
 //
 // A delegating client forms a Client by composing separate reader, writer and
 // statusclient interfaces.  This way, you can have an Client that reads from a
 // cache and writes to the API server.
-func NewDelegatingClient(in NewDelegatingClientInput) Client {
+func NewDelegatingClient(in NewDelegatingClientInput) (Client, error) {
+	uncachedGVKs := map[schema.GroupVersionKind]struct{}{}
+	for _, obj := range in.UncachedObjects {
+		gvk, err := apiutil.GVKForObject(obj, in.Client.Scheme())
+		if err != nil {
+			return nil, err
+		}
+		uncachedGVKs[gvk] = struct{}{}
+	}
+
 	return &delegatingClient{
 		scheme: in.Client.Scheme(),
 		mapper: in.Client.RESTMapper(),
 		Reader: &delegatingReader{
 			CacheReader:  in.CacheReader,
 			ClientReader: in.Client,
+			scheme:       in.Client.Scheme(),
+			uncachedGVKs: uncachedGVKs,
 		},
 		Writer:       in.Client,
 		StatusClient: in.Client,
-	}
+	}, nil
 }
 
 type delegatingClient struct {
@@ -75,21 +89,37 @@ func (d *delegatingClient) RESTMapper() meta.RESTMapper {
 type delegatingReader struct {
 	CacheReader  Reader
 	ClientReader Reader
+
+	uncachedGVKs map[schema.GroupVersionKind]struct{}
+	scheme       *runtime.Scheme
+}
+
+func (d *delegatingReader) isUncached(obj runtime.Object) (bool, error) {
+	gvk, err := apiutil.GVKForObject(obj, d.scheme)
+	if err != nil {
+		return false, err
+	}
+	_, isUncached := d.uncachedGVKs[gvk]
+	_, isUnstructured := obj.(*unstructured.Unstructured)
+	_, isUnstructuredList := obj.(*unstructured.UnstructuredList)
+	return isUncached || isUnstructured || isUnstructuredList, nil
 }
 
 // Get retrieves an obj for a given object key from the Kubernetes Cluster.
 func (d *delegatingReader) Get(ctx context.Context, key ObjectKey, obj Object) error {
-	_, isUnstructured := obj.(*unstructured.Unstructured)
-	if isUnstructured {
+	if isUncached, err := d.isUncached(obj); err != nil {
+		return err
+	} else if isUncached {
 		return d.ClientReader.Get(ctx, key, obj)
 	}
 	return d.CacheReader.Get(ctx, key, obj)
 }
 
 // List retrieves list of objects for a given namespace and list options.
 func (d *delegatingReader) List(ctx context.Context, list ObjectList, opts ...ListOption) error {
-	_, isUnstructured := list.(*unstructured.UnstructuredList)
-	if isUnstructured {
+	if isUncached, err := d.isUncached(list); err != nil {
+		return err
+	} else if isUncached {
 		return d.ClientReader.List(ctx, list, opts...)
 	}
 	return d.CacheReader.List(ctx, list, opts...)

pkg/manager/client.go

@@ -0,0 +1,66 @@
+/*
+Copyright 2020 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package manager
+
+import (
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// DefaultNewClient creates the default caching client
+var DefaultNewClient = NewClientFuncBuilder().Build()
+
+// NewClient builder is the interface for the builder for manager.NewClientFunc.
+type NewClientBuilder interface {
+	// WithUncached takes a list of runtime objects (plain or lists) that users don't want to cache
+	// for this client. This function can be called multiple times, it should append to an internal slice.
+	WithUncached(objs ...runtime.Object)
+
+	// Build returns the NewClientFunc that can be used when creating a new manager.
+	Build() NewClientFunc
+}
+
+// NewClientFuncBuilder returns a builder to create a new manager.NewClientFunc to be passed when creating a Manager.
+func NewClientFuncBuilder() NewClientBuilder {
+	return &newClientBuilder{}
+}
+
+type newClientBuilder struct {
+	uncached []runtime.Object
+}
+
+func (n *newClientBuilder) WithUncached(objs ...runtime.Object) {
+	n.uncached = append(n.uncached, objs...)
+}
+
+func (n *newClientBuilder) Build() NewClientFunc {
+	return func(cache cache.Cache, config *rest.Config, options client.Options) (client.Client, error) {
+		// Create the Client for Write operations.
+		c, err := client.New(config, options)
+		if err != nil {
+			return nil, err
+		}
+
+		return client.NewDelegatingClient(client.NewDelegatingClientInput{
+			CacheReader:     cache,
+			Client:          c,
+			UncachedObjects: n.uncached,
+		})
+	}
+}

pkg/manager/manager.go

@@ -503,20 +503,6 @@ func (o Options) setLeaderElectionConfig(obj v1alpha1.ControllerManagerConfigura
 	return o
 }
 
-// DefaultNewClient creates the default caching client
-func DefaultNewClient(cache cache.Cache, config *rest.Config, options client.Options) (client.Client, error) {
-	// Create the Client for Write operations.
-	c, err := client.New(config, options)
-	if err != nil {
-		return nil, err
-	}
-
-	return client.NewDelegatingClient(client.NewDelegatingClientInput{
-		CacheReader: cache,
-		Client:      c,
-	}), nil
-}
-
 // defaultHealthProbeListener creates the default health probes listener bound to the given address
 func defaultHealthProbeListener(addr string) (net.Listener, error) {
 	if addr == "" || addr == "0" {

pkg/runtime/inject/inject_test.go

@@ -87,7 +87,8 @@ var _ = Describe("runtime inject", func() {
 	})
 
 	It("should set client", func() {
-		client := client.NewDelegatingClient(client.NewDelegatingClientInput{Client: fake.NewFakeClient()})
+		client, err := client.NewDelegatingClient(client.NewDelegatingClientInput{Client: fake.NewFakeClient()})
+		Expect(err).NotTo(HaveOccurred())
 
 		By("Validating injecting client")
 		res, err := ClientInto(client, instance)
@@ -152,7 +153,8 @@ var _ = Describe("runtime inject", func() {
 	})
 
 	It("should set api reader", func() {
-		apiReader := client.NewDelegatingClient(client.NewDelegatingClientInput{Client: fake.NewFakeClient()})
+		apiReader, err := client.NewDelegatingClient(client.NewDelegatingClientInput{Client: fake.NewFakeClient()})
+		Expect(err).NotTo(HaveOccurred())
 
 		By("Validating injecting client")
 		res, err := APIReaderInto(apiReader, instance)