internal/controller: add registry (#2066)

Author: s-urbaniak

internal/controller/registry.go

@@ -0,0 +1,81 @@
+package controller
+
+import (
+	"fmt"
+	"time"
+
+	"go.uber.org/zap"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/cluster"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlas"
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlasbackupcompliancepolicy"
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlascustomrole"
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlasdatabaseuser"
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlasdatafederation"
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlasdeployment"
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlasfederatedauth"
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlasprivateendpoint"
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlasproject"
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlassearchindexconfig"
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlasstream"
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/featureflags"
+)
+
+type ManagerAware interface {
+	SetupWithManager(mgr ctrl.Manager, skipNameValidation bool) error
+}
+
+type AkoReconciler interface {
+	reconcile.Reconciler
+	ManagerAware
+}
+
+type Registry struct {
+	predicates            []predicate.Predicate
+	deletionProtection    bool
+	independentSyncPeriod time.Duration
+	featureFlags          *featureflags.FeatureFlags
+
+	logger      *zap.Logger
+	reconcilers []AkoReconciler
+}
+
+func NewRegistry(predicates []predicate.Predicate, deletionProtection bool, logger *zap.Logger, independentSyncPeriod time.Duration, featureFlags *featureflags.FeatureFlags) *Registry {
+	return &Registry{
+		predicates:            predicates,
+		deletionProtection:    deletionProtection,
+		logger:                logger,
+		independentSyncPeriod: independentSyncPeriod,
+		featureFlags:          featureFlags,
+	}
+}
+
+func (r *Registry) RegisterWithManager(mgr ctrl.Manager, skipNameValidation bool, ap atlas.Provider) error {
+	r.registerControllers(mgr, ap)
+
+	for _, reconciler := range r.reconcilers {
+		if err := reconciler.SetupWithManager(mgr, skipNameValidation); err != nil {
+			return fmt.Errorf("failed to set up with manager: %w", err)
+		}
+	}
+	return nil
+}
+
+func (r *Registry) registerControllers(c cluster.Cluster, ap atlas.Provider) {
+	var reconcilers []AkoReconciler
+	reconcilers = append(reconcilers, atlasproject.NewAtlasProjectReconciler(c, r.predicates, ap, r.deletionProtection, r.logger))
+	reconcilers = append(reconcilers, atlasdeployment.NewAtlasDeploymentReconciler(c, r.predicates, ap, r.deletionProtection, r.independentSyncPeriod, r.logger))
+	reconcilers = append(reconcilers, atlasdatabaseuser.NewAtlasDatabaseUserReconciler(c, r.predicates, ap, r.deletionProtection, r.independentSyncPeriod, r.featureFlags, r.logger))
+	reconcilers = append(reconcilers, atlasdatafederation.NewAtlasDataFederationReconciler(c, r.predicates, ap, r.deletionProtection, r.logger))
+	reconcilers = append(reconcilers, atlasfederatedauth.NewAtlasFederatedAuthReconciler(c, r.predicates, ap, r.deletionProtection, r.logger))
+	reconcilers = append(reconcilers, atlasstream.NewAtlasStreamsInstanceReconciler(c, r.predicates, ap, r.deletionProtection, r.logger))
+	reconcilers = append(reconcilers, atlasstream.NewAtlasStreamsConnectionReconciler(c, r.predicates, ap, r.deletionProtection, r.logger))
+	reconcilers = append(reconcilers, atlassearchindexconfig.NewAtlasSearchIndexConfigReconciler(c, r.predicates, ap, r.deletionProtection, r.logger))
+	reconcilers = append(reconcilers, atlasbackupcompliancepolicy.NewAtlasBackupCompliancePolicyReconciler(c, r.predicates, ap, r.deletionProtection, r.logger))
+	reconcilers = append(reconcilers, atlascustomrole.NewAtlasCustomRoleReconciler(c, r.predicates, ap, r.deletionProtection, r.independentSyncPeriod, r.logger))
+	reconcilers = append(reconcilers, atlasprivateendpoint.NewAtlasPrivateEndpointReconciler(c, r.predicates, ap, r.deletionProtection, r.logger))
+	r.reconcilers = reconcilers
+}

internal/operator/builder.go

@@ -15,24 +15,16 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/cluster"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	ctrzap "sigs.k8s.io/controller-runtime/pkg/log/zap"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
 
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller"
 	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlas"
-	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlasbackupcompliancepolicy"
-	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlascustomrole"
-	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlasdatabaseuser"
-	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlasdatafederation"
-	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlasdeployment"
-	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlasfederatedauth"
-	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlasprivateendpoint"
-	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlasproject"
-	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlassearchindexconfig"
-	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/atlasstream"
 	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/connectionsecret"
 	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/controller/watch"
 	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/featureflags"
@@ -159,8 +151,8 @@ func (b *Builder) WithSkipNameValidation(skip bool) *Builder {
 	return b
 }
 
-// Build builds the controller manager and configure operator controllers
-func (b *Builder) Build(ctx context.Context) (manager.Manager, error) {
+// Build builds the cluster object and configures operator controllers
+func (b *Builder) Build(ctx context.Context) (cluster.Cluster, error) {
 	mergeDefaults(b)
 
 	if b.independentSyncPeriod < b.minimumIndependentSyncPeriod {
@@ -186,6 +178,8 @@ func (b *Builder) Build(ctx context.Context) (manager.Manager, error) {
 		}
 	}
 
+	controllerRegistry := controller.NewRegistry(b.predicates, b.deletionProtection, b.logger, b.independentSyncPeriod, b.featureFlags)
+
 	mgr, err := b.managerProvider.New(
 		b.config,
 		ctrl.Options{
@@ -213,137 +207,16 @@ func (b *Builder) Build(ctx context.Context) (manager.Manager, error) {
 		return nil, err
 	}
 
-	if err = indexer.RegisterAll(ctx, mgr, b.logger); err != nil {
-		return nil, fmt.Errorf("unable to create indexers: %w", err)
-	}
-
 	if b.atlasProvider == nil {
 		b.atlasProvider = atlas.NewProductionProvider(b.atlasDomain, b.apiSecret, mgr.GetClient(), nil)
 	}
 
-	projectReconciler := atlasproject.NewAtlasProjectReconciler(
-		mgr,
-		b.predicates,
-		b.atlasProvider,
-		b.deletionProtection,
-		b.logger,
-	)
-	if err = projectReconciler.SetupWithManager(mgr, b.skipNameValidation); err != nil {
-		return nil, fmt.Errorf("unable to create controller AtlasProject: %w", err)
-	}
-
-	deploymentReconciler := atlasdeployment.NewAtlasDeploymentReconciler(
-		mgr,
-		b.predicates,
-		b.atlasProvider,
-		b.deletionProtection,
-		b.independentSyncPeriod,
-		b.logger,
-	)
-	if err = deploymentReconciler.SetupWithManager(mgr, b.skipNameValidation); err != nil {
-		return nil, fmt.Errorf("unable to create controller AtlasDeployment: %w", err)
-	}
-
-	dbUserReconciler := atlasdatabaseuser.NewAtlasDatabaseUserReconciler(
-		mgr,
-		b.predicates,
-		b.atlasProvider,
-		b.deletionProtection,
-		b.independentSyncPeriod,
-		b.featureFlags,
-		b.logger,
-	)
-	if err = dbUserReconciler.SetupWithManager(mgr, b.skipNameValidation); err != nil {
-		return nil, fmt.Errorf("unable to create controller AtlasDatabaseUser: %w", err)
-	}
-
-	dataFedReconciler := atlasdatafederation.NewAtlasDataFederationReconciler(
-		mgr,
-		b.predicates,
-		b.atlasProvider,
-		b.deletionProtection,
-		b.logger,
-	)
-	if err = dataFedReconciler.SetupWithManager(mgr, b.skipNameValidation); err != nil {
-		return nil, fmt.Errorf("unable to create controller AtlasDataFederation: %w", err)
-	}
-
-	fedAuthReconciler := atlasfederatedauth.NewAtlasFederatedAuthReconciler(
-		mgr,
-		b.predicates,
-		b.atlasProvider,
-		b.deletionProtection,
-		b.logger,
-	)
-	if err = fedAuthReconciler.SetupWithManager(mgr, b.skipNameValidation); err != nil {
-		return nil, fmt.Errorf("unable to create controller AtlasFederatedAuth: %w", err)
-	}
-
-	streamsInstanceReconciler := atlasstream.NewAtlasStreamsInstanceReconciler(
-		mgr,
-		b.predicates,
-		b.atlasProvider,
-		b.deletionProtection,
-		b.logger,
-	)
-	if err = streamsInstanceReconciler.SetupWithManager(mgr, b.skipNameValidation); err != nil {
-		return nil, fmt.Errorf("unable to create controller AtlasStreamsInstance: %w", err)
-	}
-
-	streamsConnReconciler := atlasstream.NewAtlasStreamsConnectionReconciler(
-		mgr,
-		b.predicates,
-		b.atlasProvider,
-		b.deletionProtection,
-		b.logger,
-	)
-	if err = streamsConnReconciler.SetupWithManager(mgr, b.skipNameValidation); err != nil {
-		return nil, fmt.Errorf("unable to create controller AtlasStreamsConnection: %w", err)
-	}
-
-	searchIndexConfigReconciler := atlassearchindexconfig.NewAtlasSearchIndexConfigReconciler(
-		mgr,
-		b.predicates,
-		b.atlasProvider,
-		b.deletionProtection,
-		b.logger,
-	)
-	if err = searchIndexConfigReconciler.SetupWithManager(mgr, b.skipNameValidation); err != nil {
-		return nil, fmt.Errorf("unable to create controller AtlasSearchIndexConfig: %w", err)
-	}
-
-	bcpReconciler := atlasbackupcompliancepolicy.NewAtlasBackupCompliancePolicyReconciler(
-		mgr,
-		b.predicates,
-		b.atlasProvider,
-		b.deletionProtection,
-		b.logger,
-	)
-	if err = bcpReconciler.SetupWithManager(mgr, b.skipNameValidation); err != nil {
-		return nil, fmt.Errorf("unable to create controller AtlasBackupCompliancePolicy: %w", err)
-	}
-
-	customRolesReconciler := atlascustomrole.NewAtlasCustomRoleReconciler(
-		mgr,
-		b.predicates,
-		b.atlasProvider,
-		b.deletionProtection,
-		b.independentSyncPeriod,
-		b.logger,
-	)
-	if err = customRolesReconciler.SetupWithManager(mgr, b.skipNameValidation); err != nil {
-		return nil, fmt.Errorf("unable to create controller AtlasCustomRole: %w", err)
+	if err := controllerRegistry.RegisterWithManager(mgr, b.skipNameValidation, b.atlasProvider); err != nil {
+		return nil, err
 	}
 
-	peReconciler := atlasprivateendpoint.NewAtlasPrivateEndpointReconciler(
-		mgr,
-		b.predicates,
-		b.atlasProvider,
-		b.deletionProtection,
-		b.logger,
-	)
-	if err = peReconciler.SetupWithManager(mgr, b.skipNameValidation); err != nil {
-		return nil, fmt.Errorf("unable to create controller AtlasPrivateEndpoint: %w", err)
+	if err := indexer.RegisterAll(ctx, mgr, b.logger); err != nil {
+		return nil, fmt.Errorf("unable to create indexers: %w", err)
 	}
 
 	return mgr, nil

test/e2e/cache_watch_test.go

@@ -12,7 +12,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/cluster"
 
 	"github.com/mongodb/mongodb-atlas-kubernetes/v2/api"
 	akov2 "github.com/mongodb/mongodb-atlas-kubernetes/v2/api/v1"
@@ -44,12 +44,12 @@ var _ = Describe("Kubernetes cache watch test:", Label("cache-watch"), func() {
 			setupSecrets(ctx, testData, namespaces)
 			defer clearSecrets(ctx, testData, namespaces)
 
-			mgr, stop := setupManager(ctx, namespaces, testCase.watchNamespaces)
+			c, stop := setupCluster(ctx, namespaces, testCase.watchNamespaces)
 			defer stop()
 
 			wantToFindSet := sets.NewString(testCase.wantToFind...)
 			By("Using the manager cache to get all secrets in all namespaces and checking the expected results", func() {
-				cache := mgr.GetCache()
+				cache := c.GetCache()
 				for _, ns := range namespaces {
 					err := cache.Get(ctx, types.NamespacedName{Name: config.DefaultOperatorGlobalKey, Namespace: ns.GetName()}, &corev1.Secret{})
 
@@ -96,12 +96,12 @@ var _ = Describe("Kubernetes cache watch test:", Label("cache-watch"), func() {
 			setupSecrets(ctx, testData, namespaces)
 			defer clearSecrets(ctx, testData, namespaces)
 
-			mgr, stop := setupManager(ctx, namespaces, testCase.watchNamespaces)
+			c, stop := setupCluster(ctx, namespaces, testCase.watchNamespaces)
 			defer stop()
 
 			wantToFindSet := sets.NewString(testCase.wantToFind...)
 			By("Using the manager cache to list all secrets in all namespaces and checking the expected results", func() {
-				cache := mgr.GetCache()
+				cache := c.GetCache()
 				for _, ns := range namespaces {
 					err := cache.List(ctx, &corev1.SecretList{}, &client.ListOptions{Namespace: ns.GetName()})
 
@@ -157,7 +157,7 @@ var _ = Describe("Reconciles test:", func() {
 			setupSecrets(ctx, testData, namespaces)
 			defer clearSecrets(ctx, testData, namespaces)
 
-			_, stop := setupManager(ctx, namespaces, testCase.watchNamespaces)
+			_, stop := setupCluster(ctx, namespaces, testCase.watchNamespaces)
 			defer stop()
 
 			By("Launching an atlas project on each namespace, expect only listened namespaces to update status", func() {
@@ -286,10 +286,10 @@ func setupSecrets(ctx context.Context, testData *model.TestDataProvider, namespa
 
 type stopper func()
 
-func setupManager(ctx context.Context, namespaces []*corev1.Namespace, wantToWatch []string) (manager.Manager, stopper) {
+func setupCluster(ctx context.Context, namespaces []*corev1.Namespace, wantToWatch []string) (cluster.Cluster, stopper) {
 	var (
-		wg  sync.WaitGroup
-		mgr manager.Manager
+		wg sync.WaitGroup
+		c  cluster.Cluster
 	)
 
 	wantToWatchSet := sets.NewString(wantToWatch...)
@@ -315,13 +315,13 @@ func setupManager(ctx context.Context, namespaces []*corev1.Namespace, wantToWat
 		}
 
 		var err error
-		mgr, err = k8s.BuildManager(managerConfig)
+		c, err = k8s.BuildCluster(managerConfig)
 		Expect(err).NotTo(HaveOccurred())
 
 		wg.Add(1)
 		go func(ctx context.Context) {
 			defer wg.Done()
-			err := mgr.Start(ctx)
+			err := c.Start(ctx)
 			Expect(err).NotTo(HaveOccurred())
 		}(mgrCtx)
 	})
@@ -330,7 +330,7 @@ func setupManager(ctx context.Context, namespaces []*corev1.Namespace, wantToWat
 		// the first namespace should always be cache-accessible
 		Eventually(func(g Gomega) bool {
 			return g.Expect(
-				mgr.GetCache().Get(ctx, types.NamespacedName{Name: namespaces[0].GetName()}, &corev1.Namespace{}),
+				c.GetCache().Get(ctx, types.NamespacedName{Name: namespaces[0].GetName()}, &corev1.Namespace{}),
 			).To(Succeed())
 		}).WithTimeout(time.Minute).Should(BeTrue())
 	})
@@ -340,7 +340,7 @@ func setupManager(ctx context.Context, namespaces []*corev1.Namespace, wantToWat
 		wg.Wait()
 	}
 
-	return mgr, stopper
+	return c, stopper
 }
 
 func clearSecrets(ctx context.Context, testData *model.TestDataProvider, namespaces []*corev1.Namespace) {

test/e2e/configuration_test.go

@@ -159,10 +159,10 @@ var _ = Describe("Configuration namespaced. Deploy deployment", Label("deploymen
 })
 
 func mainCycle(testData *model.TestDataProvider) {
-	mgr := actions.PrepareOperatorConfigurations(testData)
+	r := actions.PrepareOperatorConfigurations(testData)
 	ctx := context.Background()
 	go func(ctx context.Context) {
-		err := mgr.Start(ctx)
+		err := r.Start(ctx)
 		Expect(err).NotTo(HaveOccurred())
 	}(ctx)