CLOUDP-334941: Refactored Experimental Connection Secret Controller

Makefile

@@ -258,7 +258,11 @@ manifests: fmt ## Generate manifests e.g. CRD, RBAC etc.
 	controller-gen $(CRD_OPTIONS) rbac:roleName=manager-role webhook paths="./api/..." paths="./internal/controller/..." output:crd:artifacts:config=config/crd/bases
 	@./scripts/split_roles_yaml.sh
 ifdef EXPERIMENTAL
-	controller-gen crd paths="./internal/nextapi/v1" output:crd:artifacts:config=internal/next-crds
+	@if [ -d internal/next-crds ] && find internal/next-crds -maxdepth 1 -name '*.yaml' | grep -q .; then \
+	controller-gen crd paths="./internal/nextapi/v1" output:crd:artifacts:config=internal/next-crds; \
+	else \
+	echo "No experimental CRDs found, skipping apply."; \
+	fi
 endif
 
 .PHONY: lint
@@ -553,7 +557,11 @@ clear-e2e-leftovers: ## Clear the e2e test leftovers quickly
 install-crds: ## Install CRDs in Kubernetes
 	kubectl apply -k config/crd
 ifdef EXPERIMENTAL
-	kubectl apply -f internal/next-crds/*.yaml
+	@if [ -d internal/next-crds ] && find internal/next-crds -maxdepth 1 -name '*.yaml' | grep -q .; then \
+	kubectl apply -f internal/next-crds/*.yaml; \
+	else \
+	echo "No experimental CRDs found, skipping apply."; \
+	fi
 endif
 
 .PHONY: set-namespace
@@ -572,7 +580,7 @@ install-credentials: set-namespace ## Install the Atlas credentials for the Oper
 
 .PHONY: prepare-run
 prepare-run: generate vet manifests run-kind install-crds install-credentials
-	rm bin/manager
+	rm -f bin/manager
 	$(MAKE) manager VERSION=$(NEXT_VERSION)
 
 .PHONY: run

api/condition.go

@@ -177,6 +177,15 @@ func HasConditionType(typ ConditionType, source []Condition) bool {
 	return false
 }
 
+func HasReadyCondition(conditions []Condition) bool {
+	for _, c := range conditions {
+		if c.Type == ReadyType && c.Status == corev1.ConditionTrue {
+			return true
+		}
+	}
+	return false
+}
+
 // EnsureConditionExists adds or updates the condition in the copy of a 'source' slice
 func EnsureConditionExists(condition Condition, source []Condition) []Condition {
 	condition.LastTransitionTime = metav1.Now()

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/pointer"
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/v3/controller/connectionsecret"
 	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/version"
 	ctrlstate "github.com/mongodb/mongodb-atlas-kubernetes/v2/pkg/controller/state"
 	"github.com/mongodb/mongodb-atlas-kubernetes/v2/pkg/ratelimit"
@@ -134,6 +135,7 @@ func (r *Registry) registerControllers(c cluster.Cluster, ap atlas.Provider) {
 
 	if version.IsExperimental() {
 		// Add experimental controllers here
+		reconcilers = append(reconcilers, connectionsecret.NewConnectionSecretReconciler(c, r.defaultPredicates(), ap, r.logger, r.globalSecretRef))
 	}
 	r.reconcilers = reconcilers
 }

internal/controller/watch/predicates.go

@@ -84,3 +84,30 @@ func DefaultPredicates[T metav1.Object]() predicate.TypedPredicate[T] {
 		IgnoreDeletedPredicate[T](),
 	)
 }
+
+type ReadyFunc[T any] func(obj T) bool
+
+// ReadyTransitionPredicate filters out only those objects where the previous
+// oldObject was not ready, but the new one is, or the object was deleted.
+func ReadyTransitionPredicate[T any](ready ReadyFunc[T]) predicate.Predicate {
+	return predicate.Funcs{
+		CreateFunc:  func(e event.CreateEvent) bool { return false },
+		GenericFunc: func(e event.GenericEvent) bool { return false },
+		DeleteFunc:  func(e event.DeleteEvent) bool { return true },
+		UpdateFunc: func(e event.UpdateEvent) bool {
+			if e.ObjectNew == nil || e.ObjectOld == nil {
+				return false
+			}
+
+			newObj, ok := e.ObjectNew.(T)
+			if !ok {
+				return false
+			}
+			oldObj, ok := e.ObjectOld.(T)
+			if !ok {
+				return false
+			}
+			return !ready(oldObj) && ready(newObj)
+		},
+	}
+}

internal/controller/workflow/reason.go

@@ -193,3 +193,18 @@ const (
 	NetworkPeeringConnectionPending  ConditionReason = "NetworkPeeringConnectionPending"
 	NetworkPeeringConnectionClosing  ConditionReason = "NetworkPeeringConnectionClosing"
 )
+
+// ConnectionSecret reasons
+const (
+	ConnectionSecretInvalidUsername            ConditionReason = "ConnectionSecretInvalidUsername"
+	ConnectionSecretStaleSecretsNotCleaned     ConditionReason = "ConnectionSecretStaleSecretsNotCleaned"
+	ConnectionSecretProjectIDNotLoaded         ConditionReason = "ConnectionSecretProjectIDNotLoaded"
+	ConnectionSecretConnectionTargetsNotLoaded ConditionReason = "ConnectionSecretConnectionTargetsNotLoaded"
+	ConnectionSecretUserExpired                ConditionReason = "ConnectionSecretUserExpired"
+	ConnectionSecretNotReady                   ConditionReason = "ConnectionSecretNotReady"
+	ConnectionSecretFailedToBuildData          ConditionReason = "ConnectionSecretFailedToBuildData"
+	ConnectionSecretFailedToFillData           ConditionReason = "ConnectionSecretFailedToFillData"
+	ConnectionSecretFailedDeletion             ConditionReason = "ConnectionSecretFailedDeletion"
+	ConnectionSecretFailedToUpsertSecret       ConditionReason = "ConnectionSecretFailedToUpsertSecret"
+	ConnectionSecretFailedToSetOwnerReferences ConditionReason = "ConnectionSecretFailedToSetOwnerReferences"
+)

internal/indexer/atlasdatafederationbyprojectid.go

@@ -0,0 +1,72 @@
+// 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.
+
+//nolint:dupl
+package indexer
+
+import (
+	"context"
+
+	"go.uber.org/zap"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	akov2 "github.com/mongodb/mongodb-atlas-kubernetes/v2/api/v1"
+)
+
+const (
+	AtlasDataFederationByProjectID = "atlasdatafederation.spec.projectID"
+)
+
+type AtlasDataFederationByProjectIDIndexer struct {
+	ctx    context.Context
+	client client.Client
+	logger *zap.SugaredLogger
+}
+
+func NewAtlasDataFederationByProjectIDIndexer(ctx context.Context, c client.Client, logger *zap.Logger) *AtlasDataFederationByProjectIDIndexer {
+	return &AtlasDataFederationByProjectIDIndexer{
+		ctx:    ctx,
+		client: c,
+		logger: logger.Named(AtlasDataFederationByProjectID).Sugar(),
+	}
+}
+
+func (*AtlasDataFederationByProjectIDIndexer) Object() client.Object {
+	return &akov2.AtlasDataFederation{}
+}
+
+func (*AtlasDataFederationByProjectIDIndexer) Name() string {
+	return AtlasDataFederationByProjectID
+}
+
+func (a *AtlasDataFederationByProjectIDIndexer) Keys(object client.Object) []string {
+	df, ok := object.(*akov2.AtlasDataFederation)
+	if !ok {
+		a.logger.Errorf("expected *v1.AtlasDataFederation but got %T", object)
+		return nil
+	}
+
+	if df.Spec.Project.Name != "" {
+		project := &akov2.AtlasProject{}
+		if err := a.client.Get(a.ctx, *df.Spec.Project.GetObject(df.Namespace), project); err != nil {
+			a.logger.Errorf("unable to find project to index: %s", err)
+			return nil
+		}
+		if project.ID() != "" {
+			return []string{project.ID()}
+		}
+	}
+
+	return nil
+}

internal/indexer/atlasdatafederationbyprojectid_test.go

@@ -0,0 +1,118 @@
+// 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 indexer
+
+import (
+	"context"
+	"sort"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+	"go.uber.org/zap/zaptest/observer"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	akov2 "github.com/mongodb/mongodb-atlas-kubernetes/v2/api/v1"
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/api/v1/common"
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/api/v1/status"
+)
+
+func TestAtlasDataFederationByProjectIDIndexer(t *testing.T) {
+	tests := map[string]struct {
+		object       client.Object
+		expectedKeys []string
+		expectedLogs []observer.LoggedEntry
+	}{
+		"should return nil on wrong type": {
+			object: &akov2.AtlasStreamInstance{},
+			expectedLogs: []observer.LoggedEntry{
+				{
+					Context: []zapcore.Field{},
+					Entry:   zapcore.Entry{LoggerName: AtlasDataFederationByProjectID, Level: zap.ErrorLevel, Message: "expected *v1.AtlasDataFederation but got *v1.AtlasStreamInstance"},
+				},
+			},
+		},
+		"should return nil when there is an empty project reference": {
+			object:       &akov2.AtlasDataFederation{},
+			expectedLogs: []observer.LoggedEntry{},
+		},
+		"should return nil when referenced project was not found": {
+			object: &akov2.AtlasDataFederation{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "df",
+					Namespace: "ns",
+				},
+				Spec: akov2.DataFederationSpec{
+					Project: common.ResourceRefNamespaced{
+						Name: "not-found-project",
+					},
+				},
+			},
+			expectedLogs: []observer.LoggedEntry{
+				{
+					Context: []zapcore.Field{},
+					Entry:   zapcore.Entry{LoggerName: AtlasDataFederationByProjectID, Level: zap.ErrorLevel, Message: "unable to find project to index: atlasprojects.atlas.mongodb.com \"not-found-project\" not found"},
+				},
+			},
+		},
+		"should return project ID using DataFederation namespace": {
+			object: &akov2.AtlasDataFederation{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "df",
+					Namespace: "ns",
+				},
+				Spec: akov2.DataFederationSpec{
+					Project: common.ResourceRefNamespaced{
+						Name: "internal-project",
+					},
+				},
+			},
+			expectedKeys: []string{"external-project-id"},
+			expectedLogs: []observer.LoggedEntry{},
+		},
+	}
+
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			project := &akov2.AtlasProject{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "internal-project",
+					Namespace: "ns",
+				},
+				Status: status.AtlasProjectStatus{
+					ID: "external-project-id",
+				},
+			}
+
+			testScheme := runtime.NewScheme()
+			assert.NoError(t, akov2.AddToScheme(testScheme))
+
+			builder := fake.NewClientBuilder().WithScheme(testScheme)
+			k8sClient := builder.WithObjects(project).Build()
+			core, logs := observer.New(zap.DebugLevel)
+
+			indexer := NewAtlasDataFederationByProjectIDIndexer(context.Background(), k8sClient, zap.New(core))
+			keys := indexer.Keys(tt.object)
+			sort.Strings(keys)
+
+			assert.Equal(t, tt.expectedKeys, keys)
+			assert.Equal(t, tt.expectedLogs, logs.AllUntimed())
+		})
+	}
+}

internal/indexer/indexer.go

@@ -74,6 +74,9 @@ func RegisterAll(ctx context.Context, c cluster.Cluster, logger *zap.Logger) err
 	)
 	if version.IsExperimental() {
 		// add experimental indexers here
+		indexers = append(indexers,
+			NewAtlasDataFederationByProjectIDIndexer(ctx, c.GetClient(), logger),
+		)
 	}
 	return Register(ctx, c, indexers...)
 }

internal/operator/builder_test.go

@@ -24,6 +24,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap/zaptest"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/record"
@@ -161,6 +162,7 @@ func TestBuildManager(t *testing.T) {
 		t.Run(name, func(t *testing.T) {
 			akoScheme := runtime.NewScheme()
 			require.NoError(t, akov2.AddToScheme(akoScheme))
+			require.NoError(t, corev1.AddToScheme(akoScheme))
 
 			mgrMock := &managerMock{}
 			builder := NewBuilder(mgrMock, akoScheme, 5*time.Minute)

internal/timeutil/timeutil.go

@@ -61,3 +61,18 @@ func MustParseISO8601(dateTime string) time.Time {
 func FormatISO8601(dateTime time.Time) string {
 	return dateTime.Format("2006-01-02T15:04:05.999Z")
 }
+
+// IsExpired parses the given ISO8601 date string and returns whether it is before now.
+// Returns an error if the string cannot be parsed.
+func IsExpired(deleteAfterDate string) (bool, error) {
+	if deleteAfterDate == "" {
+		return false, nil
+	}
+
+	deleteAfter, err := ParseISO8601(deleteAfterDate)
+	if err != nil {
+		return false, err
+	}
+
+	return deleteAfter.Before(time.Now()), nil
+}