dry run manager (#2051)

Co-authored-by: Sergiusz Urbaniak <[email protected]>

Author: igor-karpukhin

.github/workflows/test-e2e.yml

@@ -184,6 +184,7 @@ jobs:
             "backup-compliance",
             "flex",
             "ip-access-list",
+            "dry-run",
           ]
     steps:
       - name: Get repo files from cache

cmd/main.go

@@ -75,7 +75,7 @@ func main() {
 	setupLog := logger.Named("setup").Sugar()
 	setupLog.Info("starting with configuration", zap.Any("config", config), zap.Any("version", version.Version))
 
-	mgr, err := operator.NewBuilder(operator.ManagerProviderFunc(ctrl.NewManager), akoScheme, time.Duration(minimumIndependentSyncPeriod)*time.Minute).
+	runnable, err := operator.NewBuilder(operator.ManagerProviderFunc(ctrl.NewManager), akoScheme, time.Duration(minimumIndependentSyncPeriod)*time.Minute).
 		WithConfig(ctrl.GetConfigOrDie()).
 		WithNamespaces(collection.Keys(config.WatchedNamespaces)...).
 		WithLogger(logger).
@@ -86,6 +86,7 @@ func main() {
 		WithAPISecret(config.GlobalAPISecret).
 		WithDeletionProtection(config.ObjectDeletionProtection).
 		WithIndependentSyncPeriod(time.Duration(config.IndependentSyncPeriod) * time.Minute).
+		WithDryRun(config.DryRun).
 		Build(ctx)
 	if err != nil {
 		setupLog.Error(err, "unable to start operator")
@@ -94,8 +95,8 @@ func main() {
 
 	setupLog.Info(subobjectDeletionProtectionMessage)
 	setupLog.Info("starting manager")
-	if err = mgr.Start(ctx); err != nil {
-		setupLog.Error(err, "problem running manager")
+	if err = runnable.Start(ctx); err != nil {
+		setupLog.Errorf("error running manager: %v", err)
 		os.Exit(1)
 	}
 }
@@ -113,6 +114,7 @@ type Config struct {
 	SubObjectDeletionProtection bool
 	IndependentSyncPeriod       int
 	FeatureFlags                *featureflags.FeatureFlags
+	DryRun                      bool
 }
 
 // ParseConfiguration fills the 'OperatorConfig' from the flags passed to the program
@@ -139,6 +141,8 @@ func parseConfiguration() Config {
 		independentSyncPeriod,
 		fmt.Sprintf("The default time, in minutes,  between reconciliations for independent custom resources. (default %d, minimum %d)", independentSyncPeriod, minimumIndependentSyncPeriod),
 	)
+	flag.BoolVar(&config.DryRun, "dry-run", false, "If set, the operator will not perform any changes to the Atlas resources, run all reconcilers only Once and emit events for all planned changes")
+
 	appVersion := flag.Bool("v", false, "prints application version")
 	flag.Parse()
 

internal/controller/atlasproject/project.go

@@ -36,7 +36,10 @@ func (r *AtlasProjectReconciler) handleProject(ctx *workflow.Context, orgID stri
 	case !existInAtlas && wasDeleted:
 		return r.release(ctx, atlasProject)
 	case existInAtlas && !wasDeleted && atlasProject.Status.ID == "":
-		return r.manage(ctx, atlasProject, projectInAtlas.ID)
+		// short circuit the "manage" state,
+		// there is no need to wait another reconcile cycle to continue.
+		_, _ = r.manage(ctx, atlasProject, projectInAtlas.ID)
+		atlasProject.Status.ID = projectInAtlas.ID
 	}
 
 	ctx.SetConditionTrue(api.ProjectReadyType)

internal/controller/registry.go

@@ -6,6 +6,7 @@ import (
 
 	"go.uber.org/zap"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/cluster"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
@@ -24,16 +25,14 @@ import (
 	"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/watch"
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/dryrun"
 	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/featureflags"
 )
 
-type ManagerAware interface {
-	SetupWithManager(mgr ctrl.Manager, skipNameValidation bool) error
-}
-
-type AkoReconciler interface {
+type Reconciler interface {
 	reconcile.Reconciler
-	ManagerAware
+	For() (client.Object, builder.Predicates)
+	SetupWithManager(mgr ctrl.Manager, skipNameValidation bool) error
 }
 
 type Registry struct {
@@ -43,7 +42,7 @@ type Registry struct {
 	featureFlags          *featureflags.FeatureFlags
 
 	logger      *zap.Logger
-	reconcilers []AkoReconciler
+	reconcilers []Reconciler
 }
 
 func NewRegistry(predicates []predicate.Predicate, deletionProtection bool, logger *zap.Logger, independentSyncPeriod time.Duration, featureFlags *featureflags.FeatureFlags) *Registry {
@@ -56,6 +55,16 @@ func NewRegistry(predicates []predicate.Predicate, deletionProtection bool, logg
 	}
 }
 
+func (r *Registry) RegisterWithDryRunManager(mgr *dryrun.Manager, ap atlas.Provider) error {
+	r.registerControllers(mgr, ap)
+
+	for _, reconciler := range r.reconcilers {
+		mgr.SetupReconciler(reconciler)
+	}
+
+	return nil
+}
+
 func (r *Registry) RegisterWithManager(mgr ctrl.Manager, skipNameValidation bool, ap atlas.Provider) error {
 	r.registerControllers(mgr, ap)
 
@@ -68,7 +77,11 @@ func (r *Registry) RegisterWithManager(mgr ctrl.Manager, skipNameValidation bool
 }
 
 func (r *Registry) registerControllers(c cluster.Cluster, ap atlas.Provider) {
-	var reconcilers []AkoReconciler
+	if len(r.reconcilers) > 0 {
+		return
+	}
+
+	var reconcilers []Reconciler
 	reconcilers = append(reconcilers, atlasproject.NewAtlasProjectReconciler(c, r.deprecatedPredicates(), ap, r.deletionProtection, r.logger))
 	reconcilers = append(reconcilers, atlasdeployment.NewAtlasDeploymentReconciler(c, r.deprecatedPredicates(), ap, r.deletionProtection, r.independentSyncPeriod, r.logger))
 	reconcilers = append(reconcilers, atlasdatabaseuser.NewAtlasDatabaseUserReconciler(c, r.deprecatedPredicates(), ap, r.deletionProtection, r.independentSyncPeriod, r.featureFlags, r.logger))

internal/controller/workflow/result.go

@@ -4,6 +4,8 @@ import (
 	"time"
 
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/dryrun"
 )
 
 const (
@@ -41,6 +43,8 @@ func Requeue(period time.Duration) Result {
 // 'reason' and 'message' indicate the error state and are supposed to be reflected in the `conditions` for the
 // reconciled Custom Resource.
 func Terminate(reason ConditionReason, err error) Result {
+	dryrun.AddTerminationError(err) // TODO: factor this in favor of controller-runtime error handling
+
 	return Result{
 		terminated:   true,
 		requeueAfter: DefaultRetry,
@@ -78,6 +82,8 @@ func (r Result) IsDeleted() bool {
 // TerminateSilently indicates that the reconciliation logic cannot proceed and needs to be finished (and possibly requeued)
 // The status of the reconciled Custom Resource is not supposed to be updated.
 func TerminateSilently(err error) Result {
+	dryrun.AddTerminationError(err)
+
 	return Result{terminated: true, requeueAfter: DefaultRetry}
 }
 

internal/dryrun/error.go

@@ -1,45 +1,33 @@
 package dryrun
 
 import (
+	"errors"
 	"fmt"
-
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime/schema"
 )
 
+const dryRunErrorPrefix = "DryRun event: "
+
 type DryRunError struct {
-	GVK                    string
-	Namespace, Name        string
-	EventType, Reason, Msg string
+	Msg string
 }
 
-func NewDryRunError(kind schema.ObjectKind, meta metav1.ObjectMetaAccessor, eventtype, reason, messageFmt string, args ...interface{}) error {
-	gvk := "unknown"
-	if kind != nil {
-		gvk = kind.GroupVersionKind().String()
-	}
-
-	namespace, name := "unknown", "unknown"
-	if meta != nil {
-		namespace = meta.GetObjectMeta().GetNamespace()
-		name = meta.GetObjectMeta().GetName()
-	}
-
+func NewDryRunError(messageFmt string, args ...interface{}) error {
 	msg := fmt.Sprintf(messageFmt, args...)
 
 	return &DryRunError{
-		GVK:       gvk,
-		Namespace: namespace,
-		Name:      name,
-		EventType: eventtype,
-		Reason:    reason,
-		Msg:       msg,
+		Msg: msg,
 	}
 }
 
 func (e *DryRunError) Error() string {
-	return fmt.Sprintf(
-		"DryRun event GVK=%v, Namespace=%v, Name=%v, EventType=%v, Reason=%v, Message=%v",
-		e.GVK, e.Namespace, e.Name, e.EventType, e.Reason, e.Msg,
-	)
+	return dryRunErrorPrefix + e.Msg
+}
+
+// containsDryRunErrors returns true if the given error contains at least one DryRunError.
+//
+// Note: we DO NOT want to export this as we do not want "special dry-run" cases in reconcilers.
+// Reconcilers should behave exactly the same during dry-run as during regular reconciles.
+func containsDryRunErrors(err error) bool {
+	dErr := &DryRunError{}
+	return errors.As(err, &dErr)
 }

internal/dryrun/error_queue.go

@@ -0,0 +1,50 @@
+package dryrun
+
+import (
+	"errors"
+	"sync"
+)
+
+type errorQueue struct {
+	mu sync.Mutex // protects fields below
+
+	active bool
+	errs   []error
+}
+
+var reconcileErrors = &errorQueue{}
+
+func AddTerminationError(err error) {
+	reconcileErrors.mu.Lock()
+	defer reconcileErrors.mu.Unlock()
+
+	if !reconcileErrors.active {
+		return
+	}
+
+	reconcileErrors.errs = append(reconcileErrors.errs, err)
+}
+
+func terminationError() error {
+	reconcileErrors.mu.Lock()
+	defer reconcileErrors.mu.Unlock()
+
+	result := make([]error, 0, len(reconcileErrors.errs))
+	result = append(result, reconcileErrors.errs...)
+
+	return errors.Join(result...)
+}
+
+func clearTerminationErrors() {
+	reconcileErrors.mu.Lock()
+	defer reconcileErrors.mu.Unlock()
+
+	reconcileErrors.errs = nil
+}
+
+func enableErrors() {
+	reconcileErrors.mu.Lock()
+	defer reconcileErrors.mu.Unlock()
+
+	reconcileErrors.active = true
+}