CLOUDP-359047: Add Flex Reconciler (#2955)

* CLOUDP-359047: Add Flex Cluster reconciler support

Signed-off-by: jose.vazquez <[email protected]>

* Register flexcluster

* Add manual group id resolution

* Fix time until lint hint

---------

Signed-off-by: jose.vazquez <[email protected]>

Author: josvazg

internal/controller/registry.go

@@ -47,6 +47,7 @@ import (
 	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/dryrun"
 	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/featureflags"
 	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/generated/controller/connectionsecret"
+	"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"
 	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/version"
@@ -147,7 +148,14 @@ func (r *Registry) registerControllers(c cluster.Cluster, ap atlas.Provider) err
 		if err != nil {
 			return fmt.Errorf("error creating group reconciler: %w", err)
 		}
-		reconcilers = append(reconcilers, newCtrlStateReconciler(groupReconciler, r.maxConcurrentReconciles))
+		flexController, err := flexcluster.NewFlexClusterReconciler(c, ap, r.logger, r.globalSecretRef, r.deletionProtection, true, r.defaultPredicates())
+		if err != nil {
+			return fmt.Errorf("error creating group reconciler: %w", err)
+		}
+		reconcilers = append(reconcilers,
+			newCtrlStateReconciler(groupReconciler, r.maxConcurrentReconciles),
+			newCtrlStateReconciler(flexController, r.maxConcurrentReconciles),
+		)
 	}
 
 	r.reconcilers = reconcilers

internal/generated/controller/flexcluster/handler_v20250312.go

@@ -16,6 +16,8 @@ package flexcluster
 
 import (
 	"context"
+	"errors"
+	"fmt"
 
 	v20250312sdk "go.mongodb.org/atlas-sdk/v20250312006/admin"
 	controllerruntime "sigs.k8s.io/controller-runtime"
@@ -24,7 +26,9 @@ import (
 	controller "sigs.k8s.io/controller-runtime/pkg/controller"
 	reconcile "sigs.k8s.io/controller-runtime/pkg/reconcile"
 
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/customresource"
 	crapi "github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/crapi"
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/generated/references"
 	akov2generated "github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/nextapi/generated/v1"
 	ctrlstate "github.com/mongodb/mongodb-atlas-kubernetes/v2/pkg/controller/state"
 	result "github.com/mongodb/mongodb-atlas-kubernetes/v2/pkg/result"
@@ -47,67 +51,138 @@ func NewHandlerv20250312(kubeClient client.Client, atlasClient *v20250312sdk.API
 	}
 }
 
+// TODO: Autogenerate with Scaffolder
+func (h *Handlerv20250312) getGroupId(ctx context.Context, flexcluster *akov2generated.FlexCluster) (string, error) {
+	groupID := ""
+
+	if flexcluster.Spec.V20250312.GroupId != nil {
+		groupID = *flexcluster.Spec.V20250312.GroupId
+	} else {
+		var err error
+		groupID, err = references.GetGroupID(
+			ctx, h.kubeClient, flexcluster.Spec.V20250312.GroupRef, flexcluster.GetNamespace())
+		if err != nil {
+			return "", err
+		}
+	}
+
+	return groupID, nil
+}
+
 // HandleInitial handles the initial state for version v20250312
 func (h *Handlerv20250312) HandleInitial(ctx context.Context, flexcluster *akov2generated.FlexCluster) (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 AtlasFlexCluster.")
+	groupID, err := h.getGroupId(ctx, flexcluster)
+	if err != nil {
+		return result.Error(state.StateInitial, reconcile.TerminalError(err))
+	}
+	flexClusterRequest := v20250312sdk.FlexClusterDescriptionCreate20241113{}
+	if err := h.translator.ToAPI(&flexClusterRequest, flexcluster); err != nil {
+		return result.Error(state.StateInitial, fmt.Errorf("failed to translate flex create request: %w", err))
+	}
+	atlasFlexCluster, _, err := h.atlasClient.FlexClustersApi.CreateFlexCluster(ctx, groupID, &flexClusterRequest).Execute()
+	if err != nil {
+		return result.Error(state.StateInitial, fmt.Errorf("failed to create flex cluster: %w", err))
+	}
+	newFlexCluster := flexcluster.DeepCopy()
+	if _, err := h.translator.FromAPI(newFlexCluster, atlasFlexCluster); err != nil {
+		return result.Error(state.StateInitial, fmt.Errorf("failed to translate flex create response: %w", err))
+	}
+
+	err = h.kubeClient.Status().Patch(ctx, newFlexCluster, client.MergeFrom(flexcluster))
+	if err != nil {
+		return result.Error(state.StateInitial, fmt.Errorf("failed to patch flex cluster status: %w", err))
+	}
+	return result.NextState(state.StateCreating, "Creating Flex Cluster.")
 }
 
 // HandleImportRequested handles the importrequested state for version v20250312
 func (h *Handlerv20250312) HandleImportRequested(ctx context.Context, flexcluster *akov2generated.FlexCluster) (ctrlstate.Result, error) {
-	// TODO: Implement importrequested state logic
-	// TODO: Use h.atlasProvider.SdkClientSet(ctx, h.globalSecretRef, h.log) to get Atlas SDK client
-	return result.NextState(state.StateImported, "Import completed")
+	externalName, ok := flexcluster.GetAnnotations()["mongodb.com/external-name"]
+	if !ok {
+		return result.Error(state.StateImportRequested, errors.New("missing mongodb.com/external-name"))
+	}
+
+	externalGroupID, ok := flexcluster.GetAnnotations()["mongodb.com/external-group-id"]
+	if !ok {
+		return result.Error(state.StateImportRequested, errors.New("missing mongodb.com/external-group-id"))
+	}
+	flexClusterCopy := flexcluster.DeepCopy()
+	flexClusterCopy.Spec.V20250312.Entry.Name = externalName
+	flexClusterCopy.Spec.V20250312.GroupId = &externalGroupID
+	_, err := h.getToPatchStatus(ctx, flexClusterCopy)
+	if err != nil {
+		return result.Error(state.StateImportRequested, err)
+	}
+	return result.NextState(state.StateImported, "Imported Flex Cluster.")
 }
 
 // HandleImported handles the imported state for version v20250312
 func (h *Handlerv20250312) HandleImported(ctx context.Context, flexcluster *akov2generated.FlexCluster) (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, flexcluster, state.StateCreated, state.StateUpdating)
 }
 
 // HandleCreating handles the creating state for version v20250312
 func (h *Handlerv20250312) HandleCreating(ctx context.Context, flexcluster *akov2generated.FlexCluster) (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")
+	return h.handleUpserting(ctx, flexcluster, state.StateCreating, state.StateCreated)
 }
 
 // HandleCreated handles the created state for version v20250312
 func (h *Handlerv20250312) HandleCreated(ctx context.Context, flexcluster *akov2generated.FlexCluster) (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, flexcluster, state.StateCreated, state.StateUpdating)
 }
 
 // HandleUpdating handles the updating state for version v20250312
 func (h *Handlerv20250312) HandleUpdating(ctx context.Context, flexcluster *akov2generated.FlexCluster) (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")
+	return h.handleUpserting(ctx, flexcluster, state.StateUpdating, state.StateUpdated)
 }
 
 // HandleUpdated handles the updated state for version v20250312
 func (h *Handlerv20250312) HandleUpdated(ctx context.Context, flexcluster *akov2generated.FlexCluster) (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, flexcluster, state.StateUpdated, state.StateUpdating)
 }
 
 // HandleDeletionRequested handles the deletionrequested state for version v20250312
 func (h *Handlerv20250312) HandleDeletionRequested(ctx context.Context, flexcluster *akov2generated.FlexCluster) (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")
+	if customresource.IsResourcePolicyKeepOrDefault(flexcluster, h.deletionProtection) {
+		return result.NextState(state.StateDeleted, "Flex Cluster deleted.")
+	}
+
+	if flexcluster.Status.V20250312 == nil {
+		return result.NextState(state.StateDeleted, "Flex Cluster is unamanged.")
+	}
+
+	groupID, err := h.getGroupId(ctx, flexcluster)
+	if err != nil {
+		return result.Error(state.StateInitial, reconcile.TerminalError(err))
+	}
+	_, err = h.atlasClient.FlexClustersApi.DeleteFlexCluster(
+		ctx, groupID, flexcluster.Spec.V20250312.Entry.Name).Execute()
+
+	switch {
+	case v20250312sdk.IsErrorCode(err, "CLUSTER_NOT_FOUND"):
+		return result.NextState(state.StateDeleted, "Flex Cluster was deleted in Atlas.")
+	case err != nil:
+		return result.Error(state.StateDeletionRequested, fmt.Errorf("failed to delete flex cluster: %w", err))
+	}
+
+	return result.NextState(state.StateDeleting, "Deleting Flex Cluster.")
 }
 
 // HandleDeleting handles the deleting state for version v20250312
 func (h *Handlerv20250312) HandleDeleting(ctx context.Context, flexcluster *akov2generated.FlexCluster) (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")
+	groupID, err := h.getGroupId(ctx, flexcluster)
+	if err != nil {
+		return result.Error(state.StateInitial, reconcile.TerminalError(err))
+	}
+	_, _, err = h.atlasClient.FlexClustersApi.GetFlexCluster(
+		ctx, groupID, flexcluster.Spec.V20250312.Entry.Name).Execute()
+	switch {
+	case v20250312sdk.IsErrorCode(err, "CLUSTER_NOT_FOUND"):
+		return result.NextState(state.StateDeleted, "Deleted")
+	case err != nil:
+		return result.Error(state.StateDeletionRequested, fmt.Errorf("failed to delete flexcluster: %w", err))
+	}
+	return result.NextState(state.StateDeleting, "Deleting Flex Cluster.")
 }
 
 // For returns the resource and predicates for the controller
