CLOUDP-80765: Handle Cluster pause/unpause (#89)

Author: vasilevp

config/crd/bases/atlas.mongodb.com_atlasclusters.yaml

@@ -143,6 +143,9 @@ spec:
                 maximum: 50
                 minimum: 1
                 type: integer
+              paused:
+                description: Flag that indicates whether the cluster should be paused.
+                type: boolean
               pitEnabled:
                 description: Flag that indicates the cluster uses continuous cloud
                   backups.

pkg/api/v1/atlascluster_types.go

@@ -83,6 +83,9 @@ type AtlasClusterSpec struct {
 	// +optional
 	NumShards *int `json:"numShards,omitempty"`
 
+	// Flag that indicates whether the cluster should be paused.
+	Paused *bool `json:"paused,omitempty"`
+
 	// Flag that indicates the cluster uses continuous cloud backups.
 	// +optional
 	PitEnabled *bool `json:"pitEnabled,omitempty"`

pkg/api/v1/atlascluster_types_test.go

@@ -10,8 +10,10 @@ import (
 	"go.mongodb.org/atlas/mongodbatlas"
 )
 
-var excludedClusterFieldsOurs = map[string]bool{}
-var excludedClusterFieldsTheirs = map[string]bool{}
+var (
+	excludedClusterFieldsOurs   = map[string]bool{}
+	excludedClusterFieldsTheirs = map[string]bool{}
+)
 
 func init() {
 	excludedClusterFieldsOurs["projectRef"] = true
@@ -32,9 +34,6 @@ func init() {
 	excludedClusterFieldsTheirs["connectionStrings"] = true
 	excludedClusterFieldsTheirs["srvAddress"] = true
 	excludedClusterFieldsTheirs["stateName"] = true
-
-	// CLOUDP-80765
-	excludedClusterFieldsTheirs["paused"] = true
 }
 
 func TestCompatibility(t *testing.T) {

pkg/api/v1/zz_generated.deepcopy.go

@@ -48,15 +48,28 @@ func (r *AtlasClusterReconciler) ensureClusterState(log *zap.SugaredLogger, conn
 
 	switch c.StateName {
 	case "IDLE":
+		if done, err := clusterMatchesSpec(log, c, cluster.Spec); err != nil {
+			return c, workflow.Terminate(workflow.Internal, err.Error())
+		} else if done {
+			return c, workflow.OK()
+		}
+
 		spec, err := cluster.Spec.Cluster()
 		if err != nil {
 			return c, workflow.Terminate(workflow.Internal, err.Error())
 		}
 
-		if done, err := clusterMatchesSpec(log, c, cluster.Spec); err != nil {
-			return c, workflow.Terminate(workflow.Internal, err.Error())
-		} else if done {
-			return c, workflow.OK()
+		if cluster.Spec.Paused != nil {
+			if c.Paused == nil || *c.Paused != *cluster.Spec.Paused {
+				// paused is different from Atlas
+				// we need to first send a special (un)pause request before reconciling everything else
+				spec = &mongodbatlas.Cluster{
+					Paused: cluster.Spec.Paused,
+				}
+			} else {
+				// otherwise, don't send the paused field
+				spec.Paused = nil
+			}
 		}
 
 		c, _, err = client.Clusters.Update(ctx, project.Status.ID, cluster.Spec.Name, spec)

pkg/controller/atlascluster/cluster.go

@@ -89,7 +89,6 @@ var _ = Describe("AtlasCluster", func() {
 			)))
 			Expect(createdCluster.Status.ObservedGeneration).To(Equal(createdCluster.Generation))
 			Expect(createdCluster.Status.ObservedGeneration).To(Equal(lastGeneration + 1))
-			lastGeneration++
 		})
 	}
 
@@ -119,7 +118,7 @@ var _ = Describe("AtlasCluster", func() {
 		Eventually(testutil.WaitFor(k8sClient, createdCluster, status.TrueCondition(status.ReadyType), validateClusterUpdatingFunc()),
 			1200, interval).Should(BeTrue())
 
-		doCommonChecks()
+		lastGeneration++
 	}
 
 	Describe("Create/Update the cluster", func() {
@@ -142,12 +141,14 @@ var _ = Describe("AtlasCluster", func() {
 			By("Updating the Cluster labels", func() {
 				createdCluster.Spec.Labels = []mdbv1.LabelSpec{{Key: "int-test", Value: "true"}}
 				performUpdate()
+				doCommonChecks()
 				checkAtlasState()
 			})
 
 			By("Updating the Cluster backups settings", func() {
 				createdCluster.Spec.ProviderBackupEnabled = boolptr(true)
 				performUpdate()
+				doCommonChecks()
 				checkAtlasState(func(c *mongodbatlas.Cluster) {
 					Expect(c.ProviderBackupEnabled).To(Equal(createdCluster.Spec.ProviderBackupEnabled))
 				})
@@ -156,13 +157,52 @@ var _ = Describe("AtlasCluster", func() {
 			By("Decreasing the Cluster disk size", func() {
 				createdCluster.Spec.DiskSizeGB = intptr(10)
 				performUpdate()
+				doCommonChecks()
 				checkAtlasState(func(c *mongodbatlas.Cluster) {
 					Expect(*c.DiskSizeGB).To(BeEquivalentTo(*createdCluster.Spec.DiskSizeGB))
 
 					// check whether https://github.com/mongodb/go-client-mongodb-atlas/issues/140 is fixed
 					Expect(c.DiskSizeGB).To(BeAssignableToTypeOf(float64ptr(0)), "DiskSizeGB is no longer a *float64, please check the spec!")
 				})
 			})
+
+			By("Pausing the cluster", func() {
+				createdCluster.Spec.Paused = boolptr(true)
+				performUpdate()
+				doCommonChecks()
+				checkAtlasState(func(c *mongodbatlas.Cluster) {
+					Expect(c.Paused).To(Equal(createdCluster.Spec.Paused))
+				})
+			})
+
+			By("Updating the Cluster configuration while paused (should fail)", func() {
+				createdCluster.Spec.ProviderBackupEnabled = boolptr(false)
+
+				Expect(k8sClient.Update(context.Background(), createdCluster)).To(Succeed())
+				Eventually(
+					testutil.WaitFor(k8sClient, createdCluster, status.FalseCondition(status.ClusterReadyType).WithReason(string(workflow.ClusterNotCreatedInAtlas))),
+					60,
+					interval,
+				).Should(BeTrue())
+
+				lastGeneration++
+			})
+
+			By("Unpausing the cluster", func() {
+				createdCluster.Spec.Paused = boolptr(false)
+				performUpdate()
+				doCommonChecks()
+				checkAtlasState(func(c *mongodbatlas.Cluster) {
+					Expect(c.Paused).To(Equal(createdCluster.Spec.Paused))
+				})
+			})
+
+			By("Checking that modifications were applied after unpausing", func() {
+				doCommonChecks()
+				checkAtlasState(func(c *mongodbatlas.Cluster) {
+					Expect(c.ProviderBackupEnabled).To(Equal(createdCluster.Spec.ProviderBackupEnabled))
+				})
+			})
 		})
 	})
 })
@@ -200,7 +240,7 @@ func validateClusterUpdatingFunc() func(a mdbv1.AtlasCustomResource) {
 		}
 		// When the create request has been made to Atlas - we expect the following status
 		if !isIdle {
-			Expect(c.Status.StateName).To(Equal("UPDATING"), fmt.Sprintf("Current conditions: %+v", c.Status.Conditions))
+			Expect(c.Status.StateName).To(Or(Equal("UPDATING"), Equal("REPAIRING")), fmt.Sprintf("Current conditions: %+v", c.Status.Conditions))
 			expectedConditionsMatchers := testutil.MatchConditions(
 				status.FalseCondition(status.ClusterReadyType).WithReason(string(workflow.ClusterUpdating)).WithMessageRegexp("cluster is updating"),
 				status.FalseCondition(status.ReadyType),