Repository: rancher/fleet
Pull request #1809 (2023-09-26 14:16:40)

Helm Storage Backend With Caching
Helm is internally listing all secrets each time we retrieve information about the release. We were using the helm storage secrets driver, which is meant to be used as a cli. It uses a k8s client to get/list/create.. secrets, and it doesn't use any cache (see implementation [here](https://github.com/helm/helm/blob/v3.12.3/pkg/action/lazyclient.go#L60-L125))

This PR adds a new helm storage driver that uses the `SecretCache` from wrangler for getting and listing secrets. All other operations uses a k8s client like the helm secrets driver does.  `SecretCache` from wrangler is internally using a `ShareInformer` for creating the cache.

This PR undoes the cache implemented in https://github.com/rancher/fleet/commit/8b0d9b7031570b7f6c0b8b5e969e34542339d181

The helm globalCfg is created with an [empty namespace](https://github.com/rancher/fleet/blob/v0.8.0/internal/helmdeployer/deployer.go#L116), which means that it will list secrets from all namespaces. We are using it in some [places](https://github.com/rancher/fleet/blob/v0.8.0/internal/helmdeployer/deployer.go#L598) for retrieving releases information. That means we are unnecessarily listing secrets from all namespaces, and that could cause issues if two releases have the same name in two different namespaces. This will be fixed in a separate PR. 

refers to https://github.com/rancher/fleet/issues/1738


Patch:
diff --git a/internal/helmdeployer/deployer.go b/internal/helmdeployer/deployer.go
index f1db0ee9..30339aa0 100644
--- a/internal/helmdeployer/deployer.go
+++ b/internal/helmdeployer/deployer.go
@@ -9,15 +9,15 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/rancher/fleet/internal/config"
-	"github.com/rancher/fleet/pkg/durations"
+	"github.com/rancher/fleet/internal/helmdeployer/helmcache"
 	"github.com/sirupsen/logrus"
 	"helm.sh/helm/v3/pkg/action"
 	"helm.sh/helm/v3/pkg/chart"
 	"helm.sh/helm/v3/pkg/chart/loader"
 	"helm.sh/helm/v3/pkg/kube"
 	"helm.sh/helm/v3/pkg/release"
+	"helm.sh/helm/v3/pkg/storage"
 	"helm.sh/helm/v3/pkg/storage/driver"
-	"k8s.io/client-go/tools/cache"
 
 	"github.com/rancher/fleet/internal/helmdeployer/kustomize"
 	"github.com/rancher/fleet/internal/helmdeployer/rawyaml"
@@ -77,12 +77,6 @@ type Helm struct {
 	defaultNamespace string
 	labelPrefix      string
 	labelSuffix      string
-	releaseCache     cache.Store
-}
-
-func releaseKeyfunc(obj interface{}) (string, error) {
-	r := obj.(*release.Release)
-	return fmt.Sprintf("%s/%s-%d", r.Namespace, r.Name, r.Version), nil
 }
 
 type Resources struct {
@@ -109,14 +103,15 @@ func NewHelm(namespace, defaultNamespace, labelPrefix, labelSuffix string, gette
 		serviceAccountCache: serviceAccountCache,
 		configmapCache:      configmapCache,
 		secretCache:         secretCache,
-		releaseCache:        cache.NewTTLStore(releaseKeyfunc, durations.ReleaseCacheTTL),
 		labelPrefix:         labelPrefix,
 		labelSuffix:         labelSuffix,
 	}
-	if err := h.globalCfg.Init(getter, "", "secrets", logrus.Infof); err != nil {
+	cfg, err := h.createCfg("")
+	if err != nil {
 		return nil, err
 	}
-	h.globalCfg.Releases.MaxHistory = MaxHelmHistory
+	h.globalCfg = cfg
+
 	return h, nil
 }
 
@@ -344,7 +339,7 @@ func (h *Helm) getCfg(namespace, serviceAccountName string) (action.Configuratio
 	kClient := kube.New(getter)
 	kClient.Namespace = namespace
 
-	err = cfg.Init(getter, namespace, "secrets", logrus.Infof)
+	cfg, err = h.createCfg(namespace)
 	cfg.Releases.MaxHistory = MaxHelmHistory
 	cfg.KubeClient = kClient
 
@@ -589,63 +584,22 @@ func getReleaseNameVersionAndNamespace(bundleID, resourcesID string) (string, in
 }
 
 func (h *Helm) getRelease(releaseName, namespace string, version int) (*release.Release, error) {
-	key := &release.Release{
-		Namespace: namespace,
-		Name:      releaseName,
-		Version:   version,
-	}
+	hist := action.NewHistory(&h.globalCfg)
 
-	object, found, err := h.releaseCache.Get(key)
-	if err != nil {
+	releases, err := hist.Run(releaseName)
+	if err == driver.ErrReleaseNotFound {
+		return nil, ErrNoRelease
+	} else if err != nil {
 		return nil, err
 	}
-	if !found {
-		// cache MISS, query Helm's storage
-		hist := action.NewHistory(&h.globalCfg)
-		releases, err := hist.Run(releaseName)
-		if err == driver.ErrReleaseNotFound {
-			return nil, ErrNoRelease
-		} else if err != nil {
-			return nil, err
-		}
 
-		// opportunistically cache all releases in the storage with that name
-		for _, release := range releases {
-			err := h.releaseCache.Add(release)
-			if err != nil {
-				return nil, err
-			}
-		}
-
-		// return the right one if found
-		for _, release := range releases {
-			if release.Name == releaseName && release.Version == version && release.Namespace == namespace {
-				return release, nil
-			}
-		}
-
-		// desired release was not found in Helm's storage. Put a "dummy" in the cache to remember it was not found
-		err = h.releaseCache.Add(&release.Release{
-			Namespace: namespace,
-			Name:      releaseName,
-			Version:   version,
-			Manifest:  "release not found",
-		})
-		if err != nil {
-			return nil, err
+	for _, release := range releases {
+		if release.Name == releaseName && release.Version == version && release.Namespace == namespace {
+			return release, nil
 		}
-		return nil, ErrNoRelease
 	}
 
-	// cache HIT
-	rel := object.(*release.Release)
-	if rel.Manifest == "release not found" {
-		// but it was actually a "dummy". Return not found
-		return nil, ErrNoRelease
-	}
-
-	// happy case
-	return rel, nil
+	return nil, ErrNoRelease
 }
 
 func (h *Helm) EnsureInstalled(bundleID, resourcesID string) (bool, error) {
@@ -810,6 +764,26 @@ func (h *Helm) delete(bundleID string, options fleet.BundleDeploymentOptions, dr
 	return err
 }
 
+func (h *Helm) createCfg(namespace string) (action.Configuration, error) {
+	kc := kube.New(h.getter)
+	kc.Log = logrus.Infof
+	clientSet, err := kc.Factory.KubernetesClientSet()
+	if err != nil {
+		return action.Configuration{}, err
+	}
+	driver := driver.NewSecrets(helmcache.NewSecretClient(h.secretCache, clientSet, namespace))
+	driver.Log = logrus.Infof
+	store := storage.Init(driver)
+	store.MaxHistory = MaxHelmHistory
+
+	return action.Configuration{
+		RESTClientGetter: h.getter,
+		Releases:         store,
+		KubeClient:       kc,
+		Log:              logrus.Infof,
+	}, nil
+}
+
 func deleteHistory(cfg action.Configuration, bundleID string) error {
 	releases, err := cfg.Releases.List(func(r *release.Release) bool {
 		return r.Name == bundleID && r.Chart.Metadata.Annotations[BundleIDAnnotation] == bundleID
diff --git a/internal/helmdeployer/helmcache/secret.go b/internal/helmdeployer/helmcache/secret.go
new file mode 100644
index 00000000..e9e8d6a5
--- /dev/null
+++ b/internal/helmdeployer/helmcache/secret.go
@@ -0,0 +1,87 @@
+package helmcache
+
+import (
+	"context"
+
+	corecontrollers "github.com/rancher/wrangler/pkg/generated/controllers/core/v1"
+	v1 "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/watch"
+	applycorev1 "k8s.io/client-go/applyconfigurations/core/v1"
+	"k8s.io/client-go/kubernetes"
+)
+
+// SecretClient implements methods to handle secrets. Get and list will be retrieved from the wrangler cache, the other calls
+// will be make to the Kubernetes API server.
+type SecretClient struct {
+	cache     corecontrollers.SecretCache
+	client    kubernetes.Interface
+	namespace string
+}
+
+func NewSecretClient(cache corecontrollers.SecretCache, client kubernetes.Interface, namespace string) *SecretClient {
+	return &SecretClient{cache, client, namespace}
+}
+
+// Create creates a secret using a k8s client that calls the Kubernetes API server
+func (s *SecretClient) Create(ctx context.Context, secret *v1.Secret, opts metav1.CreateOptions) (*v1.Secret, error) {
+	return s.client.CoreV1().Secrets(s.namespace).Create(ctx, secret, opts)
+}
+
+// Update updates a secret using a k8s client that calls the Kubernetes API server
+func (s *SecretClient) Update(ctx context.Context, secret *v1.Secret, opts metav1.UpdateOptions) (*v1.Secret, error) {
+	return s.client.CoreV1().Secrets(s.namespace).Update(ctx, secret, opts)
+}
+
+// Delete deletes a secret using a k8s client that calls the Kubernetes API server
+func (s *SecretClient) Delete(ctx context.Context, name string, opts metav1.DeleteOptions) error {
+	return s.client.CoreV1().Secrets(s.namespace).Delete(ctx, name, opts)
+}
+
+// DeleteCollection deletes a secret collection using a k8s client that calls the Kubernetes API server
+func (s *SecretClient) DeleteCollection(ctx context.Context, opts metav1.DeleteOptions, listOpts metav1.ListOptions) error {
+	return s.client.CoreV1().Secrets(s.namespace).DeleteCollection(ctx, opts, listOpts)
+}
+
+// Get gets a secret from the wrangler cache.
+func (s *SecretClient) Get(ctx context.Context, name string, opts metav1.GetOptions) (*v1.Secret, error) {
+	return s.cache.Get(s.namespace, name)
+}
+
+// List lists secrets from the wrangler cache.
+func (s *SecretClient) List(ctx context.Context, opts metav1.ListOptions) (*v1.SecretList, error) {
+	labels, err := labels.Parse(opts.LabelSelector)
+	if err != nil {
+		return nil, err
+	}
+	secrets, err := s.cache.List(s.namespace, labels)
+	if err != nil {
+		return nil, err
+	}
+
+	var items []v1.Secret
+	for _, secret := range secrets {
+		items = append(items, *secret)
+	}
+
+	return &v1.SecretList{
+		Items: items,
+	}, nil
+}
+
+// Watch watches a secret using a k8s client that calls the Kubernetes API server
+func (s *SecretClient) Watch(ctx context.Context, opts metav1.ListOptions) (watch.Interface, error) {
+	return s.client.CoreV1().Secrets(s.namespace).Watch(ctx, opts)
+}
+
+// Patch patches a secret using a k8s client that calls the Kubernetes API server
+func (s *SecretClient) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts metav1.PatchOptions, subresources ...string) (*v1.Secret, error) {
+	return s.client.CoreV1().Secrets(s.namespace).Patch(ctx, name, pt, data, opts, subresources...)
+}
+
+// Apply applies a secret using a k8s client that calls the Kubernetes API server
+func (s *SecretClient) Apply(ctx context.Context, secretConfiguration *applycorev1.SecretApplyConfiguration, opts metav1.ApplyOptions) (*v1.Secret, error) {
+	return s.client.CoreV1().Secrets(s.namespace).Apply(ctx, secretConfiguration, opts)
+}


Test patch:
diff --git a/integrationtests/agent/suite_test.go b/integrationtests/agent/suite_test.go
index 5ab668c0..5b797de0 100644
--- a/integrationtests/agent/suite_test.go
+++ b/integrationtests/agent/suite_test.go
@@ -162,6 +162,9 @@ func registerBundleDeploymentController(cfg *rest.Config, namespace string, look
 	err = factory.Start(ctx, 50)
 	Expect(err).ToNot(HaveOccurred())
 
+	err = coreFactory.Start(ctx, 50)
+	Expect(err).ToNot(HaveOccurred())
+
 	return factory.Fleet().V1alpha1().BundleDeployment()
 }
 
diff --git a/internal/helmdeployer/helmcache/secret_test.go b/internal/helmdeployer/helmcache/secret_test.go
new file mode 100644
index 00000000..802bd99d
--- /dev/null
+++ b/internal/helmdeployer/helmcache/secret_test.go
@@ -0,0 +1,178 @@
+package helmcache
+
+import (
+	"context"
+	"testing"
+
+	"github.com/golang/mock/gomock"
+	"github.com/google/go-cmp/cmp"
+	"github.com/rancher/fleet/internal/cmd/controller/mocks"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
+	applycorev1 "k8s.io/client-go/applyconfigurations/core/v1"
+	v1 "k8s.io/client-go/applyconfigurations/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+const (
+	secretName      = "test"
+	secretNamespace = "test-ns"
+)
+
+var secret = corev1.Secret{
+	ObjectMeta: metav1.ObjectMeta{
+		Name:      secretName,
+		Namespace: secretNamespace,
+	},
+}
+
+func TestGet(t *testing.T) {
+	ctlr := gomock.NewController(t)
+	mockCache := mocks.NewMockSecretCache(ctlr)
+	secretClient := NewSecretClient(mockCache, nil, secretNamespace)
+	mockCache.EXPECT().Get(secretNamespace, secretName).Return(&secret, nil)
+
+	secretGot, err := secretClient.Get(context.TODO(), secretName, metav1.GetOptions{})
+
+	if err != nil {
+		t.Errorf("unexpected error %v", err)
+	}
+	if !cmp.Equal(&secret, secretGot) {
+		t.Errorf("expected secret %v, got %v", secret, secretGot)
+	}
+}
+
+func TestList(t *testing.T) {
+	ctlr := gomock.NewController(t)
+	mockCache := mocks.NewMockSecretCache(ctlr)
+	secretClient := NewSecretClient(mockCache, nil, secretNamespace)
+	labelSelector, err := labels.Parse("foo=bar")
+	if err != nil {
+		t.Errorf("unexpected error %v", err)
+	}
+	mockCache.EXPECT().List(secretNamespace, labelSelector).Return([]*corev1.Secret{&secret}, nil)
+	secretList, err := secretClient.List(context.TODO(), metav1.ListOptions{LabelSelector: "foo=bar"})
+
+	if err != nil {
+		t.Errorf("unexpected error %v", err)
+	}
+	secretListExpected := &corev1.SecretList{Items: []corev1.Secret{secret}}
+	if !cmp.Equal(secretListExpected, secretList) {
+		t.Errorf("expected secret %v, got %v", secretListExpected, secretList)
+	}
+}
+
+func TestCreate(t *testing.T) {
+	client := fake.NewSimpleClientset()
+	secretClient := NewSecretClient(nil, client, secretNamespace)
+	secretCreated, err := secretClient.Create(context.TODO(), &secret, metav1.CreateOptions{})
+
+	if err != nil {
+		t.Errorf("unexpected error %v", err)
+	}
+	if !cmp.Equal(&secret, secretCreated) {
+		t.Errorf("expected secret %v, got %v", secret, secretCreated)
+	}
+}
+
+func TestUpdate(t *testing.T) {
+	secretUpdate := corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      secretName,
+			Namespace: secretNamespace,
+		},
+		Data: map[string][]byte{"test": []byte("data")},
+	}
+	client := fake.NewSimpleClientset(&secret)
+	secretClient := NewSecretClient(nil, client, secretNamespace)
+	secretUpdated, err := secretClient.Update(context.TODO(), &secretUpdate, metav1.UpdateOptions{})
+
+	if err != nil {
+		t.Errorf("unexpected error %v", err)
+	}
+	if !cmp.Equal(&secretUpdate, secretUpdated) {
+		t.Errorf("expected secret %v, got %v", secretUpdate, secretUpdated)
+	}
+}
+
+func TestDelete(t *testing.T) {
+	client := fake.NewSimpleClientset(&secret)
+	secretClient := NewSecretClient(nil, client, secretNamespace)
+	err := secretClient.Delete(context.TODO(), secretName, metav1.DeleteOptions{})
+
+	if err != nil {
+		t.Errorf("unexpected error %v", err)
+	}
+}
+
+func TestDeleteCollection(t *testing.T) {
+	client := fake.NewSimpleClientset(&secret)
+	secretClient := NewSecretClient(nil, client, secretNamespace)
+	err := secretClient.DeleteCollection(context.TODO(), metav1.DeleteOptions{}, metav1.ListOptions{FieldSelector: "name=" + secretName})
+
+	if err != nil {
+		t.Errorf("unexpected error %v", err)
+	}
+}
+
+func TestWatch(t *testing.T) {
+	client := fake.NewSimpleClientset(&secret)
+	secretClient := NewSecretClient(nil, client, secretNamespace)
+	watch, err := secretClient.Watch(context.TODO(), metav1.ListOptions{FieldSelector: "name=" + secretName})
+
+	if err != nil {
+		t.Errorf("unexpected error %v", err)
+	}
+	if watch == nil {
+		t.Errorf("watch should not be nil")
+	}
+}
+
+func TestPatch(t *testing.T) {
+	secretPatch := corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      secretName,
+			Namespace: secretNamespace,
+		},
+		Data: map[string][]byte{"test": []byte("content")},
+	}
+	client := fake.NewSimpleClientset(&secret)
+	secretClient := NewSecretClient(nil, client, secretNamespace)
+	patch := []byte(`{"data":{"test":"Y29udGVudA=="}}`) // "content", base64-encoded
+	secretPatched, err := secretClient.Patch(context.TODO(), secretName, types.StrategicMergePatchType, patch, metav1.PatchOptions{})
+
+	if err != nil {
+		t.Errorf("unexpected error %v", err)
+	}
+	if !cmp.Equal(&secretPatch, secretPatched) {
+		t.Errorf("expected secret %v, got %v", secretPatch, secretPatched)
+	}
+}
+
+func TestApply(t *testing.T) {
+	secretApply := corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      secretName,
+			Namespace: secretNamespace,
+		},
+		Data: map[string][]byte{"test": []byte("content")},
+	}
+	client := fake.NewSimpleClientset(&secret)
+	secretClient := NewSecretClient(nil, client, secretNamespace)
+	secretName := "test"
+	secretApplied, err := secretClient.Apply(context.TODO(), &applycorev1.SecretApplyConfiguration{
+		ObjectMetaApplyConfiguration: &v1.ObjectMetaApplyConfiguration{
+			Name: &secretName,
+		},
+		Data: map[string][]byte{"test": []byte("content")},
+	}, metav1.ApplyOptions{})
+
+	if err != nil {
+		t.Errorf("unexpected error %v", err)
+	}
+	if !cmp.Equal(&secretApply, secretApplied) {
+		t.Errorf("expected secret %v, got %v", secretApply, secretApplied)
+	}
+}