@@ -117,6 +192,75 @@ func (h *Handlerv20250312) For() (client.Object, builder.Predicates) {
 
 // SetupWithManager sets up the controller with the Manager
 func (h *Handlerv20250312) SetupWithManager(mgr controllerruntime.Manager, rec reconcile.Reconciler, defaultOptions controller.Options) error {
-	// This method is not used for version-specific handlers but required by StateHandler interface
-	return nil
+	panic("do not setup Handlerv20250312")
+}
+
+// HandleUpserting handles the creating and updating state for flex version v20250312
+func (h *Handlerv20250312) handleUpserting(ctx context.Context, flexcluster *akov2generated.FlexCluster, currentState, finalState state.ResourceState) (ctrlstate.Result, error) {
+	atlasFlexCluster, err := h.getToPatchStatus(ctx, flexcluster)
+	if err != nil {
+		return result.Error(currentState, err)
+	}
+	if atlasFlexCluster.GetStateName() == "CREATING" || atlasFlexCluster.GetStateName() == "UPDATING" {
+		return result.NextState(currentState, "Upserting Flex Cluster.")
+	}
+	return result.NextState(finalState, "Upserted Flex Cluster.")
+}
+
+// HandleIdle handles the creating and updating state for flex version v20250312
+func (h *Handlerv20250312) handleIdle(ctx context.Context, flexcluster *akov2generated.FlexCluster, currentState, finalState state.ResourceState) (ctrlstate.Result, error) {
+	update, err := ctrlstate.ShouldUpdate(flexcluster)
+	if err != nil {
+		return result.Error(currentState, reconcile.TerminalError(err))
+	}
+
+	if !update {
+		return result.NextState(currentState, "Flex cluster up to date. No update required.")
+	}
+
+	groupID, err := h.getGroupId(ctx, flexcluster)
+	if err != nil {
+		return result.Error(state.StateInitial, reconcile.TerminalError(err))
+	}
+	flexClusterUpdate := v20250312sdk.FlexClusterDescriptionUpdate20241113{}
+	if err := h.translator.ToAPI(&flexClusterUpdate, flexcluster); err != nil {
+		return result.Error(state.StateInitial, fmt.Errorf("failed to translate update flex cluster request: %w", err))
+	}
+	atlasFlexCluster, _, err := h.atlasClient.FlexClustersApi.UpdateFlexCluster(
+		ctx, groupID, flexcluster.Spec.V20250312.Entry.Name, &flexClusterUpdate).Execute()
+	if err != nil {
+		return result.Error(currentState, fmt.Errorf("failed to get update cluster: %w", err))
+	}
+	flexclusterCopy := flexcluster.DeepCopy()
+	if _, err := h.translator.FromAPI(flexclusterCopy, atlasFlexCluster); err != nil {
+		return result.Error(currentState, fmt.Errorf("failed to translate update cluster response: %w", err))
+	}
+
+	err = h.kubeClient.Status().Patch(ctx, flexclusterCopy, client.MergeFrom(flexcluster))
+	if err != nil {
+		return result.Error(currentState, fmt.Errorf("failed to patch cluster status: %w", err))
+	}
+	return result.NextState(finalState, "Updating Flex Cluster.")
+}
+
+func (h *Handlerv20250312) getToPatchStatus(ctx context.Context, flexcluster *akov2generated.FlexCluster) (*v20250312sdk.FlexClusterDescription20241113, error) {
+	groupID, err := h.getGroupId(ctx, flexcluster)
+	if err != nil {
+		return nil, err
+	}
+	atlasFlexCluster, _, err := h.atlasClient.FlexClustersApi.GetFlexCluster(
+		ctx, groupID, flexcluster.Spec.V20250312.Entry.Name).Execute()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get cluster: %w", err)
+	}
+	flexclusterCopy := flexcluster.DeepCopy()
+	if _, err := h.translator.FromAPI(flexclusterCopy, atlasFlexCluster); err != nil {
+		return nil, fmt.Errorf("failed to translate get cluster response: %w", err)
+	}
+
+	err = h.kubeClient.Status().Patch(ctx, flexclusterCopy, client.MergeFrom(flexcluster))
+	if err != nil {
+		return nil, fmt.Errorf("failed to patch cluster status: %w", err)
+	}
+	return atlasFlexCluster, nil
 }

