Helm: Use informer to list helm secrets to improve performance

Helm stores its state in secrets inside the cluster.
Instead of listing these secrets before every reconciliation of every release, we use an informer to query a local secrets list.
This significantly reduced the load on the kubernetes apiserver and etcd

Author: Lucaber

changelog/fragments/05-template.yaml

@@ -0,0 +1,43 @@
+# entries is a list of entries to include in
+# release notes and/or the migration guide
+entries:
+  - description: >
+      (helm): Use informer to list helm secrets to improve performance
+
+    # kind is one of:
+    # - addition
+    # - change
+    # - deprecation
+    # - removal
+    # - bugfix
+    kind: "change"
+
+    # Is this a breaking change?
+    breaking: false
+
+    # NOTE: ONLY USE `pull_request_override` WHEN ADDING THIS
+    # FILE FOR A PREVIOUSLY MERGED PULL_REQUEST!
+    #
+    # The generator auto-detects the PR number from the commit
+    # message in which this file was originally added.
+    #
+    # What is the pull request number (without the "#")?
+    # pull_request_override: 0
+
+
+    # Migration can be defined to automatically add a section to
+    # the migration guide. This is required for breaking changes.
+    migration:
+      header: Require `watch` on `secrets`
+      body: |
+        The operator now requires the watch operation on secrets.
+        When using a custom ServiceAccount for deployment, add following additional role is now required:
+        ```
+        rules:
+          - apiGroups:
+              - ""
+            resources:
+              - secrets
+            verbs:
+              - watch
+        ```

internal/helm/client/actionconfig.go

