Add support for a chunked release storage driver

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 <[email protected]>

Author: joelanford

pkg/client/actionconfig.go

@@ -33,6 +33,8 @@ import (
 	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/rest"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	customstorage "github.com/operator-framework/helm-operator-plugins/pkg/storage"
 )
 
 type ActionConfigGetter interface {
@@ -57,14 +59,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 +86,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 +144,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
 }
@@ -177,15 +209,82 @@ var _ v1.SecretInterface = &ownerRefSecretClient{}
 
 type ownerRefSecretClient struct {
 	v1.SecretInterface
-	refs []metav1.OwnerReference
+	match func(secret *corev1.Secret) bool
+	refs  []metav1.OwnerReference
 }
 
 func (c *ownerRefSecretClient) Create(ctx context.Context, in *corev1.Secret, opts metav1.CreateOptions) (*corev1.Secret, error) {
-	in.OwnerReferences = append(in.OwnerReferences, c.refs...)
+	if c.match == nil || c.match(in) {
+		in.OwnerReferences = append(in.OwnerReferences, c.refs...)
+	}
 	return c.SecretInterface.Create(ctx, in, opts)
 }
 
 func (c *ownerRefSecretClient) Update(ctx context.Context, in *corev1.Secret, opts metav1.UpdateOptions) (*corev1.Secret, error) {
-	in.OwnerReferences = append(in.OwnerReferences, c.refs...)
+	if c.match == nil || c.match(in) {
+		in.OwnerReferences = append(in.OwnerReferences, c.refs...)
+	}
 	return c.SecretInterface.Update(ctx, in, opts)
 }
+
+type SecretsStorageDriverOpts struct {
+	DisableOwnerRefInjection bool
+	StorageNamespaceMapper   ObjectToStringMapper
+}
+
+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)
+		}
+
+		secretClient := secretsInterface.Secrets(storageNamespace)
+		if !opts.DisableOwnerRefInjection {
+			ownerRef := metav1.NewControllerRef(obj, obj.GetObjectKind().GroupVersionKind())
+			secretClient = &ownerRefSecretClient{
+				SecretInterface: secretClient,
+				refs:            []metav1.OwnerReference{*ownerRef},
+			}
+		}
+		d := driver.NewSecrets(secretClient)
+		d.Log = getDebugLogger(ctx)
+		return d, nil
+	}
+}
+
+func ChunkedSecretsStorageDriver(owner string, chunkSize int, 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)
+		}
+
+		secretClient := secretsInterface.Secrets(storageNamespace)
+		if !opts.DisableOwnerRefInjection {
+			ownerRef := metav1.NewControllerRef(obj, obj.GetObjectKind().GroupVersionKind())
+			secretClient = &ownerRefSecretClient{
+				SecretInterface: secretClient,
+				match:           func(secret *corev1.Secret) bool { return secret.Type == customstorage.SecretTypeChunkedIndex },
+				refs:            []metav1.OwnerReference{*ownerRef},
+			}
+		}
+		d := customstorage.NewChunkedSecrets(secretClient, chunkSize, owner)
+		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,31 @@ 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 := ac.Releases.Get(expected.Name, expected.Version)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(actual).To(Equal(expected))
+			})
 		})
 	})