internal/generated/controller/group/handler_v20250312.go

@@ -150,15 +150,13 @@ func (h *Handlerv20250312) HandleDeletionRequested(ctx context.Context, group *a
 // HandleDeleting handles the deleting state for version v20250312
 func (h *Handlerv20250312) HandleDeleting(ctx context.Context, group *akov2generated.Group) (ctrlstate.Result, error) {
 	_, _, err := h.atlasClient.ProjectsApi.GetProject(ctx, *group.Status.V20250312.Id).Execute()
-	if err == nil {
-		return result.Error(state.StateDeletionRequested, fmt.Errorf("failed to delete group: %w deletion", err))
-	}
-
-	if !v20250312sdk.IsErrorCode(err, "GROUP_NOT_FOUND") {
+	switch {
+	case v20250312sdk.IsErrorCode(err, "GROUP_NOT_FOUND"):
+		return result.NextState(state.StateDeleted, "Deleted")
+	case err != nil:
 		return result.Error(state.StateDeletionRequested, fmt.Errorf("failed to delete group: %w", err))
 	}
-
-	return result.NextState(state.StateDeleted, "Group deleted.")
+	return result.NextState(state.StateDeleting, "Deleting Group.")
 }
 
 // For returns the resource and predicates for the controller

internal/generated/references/references.go

@@ -0,0 +1,51 @@
+// Copyright 2025 MongoDB Inc
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package references
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/crd2go/crd2go/k8s"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	akov2generated "github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/nextapi/generated/v1"
+)
+
+// GetGroupID returns the group IP (if present) from a given Kubernetes group reference
+// TODO: Autogenerate with Scaffolder
+func GetGroupID(ctx context.Context, c client.Client, groupRef *k8s.LocalReference, namespace string) (string, error) {
+	if groupRef == nil {
+		return "", errors.New("group reference is nil")
+	}
+
+	group := &akov2generated.Group{}
+	err := c.Get(ctx, client.ObjectKey{
+		Namespace: groupRef.Name,
+		Name:      namespace,
+	}, group)
+
+	if err != nil {
+		return "", fmt.Errorf("failed to get object: %w", err)
+	}
+
+	// for each suported Group version...
+	if group.Status.V20250312 != nil && group.Status.V20250312.Id != nil {
+		return *group.Status.V20250312.Id, nil
+	}
+
+	return "", errors.New("group ID is not available")
+}

pkg/controller/state/reapply.go

@@ -48,7 +48,7 @@ func ShouldReapply(obj metav1.Object) (bool, error) {
 		return false, nil
 	}
 
-	diff := timestamp.Add(period).Sub(time.Now())
+	diff := time.Until(timestamp.Add(period))
 
 	return diff <= 0, nil
 }