Issue 94: Fix for zk scale up problem

pkg/controller/zookeepercluster/zookeepercluster_controller.go

@@ -201,7 +201,7 @@ func (r *ReconcileZookeeperCluster) reconcileStatefulSet(instance *zookeeperv1be
 				failing which `zookeeperTeardown.sh` won't get invoked for the pods that are being scaled down
 				and these will stay in the ensemble config forever.
 				For details see:
-				//https://kubernetes.io/docs/tasks/configure-pod-container/configure-pod-configmap/#mounted-configmaps-are-updated-automatically
+				https://kubernetes.io/docs/tasks/configure-pod-container/configure-pod-configmap/#mounted-configmaps-are-updated-automatically
 			*/
 			r.skipSTSReconcile++
 			if r.skipSTSReconcile < 6 {
@@ -532,9 +532,10 @@ func (r *ReconcileZookeeperCluster) reconcileFinalizers(instance *zookeeperv1bet
 				return err
 			}
 		}
+		return r.cleanupOrphanPVCs(instance)
 	} else {
 		if utils.ContainsString(instance.ObjectMeta.Finalizers, utils.ZkFinalizer) {
-			if err = r.cleanUpZookeeperPVC(instance); err != nil {
+			if err = r.cleanUpAllPVCs(instance); err != nil {
 				return err
 			}
 			instance.ObjectMeta.Finalizers = utils.RemoveString(instance.ObjectMeta.Finalizers, utils.ZkFinalizer)
@@ -546,7 +547,40 @@ func (r *ReconcileZookeeperCluster) reconcileFinalizers(instance *zookeeperv1bet
 	return nil
 }
 
-func (r *ReconcileZookeeperCluster) cleanUpZookeeperPVC(instance *zookeeperv1beta1.ZookeeperCluster) (err error) {
+func (r *ReconcileZookeeperCluster) getPVCCount(instance *zookeeperv1beta1.ZookeeperCluster) (pvcCount int, err error) {
+	pvcList, err := r.getPVCList(instance)
+	if err != nil {
+		return -1, err
+	}
+	pvcCount = len(pvcList.Items)
+	return pvcCount, nil
+}
+
+func (r *ReconcileZookeeperCluster) cleanupOrphanPVCs(instance *zookeeperv1beta1.ZookeeperCluster) (err error) {
+	// this check should make sure we do not delete the PVCs before the STS has scaled down
+	if instance.Status.ReadyReplicas == instance.Spec.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) {
+			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) {
+					r.deletePVC(pvcItem)
+				}
+			}
+		}
+	}
+	return nil
+}
+
+func (r *ReconcileZookeeperCluster) getPVCList(instance *zookeeperv1beta1.ZookeeperCluster) (pvList corev1.PersistentVolumeClaimList, err error) {
 	selector, err := metav1.LabelSelectorAsSelector(&metav1.LabelSelector{
 		MatchLabels: map[string]string{"app": instance.GetName()},
 	})
@@ -556,20 +590,30 @@ func (r *ReconcileZookeeperCluster) cleanUpZookeeperPVC(instance *zookeeperv1bet
 	}
 	pvcList := &corev1.PersistentVolumeClaimList{}
 	err = r.client.List(context.TODO(), pvclistOps, pvcList)
+	return *pvcList, err
+}
+
+func (r *ReconcileZookeeperCluster) cleanUpAllPVCs(instance *zookeeperv1beta1.ZookeeperCluster) (err error) {
+	pvcList, err := r.getPVCList(instance)
 	if err != nil {
 		return err
 	}
 	for _, pvcItem := range pvcList.Items {
-		pvcDelete := &corev1.PersistentVolumeClaim{
-			ObjectMeta: metav1.ObjectMeta{
-				Name:      pvcItem.Name,
-				Namespace: pvcItem.Namespace,
-			},
-		}
-		err = r.client.Delete(context.TODO(), pvcDelete)
-		if err != nil {
-			return err
-		}
+		r.deletePVC(pvcItem)
 	}
 	return nil
 }
+
+func (r *ReconcileZookeeperCluster) deletePVC(pvcItem corev1.PersistentVolumeClaim) {
+	pvcDelete := &corev1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      pvcItem.Name,
+			Namespace: pvcItem.Namespace,
+		},
+	}
+	r.log.Info("Deleting PVC", "With Name", pvcItem.Name)
+	err := r.client.Delete(context.TODO(), pvcDelete)
+	if err != nil {
+		r.log.Error(err, "Error deleteing PVC.", "Name", pvcDelete.Name)
+	}
+}

pkg/utils/finalizer_utils.go

@@ -10,6 +10,11 @@
 
 package utils
 
+import (
+	"strconv"
+	"strings"
+)
+
 const (
 	ZkFinalizer = "cleanUpZookeeperPVC"
 )
@@ -32,3 +37,17 @@ func RemoveString(slice []string, str string) (result []string) {
 	}
 	return result
 }
+
+func IsPVCOrphan(zkPvcName string, replicas int32) bool {
+	index := strings.LastIndexAny(zkPvcName, "-")
+	if index == -1 {
+		return false
+	}
+
+	ordinal, err := strconv.Atoi(zkPvcName[index+1:])
+	if err != nil {
+		return false
+	}
+
+	return int32(ordinal) >= replicas
+}