Secrets manager: create secure value using the active keeper (#114039)

* Secrets manager: create secure value using the active keeper

* SecureValueService.Update: fetch secure value from db to get the keeper

* fix keeper_store_test.go

* SecureValueService: fix bug in update where the current version keeper wasn't being passed to the createNewVersion method

* make gofmt

* remove outdated test

* update TestModel

* undo enterprise_imports changes

* use xkube.Namespace

* migrator: set secret_secure_value.keeper to 'system' when the column is null

* indent cue

* fix tests

* fix enterprise imports

* properly fix enterprise imports

* make update-workspace

* go mod tidy

---------

Co-authored-by: Matheus Macabu <macabu.matheus@gmail.com>

apps/secret/kinds/v1beta1/securevalue.cue

@@ -29,13 +29,6 @@ SecureValueSpec: {
 	// +optional
 	ref?: string & strings.MinRunes(1) & strings.MaxRunes(1024)
 
-	// Name of the keeper, being the actual storage of the secure value.
-	// If not specified, the default keeper for the namespace will be used.
-	// +k8s:validation:minLength=1
-	// +k8s:validation:maxLength=253
-	// +optional
-	keeper?: string & strings.MinRunes(1) & strings.MaxRunes(253)
-
 	// The Decrypters that are allowed to decrypt this secret.
 	// An empty list means no service can decrypt it.
 	// +k8s:validation:maxItems=64
@@ -53,4 +46,7 @@ SecureValueStatus: {
 	// External ID where the secret is stored. Cannot be set.
 	// +optional
 	externalID: string
+
+	// The name of the keeper used to create the secure value. Cannot be set.
+	keeper: string
 }

apps/secret/pkg/apis/secret/v1beta1/register.go

@@ -28,7 +28,7 @@ var SecureValuesResourceInfo = utils.NewResourceInfo(
 		},
 		Reader: func(obj any) ([]any, error) {
 			if r, ok := obj.(*SecureValue); ok {
-				return []any{r.Name, r.Spec.Description, r.Spec.Keeper, r.Spec.Ref}, nil
+				return []any{r.Name, r.Spec.Description, r.Status.Keeper, r.Spec.Ref}, nil
 			}
 
 			return nil, fmt.Errorf("expected SecureValue but got %T", obj)

apps/secret/pkg/apis/secret/v1beta1/securevalue_spec_gen.go

@@ -25,12 +25,6 @@ type SecureValueSpec struct {
 	// +k8s:validation:maxLength=1024
 	// +optional
 	Ref *string `json:"ref,omitempty"`
-	// Name of the keeper, being the actual storage of the secure value.
-	// If not specified, the default keeper for the namespace will be used.
-	// +k8s:validation:minLength=1
-	// +k8s:validation:maxLength=253
-	// +optional
-	Keeper *string `json:"keeper,omitempty"`
 	// The Decrypters that are allowed to decrypt this secret.
 	// An empty list means no service can decrypt it.
 	// +k8s:validation:maxItems=64

apps/secret/pkg/apis/secret/v1beta1/securevalue_status_gen.go

@@ -25,12 +25,14 @@ type SecureValueStatus struct {
 	// Version of the secure value. Cannot be set.
 	// +optional
 	Version int64 `json:"version"`
-	// operatorStates is a map of operator ID to operator state evaluations.
-	// Any operator which consumes this kind SHOULD add its state evaluation information to this field.
-	OperatorStates map[string]SecureValuestatusOperatorState `json:"operatorStates,omitempty"`
 	// External ID where the secret is stored. Cannot be set.
 	// +optional
 	ExternalID string `json:"externalID"`
+	// operatorStates is a map of operator ID to operator state evaluations.
+	// Any operator which consumes this kind SHOULD add its state evaluation information to this field.
+	OperatorStates map[string]SecureValuestatusOperatorState `json:"operatorStates,omitempty"`
+	// The name of the keeper used to create the secure value. Cannot be set.
+	Keeper string `json:"keeper"`
 	// additionalFields is reserved for future use
 	AdditionalFields map[string]interface{} `json:"additionalFields,omitempty"`
 }

apps/secret/pkg/apis/secret/v1beta1/zz_openapi_gen.go

@@ -587,15 +587,6 @@ func schema_pkg_apis_secret_v1beta1_SecureValueSpec(ref common.ReferenceCallback
 							Format:      "",
 						},
 					},
-					"keeper": {
-						SchemaProps: spec.SchemaProps{
-							Description: "Name of the keeper, being the actual storage of the secure value. If not specified, the default keeper for the namespace will be used.",
-							MinLength:   ptr.To[int64](1),
-							MaxLength:   ptr.To[int64](253),
-							Type:        []string{"string"},
-							Format:      "",
-						},
-					},
 					"decrypters": {
 						VendorExtensible: spec.VendorExtensible{
 							Extensions: spec.Extensions{
@@ -639,6 +630,14 @@ func schema_pkg_apis_secret_v1beta1_SecureValueStatus(ref common.ReferenceCallba
 							Format:      "int64",
 						},
 					},
+					"externalID": {
+						SchemaProps: spec.SchemaProps{
+							Description: "External ID where the secret is stored. Cannot be set.",
+							Default:     "",
+							Type:        []string{"string"},
+							Format:      "",
+						},
+					},
 					"operatorStates": {
 						SchemaProps: spec.SchemaProps{
 							Description: "operatorStates is a map of operator ID to operator state evaluations. Any operator which consumes this kind SHOULD add its state evaluation information to this field.",
@@ -654,9 +653,9 @@ func schema_pkg_apis_secret_v1beta1_SecureValueStatus(ref common.ReferenceCallba
 							},
 						},
 					},
-					"externalID": {
+					"keeper": {
 						SchemaProps: spec.SchemaProps{
-							Description: "External ID where the secret is stored. Cannot be set.",
+							Description: "The name of the keeper used to create the secure value. Cannot be set.",
 							Default:     "",
 							Type:        []string{"string"},
 							Format:      "",
@@ -678,6 +677,7 @@ func schema_pkg_apis_secret_v1beta1_SecureValueStatus(ref common.ReferenceCallba
 						},
 					},
 				},
+				Required: []string{"keeper"},
 			},
 		},
 		Dependencies: []string{

pkg/registry/apis/secret/contracts/keeper.go

@@ -10,6 +10,9 @@ import (
 )
 
 var (
+	// The name used to refer to the system keeper
+	SystemKeeperName = "system"
+
 	ErrKeeperNotFound      = errors.New("keeper not found")
 	ErrKeeperAlreadyExists = errors.New("keeper already exists")
 )
@@ -21,7 +24,9 @@ type KeeperMetadataStorage interface {
 	Update(ctx context.Context, keeper *secretv1beta1.Keeper, actorUID string) (*secretv1beta1.Keeper, error)
 	Delete(ctx context.Context, namespace xkube.Namespace, name string) error
 	List(ctx context.Context, namespace xkube.Namespace) ([]secretv1beta1.Keeper, error)
-	GetKeeperConfig(ctx context.Context, namespace string, name *string, opts ReadOpts) (secretv1beta1.KeeperConfig, error)
+	GetKeeperConfig(ctx context.Context, namespace string, name string, opts ReadOpts) (secretv1beta1.KeeperConfig, error)
+	SetAsActive(ctx context.Context, namespace xkube.Namespace, name string) error
+	GetActiveKeeperConfig(ctx context.Context, namespace string) (string, secretv1beta1.KeeperConfig, error)
 }
 
 // ErrKeeperInvalidSecureValues is returned when a Keeper references SecureValues that do not exist.

pkg/registry/apis/secret/contracts/secure_value.go

@@ -31,7 +31,7 @@ type ReadOpts struct {
 
 // SecureValueMetadataStorage is the interface for wiring and dependency injection.
 type SecureValueMetadataStorage interface {
-	Create(ctx context.Context, sv *secretv1beta1.SecureValue, actorUID string) (*secretv1beta1.SecureValue, error)
+	Create(ctx context.Context, keeper string, sv *secretv1beta1.SecureValue, actorUID string) (*secretv1beta1.SecureValue, error)
 	Read(ctx context.Context, namespace xkube.Namespace, name string, opts ReadOpts) (*secretv1beta1.SecureValue, error)
 	List(ctx context.Context, namespace xkube.Namespace) ([]secretv1beta1.SecureValue, error)
 	SetVersionToActive(ctx context.Context, namespace xkube.Namespace, name string, version int64) error
@@ -47,6 +47,7 @@ type SecureValueService interface {
 	List(ctx context.Context, namespace xkube.Namespace) (*secretv1beta1.SecureValueList, error)
 	Update(ctx context.Context, newSecureValue *secretv1beta1.SecureValue, actorUID string) (*secretv1beta1.SecureValue, bool, error)
 	Delete(ctx context.Context, namespace xkube.Namespace, name string) (*secretv1beta1.SecureValue, error)
+	SetKeeperAsActive(ctx context.Context, namespace xkube.Namespace, keeperName string) error
 }
 
 type SecureValueClient interface {

pkg/registry/apis/secret/garbagecollectionworker/worker.go

@@ -93,14 +93,14 @@ func (w *Worker) CleanupInactiveSecureValues(ctx context.Context) ([]secretv1bet
 }
 
 func (w *Worker) Cleanup(ctx context.Context, sv *secretv1beta1.SecureValue) error {
-	keeperCfg, err := w.keeperMetadataStorage.GetKeeperConfig(ctx, sv.Namespace, sv.Spec.Keeper, contracts.ReadOpts{ForUpdate: false})
+	keeperCfg, err := w.keeperMetadataStorage.GetKeeperConfig(ctx, sv.Namespace, sv.Status.Keeper, contracts.ReadOpts{ForUpdate: false})
 	if err != nil {
-		return fmt.Errorf("fetching keeper config: namespace=%+v keeperName=%+v %w", sv.Namespace, sv.Spec.Keeper, err)
+		return fmt.Errorf("fetching keeper config: namespace=%+v keeperName=%+v %w", sv.Namespace, sv.Status.Keeper, err)
 	}
 
 	keeper, err := w.keeperService.KeeperForConfig(keeperCfg)
 	if err != nil {
-		return fmt.Errorf("getting keeper for config: namespace=%+v keeperName=%+v %w", sv.Namespace, sv.Spec.Keeper, err)
+		return fmt.Errorf("getting keeper for config: namespace=%+v keeperName=%+v %w", sv.Namespace, sv.Status.Keeper, err)
 	}
 
 	// Keeper deletion is idempotent

pkg/registry/apis/secret/garbagecollectionworker/worker_test.go

@@ -52,7 +52,7 @@ func TestBasic(t *testing.T) {
 		sv, err := sut.CreateSv(t.Context())
 		require.NoError(t, err)
 
-		keeperCfg, err := sut.KeeperMetadataStorage.GetKeeperConfig(t.Context(), sv.Namespace, sv.Spec.Keeper, contracts.ReadOpts{ForUpdate: false})
+		keeperCfg, err := sut.KeeperMetadataStorage.GetKeeperConfig(t.Context(), sv.Namespace, sv.Status.Keeper, contracts.ReadOpts{ForUpdate: false})
 		require.NoError(t, err)
 
 		keeper, err := sut.KeeperService.KeeperForConfig(keeperCfg)

pkg/registry/apis/secret/service/secure_value.go

@@ -93,7 +93,13 @@ func (s *SecureValueService) Create(ctx context.Context, sv *secretv1beta1.Secur
 		s.metrics.SecureValueCreateDuration.WithLabelValues(strconv.FormatBool(success)).Observe(time.Since(start).Seconds())
 	}()
 
-	return s.createNewVersion(ctx, sv, actorUID)
+	// Secure value creation uses the active keeper
+	keeperName, keeperCfg, err := s.keeperMetadataStorage.GetActiveKeeperConfig(ctx, sv.Namespace)
+	if err != nil {
+		return nil, fmt.Errorf("fetching active keeper config: namespace=%+v %w", sv.Namespace, err)
+	}
+
+	return s.createNewVersion(ctx, keeperName, keeperCfg, sv, actorUID)
 }
 
 func (s *SecureValueService) Update(ctx context.Context, newSecureValue *secretv1beta1.SecureValue, actorUID string) (_ *secretv1beta1.SecureValue, sync bool, updateErr error) {
@@ -128,23 +134,22 @@ func (s *SecureValueService) Update(ctx context.Context, newSecureValue *secretv
 		s.metrics.SecureValueUpdateDuration.WithLabelValues(strconv.FormatBool(success)).Observe(time.Since(start).Seconds())
 	}()
 
+	currentVersion, err := s.secureValueMetadataStorage.Read(ctx, xkube.Namespace(newSecureValue.Namespace), newSecureValue.Name, contracts.ReadOpts{})
+	if err != nil {
+		return nil, false, fmt.Errorf("reading secure value secret: %+w", err)
+	}
+
+	keeperCfg, err := s.keeperMetadataStorage.GetKeeperConfig(ctx, currentVersion.Namespace, currentVersion.Status.Keeper, contracts.ReadOpts{})
+	if err != nil {
+		return nil, false, fmt.Errorf("fetching keeper config: namespace=%+v keeper: %q %w", newSecureValue.Namespace, currentVersion.Status.Keeper, err)
+	}
+
 	if newSecureValue.Spec.Value == nil {
-		currentVersion, err := s.secureValueMetadataStorage.Read(ctx, xkube.Namespace(newSecureValue.Namespace), newSecureValue.Name, contracts.ReadOpts{})
-		if err != nil {
-			return nil, false, fmt.Errorf("reading secure value secret: %+w", err)
-		}
-
-		// TODO: does this need to be for update?
-		keeperCfg, err := s.keeperMetadataStorage.GetKeeperConfig(ctx, newSecureValue.Namespace, newSecureValue.Spec.Keeper, contracts.ReadOpts{ForUpdate: true})
-		if err != nil {
-			return nil, false, fmt.Errorf("fetching keeper config: namespace=%+v keeperName=%+v %w", newSecureValue.Namespace, newSecureValue.Spec.Keeper, err)
-		}
-
 		keeper, err := s.keeperService.KeeperForConfig(keeperCfg)
 		if err != nil {
-			return nil, false, fmt.Errorf("getting keeper for config: namespace=%+v keeperName=%+v %w", newSecureValue.Namespace, newSecureValue.Spec.Keeper, err)
+			return nil, false, fmt.Errorf("getting keeper for config: namespace=%+v keeperName=%+v %w", newSecureValue.Namespace, newSecureValue.Status.Keeper, err)
 		}
-		logging.FromContext(ctx).Debug("retrieved keeper", "namespace", newSecureValue.Namespace, "keeperName", newSecureValue.Spec.Keeper, "type", keeperCfg.Type())
+		logging.FromContext(ctx).Debug("retrieved keeper", "namespace", newSecureValue.Namespace, "type", keeperCfg.Type())
 
 		secret, err := keeper.Expose(ctx, keeperCfg, xkube.Namespace(newSecureValue.Namespace), newSecureValue.Name, currentVersion.Status.Version)
 		if err != nil {
@@ -154,12 +159,16 @@ func (s *SecureValueService) Update(ctx context.Context, newSecureValue *secretv
 		newSecureValue.Spec.Value = &secret
 	}
 
+	// Secure value updates use the keeper used to create the secure value
 	const updateIsSync = true
-	createdSv, err := s.createNewVersion(ctx, newSecureValue, actorUID)
+	createdSv, err := s.createNewVersion(ctx, currentVersion.Status.Keeper, keeperCfg, newSecureValue, actorUID)
 	return createdSv, updateIsSync, err
 }
 
-func (s *SecureValueService) createNewVersion(ctx context.Context, sv *secretv1beta1.SecureValue, actorUID string) (*secretv1beta1.SecureValue, error) {
+func (s *SecureValueService) createNewVersion(ctx context.Context, keeperName string, keeperCfg secretv1beta1.KeeperConfig, sv *secretv1beta1.SecureValue, actorUID string) (*secretv1beta1.SecureValue, error) {
+	if keeperName == "" {
+		return nil, fmt.Errorf("keeper name is required, got empty string")
+	}
 	if err := s.secureValueMutator.Mutate(sv, admission.Create); err != nil {
 		return nil, err
 	}
@@ -168,25 +177,21 @@ func (s *SecureValueService) createNewVersion(ctx context.Context, sv *secretv1b
 		return nil, contracts.NewErrValidateSecureValue(errorList)
 	}
 
-	createdSv, err := s.secureValueMetadataStorage.Create(ctx, sv, actorUID)
+	createdSv, err := s.secureValueMetadataStorage.Create(ctx, keeperName, sv, actorUID)
 	if err != nil {
 		return nil, fmt.Errorf("creating secure value: %w", err)
 	}
+
 	createdSv.Status = secretv1beta1.SecureValueStatus{
 		Version: createdSv.Status.Version,
-	}
-
-	// TODO: does this need to be for update?
-	keeperCfg, err := s.keeperMetadataStorage.GetKeeperConfig(ctx, createdSv.Namespace, createdSv.Spec.Keeper, contracts.ReadOpts{ForUpdate: true})
-	if err != nil {
-		return nil, fmt.Errorf("fetching keeper config: namespace=%+v keeperName=%+v %w", createdSv.Namespace, createdSv.Spec.Keeper, err)
+		Keeper:  keeperName,
 	}
 
 	keeper, err := s.keeperService.KeeperForConfig(keeperCfg)
 	if err != nil {
-		return nil, fmt.Errorf("getting keeper for config: namespace=%+v keeperName=%+v %w", createdSv.Namespace, createdSv.Spec.Keeper, err)
+		return nil, fmt.Errorf("getting keeper for config: namespace=%+v keeperName=%+v %w", createdSv.Namespace, keeperName, err)
 	}
-	logging.FromContext(ctx).Debug("retrieved keeper", "namespace", createdSv.Namespace, "keeperName", createdSv.Spec.Keeper, "type", keeperCfg.Type())
+	logging.FromContext(ctx).Debug("retrieved keeper", "namespace", createdSv.Namespace, "type", keeperCfg.Type())
 
 	// TODO: can we stop using external id?
 	// TODO: store uses only the namespace and returns and id. It could be a kv instead.
@@ -364,3 +369,10 @@ func (s *SecureValueService) Delete(ctx context.Context, namespace xkube.Namespa
 
 	return sv, nil
 }
+
+func (s *SecureValueService) SetKeeperAsActive(ctx context.Context, namespace xkube.Namespace, name string) error {
+	if err := s.keeperMetadataStorage.SetAsActive(ctx, namespace, name); err != nil {
+		return fmt.Errorf("calling keeper metadata storage to set keeper as active: %w", err)
+	}
+	return nil
+}

pkg/registry/apis/secret/validator/keeper.go

@@ -41,6 +41,10 @@ func (v *keeperValidator) Validate(keeper *secretv1beta1.Keeper, oldKeeper *secr
 		return errs
 	}
 
+	if keeper.Name == contracts.SystemKeeperName {
+		errs = append(errs, field.Forbidden(field.NewPath("name"), "the keeper name `system` is reserved"))
+	}
+
 	if keeper.Spec.Description == "" {
 		errs = append(errs, field.Required(field.NewPath("spec", "description"), "a `description` is required"))
 	}

pkg/registry/apis/secret/validator/keeper_test.go

@@ -35,36 +35,6 @@ func TestValidateKeeper(t *testing.T) {
 		})
 	})
 
-	t.Run("only one `keeper` must be present", func(t *testing.T) {
-		keeper := &secretv1beta1.Keeper{
-			ObjectMeta: objectMeta,
-			Spec: secretv1beta1.KeeperSpec{
-				Description:    "short description",
-				Aws:            &secretv1beta1.KeeperAWSConfig{},
-				Azure:          &secretv1beta1.KeeperAzureConfig{},
-				Gcp:            &secretv1beta1.KeeperGCPConfig{},
-				HashiCorpVault: &secretv1beta1.KeeperHashiCorpConfig{},
-			},
-		}
-
-		errs := validator.Validate(keeper, nil, admission.Create)
-		require.Len(t, errs, 1)
-		require.Equal(t, "spec", errs[0].Field)
-	})
-
-	t.Run("at least one `keeper` must be present", func(t *testing.T) {
-		keeper := &secretv1beta1.Keeper{
-			ObjectMeta: objectMeta,
-			Spec: secretv1beta1.KeeperSpec{
-				Description: "description",
-			},
-		}
-
-		errs := validator.Validate(keeper, nil, admission.Create)
-		require.Len(t, errs, 1)
-		require.Equal(t, "spec", errs[0].Field)
-	})
-
 	t.Run("aws keeper validation", func(t *testing.T) {
 		validKeeperAWS := &secretv1beta1.Keeper{
 			ObjectMeta: objectMeta,
@@ -341,4 +311,27 @@ func TestValidateKeeper(t *testing.T) {
 		require.Len(t, errs, 1)
 		require.Equal(t, "metadata.namespace", errs[0].Field)
 	})
+
+	t.Run("keeper name `system` is reserved", func(t *testing.T) {
+		keeper := &secretv1beta1.Keeper{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "system",
+				Namespace: "ns1",
+			},
+			Spec: secretv1beta1.KeeperSpec{
+				Description: "description",
+				HashiCorpVault: &secretv1beta1.KeeperHashiCorpConfig{
+					Address: "http://address",
+					Token: secretv1beta1.KeeperCredentialValue{
+						ValueFromConfig: "config.path.value",
+					},
+				},
+			},
+		}
+
+		errs := validator.Validate(keeper, nil, admission.Create)
+		require.Len(t, errs, 1)
+		require.Equal(t, "name", errs[0].Field)
+		require.Equal(t, "the keeper name `system` is reserved", errs[0].Detail)
+	})
 }

pkg/registry/apis/secret/validator/secure_value.go

@@ -110,11 +110,6 @@ func validateSecureValueUpdate(sv, oldSv *secretv1beta1.SecureValue) field.Error
 		}
 	}
 
-	// Keeper cannot be changed.
-	if sv.Spec.Keeper != oldSv.Spec.Keeper {
-		errs = append(errs, field.Forbidden(field.NewPath("spec"), "the `keeper` cannot be changed"))
-	}
-
 	return errs
 }
 

pkg/registry/apis/secret/validator/secure_value_test.go

@@ -25,9 +25,9 @@ func TestValidateSecureValue(t *testing.T) {
 			Spec: secretv1beta1.SecureValueSpec{
 				Description: "description",
 				Value:       ptr.To(secretv1beta1.NewExposedSecureValue("value")),
-				Keeper:      &keeper,
 				Decrypters:  []string{"app1", "app2"},
 			},
+			Status: secretv1beta1.SecureValueStatus{Keeper: keeper},
 		}
 
 		t.Run("the `description` must be present", func(t *testing.T) {
@@ -182,28 +182,6 @@ func TestValidateSecureValue(t *testing.T) {
 			require.Len(t, errs, 1)
 			require.Equal(t, "spec", errs[0].Field)
 		})
-
-		t.Run("when trying to change the `keeper`, it returns an error", func(t *testing.T) {
-			keeperA := "a-keeper"
-			keeperAnother := "another-keeper"
-			oldSv := &secretv1beta1.SecureValue{
-				ObjectMeta: objectMeta,
-				Spec: secretv1beta1.SecureValueSpec{
-					Keeper: &keeperA,
-				},
-			}
-
-			sv := &secretv1beta1.SecureValue{
-				ObjectMeta: objectMeta,
-				Spec: secretv1beta1.SecureValueSpec{
-					Keeper: &keeperAnother,
-				},
-			}
-
-			errs := validator.Validate(sv, oldSv, admission.Update)
-			require.Len(t, errs, 1)
-			require.Equal(t, "spec", errs[0].Field)
-		})
 	})
 
 	t.Run("`decrypters` must have unique items", func(t *testing.T) {

pkg/storage/secret/metadata/data/keeper_read_active.sql

@@ -0,0 +1,20 @@
+SELECT
+  {{ .Ident "guid" }},
+  {{ .Ident "name" }},
+  {{ .Ident "namespace" }},
+  {{ .Ident "annotations" }},
+  {{ .Ident "labels" }},
+  {{ .Ident "created" }},
+  {{ .Ident "created_by" }},
+  {{ .Ident "updated" }},
+  {{ .Ident "updated_by" }},
+  {{ .Ident "description" }},
+  {{ .Ident "type" }},
+  {{ .Ident "payload" }}
+FROM
+  {{ .Ident "secret_keeper" }}
+WHERE 
+  {{ .Ident "namespace" }} = {{ .Arg .Namespace }} AND
+  {{ .Ident "active" }} = true
+LIMIT 1
+;

pkg/storage/secret/metadata/data/keeper_set_as_active.sql

@@ -0,0 +1,5 @@
+UPDATE {{ .Ident "secret_keeper" }}
+SET {{ .Ident "active" }} = ({{ .Ident "name" }} = {{ .Arg .Name }})
+WHERE 
+  {{ .Ident "namespace" }} = {{ .Arg .Namespace }}
+;

pkg/storage/secret/metadata/decrypt_store.go

@@ -135,7 +135,7 @@ func (s *decryptStorage) Decrypt(ctx context.Context, namespace xkube.Namespace,
 		return "", fmt.Errorf("failed to authorize decryption with reason %v (%w)", reason, contracts.ErrDecryptNotAuthorized)
 	}
 
-	keeperConfig, err := s.keeperMetadataStorage.GetKeeperConfig(ctx, namespace.String(), sv.Spec.Keeper, contracts.ReadOpts{})
+	keeperConfig, err := s.keeperMetadataStorage.GetKeeperConfig(ctx, namespace.String(), sv.Status.Keeper, contracts.ReadOpts{})
 	if err != nil {
 		return "", fmt.Errorf("failed to read keeper config metadata storage: %v (%w)", err, contracts.ErrDecryptFailed)
 	}

pkg/storage/secret/metadata/keeper_model.go

@@ -280,3 +280,18 @@ func extractSecureValues(kp *secretv1beta1.Keeper) map[string]struct{} {
 
 	return nil
 }
+
+func getKeeperConfig(keeper *secretv1beta1.Keeper) secretv1beta1.KeeperConfig {
+	switch keeper.Spec.GetType() {
+	case secretv1beta1.AWSKeeperType:
+		return keeper.Spec.Aws
+	case secretv1beta1.AzureKeeperType:
+		return keeper.Spec.Azure
+	case secretv1beta1.GCPKeeperType:
+		return keeper.Spec.Gcp
+	case secretv1beta1.HashiCorpKeeperType:
+		return keeper.Spec.HashiCorpVault
+	default:
+		return nil
+	}
+}

pkg/storage/secret/metadata/keeper_store.go

@@ -2,6 +2,7 @@ package metadata
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strconv"
 	"time"
@@ -186,7 +187,7 @@ func (s *keeperMetadataStorage) read(ctx context.Context, namespace, name string
 	defer func() { _ = res.Close() }()
 
 	if !res.Next() {
-		return nil, contracts.ErrKeeperNotFound
+		return nil, fmt.Errorf("keeper=%s: %w", name, contracts.ErrKeeperNotFound)
 	}
 
 	var keeper keeperDB
@@ -568,9 +569,10 @@ func (s *keeperMetadataStorage) validateSecureValueReferences(ctx context.Contex
 	return nil
 }
 
-func (s *keeperMetadataStorage) GetKeeperConfig(ctx context.Context, namespace string, name *string, opts contracts.ReadOpts) (_ secretv1beta1.KeeperConfig, getErr error) {
+func (s *keeperMetadataStorage) GetKeeperConfig(ctx context.Context, namespace string, name string, opts contracts.ReadOpts) (_ secretv1beta1.KeeperConfig, getErr error) {
 	ctx, span := s.tracer.Start(ctx, "KeeperMetadataStorage.GetKeeperConfig", trace.WithAttributes(
 		attribute.String("namespace", namespace),
+		attribute.String("name", name),
 		attribute.Bool("isForUpdate", opts.ForUpdate),
 	))
 	start := time.Now()
@@ -581,6 +583,7 @@ func (s *keeperMetadataStorage) GetKeeperConfig(ctx context.Context, namespace s
 
 		args := []any{
 			"namespace", namespace,
+			"name", name,
 			"isForUpdate", strconv.FormatBool(opts.ForUpdate),
 		}
 
@@ -597,14 +600,12 @@ func (s *keeperMetadataStorage) GetKeeperConfig(ctx context.Context, namespace s
 	}()
 
 	// Check if keeper is the systemwide one.
-	if name == nil {
+	if name == contracts.SystemKeeperName {
 		return &secretv1beta1.SystemKeeperConfig{}, nil
 	}
 
-	span.SetAttributes(attribute.String("name", *name))
-
 	// Load keeper config from metadata store, or TODO: keeper cache.
-	kp, err := s.read(ctx, namespace, *name, opts)
+	kp, err := s.read(ctx, namespace, name, opts)
 	if err != nil {
 		return nil, err
 	}
@@ -614,3 +615,109 @@ func (s *keeperMetadataStorage) GetKeeperConfig(ctx context.Context, namespace s
 	// TODO: this would be a good place to check if credentials are secure values and load them.
 	return keeperConfig, nil
 }
+
+func (s *keeperMetadataStorage) SetAsActive(ctx context.Context, namespace xkube.Namespace, name string) error {
+	req := setKeeperAsActive{
+		SQLTemplate: sqltemplate.New(s.dialect),
+		Namespace:   namespace.String(),
+		Name:        name,
+	}
+
+	query, err := sqltemplate.Execute(sqlKeeperSetAsActive, req)
+	if err != nil {
+		return fmt.Errorf("template %q: %w", sqlKeeperSetAsActive.Name(), err)
+	}
+
+	// Check keeper exists. No need to worry about time of check to time of use
+	// since trying to activate a just deleted keeper will result in all
+	// keepers being inactive and defaulting to the system keeper.
+	if _, err := s.read(ctx, namespace.String(), name, contracts.ReadOpts{}); err != nil {
+		return fmt.Errorf("reading keeper before setting as active: %w", err)
+	}
+
+	_, err = s.db.ExecContext(ctx, query, req.GetArgs()...)
+	if err != nil {
+		return fmt.Errorf("setting keeper as active %q: %w", query, err)
+	}
+
+	return nil
+}
+
+func (s *keeperMetadataStorage) GetActiveKeeper(ctx context.Context, namespace string) (keeper *secretv1beta1.Keeper, readErr error) {
+	start := time.Now()
+	ctx, span := s.tracer.Start(ctx, "KeeperMetadataStorage.GetActiveKeeper", trace.WithAttributes(
+		attribute.String("namespace", namespace),
+	))
+	defer span.End()
+
+	defer func() {
+		success := readErr == nil
+
+		args := []any{
+			"namespace", namespace,
+		}
+
+		args = append(args, "success", success)
+		if !success {
+			span.SetStatus(codes.Error, "KeeperMetadataStorage.GetActiveKeeper failed")
+			span.RecordError(readErr)
+			args = append(args, "error", readErr)
+		}
+
+		logging.FromContext(ctx).Info("KeeperMetadataStorage.GetActiveKeeper", args...)
+
+		s.metrics.KeeperMetadataGetDuration.WithLabelValues(strconv.FormatBool(success)).Observe(time.Since(start).Seconds())
+	}()
+
+	req := &readActiveKeeper{
+		SQLTemplate: sqltemplate.New(s.dialect),
+		Namespace:   namespace,
+	}
+
+	query, err := sqltemplate.Execute(sqlKeeperReadActive, req)
+	if err != nil {
+		return nil, fmt.Errorf("execute template %q: %w", sqlKeeperReadActive.Name(), err)
+	}
+
+	res, err := s.db.QueryContext(ctx, query, req.GetArgs()...)
+	if err != nil {
+		return nil, fmt.Errorf("executing query to fetch active keeper in namespace %s: %w", namespace, err)
+	}
+	defer func() { _ = res.Close() }()
+
+	if !res.Next() {
+		return nil, contracts.ErrKeeperNotFound
+	}
+
+	var keeperDB keeperDB
+	err = res.Scan(
+		&keeperDB.GUID, &keeperDB.Name, &keeperDB.Namespace, &keeperDB.Annotations, &keeperDB.Labels, &keeperDB.Created,
+		&keeperDB.CreatedBy, &keeperDB.Updated, &keeperDB.UpdatedBy, &keeperDB.Description, &keeperDB.Type, &keeperDB.Payload,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to scan keeper row: %w", err)
+	}
+	if err := res.Err(); err != nil {
+		return nil, fmt.Errorf("read rows error: %w", err)
+	}
+
+	keeper, readErr = keeperDB.toKubernetes()
+	if readErr != nil {
+		return keeper, fmt.Errorf("converting from keeperDB to kubernetes struct: %w", err)
+	}
+
+	return keeper, nil
+}
+
+func (s *keeperMetadataStorage) GetActiveKeeperConfig(ctx context.Context, namespace string) (string, secretv1beta1.KeeperConfig, error) {
+	keeper, err := s.GetActiveKeeper(ctx, namespace)
+	if err != nil {
+		// When there are not active keepers, default to the system keeper
+		if errors.Is(err, contracts.ErrKeeperNotFound) {
+			return contracts.SystemKeeperName, &secretv1beta1.SystemKeeperConfig{}, nil
+		}
+		return "", nil, fmt.Errorf("fetching active keeper from db: %w", err)
+	}
+
+	return keeper.Name, getKeeperConfig(keeper), nil
+}

pkg/storage/secret/metadata/keeper_store_test.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/otel/trace/noop"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/ptr"
 
 	secretv1beta1 "github.com/grafana/grafana/apps/secret/pkg/apis/secret/v1beta1"
@@ -40,7 +41,7 @@ func Test_KeeperMetadataStorage_GetKeeperConfig(t *testing.T) {
 		keeperMetadataStorage := initStorage(t)
 
 		// get system keeper config
-		keeperConfig, err := keeperMetadataStorage.GetKeeperConfig(ctx, defaultKeeperNS, nil, contracts.ReadOpts{})
+		keeperConfig, err := keeperMetadataStorage.GetKeeperConfig(ctx, defaultKeeperNS, contracts.SystemKeeperName, contracts.ReadOpts{})
 		require.NoError(t, err)
 		require.IsType(t, &secretv1beta1.SystemKeeperConfig{}, keeperConfig)
 	})
@@ -55,7 +56,7 @@ func Test_KeeperMetadataStorage_GetKeeperConfig(t *testing.T) {
 		_, err := keeperMetadataStorage.Create(ctx, testKeeper, "testuser")
 		require.NoError(t, err)
 
-		keeperConfig, err := keeperMetadataStorage.GetKeeperConfig(ctx, defaultKeeperNS, &defaultKeeperName, contracts.ReadOpts{})
+		keeperConfig, err := keeperMetadataStorage.GetKeeperConfig(ctx, defaultKeeperNS, defaultKeeperName, contracts.ReadOpts{})
 		require.NoError(t, err)
 		require.NotNil(t, keeperConfig)
 		require.NotEmpty(t, keeperConfig.Type())
@@ -105,7 +106,7 @@ func Test_KeeperMetadataStorage_GetKeeperConfig(t *testing.T) {
 		require.NoError(t, err)
 
 		// we are able to get it
-		keeperConfig, err := keeperMetadataStorage.GetKeeperConfig(ctx, keeperNamespaceTest, &keeperTest, contracts.ReadOpts{})
+		keeperConfig, err := keeperMetadataStorage.GetKeeperConfig(ctx, keeperNamespaceTest, keeperTest, contracts.ReadOpts{})
 		require.NoError(t, err)
 		require.NotNil(t, keeperConfig)
 		require.NotEmpty(t, keeperConfig.Type())
@@ -115,7 +116,7 @@ func Test_KeeperMetadataStorage_GetKeeperConfig(t *testing.T) {
 		require.NoError(t, delErr)
 
 		// and we shouldn't be able to get it again
-		_, getErr := keeperMetadataStorage.GetKeeperConfig(ctx, keeperNamespaceTest, &keeperTest, contracts.ReadOpts{})
+		_, getErr := keeperMetadataStorage.GetKeeperConfig(ctx, keeperNamespaceTest, keeperTest, contracts.ReadOpts{})
 		require.Errorf(t, getErr, "keeper not found")
 	})
 
@@ -162,7 +163,7 @@ func Test_KeeperMetadataStorage_GetKeeperConfig(t *testing.T) {
 		require.NoError(t, err)
 
 		// Validate updated values
-		updatedConfig, err := keeperMetadataStorage.GetKeeperConfig(ctx, keeperNamespaceTest, &keeperTest, contracts.ReadOpts{})
+		updatedConfig, err := keeperMetadataStorage.GetKeeperConfig(ctx, keeperNamespaceTest, keeperTest, contracts.ReadOpts{})
 		require.NoError(t, err)
 		require.NotNil(t, updatedConfig)
 		require.NotEmpty(t, updatedConfig.Type())
@@ -260,7 +261,7 @@ func Test_KeeperMetadataStorage_GetKeeperConfig(t *testing.T) {
 
 		_, err := keeperMetadataStorage.Read(ctx, "ns", "non-existent", contracts.ReadOpts{})
 		require.Error(t, err)
-		require.Equal(t, contracts.ErrKeeperNotFound, err)
+		require.ErrorIs(t, err, contracts.ErrKeeperNotFound)
 	})
 
 	t.Run("update keeper with different namespace", func(t *testing.T) {
@@ -329,6 +330,40 @@ func Test_KeeperMetadataStorage_GetKeeperConfig(t *testing.T) {
 	})
 }
 
+func Test_KeeperMetadataStorage_SetAsActive(t *testing.T) {
+	t.Parallel()
+
+	keeperMetadataStorage := initStorage(t)
+
+	k1, err := keeperMetadataStorage.Create(t.Context(), &secretv1beta1.Keeper{
+		ObjectMeta: v1.ObjectMeta{Namespace: "ns1", Name: "k1"},
+		Spec: secretv1beta1.KeeperSpec{
+			Description: "description",
+			Aws:         &secretv1beta1.KeeperAWSConfig{},
+		},
+	}, "actor-uid")
+	require.NoError(t, err)
+
+	k2, err := keeperMetadataStorage.Create(t.Context(), &secretv1beta1.Keeper{
+		ObjectMeta: v1.ObjectMeta{Namespace: "ns1", Name: "k2"},
+		Spec: secretv1beta1.KeeperSpec{
+			Description: "description",
+			Aws:         &secretv1beta1.KeeperAWSConfig{},
+		},
+	}, "actor-uid")
+	require.NoError(t, err)
+
+	require.NoError(t, keeperMetadataStorage.SetAsActive(t.Context(), xkube.Namespace(k1.Namespace), k1.Name))
+	keeperName, _, err := keeperMetadataStorage.GetActiveKeeperConfig(t.Context(), k1.Namespace)
+	require.NoError(t, err)
+	require.Equal(t, k1.Name, keeperName)
+
+	require.NoError(t, keeperMetadataStorage.SetAsActive(t.Context(), xkube.Namespace(k2.Namespace), k2.Name))
+	keeperName, _, err = keeperMetadataStorage.GetActiveKeeperConfig(t.Context(), k2.Namespace)
+	require.NoError(t, err)
+	require.Equal(t, k2.Name, keeperName)
+}
+
 func initStorage(t *testing.T) contracts.KeeperMetadataStorage {
 	testDB := sqlstore.NewTestStore(t, sqlstore.WithMigrator(migrator.New()))
 	tracer := noop.NewTracerProvider().Tracer("test")

pkg/storage/secret/metadata/query.go

@@ -15,11 +15,13 @@ var (
 	sqlTemplates = template.Must(template.New("sql").ParseFS(sqlTemplatesFS, `data/*.sql`))
 
 	// The SQL Commands
-	sqlKeeperCreate = mustTemplate("keeper_create.sql")
-	sqlKeeperRead   = mustTemplate("keeper_read.sql")
-	sqlKeeperUpdate = mustTemplate("keeper_update.sql")
-	sqlKeeperList   = mustTemplate("keeper_list.sql")
-	sqlKeeperDelete = mustTemplate("keeper_delete.sql")
+	sqlKeeperCreate      = mustTemplate("keeper_create.sql")
+	sqlKeeperRead        = mustTemplate("keeper_read.sql")
+	sqlKeeperReadActive  = mustTemplate("keeper_read_active.sql")
+	sqlKeeperUpdate      = mustTemplate("keeper_update.sql")
+	sqlKeeperList        = mustTemplate("keeper_list.sql")
+	sqlKeeperDelete      = mustTemplate("keeper_delete.sql")
+	sqlKeeperSetAsActive = mustTemplate("keeper_set_as_active.sql")
 
 	sqlKeeperListByName      = mustTemplate("keeper_listByName.sql")
 	sqlSecureValueListByName = mustTemplate("secure_value_listByName.sql")
@@ -48,6 +50,18 @@ func mustTemplate(filename string) *template.Template {
 /**-- Keeper Queries --**/
 /************************/
 
+// Set as active
+type setKeeperAsActive struct {
+	sqltemplate.SQLTemplate
+	Namespace string
+	Name      string
+}
+
+// Validate is only used if we use `dbutil` from `unifiedstorage`
+func (r setKeeperAsActive) Validate() error {
+	return nil // TODO
+}
+
 // Create
 type createKeeper struct {
 	sqltemplate.SQLTemplate
@@ -72,6 +86,17 @@ func (r readKeeper) Validate() error {
 	return nil // TODO
 }
 
+// Read active keeper
+type readActiveKeeper struct {
+	sqltemplate.SQLTemplate
+	Namespace string
+}
+
+// Validate is only used if we use `dbutil` from `unifiedstorage`
+func (r readActiveKeeper) Validate() error {
+	return nil // TODO
+}
+
 // Update
 type updateKeeper struct {
 	sqltemplate.SQLTemplate

pkg/storage/secret/metadata/query_test.go

@@ -13,6 +13,15 @@ func TestKeeperQueries(t *testing.T) {
 	mocks.CheckQuerySnapshots(t, mocks.TemplateTestSetup{
 		RootDir: "testdata",
 		Templates: map[*template.Template][]mocks.TemplateTestCase{
+			sqlKeeperSetAsActive: {
+				{
+					Name: "keeper set as active",
+					Data: &setKeeperAsActive{
+						SQLTemplate: mocks.NewTestingSQLTemplate(), Name: "name",
+						Namespace: "ns",
+					},
+				},
+			},
 			sqlKeeperCreate: {
 				{
 					Name: "create",
@@ -54,6 +63,15 @@ func TestKeeperQueries(t *testing.T) {
 					},
 				},
 			},
+			sqlKeeperReadActive: {
+				{
+					Name: "read active",
+					Data: &readActiveKeeper{
+						SQLTemplate: mocks.NewTestingSQLTemplate(),
+						Namespace:   "ns",
+					},
+				},
+			},
 			sqlKeeperRead: {
 				{
 					Name: "read",

pkg/storage/secret/metadata/secure_value_model.go

@@ -79,7 +79,7 @@ func (sv *secureValueDB) toKubernetes() (*secretv1beta1.SecureValue, error) {
 	}
 
 	if sv.Keeper.Valid {
-		resource.Spec.Keeper = &sv.Keeper.String
+		resource.Status.Keeper = sv.Keeper.String
 	}
 	if sv.Ref.Valid {
 		resource.Spec.Ref = &sv.Ref.String
@@ -122,14 +122,13 @@ func (sv *secureValueDB) toKubernetes() (*secretv1beta1.SecureValue, error) {
 }
 
 // toCreateRow maps a Kubernetes resource into a DB row for new resources being created/inserted.
-func toCreateRow(now time.Time, sv *secretv1beta1.SecureValue, actorUID string) (*secureValueDB, error) {
-	row, err := toRow(sv, "")
+func toCreateRow(now time.Time, keeper string, sv *secretv1beta1.SecureValue, actorUID string) (*secureValueDB, error) {
+	row, err := toRow(keeper, sv, "")
 	if err != nil {
 		return nil, fmt.Errorf("failed to convert SecureValue to secureValueDB: %w", err)
 	}
 
 	timestamp := now.UTC().Unix()
-
 	row.GUID = uuid.New().String()
 	row.Created = timestamp
 	row.CreatedBy = actorUID
@@ -140,7 +139,7 @@ func toCreateRow(now time.Time, sv *secretv1beta1.SecureValue, actorUID string)
 }
 
 // toRow maps a Kubernetes resource into a DB row.
-func toRow(sv *secretv1beta1.SecureValue, externalID string) (*secureValueDB, error) {
+func toRow(keeper string, sv *secretv1beta1.SecureValue, externalID string) (*secureValueDB, error) {
 	var annotations string
 	if len(sv.Annotations) > 0 {
 		cleanedAnnotations := xkube.CleanAnnotations(sv.Annotations)
@@ -237,7 +236,7 @@ func toRow(sv *secretv1beta1.SecureValue, externalID string) (*secureValueDB, er
 		Version: sv.Status.Version,
 
 		Description: sv.Spec.Description,
-		Keeper:      toNullString(sv.Spec.Keeper),
+		Keeper:      toNullString(&keeper),
 		Decrypters:  toNullString(decrypters),
 		Ref:         toNullString(sv.Spec.Ref),
 		ExternalID:  externalID,

pkg/storage/secret/metadata/secure_value_store.go

@@ -47,13 +47,14 @@ type secureValueMetadataStorage struct {
 	tracer  trace.Tracer
 }
 
-func (s *secureValueMetadataStorage) Create(ctx context.Context, sv *secretv1beta1.SecureValue, actorUID string) (_ *secretv1beta1.SecureValue, svmCreateErr error) {
+func (s *secureValueMetadataStorage) Create(ctx context.Context, keeper string, sv *secretv1beta1.SecureValue, actorUID string) (_ *secretv1beta1.SecureValue, svmCreateErr error) {
 	start := s.clock.Now()
 	name := sv.GetName()
 	namespace := sv.GetNamespace()
 	ctx, span := s.tracer.Start(ctx, "SecureValueMetadataStorage.Create", trace.WithAttributes(
 		attribute.String("name", name),
 		attribute.String("namespace", namespace),
+		attribute.String("keeper", keeper),
 		attribute.String("actorUID", actorUID),
 	))
 	defer span.End()
@@ -64,6 +65,7 @@ func (s *secureValueMetadataStorage) Create(ctx context.Context, sv *secretv1bet
 		args := []any{
 			"name", name,
 			"namespace", namespace,
+			"keeper", keeper,
 			"actorUID", actorUID,
 		}
 
@@ -83,33 +85,6 @@ func (s *secureValueMetadataStorage) Create(ctx context.Context, sv *secretv1bet
 	var row *secureValueDB
 
 	err := s.db.Transaction(ctx, func(ctx context.Context) error {
-		if sv.Spec.Keeper != nil {
-			// Validate before inserting that the chosen `keeper` exists.
-
-			// -- This is a copy of KeeperMetadataStore.read, which is not public at the moment, and is not defined in contract.KeeperMetadataStorage
-			req := &readKeeper{
-				SQLTemplate: sqltemplate.New(s.dialect),
-				Namespace:   sv.Namespace,
-				Name:        *sv.Spec.Keeper,
-				IsForUpdate: true,
-			}
-
-			query, err := sqltemplate.Execute(sqlKeeperRead, req)
-			if err != nil {
-				return fmt.Errorf("execute template %q: %w", sqlKeeperRead.Name(), err)
-			}
-
-			res, err := s.db.QueryContext(ctx, query, req.GetArgs()...)
-			if err != nil {
-				return fmt.Errorf("getting row: %w", err)
-			}
-			defer func() { _ = res.Close() }()
-
-			if !res.Next() {
-				return contracts.ErrKeeperNotFound
-			}
-		}
-
 		latestVersion, err := s.getLatestVersion(ctx, xkube.Namespace(sv.Namespace), sv.Name)
 		if err != nil {
 			return fmt.Errorf("fetching latest secure value version: %w", err)
@@ -127,7 +102,7 @@ func (s *secureValueMetadataStorage) Create(ctx context.Context, sv *secretv1bet
 		for {
 			sv.Status.Version = version
 
-			row, err = toCreateRow(s.clock.Now(), sv, actorUID)
+			row, err = toCreateRow(s.clock.Now(), keeper, sv, actorUID)
 			if err != nil {
 				return fmt.Errorf("to create row: %w", err)
 			}

pkg/storage/secret/metadata/secure_value_store_test.go

@@ -63,20 +63,20 @@ func Test_SecureValueMetadataStorage_CreateAndRead(t *testing.T) {
 			Spec: secretv1beta1.SecureValueSpec{
 				Description: "test description",
 				Value:       ptr.To(secretv1beta1.NewExposedSecureValue("test-value")),
-				Keeper:      &keeperName,
 			},
+			Status: secretv1beta1.SecureValueStatus{Keeper: keeperName},
 		}
 		testSecureValue.Name = "sv-test"
 		testSecureValue.Namespace = "default"
 
 		// Create the secure value
-		createdSecureValue, err := secureValueStorage.Create(ctx, testSecureValue, "testuser")
+		createdSecureValue, err := secureValueStorage.Create(ctx, keeperName, testSecureValue, "testuser")
 		require.NoError(t, err)
 		require.NotNil(t, createdSecureValue)
 		require.Equal(t, "sv-test", createdSecureValue.Name)
 		require.Equal(t, "default", createdSecureValue.Namespace)
 		require.Equal(t, "test description", createdSecureValue.Spec.Description)
-		require.Equal(t, keeperName, *createdSecureValue.Spec.Keeper)
+		require.Equal(t, keeperName, createdSecureValue.Status.Keeper)
 
 		require.NoError(t, secureValueStorage.SetVersionToActive(ctx, xkube.Namespace(createdSecureValue.Namespace), createdSecureValue.Name, createdSecureValue.Status.Version))
 
@@ -87,7 +87,7 @@ func Test_SecureValueMetadataStorage_CreateAndRead(t *testing.T) {
 		require.Equal(t, "sv-test", readSecureValue.Name)
 		require.Equal(t, "default", readSecureValue.Namespace)
 		require.Equal(t, "test description", readSecureValue.Spec.Description)
-		require.Equal(t, keeperName, *readSecureValue.Spec.Keeper)
+		require.Equal(t, keeperName, readSecureValue.Status.Keeper)
 
 		// List secure values and verify our value is in the list
 		secureValues, err := secureValueStorage.List(ctx, xkube.Namespace("default"))
@@ -101,7 +101,7 @@ func Test_SecureValueMetadataStorage_CreateAndRead(t *testing.T) {
 				found = true
 				require.Equal(t, "default", sv.Namespace)
 				require.Equal(t, "test description", sv.Spec.Description)
-				require.Equal(t, keeperName, *sv.Spec.Keeper)
+				require.Equal(t, keeperName, sv.Status.Keeper)
 				break
 			}
 		}
@@ -117,14 +117,14 @@ func Test_SecureValueMetadataStorage_CreateAndRead(t *testing.T) {
 			Spec: secretv1beta1.SecureValueSpec{
 				Description: "test description 2",
 				Value:       ptr.To(secretv1beta1.NewExposedSecureValue("test-value-2")),
-				Keeper:      &keeperName,
 			},
+			Status: secretv1beta1.SecureValueStatus{Keeper: keeperName},
 		}
 		testSecureValue.Name = "sv-test-2"
 		testSecureValue.Namespace = "default"
 
 		// Create the secure value
-		createdSecureValue, err := secureValueStorage.Create(ctx, testSecureValue, "testuser")
+		createdSecureValue, err := secureValueStorage.Create(ctx, keeperName, testSecureValue, "testuser")
 		require.NoError(t, err)
 		require.NotNil(t, createdSecureValue)
 
@@ -211,7 +211,7 @@ func TestPropertySecureValueMetadataStorage(t *testing.T) {
 			},
 			"delete": func(t *rapid.T) {
 				ns := namespaceGen.Draw(t, "ns")
-				name := nameGen.Draw(t, "name")
+				name := secureValueNameGen.Draw(t, "name")
 				modelSv, modelErr := model.delete(ns, name)
 				sv, err := sut.DeleteSv(t.Context(), ns, name)
 				if err != nil || modelErr != nil {

pkg/storage/secret/metadata/secure_value_test.go

@@ -26,9 +26,16 @@ type modelSecureValue struct {
 	leaseCreated time.Time
 }
 
+type modelKeeper struct {
+	namespace string
+	name      string
+	active    bool
+}
+
 // A simplified model of the grafana secrets manager
 type model struct {
 	secureValues []*modelSecureValue
+	keepers      []*modelKeeper
 }
 
 func newModel() *model {
@@ -73,24 +80,91 @@ func (m *model) readActiveVersion(namespace, name string) *modelSecureValue {
 }
 
 func (m *model) create(now time.Time, sv *secretv1beta1.SecureValue) (*secretv1beta1.SecureValue, error) {
+	keeper := m.getActiveKeeper(sv.Namespace)
+	sv = deepCopy(sv)
 	modelSv := &modelSecureValue{SecureValue: sv, active: false, created: now}
 	modelSv.Status.Version = m.getNewVersionNumber(modelSv.Namespace, modelSv.Name)
 	modelSv.Status.ExternalID = fmt.Sprintf("%d", modelSv.Status.Version)
+	modelSv.Status.Keeper = keeper.name
 	m.secureValues = append(m.secureValues, modelSv)
 	m.setVersionToActive(modelSv.Namespace, modelSv.Name, modelSv.Status.Version)
 	return modelSv.SecureValue, nil
 }
 
+func (m *model) getActiveKeeper(namespace string) *modelKeeper {
+	for _, k := range m.keepers {
+		if k.namespace == namespace && k.active {
+			return k
+		}
+	}
+
+	// Default to the system keeper when there are no active keepers in the namespace
+	return &modelKeeper{namespace: namespace, name: contracts.SystemKeeperName, active: true}
+}
+
+func (m *model) keeperExists(namespace, name string) bool {
+	return m.findKeeper(namespace, name) != nil
+}
+
+func (m *model) findKeeper(namespace, name string) *modelKeeper {
+	// The system keeper is not in the list of keepers
+	if name == contracts.SystemKeeperName {
+		return &modelKeeper{namespace: namespace, name: contracts.SystemKeeperName, active: true}
+	}
+	for _, k := range m.keepers {
+		if k.namespace == namespace && k.name == name {
+			return k
+		}
+	}
+	return nil
+}
+
+func (m *model) createKeeper(keeper *secretv1beta1.Keeper) (*secretv1beta1.Keeper, error) {
+	if m.keeperExists(keeper.Namespace, keeper.Name) {
+		return nil, contracts.ErrKeeperAlreadyExists
+	}
+
+	m.keepers = append(m.keepers, &modelKeeper{namespace: keeper.Namespace, name: keeper.Name})
+
+	return deepCopy(keeper), nil
+}
+
+func (m *model) setKeeperAsActive(namespace, keeperName string) error {
+	keeper := m.findKeeper(namespace, keeperName)
+	if keeper == nil {
+		return contracts.ErrKeeperNotFound
+	}
+	// Set the keeper as active
+	keeper.active = true
+
+	// Set every other keeper in the namespace as inactive
+	for _, k := range m.keepers {
+		if k.namespace == namespace && k.name != keeperName {
+			k.active = false
+		}
+	}
+
+	return nil
+}
+
 func (m *model) update(now time.Time, newSecureValue *secretv1beta1.SecureValue) (*secretv1beta1.SecureValue, bool, error) {
+	sv := m.readActiveVersion(newSecureValue.Namespace, newSecureValue.Name)
+	if sv == nil {
+		return nil, false, contracts.ErrSecureValueNotFound
+	}
+
+	// If the keeper doesn't exist, return an error
+	if !m.keeperExists(sv.Namespace, sv.Status.Keeper) {
+		return nil, false, contracts.ErrKeeperNotFound
+	}
+
 	// If the payload doesn't contain a value, get the value from current version
 	if newSecureValue.Spec.Value == nil {
-		sv := m.readActiveVersion(newSecureValue.Namespace, newSecureValue.Name)
-		if sv == nil {
-			return nil, false, contracts.ErrSecureValueNotFound
-		}
 		newSecureValue.Spec.Value = sv.Spec.Value
 	}
+
 	createdSv, err := m.create(now, newSecureValue)
+
 	return createdSv, true, err
 }
 
@@ -161,13 +235,14 @@ func (m *model) leaseInactiveSecureValues(now time.Time, minAge, leaseTTL time.D
 }
 
 var (
-	decryptersGen     = rapid.SampledFrom([]string{"svc1", "svc2", "svc3", "svc4", "svc5"})
-	nameGen           = rapid.SampledFrom([]string{"n1", "n2", "n3", "n4", "n5"})
-	namespaceGen      = rapid.SampledFrom([]string{"ns1", "ns2", "ns3", "ns4", "ns5"})
-	anySecureValueGen = rapid.Custom(func(t *rapid.T) *secretv1beta1.SecureValue {
+	decryptersGen      = rapid.SampledFrom([]string{"svc1", "svc2", "svc3", "svc4", "svc5"})
+	secureValueNameGen = rapid.SampledFrom([]string{"n1", "n2", "n3", "n4", "n5"})
+	keeperNameGen      = rapid.SampledFrom([]string{"k1", "k2", "k3", "k4", "k5"})
+	namespaceGen       = rapid.SampledFrom([]string{"ns1", "ns2", "ns3", "ns4", "ns5"})
+	anySecureValueGen  = rapid.Custom(func(t *rapid.T) *secretv1beta1.SecureValue {
 		return &secretv1beta1.SecureValue{
 			ObjectMeta: metav1.ObjectMeta{
-				Name:      nameGen.Draw(t, "name"),
+				Name:      secureValueNameGen.Draw(t, "name"),
 				Namespace: namespaceGen.Draw(t, "ns"),
 			},
 			Spec: secretv1beta1.SecureValueSpec{
@@ -191,10 +266,37 @@ var (
 	decryptGen           = rapid.Custom(func(t *rapid.T) decryptInput {
 		return decryptInput{
 			namespace: namespaceGen.Draw(t, "ns"),
-			name:      nameGen.Draw(t, "name"),
+			name:      secureValueNameGen.Draw(t, "name"),
 			decrypter: decryptersGen.Draw(t, "decrypter"),
 		}
 	})
+	anyKeeperGen = rapid.Custom(func(t *rapid.T) *secretv1beta1.Keeper {
+		spec := secretv1beta1.KeeperSpec{
+			Description: rapid.String().Draw(t, "description"),
+		}
+
+		keeperType := rapid.SampledFrom([]string{"isAwsKeeper", "isAzureKeeper", "isGcpKeeper", "isVaultKeeper"}).Draw(t, "keeperType")
+		switch keeperType {
+		case "isAwsKeeper":
+			spec.Aws = &secretv1beta1.KeeperAWSConfig{}
+		case "isAzureKeeper":
+			spec.Azure = &secretv1beta1.KeeperAzureConfig{}
+		case "isGcpKeeper":
+			spec.Gcp = &secretv1beta1.KeeperGCPConfig{}
+		case "isVaultKeeper":
+			spec.HashiCorpVault = &secretv1beta1.KeeperHashiCorpConfig{}
+		default:
+			panic(fmt.Sprintf("unhandled keeper type '%+v', did you forget a switch case?", keeperType))
+		}
+
+		return &secretv1beta1.Keeper{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      keeperNameGen.Draw(t, "name"),
+				Namespace: namespaceGen.Draw(t, "ns"),
+			},
+			Spec: spec,
+		}
+	})
 )
 
 type decryptInput struct {
@@ -268,9 +370,8 @@ func TestModel(t *testing.T) {
 		sv4 := deepCopy(sv3)
 		sv4.Name = "i_dont_exist"
 		sv4.Spec.Value = ptr.To(secretv1beta1.NewExposedSecureValue("sv4"))
-		sv4, _, err = m.update(now, sv4)
-		require.NoError(t, err)
-		require.EqualValues(t, 1, sv4.Status.Version)
+		_, _, err = m.update(now, sv4)
+		require.ErrorIs(t, err, contracts.ErrSecureValueNotFound)
 	})
 
 	t.Run("deleting a secure value", func(t *testing.T) {
@@ -359,7 +460,6 @@ func TestStateMachine(t *testing.T) {
 				sv := anySecureValueGen.Draw(t, "sv")
 
 				modelCreatedSv, modelErr := model.create(sut.Clock.Now(), deepCopy(sv))
-
 				createdSv, err := sut.CreateSv(t.Context(), testutils.CreateSvWithSv(deepCopy(sv)))
 				if err != nil || modelErr != nil {
 					require.ErrorIs(t, err, modelErr)
@@ -412,7 +512,8 @@ func TestStateMachine(t *testing.T) {
 					if !slices.ContainsFunc(list.Items, func(v2 secretv1beta1.SecureValue) bool {
 						return v2.Namespace == v1.Namespace && v2.Name == v1.Name && v2.Status.Version == v1.Status.Version
 					}) {
-						t.Fatalf("expected sut to return secure value ns=%+v name=%+v version=%+v in the result", v1.Namespace, v1.Name, v1.Status.Version)
+						t.Fatalf("expected sut to return secure value ns=%+v name=%+v version=%+v in the result",
+							v1.Namespace, v1.Name, v1.Status.Version)
 					}
 				}
 			},
@@ -439,8 +540,28 @@ func TestStateMachine(t *testing.T) {
 
 				require.Equal(t, len(modelResult), len(result))
 				for name := range modelResult {
-					require.Equal(t, modelResult[name].Value(), result[name].Value())
 					require.Equal(t, modelResult[name].Error(), result[name].Error())
+					require.Equal(t, modelResult[name].Value(), result[name].Value())
+				}
+			},
+			"createKeeper": func(t *rapid.T) {
+				input := anyKeeperGen.Draw(t, "keeper")
+				modelKeeper, modelErr := model.createKeeper(input)
+				keeper, err := sut.KeeperMetadataStorage.Create(t.Context(), input, "actor-uid")
+				if err != nil || modelErr != nil {
+					require.ErrorIs(t, err, modelErr)
+					return
+				}
+				require.Equal(t, modelKeeper.Name, keeper.Name)
+			},
+			"setKeeperAsActive": func(t *rapid.T) {
+				namespace := namespaceGen.Draw(t, "namespace")
+				keeper := keeperNameGen.Draw(t, "keeper")
+				modelErr := model.setKeeperAsActive(namespace, keeper)
+				err := sut.KeeperMetadataStorage.SetAsActive(t.Context(), xkube.Namespace(namespace), keeper)
+				if err != nil || modelErr != nil {
+					require.ErrorIs(t, err, modelErr)
+					return
 				}
 			},
 		})
@@ -471,6 +592,69 @@ func TestSecureValueServiceExampleBased(t *testing.T) {
 		require.Equal(t, 1, len(result))
 		require.ErrorIs(t, result[sv.Name].Error(), contracts.ErrDecryptNotFound)
 	})
+
+	t.Run("should be able to use secrets that were created with a keeper that's inactive", func(t *testing.T) {
+		t.Parallel()
+
+		sut := testutils.Setup(t)
+
+		// - Create a secret with k1
+		k1, err := sut.KeeperMetadataStorage.Create(t.Context(), &secretv1beta1.Keeper{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: "n1",
+				Name:      "k1",
+			},
+			Spec: secretv1beta1.KeeperSpec{
+				Description: "description",
+				Aws:         &secretv1beta1.KeeperAWSConfig{},
+			},
+		}, "actor-uid")
+		require.NoError(t, err)
+
+		require.NoError(t, sut.SecureValueService.SetKeeperAsActive(t.Context(), xkube.Namespace(k1.Namespace), k1.Name))
+
+		value := secretv1beta1.NewExposedSecureValue("v1")
+		sv1, err := sut.CreateSv(t.Context(), testutils.CreateSvWithSv(&secretv1beta1.SecureValue{
+			ObjectMeta: metav1.ObjectMeta{Namespace: k1.Namespace, Name: "s1"},
+			Spec: secretv1beta1.SecureValueSpec{
+				Description: "desc",
+				Value:       &value,
+			},
+		}))
+		require.NoError(t, err)
+		require.Equal(t, k1.Name, sv1.Status.Keeper)
+
+		// - Set a new keeper as active
+		k2, err := sut.KeeperMetadataStorage.Create(t.Context(), &secretv1beta1.Keeper{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: "n1",
+				Name:      "k2",
+			},
+			Spec: secretv1beta1.KeeperSpec{
+				Description: "description",
+				Aws:         &secretv1beta1.KeeperAWSConfig{},
+			},
+		}, "actor-uid")
+		require.NoError(t, err)
+		require.NoError(t, sut.SecureValueService.SetKeeperAsActive(t.Context(), xkube.Namespace(k2.Namespace), k2.Name))
+
+		// - Read secure value created with inactive keeper
+		readSv, err := sut.SecureValueService.Read(t.Context(), xkube.Namespace(sv1.Namespace), sv1.Name)
+		require.NoError(t, err)
+		require.Equal(t, sv1.Namespace, readSv.Namespace)
+		require.Equal(t, sv1.Name, readSv.Name)
+		require.Equal(t, k1.Name, readSv.Status.Keeper)
+
+		// - Update secure value created with inactive keeper
+		newSv1 := sv1.DeepCopy()
+		newSv1.Spec.Description = "updated desc"
+		updatedSv, _, err := sut.SecureValueService.Update(t.Context(), newSv1, "actor-uid")
+		require.NoError(t, err)
+		require.Equal(t, sv1.Namespace, updatedSv.Namespace)
+		require.Equal(t, sv1.Name, updatedSv.Name)
+		require.Equal(t, k1.Name, updatedSv.Status.Keeper)
+		require.Equal(t, newSv1.Spec.Description, updatedSv.Spec.Description)
+	})
 }
 
 func deepCopy[T any](sv T) T {

pkg/storage/secret/metadata/testdata/mysql--keeper_read_active-read active.sql

@@ -0,0 +1,20 @@
+SELECT
+  `guid`,
+  `name`,
+  `namespace`,
+  `annotations`,
+  `labels`,
+  `created`,
+  `created_by`,
+  `updated`,
+  `updated_by`,
+  `description`,
+  `type`,
+  `payload`
+FROM
+  `secret_keeper`
+WHERE 
+  `namespace` = 'ns' AND
+  `active` = true
+LIMIT 1
+;

pkg/storage/secret/metadata/testdata/mysql--keeper_set_as_active-keeper set as active.sql

@@ -0,0 +1,5 @@
+UPDATE `secret_keeper`
+SET `active` = (`name` = 'name')
+WHERE 
+  `namespace` = 'ns'
+;

pkg/storage/secret/metadata/testdata/postgres--keeper_read_active-read active.sql

@@ -0,0 +1,20 @@
+SELECT
+  "guid",
+  "name",
+  "namespace",
+  "annotations",
+  "labels",
+  "created",
+  "created_by",
+  "updated",
+  "updated_by",
+  "description",
+  "type",
+  "payload"
+FROM
+  "secret_keeper"
+WHERE 
+  "namespace" = 'ns' AND
+  "active" = true
+LIMIT 1
+;

pkg/storage/secret/metadata/testdata/postgres--keeper_set_as_active-keeper set as active.sql

@@ -0,0 +1,5 @@
+UPDATE "secret_keeper"
+SET "active" = ("name" = 'name')
+WHERE 
+  "namespace" = 'ns'
+;

pkg/storage/secret/metadata/testdata/sqlite--keeper_read_active-read active.sql

@@ -0,0 +1,20 @@
+SELECT
+  "guid",
+  "name",
+  "namespace",
+  "annotations",
+  "labels",
+  "created",
+  "created_by",
+  "updated",
+  "updated_by",
+  "description",
+  "type",
+  "payload"
+FROM
+  "secret_keeper"
+WHERE 
+  "namespace" = 'ns' AND
+  "active" = true
+LIMIT 1
+;

pkg/storage/secret/metadata/testdata/sqlite--keeper_set_as_active-keeper set as active.sql

@@ -0,0 +1,5 @@
+UPDATE "secret_keeper"
+SET "active" = ("name" = 'name')
+WHERE 
+  "namespace" = 'ns'
+;

pkg/storage/secret/migrator/migrator.go

@@ -77,7 +77,7 @@ func (*SecretDB) AddMigration(mg *migrator.Migrator) {
 	}
 	tables = append(tables, secureValueTable)
 
-	tables = append(tables, migrator.Table{
+	keeperTable := migrator.Table{
 		Name: TableNameKeeper,
 		Columns: []*migrator.Column{
 			// Kubernetes Metadata
@@ -100,7 +100,8 @@ func (*SecretDB) AddMigration(mg *migrator.Migrator) {
 		Indices: []*migrator.Index{
 			{Cols: []string{"namespace", "name"}, Type: migrator.UniqueIndex},
 		},
-	})
+	}
+	tables = append(tables, keeperTable)
 
 	dataKeyTable := migrator.Table{
 		Name: TableNameDataKey,
@@ -211,4 +212,16 @@ func (*SecretDB) AddMigration(mg *migrator.Migrator) {
 	mg.AddMigration("add data_key_id index to "+TableNameEncryptedValue, migrator.NewAddIndexMigration(encryptedValueTable, &migrator.Index{
 		Cols: []string{"data_key_id"},
 	}))
+	mg.AddMigration("add active column to "+TableNameKeeper, migrator.NewAddColumnMigration(keeperTable, &migrator.Column{
+		Name:     "active",
+		Type:     migrator.DB_Bool,
+		Nullable: false,
+		Default:  "false",
+	}))
+	mg.AddMigration("add active column index to "+TableNameKeeper, migrator.NewAddIndexMigration(keeperTable, &migrator.Index{
+		Cols: []string{"namespace", "name", "active"},
+	}))
+	mg.AddMigration("set secret_secure_value.keeper to 'system' where keeper is null in "+TableNameSecureValue, migrator.NewRawSQLMigration(
+		fmt.Sprintf("UPDATE %s SET keeper = '%s' WHERE keeper IS NULL", TableNameSecureValue, contracts.SystemKeeperName),
+	))
 }