CLOUDP-261715: Enabled resync for Atlas custom resources (#1707)

Enabled periodic reconciliations for Atlas custom resources

pkg/controller/atlasdatabaseuser/atlasdatabaseuser_controller.go

@@ -263,7 +263,7 @@ func (r *AtlasDatabaseUserReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		Named("AtlasDatabaseUser").
 		For(&akov2.AtlasDatabaseUser{}, builder.WithPredicates(r.GlobalPredicates...)).
-		Watches(&corev1.Secret{}, watch.NewSecretHandler(&r.DeprecatedResourceWatcher)).
+		Watches(&corev1.Secret{}, watch.NewSecretHandler(&r.DeprecatedResourceWatcher), builder.WithPredicates(predicate.ResourceVersionChangedPredicate{})).
 		Complete(r)
 }
 

pkg/controller/atlasproject/atlasproject_controller.go

@@ -380,6 +380,7 @@ func (r *AtlasProjectReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		Watches(
 			&corev1.Secret{},
 			handler.EnqueueRequestsFromMapFunc(newProjectsMapFunc[corev1.Secret](indexer.AtlasProjectBySecretsIndex, r.Client, r.Log)),
+			builder.WithPredicates(predicate.ResourceVersionChangedPredicate{}),
 		).
 		Watches(
 			&akov2.AtlasTeam{},

pkg/controller/watch/predicates.go

@@ -13,6 +13,11 @@ import (
 func CommonPredicates() predicate.Funcs {
 	return predicate.Funcs{
 		UpdateFunc: func(e event.UpdateEvent) bool {
+			if e.ObjectOld.GetResourceVersion() == e.ObjectNew.GetResourceVersion() {
+				// resource version didn't change, so this is a resync, allow reconciliation.
+				return true
+			}
+
 			if e.ObjectOld.GetGeneration() == e.ObjectNew.GetGeneration() && reflect.DeepEqual(e.ObjectNew.GetFinalizers(), e.ObjectOld.GetFinalizers()) {
 				return false
 			}

test/int/integration_suite_test.go

@@ -203,6 +203,46 @@ func prepareControllers(deletionProtection bool) (*corev1.Namespace, context.Can
 	return &namespace, managerCancelFunc
 }
 
+func prepareControllersWithSyncPeriod(deletionProtection bool, syncPeriod time.Duration) (*corev1.Namespace, context.CancelFunc) {
+	var ctx context.Context
+	ctx, managerCancelFunc = context.WithCancel(context.Background())
+	namespace = corev1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace:    "test",
+			GenerateName: "test",
+		},
+	}
+
+	By("Creating the namespace " + namespace.GenerateName + "...")
+	Expect(k8sClient.Create(ctx, &namespace)).ToNot(HaveOccurred())
+	Expect(namespace.Name).ToNot(BeEmpty())
+	GinkgoWriter.Printf("Generated namespace %q\n", namespace.Name)
+
+	logger := ctrzap.NewRaw(ctrzap.UseDevMode(true), ctrzap.WriteTo(GinkgoWriter), ctrzap.StacktraceLevel(zap.ErrorLevel))
+	ctrl.SetLogger(zapr.NewLogger(logger))
+
+	// shallow copy global config
+	managerCfg := *cfg
+	managerCfg.UserAgent = "AKO"
+	mgr, err := operator.NewBuilder(operator.ManagerProviderFunc(ctrl.NewManager), scheme.Scheme).
+		WithConfig(&managerCfg).
+		WithNamespaces(namespace.Name).
+		WithLogger(logger).
+		WithAtlasDomain(atlasDomain).
+		WithSyncPeriod(syncPeriod).
+		WithAPISecret(client.ObjectKey{Name: "atlas-operator-api-key", Namespace: namespace.Name}).
+		WithDeletionProtection(deletionProtection).
+		Build(ctx)
+	Expect(err).ToNot(HaveOccurred())
+
+	go func() {
+		err = mgr.Start(ctx)
+		Expect(err).ToNot(HaveOccurred())
+	}()
+
+	return &namespace, managerCancelFunc
+}
+
 func removeControllersAndNamespace() {
 	// end the manager
 	managerCancelFunc()

test/int/syncperiod_test.go

@@ -0,0 +1,102 @@
+package int
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	corev1 "k8s.io/api/core/v1"
+
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/internal/kube"
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/pkg/api"
+	akov2 "github.com/mongodb/mongodb-atlas-kubernetes/v2/pkg/api/v1"
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/test/helper/conditions"
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/test/helper/events"
+	"github.com/mongodb/mongodb-atlas-kubernetes/v2/test/helper/resources"
+)
+
+var _ = Describe("Sync Period test", Label("int", "sync-period"), func() {
+	const interval = time.Second * 2
+	const syncInterval = 40 * time.Second
+
+	var (
+		connectionSecret        corev1.Secret
+		createdProject          *akov2.AtlasProject
+		previousResourceVersion string
+	)
+
+	BeforeEach(func() {
+		prepareControllersWithSyncPeriod(false, syncInterval)
+
+		createdProject = &akov2.AtlasProject{}
+
+		connectionSecret = buildConnectionSecret("my-atlas-key")
+		By(fmt.Sprintf("Creating the Secret %s", kube.ObjectKeyFromObject(&connectionSecret)))
+		Expect(k8sClient.Create(context.Background(), &connectionSecret)).ToNot(HaveOccurred())
+	})
+
+	AfterEach(func() {
+		if createdProject != nil && createdProject.Status.ID != "" {
+			By("Removing Atlas Project " + createdProject.Status.ID)
+			Eventually(deleteK8sObject(createdProject), 20, interval).Should(BeTrue())
+			Eventually(checkAtlasProjectRemoved(createdProject.Status.ID), 20, interval).Should(BeTrue())
+		}
+		removeControllersAndNamespace()
+	})
+	It("Should reconcile after defined SyncPeriod", func() {
+		By("Should Succeed with creating the project", func() {
+			expectedProject := akov2.DefaultProject(namespace.Name, connectionSecret.Name)
+			createdProject.ObjectMeta = expectedProject.ObjectMeta
+			Expect(k8sClient.Create(context.Background(), expectedProject)).ToNot(HaveOccurred())
+
+			Eventually(func() bool {
+				return resources.CheckCondition(k8sClient, createdProject, api.TrueCondition(api.ReadyType))
+			}).WithTimeout(ProjectCreationTimeout).WithPolling(interval).Should(BeTrue())
+
+			projectReadyConditions := conditions.MatchConditions(
+				api.TrueCondition(api.ProjectReadyType),
+				api.TrueCondition(api.ReadyType),
+				api.TrueCondition(api.ValidationSucceeded),
+			)
+			Expect(createdProject.Status.ID).NotTo(BeNil())
+			Expect(createdProject.Status.Conditions).To(ContainElements((projectReadyConditions)))
+			Expect(createdProject.Status.ObservedGeneration).To(Equal(createdProject.Generation))
+
+			atlasProject, _, err := atlasClient.ProjectsApi.
+				GetProject(context.Background(), createdProject.Status.ID).
+				Execute()
+			Expect(err).ToNot(HaveOccurred())
+
+			Expect(atlasProject.Name).To(Equal(expectedProject.Spec.Name))
+
+			events.EventExists(k8sClient, createdProject, "Normal", "Ready", "")
+
+			Eventually(func(g Gomega) bool {
+				if !resources.ReadAtlasResource(context.Background(), k8sClient, createdProject) {
+					return false
+				}
+				previousResourceVersion = createdProject.ResourceVersion
+				return true
+			}).WithTimeout(10 * time.Second).WithPolling(2 * time.Second).Should(BeTrue())
+		})
+
+		By(fmt.Sprintf("Should wait for at least %f seconds", (syncInterval*2).Seconds()), func() {
+			time.Sleep(syncInterval * 2)
+		})
+
+		By("Project resource version should be different", func() {
+			var currentResourceVersion string
+			Eventually(func(g Gomega) bool {
+				if !resources.ReadAtlasResource(context.Background(), k8sClient, createdProject) {
+					return false
+				}
+				currentResourceVersion = createdProject.ResourceVersion
+				return true
+			}).WithTimeout(10 * time.Second).WithPolling(2 * time.Second).Should(BeTrue())
+
+			Expect(currentResourceVersion).ToNot(BeEquivalentTo(previousResourceVersion))
+		})
+	})
+})