Added a selectively-caching delegating client

Avoid caching Secrets or ConfigMaps

Based on changes to the upstream controller-runtime:
kubernetes-sigs/controller-runtime#1249

Author: tvs

bootstrap/kubeadm/main.go

@@ -25,6 +25,7 @@ import (
 	"time"
 
 	"github.com/spf13/pflag"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
@@ -137,8 +138,23 @@ func main() {
 		RetryPeriod:        &leaderElectionRetryPeriod,
 		Namespace:          watchNamespace,
 		SyncPeriod:         &syncPeriod,
-		NewClient:          util.ManagerCachelessClientFunc,
-		Port:               webhookPort,
+		NewClient: func(cache cache.Cache, config *rest.Config, options client.Options) (client.Client, error) {
+			c, err := client.New(config, options)
+			if err != nil {
+				return nil, err
+			}
+
+			return util.NewSelectiveDelegatingClient(util.NewSelectiveDelegatingClientInput{
+				CacheReader: cache,
+				Client:      c,
+				Scheme:      options.Scheme,
+				UncachedObjects: []runtime.Object{
+					&corev1.Secret{},
+					&corev1.ConfigMap{},
+				},
+			})
+		},
+		Port: webhookPort,
 	})
 	if err != nil {
 		setupLog.Error(err, "unable to start manager")

controlplane/kubeadm/main.go

@@ -25,6 +25,7 @@ import (
 	"time"
 
 	"github.com/spf13/pflag"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
@@ -129,8 +130,23 @@ func main() {
 		RetryPeriod:        &leaderElectionRetryPeriod,
 		Namespace:          watchNamespace,
 		SyncPeriod:         &syncPeriod,
-		NewClient:          util.ManagerCachelessClientFunc,
-		Port:               webhookPort,
+		NewClient: func(cache cache.Cache, config *rest.Config, options client.Options) (client.Client, error) {
+			c, err := client.New(config, options)
+			if err != nil {
+				return nil, err
+			}
+
+			return util.NewSelectiveDelegatingClient(util.NewSelectiveDelegatingClientInput{
+				CacheReader: cache,
+				Client:      c,
+				Scheme:      options.Scheme,
+				UncachedObjects: []runtime.Object{
+					&corev1.Secret{},
+					&corev1.ConfigMap{},
+				},
+			})
+		},
+		Port: webhookPort,
 	})
 	if err != nil {
 		setupLog.Error(err, "unable to start manager")

main.go

@@ -24,9 +24,11 @@ import (
 	"time"
 
 	"github.com/spf13/pflag"
+	corev1 "k8s.io/api/core/v1"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
 	"k8s.io/klog"
 	"k8s.io/klog/klogr"
 	clusterv1alpha2 "sigs.k8s.io/cluster-api/api/v1alpha2"
@@ -41,6 +43,8 @@ import (
 	"sigs.k8s.io/cluster-api/feature"
 	"sigs.k8s.io/cluster-api/util"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	// +kubebuilder:scaffold:imports
@@ -155,16 +159,31 @@ func main() {
 	}
 
 	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
-		Scheme:                 scheme,
-		MetricsBindAddress:     metricsAddr,
-		LeaderElection:         enableLeaderElection,
-		LeaderElectionID:       "controller-leader-election-capi",
-		LeaseDuration:          &leaderElectionLeaseDuration,
-		RenewDeadline:          &leaderElectionRenewDeadline,
-		RetryPeriod:            &leaderElectionRetryPeriod,
-		Namespace:              watchNamespace,
-		SyncPeriod:             &syncPeriod,
-		NewClient:              util.ManagerCachelessClientFunc,
+		Scheme:             scheme,
+		MetricsBindAddress: metricsAddr,
+		LeaderElection:     enableLeaderElection,
+		LeaderElectionID:   "controller-leader-election-capi",
+		LeaseDuration:      &leaderElectionLeaseDuration,
+		RenewDeadline:      &leaderElectionRenewDeadline,
+		RetryPeriod:        &leaderElectionRetryPeriod,
+		Namespace:          watchNamespace,
+		SyncPeriod:         &syncPeriod,
+		NewClient: func(cache cache.Cache, config *rest.Config, options client.Options) (client.Client, error) {
+			c, err := client.New(config, options)
+			if err != nil {
+				return nil, err
+			}
+
+			return util.NewSelectiveDelegatingClient(util.NewSelectiveDelegatingClientInput{
+				CacheReader: cache,
+				Client:      c,
+				Scheme:      options.Scheme,
+				UncachedObjects: []runtime.Object{
+					&corev1.Secret{},
+					&corev1.ConfigMap{},
+				},
+			})
+		},
 		Port:                   webhookPort,
 		HealthProbeBindAddress: healthAddr,
 	})

util/util.go

@@ -703,12 +703,6 @@ func ManagerDelegatingClientFunc(cache cache.Cache, config *rest.Config, options
 	}, nil
 }
 
-// ManagerCachelessClientFunc returns a manager.NewClientFunc to be used when creating
-// a new controller runtime manager.
-func ManagerCachelessClientFunc(cache cache.Cache, config *rest.Config, options client.Options) (client.Client, error) {
-	return client.New(config, options)
-}
-
 // LowestNonZeroResult compares two reconciliation results
 // and returns the one with lowest requeue time.
 func LowestNonZeroResult(i, j ctrl.Result) ctrl.Result {
@@ -727,3 +721,73 @@ func LowestNonZeroResult(i, j ctrl.Result) ctrl.Result {
 		return j
 	}
 }
+
+// NewSelectiveDelegatingClientInput encapsulates the input parameters to create
+// a new selective delegating client
+type NewSelectiveDelegatingClientInput struct {
+	CacheReader     client.Reader
+	Client          client.Client
+	UncachedObjects []runtime.Object
+	Scheme          *runtime.Scheme
+}
+
+type selectiveDelegatingReader struct {
+	CacheReader  client.Reader
+	ClientReader client.Reader
+
+	uncachedGVKs map[schema.GroupVersionKind]struct{}
+	scheme       *runtime.Scheme
+}
+
+func (d *selectiveDelegatingReader) 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 *selectiveDelegatingReader) Get(ctx context.Context, key client.ObjectKey, obj runtime.Object) error {
+	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 *selectiveDelegatingReader) List(ctx context.Context, list runtime.Object, opts ...client.ListOption) error {
+	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...)
+}
+
+func NewSelectiveDelegatingClient(in NewSelectiveDelegatingClientInput) (client.Client, error) {
+	uncachedGVKs := map[schema.GroupVersionKind]struct{}{}
+	for _, obj := range in.UncachedObjects {
+		gvk, err := apiutil.GVKForObject(obj, in.Scheme)
+		if err != nil {
+			return nil, err
+		}
+		uncachedGVKs[gvk] = struct{}{}
+	}
+
+	return &client.DelegatingClient{
+		Reader: &selectiveDelegatingReader{
+			CacheReader:  in.CacheReader,
+			ClientReader: in.Client,
+			scheme:       in.Scheme,
+			uncachedGVKs: uncachedGVKs,
+		},
+		Writer:       in.Client,
+		StatusClient: in.Client,
+	}, nil
+}