@@ -19,6 +19,7 @@ package client
 import (
 	"context"
 	"fmt"
+	"sync"
 
 	"k8s.io/client-go/kubernetes"
 
@@ -57,26 +58,42 @@ func NewActionConfigGetter(cfg *rest.Config, rm meta.RESTMapper, log logr.Logger
 	}
 
 	return &actionConfigGetter{
-		kubeClient:       kc,
-		kubeClientSet:    kcs,
-		debugLog:         debugLog,
-		restClientGetter: rcg.restClientGetter,
+		kubeClient:          kc,
+		kubeClientSet:       kcs,
+		debugLog:            debugLog,
+		restClientGetter:    rcg.restClientGetter,
+		watchedSecrets:      map[string]*WatchedSecrets{},
+		watchedSecretsMutex: &sync.Mutex{},
 	}, nil
 }
 
 var _ ActionConfigGetter = &actionConfigGetter{}
 
 type actionConfigGetter struct {
-	kubeClient       *kube.Client
-	kubeClientSet    kubernetes.Interface
-	debugLog         func(string, ...interface{})
-	restClientGetter *restClientGetter
+	kubeClient          *kube.Client
+	kubeClientSet       kubernetes.Interface
+	debugLog            func(string, ...interface{})
+	restClientGetter    *restClientGetter
+	watchedSecrets      map[string]*WatchedSecrets
+	watchedSecretsMutex *sync.Mutex
+}
+
+// Creates a new watcher for each namespace to not require cluster-wide secret access
+func (acg *actionConfigGetter) getWatchedSecretsForNamespace(namespace string) *WatchedSecrets {
+	acg.watchedSecretsMutex.Lock()
+	if _, found := acg.watchedSecrets[namespace]; !found {
+		acg.watchedSecrets[namespace] = NewWatchedSecrets(acg.kubeClientSet, namespace)
+		acg.watchedSecrets[namespace].Run()
+	}
+	acg.watchedSecretsMutex.Unlock()
+	return acg.watchedSecrets[namespace]
 }
 
 func (acg *actionConfigGetter) ActionConfigFor(obj client.Object) (*action.Configuration, error) {
+	watchedSecrets := acg.getWatchedSecretsForNamespace(obj.GetNamespace())
 	ownerRef := metav1.NewControllerRef(obj, obj.GetObjectKind().GroupVersionKind())
 	d := driver.NewSecrets(&ownerRefSecretClient{
-		SecretInterface: acg.kubeClientSet.CoreV1().Secrets(obj.GetNamespace()),
+		SecretInterface: watchedSecrets,
 		refs:            []metav1.OwnerReference{*ownerRef},
 	})
 

internal/helm/client/secrets_watch.go

@@ -0,0 +1,140 @@
+/*
+Copyright 2020 The Operator-SDK 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 client
+
+import (
+	"context"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/apimachinery/pkg/watch"
+	applyconfv1 "k8s.io/client-go/applyconfigurations/core/v1"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
+	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	v12 "k8s.io/client-go/listers/core/v1"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+)
+
+var log = logf.Log.WithName("helm.watchedsecrets")
+
+// Wraps the kubernetes SecretInterface
+// Helm queries its own secrets multiple times per reconciliation. To reduce the number of lists going to the apiserver
+// we instead use an informer to watch the changes on secrets.
+type WatchedSecrets struct {
+	inner           v1.SecretInterface
+	informerFactory informers.SharedInformerFactory
+	informerLister  v12.SecretNamespaceLister
+}
+
+func (w *WatchedSecrets) Create(ctx context.Context, secret *corev1.Secret, opts metav1.CreateOptions) (*corev1.Secret, error) {
+	return w.inner.Create(ctx, secret, opts)
+}
+
+func (w *WatchedSecrets) Update(ctx context.Context, secret *corev1.Secret, opts metav1.UpdateOptions) (*corev1.Secret, error) {
+	return w.inner.Update(ctx, secret, opts)
+}
+
+func (w *WatchedSecrets) Delete(ctx context.Context, name string, opts metav1.DeleteOptions) error {
+	return w.inner.Delete(ctx, name, opts)
+}
+
+func (w *WatchedSecrets) DeleteCollection(ctx context.Context, opts metav1.DeleteOptions, listOpts metav1.ListOptions) error {
+	return w.inner.DeleteCollection(ctx, opts, listOpts)
+}
+
+func (w *WatchedSecrets) Get(ctx context.Context, name string, opts metav1.GetOptions) (*corev1.Secret, error) {
+	return w.inner.Get(ctx, name, opts)
+}
+
+func (w *WatchedSecrets) List(ctx context.Context, opts metav1.ListOptions) (*corev1.SecretList, error) {
+
+	// The informer interface only offers to filter secrets with a labelSelector
+	// Currently (helm v3.10.3) this List function is only being called in `storage/driver/secrets.go` with a
+	// labelSelector, meaning this case should never be executed. But we are able to fallback to the normal List
+	// implementation.
+	if hasListOptionsOtherThanLabelSelector(opts) {
+		log.Info("Cannot use informer to list secrets", "listOptions", opts)
+		return w.inner.List(ctx, opts)
+	}
+
+	labelSelector, err := labels.Parse(opts.LabelSelector)
+	if err != nil {
+		return nil, err
+	}
+	secrets, err := w.informerLister.List(labelSelector)
+	if err != nil {
+		return nil, err
+	}
+
+	secretList := &corev1.SecretList{
+		TypeMeta: metav1.TypeMeta{},
+		ListMeta: metav1.ListMeta{},
+		Items:    make([]corev1.Secret, len(secrets)),
+	}
+	for i, sec := range secrets {
+		secretList.Items[i] = *sec
+	}
+
+	return secretList, nil
+}
+
+func hasListOptionsOtherThanLabelSelector(opts metav1.ListOptions) bool {
+	empty := metav1.ListOptions{}
+
+	providedWithoutLabelSelector := opts
+	providedWithoutLabelSelector.LabelSelector = ""
+
+	return empty != providedWithoutLabelSelector
+}
+
+func (w *WatchedSecrets) Watch(ctx context.Context, opts metav1.ListOptions) (watch.Interface, error) {
+	return w.inner.Watch(ctx, opts)
+}
+
+func (w *WatchedSecrets) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts metav1.PatchOptions, subresources ...string) (result *corev1.Secret, err error) {
+	return w.inner.Patch(ctx, name, pt, data, opts)
+}
+
+func (w *WatchedSecrets) Apply(ctx context.Context, secret *applyconfv1.SecretApplyConfiguration, opts metav1.ApplyOptions) (result *corev1.Secret, err error) {
+	return w.inner.Apply(ctx, secret, opts)
+}
+
+var _ v1.SecretInterface = &WatchedSecrets{}
+
+func NewWatchedSecrets(clientSet kubernetes.Interface, namespace string) *WatchedSecrets {
+	log.V(2).Info("Get secrets client", "namespace", namespace)
+	informerFactory := informers.NewSharedInformerFactoryWithOptions(clientSet, time.Second*30, informers.WithNamespace(namespace))
+	secretsInformer := informerFactory.Core().V1().Secrets()
+
+	informerSecretsLister := secretsInformer.Lister().Secrets(namespace)
+
+	return &WatchedSecrets{
+		inner:           clientSet.CoreV1().Secrets(namespace),
+		informerFactory: informerFactory,
+		informerLister:  informerSecretsLister,
+	}
+}
+
+func (w *WatchedSecrets) Run() {
+	w.informerFactory.Start(wait.NeverStop)
+	_ = w.informerFactory.WaitForCacheSync(wait.NeverStop)
+}