add database user reconciler logic (#3034)

Author: s-urbaniak

config/openapi2crd.yaml

@@ -211,5 +211,7 @@ spec:
             schema: 'CloudDatabaseUser'
             filters:
               readWriteOnly: true
+              skipProperties:
+                - $.groupId
               sensitiveProperties:
                 - $.password

internal/controller/registry.go

@@ -48,6 +48,7 @@ import (
 	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/featureflags"
 	akov2generatedcluster "github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/generated/controller/cluster"
 	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/generated/controller/connectionsecret"
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/generated/controller/databaseuser"
 	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/generated/controller/flexcluster"
 	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/generated/controller/group"
 	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/pointer"
@@ -161,10 +162,16 @@ func (r *Registry) registerControllers(c cluster.Cluster, ap atlas.Provider) err
 			return fmt.Errorf("error creating flex cluster reconciler: %w", err)
 		}
 
+		databaseUserReconciler, err := databaseuser.NewDatabaseUserReconciler(c, ap, r.logger, r.globalSecretRef, r.deletionProtection, true, r.defaultPredicates())
+		if err != nil {
+			return fmt.Errorf("error creating database user reconciler: %w", err)
+		}
+
 		reconcilers = append(reconcilers,
 			newCtrlStateReconciler(groupReconciler, r.maxConcurrentReconciles),
 			newCtrlStateReconciler(clusterController, r.maxConcurrentReconciles),
 			newCtrlStateReconciler(flexController, r.maxConcurrentReconciles),
+			newCtrlStateReconciler(databaseUserReconciler, r.maxConcurrentReconciles),
 		)
 	}
 

internal/generated/controller/databaseuser/handler_v20250312.go

