fix issue-314 by checking zookeeperCluster resource version before up…

…dating sts (#326)

* fix issue-314 by checking zookeeperCluster resource version before updating sts

* move the checking earlier

* go fmt

* rename the function and add unit test

* go fmt

pkg/controller/zookeepercluster/zookeepercluster_controller.go

@@ -164,6 +164,28 @@ func (r *ReconcileZookeeperCluster) Reconcile(request reconcile.Request) (reconc
 	return reconcile.Result{RequeueAfter: ReconcileTime}, nil
 }
 
+func zookeeperClusterFresherThanSts(instance *zookeeperv1beta1.ZookeeperCluster, sts *appsv1.StatefulSet) bool {
+	labeledRV, ok := sts.Labels["owner-rv"]
+	if !ok {
+		log.Info("Fail to find owner-rv in statefulset's labels; skip checking resource version")
+		return true
+	}
+	largerRV, err := strconv.Atoi(instance.ResourceVersion)
+	if err != nil {
+		log.Info("Fail to convert %s to integer; skip checking resource version", instance.ResourceVersion)
+		return true
+	}
+	smallerRV, err := strconv.Atoi(labeledRV)
+	if err != nil {
+		log.Info("Fail to convert %s to integer; skip checking resource version", labeledRV)
+		return true
+	}
+	if largerRV < smallerRV {
+		return false
+	}
+	return true
+}
+
 func (r *ReconcileZookeeperCluster) reconcileStatefulSet(instance *zookeeperv1beta1.ZookeeperCluster) (err error) {
 
 	// we cannot upgrade if cluster is in UpgradeFailed
@@ -201,6 +223,8 @@ func (r *ReconcileZookeeperCluster) reconcileStatefulSet(instance *zookeeperv1be
 		r.log.Info("Creating a new Zookeeper StatefulSet",
 			"StatefulSet.Namespace", sts.Namespace,
 			"StatefulSet.Name", sts.Name)
+		// label the RV of the zookeeperCluster when creating the sts
+		sts.Labels["owner-rv"] = instance.ResourceVersion
 		err = r.client.Create(context.TODO(), sts)
 		if err != nil {
 			return err
@@ -209,6 +233,10 @@ func (r *ReconcileZookeeperCluster) reconcileStatefulSet(instance *zookeeperv1be
 	} else if err != nil {
 		return err
 	} else {
+		// check whether zookeeperCluster is updated before updating the sts
+		if !zookeeperClusterFresherThanSts(instance, sts) {
+			return fmt.Errorf("Staleness: cr.ResourceVersion %s is smaller than labeledRV %s", instance.ResourceVersion, sts.Labels["owner-rv"])
+		}
 		foundSTSSize := *foundSts.Spec.Replicas
 		newSTSSize := *sts.Spec.Replicas
 		if newSTSSize != foundSTSSize {

pkg/controller/zookeepercluster/zookeepercluster_controller_test.go

@@ -563,5 +563,36 @@ var _ = Describe("ZookeeperCluster Controller", func() {
 				Ω(err).To(BeNil())
 			})
 		})
+
+		Context("Checking resource version", func() {
+			var (
+				sts *appsv1.StatefulSet
+			)
+
+			BeforeEach(func() {
+				z.WithDefaults()
+				z.ResourceVersion = "100"
+				sts = &appsv1.StatefulSet{}
+				sts.Labels = make(map[string]string)
+			})
+
+			It("should return true as 99 < 100", func() {
+				sts.Labels["owner-rv"] = "99"
+				updated := zookeeperClusterFresherThanSts(z, sts)
+				Ω(updated).To(BeTrue())
+			})
+
+			It("should return true as 100 == 100", func() {
+				sts.Labels["owner-rv"] = "100"
+				updated := zookeeperClusterFresherThanSts(z, sts)
+				Ω(updated).To(BeTrue())
+			})
+
+			It("should return false as 101 > 100", func() {
+				sts.Labels["owner-rv"] = "101"
+				updated := zookeeperClusterFresherThanSts(z, sts)
+				Ω(updated).To(BeFalse())
+			})
+		})
 	})
 })