Support multiple secret managers side by side (#187)

## Overview
For managed+ customers to seamlessly migrate to using new secrets, we need to allow new secrets to be used side by side with old secret managers.

This change:
- adds a new configuration `secretManagerTypes`, which allows specifying multiple secret managers that would be executed in the order specified
- adds logic to loop through all specified secret managers.

## Test Plan
Deployed to dogfood and tested with both single and multiple secret managers side by side.

## Rollout Plan (if applicable)
This change is backwards-compatible. As is, it is a no-op. It will be rolled out via separate changes in the cloud repo (e.g., for [dogfood](https://github.com/unionai/cloud/pull/7186)).

## Upstream Changes
Should this change be upstreamed to OSS (flyteorg/flyte)? If so, please check this box for auditing. Note, this is the responsibility of each developer. See [this guide](https://unionai.atlassian.net/wiki/spaces/ENG/pages/447610883/Flyte+-+Union+Cloud+Development+Runbook/#When-are-versions-updated%3F).
- [ ] To be upstreamed

## Jira Issue
https://unionai.atlassian.net/browse/CLOUD-1743

## Checklist
* [ ] Added tests
* [ ] Ran a deploy dry run and shared the terraform plan
* [ ] Added logging and metrics
* [ ] Updated [dashboards](https://unionai.grafana.net/dashboards) and [alerts](https://unionai.grafana.net/alerting/list)
* [ ] Updated documentation

flytepropeller/pkg/webhook/aws_secret_manager.go

@@ -60,7 +60,7 @@ func (i AWSSecretManagerInjector) Type() config.SecretManagerType {
 
 func (i AWSSecretManagerInjector) Inject(ctx context.Context, secret *core.Secret, p *corev1.Pod) (newP *corev1.Pod, injected bool, err error) {
 	if len(secret.Group) == 0 || len(secret.Key) == 0 {
-		return nil, false, fmt.Errorf("AWS Secrets Webhook require both key and group to be set. "+
+		return p, false, fmt.Errorf("AWS Secrets Webhook require both key and group to be set. "+
 			"Secret: [%v]", secret)
 	}
 

flytepropeller/pkg/webhook/config/config.go

@@ -97,14 +97,16 @@ const (
 )
 
 type Config struct {
-	MetricsPrefix               string                      `json:"metrics-prefix" pflag:",An optional prefix for all published metrics."`
-	CertDir                     string                      `json:"certDir" pflag:",Certificate directory to use to write generated certs. Defaults to /etc/webhook/certs/"`
-	LocalCert                   bool                        `json:"localCert" pflag:",write certs locally. Defaults to false"`
-	ListenPort                  int                         `json:"listenPort" pflag:",The port to use to listen to webhook calls. Defaults to 9443"`
-	ServiceName                 string                      `json:"serviceName" pflag:",The name of the webhook service."`
-	ServicePort                 int32                       `json:"servicePort" pflag:",The port on the service that hosting webhook."`
-	SecretName                  string                      `json:"secretName" pflag:",Secret name to write generated certs to."`
-	SecretManagerType           SecretManagerType           `json:"secretManagerType" pflag:"-,Secret manager type to use if secrets are not found in global secrets."`
+	MetricsPrefix string `json:"metrics-prefix" pflag:",An optional prefix for all published metrics."`
+	CertDir       string `json:"certDir" pflag:",Certificate directory to use to write generated certs. Defaults to /etc/webhook/certs/"`
+	LocalCert     bool   `json:"localCert" pflag:",write certs locally. Defaults to false"`
+	ListenPort    int    `json:"listenPort" pflag:",The port to use to listen to webhook calls. Defaults to 9443"`
+	ServiceName   string `json:"serviceName" pflag:",The name of the webhook service."`
+	ServicePort   int32  `json:"servicePort" pflag:",The port on the service that hosting webhook."`
+	SecretName    string `json:"secretName" pflag:",Secret name to write generated certs to."`
+	// Deprecated: use SecretManagerTypes instead.
+	SecretManagerType           SecretManagerType           `json:"secretManagerType" pflag:"-,Deprecated. Secret manager type to use if secrets are not found in global secrets. Ignored if secretManagerTypes is set."`
+	SecretManagerTypes          []SecretManagerType         `json:"secretManagerTypes" pflag:"-,List of secret manager types to use if secrets are not found in global secrets. In order of preference. Overrides secretManagerType if set."`
 	AWSSecretManagerConfig      AWSSecretManagerConfig      `json:"awsSecretManager" pflag:",AWS Secret Manager config."`
 	GCPSecretManagerConfig      GCPSecretManagerConfig      `json:"gcpSecretManager" pflag:",GCP Secret Manager config."`
 	VaultSecretManagerConfig    VaultSecretManagerConfig    `json:"vaultSecretManager" pflag:",Vault Secret Manager config."`
@@ -120,7 +122,6 @@ const (
 )
 
 type EmbeddedSecretManagerConfig struct {
-	Enabled   bool                      `json:"enabled" pflag:",Enable secret manager service"`
 	Type      EmbeddedSecretManagerType `json:"type" pflags:"-,Type of embedded secret manager to initialize"`
 	AWSConfig AWSConfig                 `json:"awsConfig" pflag:",Config for AWS settings"`
 	GCPConfig GCPConfig                 `json:"gcpConfig" pflag:",Config for GCP settings"`

flytepropeller/pkg/webhook/embedded_secret_manager.go

@@ -115,7 +115,7 @@ func (i EmbeddedSecretManagerInjector) lookUpSecret(ctx context.Context, secret
 }
 func (i EmbeddedSecretManagerInjector) Inject(ctx context.Context, secret *core.Secret, p *corev1.Pod) (newP *corev1.Pod, injected bool, err error) {
 	if len(secret.Key) == 0 {
-		return nil, false, fmt.Errorf("EmbeddedSecretManager requires key to be set. "+
+		return p, false, fmt.Errorf("EmbeddedSecretManager requires key to be set. "+
 			"Secret: [%v]", secret)
 	}
 

flytepropeller/pkg/webhook/embedded_secret_manager_test.go

@@ -38,9 +38,7 @@ func TestEmbeddedSecretManagerInjector_Inject(t *testing.T) {
 		Project: gcpProject,
 	}, gcpClient)
 
-	injector := NewEmbeddedSecretManagerInjector(config.EmbeddedSecretManagerConfig{
-		Enabled: true,
-	}, gcpSecretsFetcher)
+	injector := NewEmbeddedSecretManagerInjector(config.EmbeddedSecretManagerConfig{}, gcpSecretsFetcher)
 
 	inputSecret := &core.Secret{
 		Key: secretIDKey,

flytepropeller/pkg/webhook/gcp_secret_manager.go

@@ -63,7 +63,7 @@ func (i GCPSecretManagerInjector) Type() config.SecretManagerType {
 
 func (i GCPSecretManagerInjector) Inject(ctx context.Context, secret *core.Secret, p *corev1.Pod) (newP *corev1.Pod, injected bool, err error) {
 	if len(secret.Group) == 0 || len(secret.GroupVersion) == 0 {
-		return nil, false, fmt.Errorf("GCP Secrets Webhook require both group and group version to be set. "+
+		return p, false, fmt.Errorf("GCP Secrets Webhook require both group and group version to be set. "+
 			"Secret: [%v]", secret)
 	}
 

flytepropeller/pkg/webhook/pod.go

@@ -221,7 +221,8 @@ func NewPodMutator(ctx context.Context, cfg *config.Config, scheme *runtime.Sche
 		cfg:     cfg,
 		Mutators: []MutatorConfig{
 			{
-				Mutator: secretsMutator,
+				Mutator:  secretsMutator,
+				Required: true,
 			},
 		},
 	}, nil

flytepropeller/pkg/webhook/secrets.go

@@ -2,6 +2,8 @@ package webhook
 
 import (
 	"context"
+	"errors"
+	"fmt"
 
 	corev1 "k8s.io/api/core/v1"
 
@@ -20,8 +22,11 @@ const (
 )
 
 type SecretsMutator struct {
-	cfg       *config.Config
-	injectors []SecretsInjector
+	// Secret manager types in order that they should be used.
+	enabledSecretManagerTypes []config.SecretManagerType
+
+	// It is expected that this map contains a key for every element in enabledSecretManagerTypes.
+	injectors map[config.SecretManagerType]SecretsInjector
 }
 
 type SecretsInjector interface {
@@ -33,55 +38,106 @@ func (s SecretsMutator) ID() string {
 	return "secrets"
 }
 
-func (s *SecretsMutator) Mutate(ctx context.Context, p *corev1.Pod) (newP *corev1.Pod, injected bool, err error) {
-	secrets, err := secretUtils.UnmarshalStringMapToSecrets(p.GetAnnotations())
+func (s *SecretsMutator) Mutate(ctx context.Context, pod *corev1.Pod) (*corev1.Pod, bool /* injected */, error) {
+	secrets, err := secretUtils.UnmarshalStringMapToSecrets(pod.GetAnnotations())
 	if err != nil {
-		return p, false, err
+		return pod, false, fmt.Errorf("failed to unmarshal secrets from pod annotations: %w", err)
 	}
 
 	for _, secret := range secrets {
-		for _, injector := range s.injectors {
-			if injector.Type() != config.SecretManagerTypeGlobal && injector.Type() != s.cfg.SecretManagerType {
-				logger.Infof(ctx, "Skipping SecretManager [%v] since it's not enabled.", injector.Type())
-				continue
-			}
-
-			p, injected, err = injector.Inject(ctx, secret, p)
-			if err != nil {
-				logger.Infof(ctx, "Failed to inject a secret using injector [%v]. Error: %v", injector.Type(), err)
-			} else if injected {
-				break
+		mutatedPod, injected, err := s.injectSecret(ctx, secret, pod)
+		if !injected {
+			if err == nil {
+				err = fmt.Errorf("none of the secret managers injected secret [%v]", secret)
+			} else {
+				err = fmt.Errorf("none of the secret managers injected secret [%v]: %w", secret, err)
 			}
+			return pod, false, err
 		}
 
+		pod = mutatedPod
+	}
+
+	return pod, len(secrets) > 0, nil
+}
+
+func (s *SecretsMutator) injectSecret(ctx context.Context, secret *core.Secret, pod *corev1.Pod) (*corev1.Pod, bool /*injected*/, error) {
+	errs := make([]error, 0)
+
+	logger.Debugf(ctx, "Injecting secret [%v].", secret)
+
+	for _, secretManagerType := range s.enabledSecretManagerTypes {
+		injector := s.injectors[secretManagerType]
+
+		mutatedPod, injected, err := injector.Inject(ctx, secret, pod)
+		logger.Debugf(
+			ctx,
+			"injection result with injector type [%v]: injected = [%v], error = [%v].",
+			injector.Type(), injected, err)
+
 		if err != nil {
-			return p, false, err
+			errs = append(errs, err)
+			continue
+		}
+		if injected {
+			return mutatedPod, true, nil
 		}
 	}
 
-	return p, injected, nil
+	return pod, false, errors.Join(errs...)
 }
 
-// NewSecretsMutator creates a new SecretsMutator with all available plugins. Depending on the selected plugins in the
-// config, only the global plugin and one other plugin can be enabled.
+// NewSecretsMutator creates a new SecretsMutator with all available plugins.
 func NewSecretsMutator(ctx context.Context, cfg *config.Config, _ promutils.Scope) (*SecretsMutator, error) {
-	var embeddedSecretsManager SecretsInjector
-	if cfg.EmbeddedSecretManagerConfig.Enabled {
-		secretFetcher, err := NewSecretFetcherManager(ctx, cfg.EmbeddedSecretManagerConfig)
+	enabledSecretManagerTypes := []config.SecretManagerType{
+		config.SecretManagerTypeGlobal,
+	}
+	if len(cfg.SecretManagerTypes) != 0 {
+		enabledSecretManagerTypes = append(enabledSecretManagerTypes, cfg.SecretManagerTypes...)
+	} else {
+		enabledSecretManagerTypes = append(enabledSecretManagerTypes, cfg.SecretManagerType)
+	}
+
+	injectors := make(map[config.SecretManagerType]SecretsInjector, len(enabledSecretManagerTypes))
+	globalSecretManagerConfig := secretmanager.GetConfig()
+	for _, secretManagerType := range enabledSecretManagerTypes {
+		injector, err := newSecretManager(ctx, secretManagerType, cfg, globalSecretManagerConfig)
 		if err != nil {
 			return nil, err
 		}
-		embeddedSecretsManager = NewEmbeddedSecretManagerInjector(cfg.EmbeddedSecretManagerConfig, secretFetcher)
+		injectors[secretManagerType] = injector
 	}
+
 	return &SecretsMutator{
-		cfg: cfg,
-		injectors: []SecretsInjector{
-			NewGlobalSecrets(secretmanager.NewFileEnvSecretManager(secretmanager.GetConfig())),
-			NewK8sSecretsInjector(),
-			NewAWSSecretManagerInjector(cfg.AWSSecretManagerConfig),
-			NewGCPSecretManagerInjector(cfg.GCPSecretManagerConfig),
-			NewVaultSecretManagerInjector(cfg.VaultSecretManagerConfig),
-			embeddedSecretsManager,
-		},
+		enabledSecretManagerTypes,
+		injectors,
 	}, nil
 }
+
+func newSecretManager(
+	ctx context.Context,
+	secretManagerType config.SecretManagerType,
+	webhookConfig *config.Config,
+	globalSecretManagerConfig *secretmanager.Config,
+) (SecretsInjector, error) {
+	switch secretManagerType {
+	case config.SecretManagerTypeGlobal:
+		return NewGlobalSecrets(secretmanager.NewFileEnvSecretManager(globalSecretManagerConfig)), nil
+	case config.SecretManagerTypeK8s:
+		return NewK8sSecretsInjector(), nil
+	case config.SecretManagerTypeAWS:
+		return NewAWSSecretManagerInjector(webhookConfig.AWSSecretManagerConfig), nil
+	case config.SecretManagerTypeGCP:
+		return NewGCPSecretManagerInjector(webhookConfig.GCPSecretManagerConfig), nil
+	case config.SecretManagerTypeVault:
+		return NewVaultSecretManagerInjector(webhookConfig.VaultSecretManagerConfig), nil
+	case config.SecretManagerTypeEmbedded:
+		secretFetcher, err := NewSecretFetcherManager(ctx, webhookConfig.EmbeddedSecretManagerConfig)
+		if err != nil {
+			return nil, err
+		}
+		return NewEmbeddedSecretManagerInjector(webhookConfig.EmbeddedSecretManagerConfig, secretFetcher), nil
+	default:
+		return nil, fmt.Errorf("unrecognized secret manager type [%v]", secretManagerType)
+	}
+}

flytepropeller/pkg/webhook/secrets_test.go

@@ -36,7 +36,10 @@ func TestSecretsWebhook_Mutate(t *testing.T) {
 		mutator.OnType().Return(config.SecretManagerTypeGlobal)
 
 		m := SecretsMutator{
-			injectors: []SecretsInjector{mutator},
+			enabledSecretManagerTypes: []config.SecretManagerType{config.SecretManagerTypeGlobal},
+			injectors: map[config.SecretManagerType]SecretsInjector{
+				config.SecretManagerTypeGlobal: mutator,
+			},
 		}
 
 		_, changed, err := m.Mutate(context.Background(), podWithAnnotations.DeepCopy())
@@ -50,7 +53,10 @@ func TestSecretsWebhook_Mutate(t *testing.T) {
 		mutator.OnType().Return(config.SecretManagerTypeGlobal)
 
 		m := SecretsMutator{
-			injectors: []SecretsInjector{mutator},
+			enabledSecretManagerTypes: []config.SecretManagerType{config.SecretManagerTypeGlobal},
+			injectors: map[config.SecretManagerType]SecretsInjector{
+				config.SecretManagerTypeGlobal: mutator,
+			},
 		}
 
 		_, changed, err := m.Mutate(context.Background(), podWithAnnotations.DeepCopy())

flytepropeller/pkg/webhook/vault_secret_manager.go

@@ -36,7 +36,7 @@ func (i VaultSecretManagerInjector) Type() config.SecretManagerType {
 
 func (i VaultSecretManagerInjector) Inject(ctx context.Context, secret *coreIdl.Secret, p *corev1.Pod) (newP *corev1.Pod, injected bool, err error) {
 	if len(secret.Group) == 0 || len(secret.Key) == 0 {
-		return nil, false, fmt.Errorf("Vault Secrets Webhook requires both key and group to be set. "+
+		return p, false, fmt.Errorf("Vault Secrets Webhook requires both key and group to be set. "+
 			"Secret: [%v]", secret)
 	}