@@ -16,8 +16,11 @@ package databaseuser
 
 import (
 	"context"
+	"errors"
+	"fmt"
 
 	v20250312sdk "go.mongodb.org/atlas-sdk/v20250312009/admin"
+	v1 "k8s.io/api/core/v1"
 	controllerruntime "sigs.k8s.io/controller-runtime"
 	builder "sigs.k8s.io/controller-runtime/pkg/builder"
 	client "sigs.k8s.io/controller-runtime/pkg/client"
@@ -49,65 +52,196 @@ func NewHandlerv20250312(kubeClient client.Client, atlasClient *v20250312sdk.API
 
 // HandleInitial handles the initial state for version v20250312
 func (h *Handlerv20250312) HandleInitial(ctx context.Context, databaseuser *akov2generated.DatabaseUser) (ctrlstate.Result, error) {
-	// TODO: Implement initial state logic
-	// TODO: Use h.atlasProvider.SdkClientSet(ctx, h.globalSecretRef, h.log) to get Atlas SDK client
-	return result.NextState(state.StateUpdated, "Updated AtlasDatabaseUser.")
+	deps, err := h.getDependencies(ctx, databaseuser)
+	if err != nil {
+		return result.Error(state.StateInitial, fmt.Errorf("Failed to get dependencies: %w", err))
+	}
+
+	body := &v20250312sdk.CloudDatabaseUser{}
+	params := &v20250312sdk.CreateDatabaseUserApiParams{
+		CloudDatabaseUser: body,
+	}
+
+	if err := h.translator.ToAPI(params, databaseuser, deps...); err != nil {
+		return result.Error(state.StateInitial, fmt.Errorf("failed to translate params: %w", err))
+	}
+
+	if err := h.translator.ToAPI(body, databaseuser, deps...); err != nil {
+		return result.Error(state.StateInitial, fmt.Errorf("failed to translate body: %w", err))
+	}
+
+	_, _, err = h.atlasClient.DatabaseUsersApi.CreateDatabaseUserWithParams(ctx, params).Execute()
+	if err != nil {
+		return result.Error(state.StateInitial, fmt.Errorf("failed to create datebaseuser: %w", err))
+	}
+
+	if err := ctrlstate.NewPatcher(databaseuser).UpdateStateTracker(deps...).Patch(ctx, h.kubeClient); err != nil {
+		return result.Error(state.StateCreated, fmt.Errorf("failed to update state tracker: %w", err))
+	}
+
+	return result.NextState(state.StateCreated, "Created Database User.")
 }
 
 // HandleImportRequested handles the importrequested state for version v20250312
 func (h *Handlerv20250312) HandleImportRequested(ctx context.Context, databaseuser *akov2generated.DatabaseUser) (ctrlstate.Result, error) {
-	// TODO: Implement importrequested state logic
-	// TODO: Use h.atlasProvider.SdkClientSet(ctx, h.globalSecretRef, h.log) to get Atlas SDK client
+	deps, err := h.getDependencies(ctx, databaseuser)
+	if err != nil {
+		return result.Error(state.StateInitial, fmt.Errorf("Failed to get dependencies: %w", err))
+	}
+
+	groupId, ok := databaseuser.GetAnnotations()["mongodb.com/external-group-id"]
+	if !ok {
+		return result.Error(state.StateImportRequested, errors.New("missing annotation mongodb.com/external-id"))
+	}
+
+	databaseName, ok := databaseuser.GetAnnotations()["mongodb.com/external-database-name"]
+	if !ok {
+		return result.Error(state.StateImportRequested, errors.New("missing annotation mongodb.com/external-database-name"))
+	}
+
+	username, ok := databaseuser.GetAnnotations()["mongodb.com/external-username"]
+	if !ok {
+		return result.Error(state.StateImportRequested, errors.New("missing annotation mongodb.com/external-username"))
+	}
+
+	params := &v20250312sdk.GetDatabaseUserApiParams{
+		GroupId:      groupId,
+		DatabaseName: databaseName,
+		Username:     username,
+	}
+	_, _, err = h.atlasClient.DatabaseUsersApi.GetDatabaseUserWithParams(ctx, params).Execute()
+	if err != nil {
+		return result.Error(state.StateInitial, fmt.Errorf("failed to create datebaseuser: %w", err))
+	}
+
+	if err := ctrlstate.NewPatcher(databaseuser).UpdateStateTracker(deps...).Patch(ctx, h.kubeClient); err != nil {
+		return result.Error(state.StateImported, fmt.Errorf("failed to update state tracker: %w", err))
+	}
+
 	return result.NextState(state.StateImported, "Import completed")
 }
 
 // HandleImported handles the imported state for version v20250312
 func (h *Handlerv20250312) HandleImported(ctx context.Context, databaseuser *akov2generated.DatabaseUser) (ctrlstate.Result, error) {
-	// TODO: Implement imported state logic
-	// TODO: Use h.atlasProvider.SdkClientSet(ctx, h.globalSecretRef, h.log) to get Atlas SDK client
-	return result.NextState(state.StateUpdated, "Ready")
+	return h.handleIdle(ctx, state.StateImported, databaseuser)
 }
 
 // HandleCreating handles the creating state for version v20250312
 func (h *Handlerv20250312) HandleCreating(ctx context.Context, databaseuser *akov2generated.DatabaseUser) (ctrlstate.Result, error) {
-	// TODO: Implement creating state logic
-	// TODO: Use h.atlasProvider.SdkClientSet(ctx, h.globalSecretRef, h.log) to get Atlas SDK client
-	return result.NextState(state.StateCreated, "Resource created")
+	panic("unsupported state")
 }
 
 // HandleCreated handles the created state for version v20250312
 func (h *Handlerv20250312) HandleCreated(ctx context.Context, databaseuser *akov2generated.DatabaseUser) (ctrlstate.Result, error) {
-	// TODO: Implement created state logic
-	// TODO: Use h.atlasProvider.SdkClientSet(ctx, h.globalSecretRef, h.log) to get Atlas SDK client
-	return result.NextState(state.StateUpdated, "Ready")
+	return h.handleIdle(ctx, state.StateCreated, databaseuser)
 }
 
 // HandleUpdating handles the updating state for version v20250312
 func (h *Handlerv20250312) HandleUpdating(ctx context.Context, databaseuser *akov2generated.DatabaseUser) (ctrlstate.Result, error) {
-	// TODO: Implement updating state logic
-	// TODO: Use h.atlasProvider.SdkClientSet(ctx, h.globalSecretRef, h.log) to get Atlas SDK client
-	return result.NextState(state.StateUpdated, "Update completed")
+	panic("unsupported state")
 }
 
 // HandleUpdated handles the updated state for version v20250312
 func (h *Handlerv20250312) HandleUpdated(ctx context.Context, databaseuser *akov2generated.DatabaseUser) (ctrlstate.Result, error) {
-	// TODO: Implement updated state logic
-	// TODO: Use h.atlasProvider.SdkClientSet(ctx, h.globalSecretRef, h.log) to get Atlas SDK client
-	return result.NextState(state.StateUpdated, "Ready")
+	return h.handleIdle(ctx, state.StateUpdated, databaseuser)
 }
 
 // HandleDeletionRequested handles the deletionrequested state for version v20250312
 func (h *Handlerv20250312) HandleDeletionRequested(ctx context.Context, databaseuser *akov2generated.DatabaseUser) (ctrlstate.Result, error) {
-	// TODO: Implement deletionrequested state logic
-	// TODO: Use h.atlasProvider.SdkClientSet(ctx, h.globalSecretRef, h.log) to get Atlas SDK client
-	return result.NextState(state.StateDeleting, "Deletion started")
+	deps, err := h.getDependencies(ctx, databaseuser)
+	if err != nil {
+		return result.Error(state.StateDeletionRequested, fmt.Errorf("Failed to get dependencies: %w", err))
+	}
+
+	params := &v20250312sdk.DeleteDatabaseUserApiParams{}
+	if err := h.translator.ToAPI(params, databaseuser, deps...); err != nil {
+		return result.Error(state.StateDeletionRequested, fmt.Errorf("failed to translate params: %w", err))
+	}
+	_, err = h.atlasClient.DatabaseUsersApi.DeleteDatabaseUserWithParams(ctx, params).Execute()
+	if err != nil {
+		return result.Error(state.StateDeletionRequested, fmt.Errorf("failed to delete datebaseuser: %w", err))
+	}
+
+	return result.NextState(state.StateDeleted, "User deleted.")
 }
 
 // HandleDeleting handles the deleting state for version v20250312
 func (h *Handlerv20250312) HandleDeleting(ctx context.Context, databaseuser *akov2generated.DatabaseUser) (ctrlstate.Result, error) {
-	// TODO: Implement deleting state logic
-	// TODO: Use h.atlasProvider.SdkClientSet(ctx, h.globalSecretRef, h.log) to get Atlas SDK client
-	return result.NextState(state.StateDeleted, "Deleted")
+	panic("unsupported state")
+}
+
+func (h *Handlerv20250312) handleIdle(ctx context.Context, currentState state.ResourceState, databaseuser *akov2generated.DatabaseUser) (ctrlstate.Result, error) {
+	deps, err := h.getDependencies(ctx, databaseuser)
+	if err != nil {
+		return result.Error(currentState, fmt.Errorf("Failed to get dependencies: %w", err))
+	}
+
+	update, err := ctrlstate.ShouldUpdate(databaseuser, deps...)
+	if err != nil {
+		return result.Error(currentState, reconcile.TerminalError(err))
+	}
+
+	if !update {
+		return result.NextState(currentState, "Database user up to date. No update required.")
+	}
+
+	body := &v20250312sdk.CloudDatabaseUser{}
+	params := &v20250312sdk.UpdateDatabaseUserApiParams{
+		CloudDatabaseUser: body,
+	}
+
+	if err := h.translator.ToAPI(params, databaseuser, deps...); err != nil {
+		return result.Error(currentState, fmt.Errorf("failed to translate params: %w", err))
+	}
+
+	if err := h.translator.ToAPI(body, databaseuser, deps...); err != nil {
+		return result.Error(currentState, fmt.Errorf("failed to translate body: %w", err))
+	}
+
+	_, _, err = h.atlasClient.DatabaseUsersApi.UpdateDatabaseUserWithParams(ctx, params).Execute()
+	if err != nil {
+		return result.Error(currentState, fmt.Errorf("failed to update datebaseuser: %w", err))
+	}
+
+	if err := ctrlstate.NewPatcher(databaseuser).UpdateStateTracker(deps...).Patch(ctx, h.kubeClient); err != nil {
+		return result.Error(currentState, fmt.Errorf("failed to update state tracker: %w", err))
+	}
+
+	return result.NextState(state.StateUpdated, "Updated Database User.")
+}
+
+func (h *Handlerv20250312) getDependencies(ctx context.Context, databaseuser *akov2generated.DatabaseUser) ([]client.Object, error) {
+	var deps []client.Object
+
+	// Check if passwordSecretRef is present
+	if databaseuser.Spec.V20250312 != nil && databaseuser.Spec.V20250312.Entry != nil && databaseuser.Spec.V20250312.Entry.PasswordSecretRef != nil {
+		secret := &v1.Secret{}
+		err := h.kubeClient.Get(ctx, client.ObjectKey{
+			Name:      databaseuser.Spec.V20250312.Entry.PasswordSecretRef.Name,
+			Namespace: databaseuser.GetNamespace(),
+		}, secret)
+		if err != nil {
+			return deps, fmt.Errorf("failed to get Secret %s/%s: %w", databaseuser.GetNamespace(), databaseuser.Spec.V20250312.Entry.PasswordSecretRef.Name, err)
+		}
+
+		deps = append(deps, secret)
+	}
+
+	// Check if groupRef is present
+	if databaseuser.Spec.V20250312 != nil && databaseuser.Spec.V20250312.GroupRef != nil {
+		group := &akov2generated.Group{}
+		err := h.kubeClient.Get(ctx, client.ObjectKey{
+			Name:      databaseuser.Spec.V20250312.GroupRef.Name,
+			Namespace: databaseuser.GetNamespace(),
+		}, group)
+		if err != nil {
+			return deps, fmt.Errorf("failed to get Group %s/%s: %w", databaseuser.GetNamespace(), databaseuser.Spec.V20250312.GroupRef.Name, err)
+		}
+
+		deps = append(deps, group)
+	}
+
+	return deps, nil
 }
 
 // For returns the resource and predicates for the controller

internal/generated/crds/crds.yaml

@@ -1962,10 +1962,6 @@ spec:
                       description:
                         description: Description of this database user.
                         type: string
-                      groupId:
-                        description: Unique 24-hexadecimal digit string that identifies
-                          the project.
-                        type: string
                       labels:
                         description: List that contains the key-value pairs for tagging
                           and categorizing the MongoDB database user. The labels that
@@ -2007,7 +2003,7 @@ spec:
                           Alphanumeric string that authenticates this database user against the database specified in `databaseName`. To authenticate with SCRAM-SHA, you must specify this parameter. This parameter doesn't appear in this response.
                         properties:
                           key:
-                            default: .data.password
+                            default: password
                             description: Key of the secret data containing the sensitive
                               field value, defaults to "password".
                             type: string
@@ -2095,7 +2091,6 @@ spec:
                         type: string
                     required:
                     - databaseName
-                    - groupId
                     - username
                     type: object
                   groupId:

internal/indexer/indexer.go

@@ -79,6 +79,8 @@ func RegisterAll(ctx context.Context, c cluster.Cluster, logger *zap.Logger) err
 			NewAtlasDataFederationByProjectIDIndexer(ctx, c.GetClient(), logger),
 			indexer.NewFlexClusterByGroupIndexer(logger),
 			indexer.NewClusterByGroupIndexer(logger),
+			indexer.NewDatabaseUserBySecretIndexer(logger),
+			indexer.NewDatabaseUserByGroupIndexer(logger),
 		)
 	}
 	return Register(ctx, c, indexers...)