Add support for a chunked release storage driver (#350)

This commit adds a custom helm release storage driver that overcomes
limitations in the size of a single value that can be stored in etcd.

In order to remain backward-compatible and also make this storage
driver available, this commit also refactors the ActionConfigGetter
options so that a custom function can be provided to the
ActionConfigGetter that can create the desired storage driver.

This commit also separates the rest config mapping into two separate
options. One for interactions with the storage backend, and the other
for managing release content.

Signed-off-by: Joe Lanford <joe.lanford@gmail.com>

pkg/client/actionconfig.go

@@ -25,7 +25,6 @@ import (
 	"helm.sh/helm/v3/pkg/kube"
 	"helm.sh/helm/v3/pkg/storage"
 	"helm.sh/helm/v3/pkg/storage/driver"
-	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/discovery"
@@ -57,14 +56,25 @@ func NewActionConfigGetter(baseRestConfig *rest.Config, rm meta.RESTMapper, opts
 	if acg.objectToClientNamespace == nil {
 		acg.objectToClientNamespace = getObjectNamespace
 	}
-	if acg.objectToStorageNamespace == nil {
-		acg.objectToStorageNamespace = getObjectNamespace
+	if acg.objectToClientRestConfig == nil {
+		acg.objectToClientRestConfig = func(_ context.Context, _ client.Object, baseRestConfig *rest.Config) (*rest.Config, error) {
+			return rest.CopyConfig(baseRestConfig), nil
+		}
 	}
-	if acg.objectToRestConfig == nil {
-		acg.objectToRestConfig = func(_ context.Context, _ client.Object, baseRestConfig *rest.Config) (*rest.Config, error) {
+	if acg.objectToStorageRestConfig == nil {
+		acg.objectToStorageRestConfig = func(_ context.Context, _ client.Object, baseRestConfig *rest.Config) (*rest.Config, error) {
 			return rest.CopyConfig(baseRestConfig), nil
 		}
 	}
+	if acg.objectToStorageDriver == nil {
+		if acg.objectToStorageNamespace == nil {
+			acg.objectToStorageNamespace = getObjectNamespace
+		}
+		acg.objectToStorageDriver = DefaultSecretsStorageDriver(SecretsStorageDriverOpts{
+			DisableOwnerRefInjection: acg.disableStorageOwnerRefInjection,
+			StorageNamespaceMapper:   acg.objectToStorageNamespace,
+		})
+	}
 	return acg, nil
 }
 
@@ -73,28 +83,52 @@ var _ ActionConfigGetter = &actionConfigGetter{}
 type ActionConfigGetterOption func(getter *actionConfigGetter)
 
 type ObjectToStringMapper func(client.Object) (string, error)
+type ObjectToRestConfigMapper func(context.Context, client.Object, *rest.Config) (*rest.Config, error)
+type ObjectToStorageDriverMapper func(context.Context, client.Object, *rest.Config) (driver.Driver, error)
+
+func ClientRestConfigMapper(f ObjectToRestConfigMapper) ActionConfigGetterOption { // nolint:revive
+	return func(getter *actionConfigGetter) {
+		getter.objectToClientRestConfig = f
+	}
+}
 
 func ClientNamespaceMapper(m ObjectToStringMapper) ActionConfigGetterOption { // nolint:revive
 	return func(getter *actionConfigGetter) {
 		getter.objectToClientNamespace = m
 	}
 }
 
+func StorageRestConfigMapper(f ObjectToRestConfigMapper) ActionConfigGetterOption {
+	return func(getter *actionConfigGetter) {
+		getter.objectToStorageRestConfig = f
+	}
+}
+
+func StorageDriverMapper(f ObjectToStorageDriverMapper) ActionConfigGetterOption {
+	return func(getter *actionConfigGetter) {
+		getter.objectToStorageDriver = f
+	}
+}
+
+// Deprecated: use StorageDriverMapper(DefaultSecretsStorageDriver(SecretsStorageDriverOpts)) instead.
 func StorageNamespaceMapper(m ObjectToStringMapper) ActionConfigGetterOption {
 	return func(getter *actionConfigGetter) {
 		getter.objectToStorageNamespace = m
 	}
 }
 
+// Deprecated: use StorageDriverMapper(DefaultSecretsStorageDriver(SecretsStorageDriverOpts)) instead.
 func DisableStorageOwnerRefInjection(v bool) ActionConfigGetterOption {
 	return func(getter *actionConfigGetter) {
 		getter.disableStorageOwnerRefInjection = v
 	}
 }
 
+// Deprecated: use ClientRestConfigMapper and StorageRestConfigMapper instead.
 func RestConfigMapper(f func(context.Context, client.Object, *rest.Config) (*rest.Config, error)) ActionConfigGetterOption {
 	return func(getter *actionConfigGetter) {
-		getter.objectToRestConfig = f
+		getter.objectToClientRestConfig = f
+		getter.objectToStorageRestConfig = f
 	}
 }
 
@@ -107,58 +141,53 @@ type actionConfigGetter struct {
 	restMapper      meta.RESTMapper
 	discoveryClient discovery.CachedDiscoveryInterface
 
-	objectToClientNamespace         ObjectToStringMapper
-	objectToStorageNamespace        ObjectToStringMapper
-	objectToRestConfig              func(context.Context, client.Object, *rest.Config) (*rest.Config, error)
+	objectToClientRestConfig ObjectToRestConfigMapper
+	objectToClientNamespace  ObjectToStringMapper
+
+	objectToStorageRestConfig ObjectToRestConfigMapper
+	objectToStorageDriver     ObjectToStorageDriverMapper
+
+	// Deprecated: only keep around for backward compatibility with StorageNamespaceMapper option.
+	objectToStorageNamespace ObjectToStringMapper
+	// Deprecated: only keep around for backward compatibility with DisableStorageOwnerRefInjection option.
 	disableStorageOwnerRefInjection bool
 }
 
 func (acg *actionConfigGetter) ActionConfigFor(ctx context.Context, obj client.Object) (*action.Configuration, error) {
-	storageNs, err := acg.objectToStorageNamespace(obj)
+	clientRestConfig, err := acg.objectToClientRestConfig(ctx, obj, acg.baseRestConfig)
 	if err != nil {
-		return nil, fmt.Errorf("get storage namespace for object: %v", err)
-	}
-
-	restConfig, err := acg.objectToRestConfig(ctx, obj, acg.baseRestConfig)
-	if err != nil {
-		return nil, fmt.Errorf("get rest config for object: %v", err)
+		return nil, fmt.Errorf("get client rest config for object: %v", err)
 	}
 
 	clientNamespace, err := acg.objectToClientNamespace(obj)
 	if err != nil {
 		return nil, fmt.Errorf("get client namespace for object: %v", err)
 	}
 
-	rcg := newRESTClientGetter(restConfig, acg.restMapper, acg.discoveryClient, clientNamespace)
-	kc := kube.New(rcg)
-	kc.Namespace = clientNamespace
-
-	kcs, err := kc.Factory.KubernetesClientSet()
-	if err != nil {
-		return nil, fmt.Errorf("create kubernetes clientset: %v", err)
-	}
+	clientRCG := newRESTClientGetter(clientRestConfig, acg.restMapper, acg.discoveryClient, clientNamespace)
+	clientKC := kube.New(clientRCG)
+	clientKC.Namespace = clientNamespace
 
 	// Setup the debug log function that Helm will use
 	debugLog := getDebugLogger(ctx)
 
-	secretClient := kcs.CoreV1().Secrets(storageNs)
-	if !acg.disableStorageOwnerRefInjection {
-		ownerRef := metav1.NewControllerRef(obj, obj.GetObjectKind().GroupVersionKind())
-		secretClient = &ownerRefSecretClient{
-			SecretInterface: secretClient,
-			refs:            []metav1.OwnerReference{*ownerRef},
-		}
+	storageRestConfig, err := acg.objectToStorageRestConfig(ctx, obj, acg.baseRestConfig)
+	if err != nil {
+		return nil, fmt.Errorf("get storage rest config for object: %v", err)
+	}
+
+	d, err := acg.objectToStorageDriver(ctx, obj, storageRestConfig)
+	if err != nil {
+		return nil, fmt.Errorf("get storage driver for object: %v", err)
 	}
-	d := driver.NewSecrets(secretClient)
-	d.Log = debugLog
 
 	// Initialize the storage backend
 	s := storage.Init(d)
 
 	return &action.Configuration{
-		RESTClientGetter: rcg,
+		RESTClientGetter: clientRCG,
 		Releases:         s,
-		KubeClient:       kc,
+		KubeClient:       clientKC,
 		Log:              debugLog,
 	}, nil
 }
@@ -173,19 +202,32 @@ func getDebugLogger(ctx context.Context) func(format string, v ...interface{}) {
 	}
 }
 
-var _ v1.SecretInterface = &ownerRefSecretClient{}
-
-type ownerRefSecretClient struct {
-	v1.SecretInterface
-	refs []metav1.OwnerReference
+type SecretsStorageDriverOpts struct {
+	DisableOwnerRefInjection bool
+	StorageNamespaceMapper   ObjectToStringMapper
 }
 
-func (c *ownerRefSecretClient) Create(ctx context.Context, in *corev1.Secret, opts metav1.CreateOptions) (*corev1.Secret, error) {
-	in.OwnerReferences = append(in.OwnerReferences, c.refs...)
-	return c.SecretInterface.Create(ctx, in, opts)
-}
+func DefaultSecretsStorageDriver(opts SecretsStorageDriverOpts) ObjectToStorageDriverMapper {
+	if opts.StorageNamespaceMapper == nil {
+		opts.StorageNamespaceMapper = getObjectNamespace
+	}
+	return func(ctx context.Context, obj client.Object, restConfig *rest.Config) (driver.Driver, error) {
+		storageNamespace, err := opts.StorageNamespaceMapper(obj)
+		if err != nil {
+			return nil, fmt.Errorf("get storage namespace for object: %v", err)
+		}
+		secretsInterface, err := v1.NewForConfig(restConfig)
+		if err != nil {
+			return nil, fmt.Errorf("create secrets client for storage: %v", err)
+		}
 
-func (c *ownerRefSecretClient) Update(ctx context.Context, in *corev1.Secret, opts metav1.UpdateOptions) (*corev1.Secret, error) {
-	in.OwnerReferences = append(in.OwnerReferences, c.refs...)
-	return c.SecretInterface.Update(ctx, in, opts)
+		secretClient := secretsInterface.Secrets(storageNamespace)
+		if !opts.DisableOwnerRefInjection {
+			ownerRef := metav1.NewControllerRef(obj, obj.GetObjectKind().GroupVersionKind())
+			secretClient = NewOwnerRefSecretClient(secretClient, []metav1.OwnerReference{*ownerRef}, MatchAllSecrets)
+		}
+		d := driver.NewSecrets(secretClient)
+		d.Log = getDebugLogger(ctx)
+		return d, nil
+	}
 }

pkg/client/actionconfig_test.go

@@ -23,8 +23,11 @@ import (
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+
 	"helm.sh/helm/v3/pkg/action"
 	"helm.sh/helm/v3/pkg/kube"
+	"helm.sh/helm/v3/pkg/release"
+	"helm.sh/helm/v3/pkg/storage/driver"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -185,6 +188,32 @@ metadata:
 				Expect(err).ToNot(HaveOccurred())
 				Expect(ac2.RESTClientGetter.ToRESTConfig()).To(WithTransform(func(c *rest.Config) string { return c.BearerToken }, Equal("test2")))
 			})
+
+			It("should use a custom storage driver", func() {
+				storageDriver := driver.NewMemory()
+
+				storageDriverMapper := func(ctx context.Context, obj client.Object, cfg *rest.Config) (driver.Driver, error) {
+					return storageDriver, nil
+				}
+				acg, err := NewActionConfigGetter(cfg, rm, StorageDriverMapper(storageDriverMapper))
+				Expect(err).ToNot(HaveOccurred())
+
+				testObject := func(name string) client.Object {
+					u := unstructured.Unstructured{}
+					u.SetName(name)
+					return &u
+				}
+
+				ac, err := acg.ActionConfigFor(context.Background(), testObject("test1"))
+				Expect(err).ToNot(HaveOccurred())
+
+				expected := &release.Release{Name: "test1", Version: 2, Info: &release.Info{Status: release.StatusDeployed}}
+				Expect(ac.Releases.Create(expected)).To(Succeed())
+				actual, err := storageDriver.List(func(r *release.Release) bool { return true })
+				Expect(err).ToNot(HaveOccurred())
+				Expect(actual).To(HaveLen(1))
+				Expect(actual[0]).To(Equal(expected))
+			})
 		})
 	})
 

pkg/client/ownerrefclient.go

@@ -0,0 +1,65 @@
+package client
+
+import (
+	"context"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	clientcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+)
+
+var _ clientcorev1.SecretInterface = &ownerRefSecretClient{}
+
+// NewOwnerRefSecretClient returns a SecretInterface that injects the provided owner references
+// to all created or updated secrets that match the provided match function. If match is nil, all
+// secrets are matched.
+func NewOwnerRefSecretClient(client clientcorev1.SecretInterface, refs []metav1.OwnerReference, match func(*corev1.Secret) bool) clientcorev1.SecretInterface {
+	if match == nil {
+		match = MatchAllSecrets
+	}
+	return &ownerRefSecretClient{
+		SecretInterface: client,
+		match:           match,
+		refs:            refs,
+	}
+}
+
+func MatchAllSecrets(_ *corev1.Secret) bool {
+	return true
+}
+
+type ownerRefSecretClient struct {
+	clientcorev1.SecretInterface
+	match func(secret *corev1.Secret) bool
+	refs  []metav1.OwnerReference
+}
+
+func (c *ownerRefSecretClient) appendMissingOwnerRefs(secret *corev1.Secret) {
+	hasOwnerRef := func(secret *corev1.Secret, ref metav1.OwnerReference) bool {
+		for _, r := range secret.OwnerReferences {
+			if r.UID == ref.UID {
+				return true
+			}
+		}
+		return false
+	}
+	for i := range c.refs {
+		if !hasOwnerRef(secret, c.refs[i]) {
+			secret.OwnerReferences = append(secret.OwnerReferences, c.refs[i])
+		}
+	}
+}
+
+func (c *ownerRefSecretClient) Create(ctx context.Context, in *corev1.Secret, opts metav1.CreateOptions) (*corev1.Secret, error) {
+	if c.match == nil || c.match(in) {
+		c.appendMissingOwnerRefs(in)
+	}
+	return c.SecretInterface.Create(ctx, in, opts)
+}
+
+func (c *ownerRefSecretClient) Update(ctx context.Context, in *corev1.Secret, opts metav1.UpdateOptions) (*corev1.Secret, error) {
+	if c.match == nil || c.match(in) {
+		c.appendMissingOwnerRefs(in)
+	}
+	return c.SecretInterface.Update(ctx, in, opts)
+}