Check Orphan PVC before updating STS

- Move Status.Repicas and Status.ReadyReplicas update to reconcileClusterStatus - Use StatefulSet Replicas when checking orphan PVC - Check need for PVC cleanup before updating STS Signed-off-by: hoyhbx <[email protected]>
pravega · Mar 18, 2023 · 6bfd811 · 6bfd811
1 parent cd732b9
commit 6bfd811
Showing 1 changed file with 41 additions and 6 deletions.
diff --git a/controllers/zookeepercluster_controller.go b/controllers/zookeepercluster_controller.go
@@ -211,6 +211,20 @@ func (r *ZookeeperClusterReconciler) reconcileStatefulSet(instance *zookeeperv1b
 	} else if err != nil {
 		return err
 	} else {
+		// check whether orphans PVCs need to be deleted before updating the sts
+		if instance.Spec.Persistence != nil &&
+			instance.Spec.Persistence.VolumeReclaimPolicy == zookeeperv1beta1.VolumeReclaimPolicyDelete {
+			pvcCount, err := r.getPVCCount(instance)
+			if err != nil {
+				return err
+			}
+			r.Log.Info("PVC count", "count", pvcCount, "replicas", foundSts.Status.Replicas, "cr replicas", instance.Spec.Replicas)
+			if pvcCount > int(foundSts.Status.Replicas) {
+				r.Log.Info("Deleting PVCs", "count", pvcCount, "replicas", instance.Status.Replicas)
+				return nil
+			}
+		}
+
 		// check whether zookeeperCluster is updated before updating the sts
 		cmp := compareResourceVersion(instance, foundSts)
 		if cmp < 0 {
@@ -258,8 +272,6 @@ func (r *ZookeeperClusterReconciler) updateStatefulSet(instance *zookeeperv1beta
 	if err != nil {
 		return err
 	}
-	instance.Status.Replicas = foundSts.Status.Replicas
-	instance.Status.ReadyReplicas = foundSts.Status.ReadyReplicas
 	return nil
 }
 
@@ -558,6 +570,15 @@ func (r *ZookeeperClusterReconciler) reconcileClusterStatus(instance *zookeeperv
 	instance.Status.Members.Ready = readyMembers
 	instance.Status.Members.Unready = unreadyMembers
 
+	foundSts := &appsv1.StatefulSet{}
+	err = r.Client.Get(context.TODO(), types.NamespacedName{
+		Name:      instance.GetName(),
+		Namespace: instance.Namespace,
+	}, foundSts)
+
+	instance.Status.Replicas = foundSts.Status.Replicas
+	instance.Status.ReadyReplicas = foundSts.Status.ReadyReplicas
+
 	// If Cluster is in a ready state...
 	if instance.Spec.Replicas == instance.Status.ReadyReplicas && (!instance.Status.MetaRootCreated) {
 		r.Log.Info("Cluster is Ready, Creating ZK Metadata...")
@@ -707,21 +728,35 @@ func (r *ZookeeperClusterReconciler) getPVCCount(instance *zookeeperv1beta1.Zook
 }
 
 func (r *ZookeeperClusterReconciler) cleanupOrphanPVCs(instance *zookeeperv1beta1.ZookeeperCluster) (err error) {
+	// get the up to date STS
+	foundSts := &appsv1.StatefulSet{}
+	err = r.Client.Get(context.TODO(), types.NamespacedName{
+		Name:      instance.GetName(),
+		Namespace: instance.Namespace,
+	}, foundSts)
+	if err != nil {
+		return err
+	}
+
 	// this check should make sure we do not delete the PVCs before the STS has scaled down
-	if instance.Status.ReadyReplicas == instance.Spec.Replicas {
+	if foundSts.Status.ReadyReplicas == foundSts.Status.Replicas {
 		pvcCount, err := r.getPVCCount(instance)
 		if err != nil {
 			return err
 		}
-		r.Log.Info("cleanupOrphanPVCs", "PVC Count", pvcCount, "ReadyReplicas Count", instance.Status.ReadyReplicas)
-		if pvcCount > int(instance.Spec.Replicas) {
+
+		r.Log.Info("cleanupOrphanPVCs",
+			"PVC Count", pvcCount,
+			"Replicas Count", foundSts.Spec.Replicas)
+		if pvcCount > int(*foundSts.Spec.Replicas) {
 			pvcList, err := r.getPVCList(instance)
 			if err != nil {
 				return err
 			}
 			for _, pvcItem := range pvcList.Items {
 				// delete only Orphan PVCs
-				if utils.IsPVCOrphan(pvcItem.Name, instance.Spec.Replicas) {
+				if utils.IsPVCOrphan(pvcItem.Name, *foundSts.Spec.Replicas) {
+					r.Log.Info("cleanupOrphanPVCs", "Deleting Orphan PVC", pvcItem.Name)
 					r.deletePVC(pvcItem)
 				}
 			}