cluster: scale-out done check fixed

when scaling the cluster out, check joining nodes before finishing action

Fixes #182

Author: pchmieli

controllers/cassandracluster/cassandra_status.go

@@ -372,14 +372,28 @@ func (rcc *CassandraClusterReconciler) UpdateStatusIfActionEnded(ctx context.Con
 			//Does the Scaling ended ?
 			if nodesPerRacks == storedStatefulSet.Status.Replicas {
 
-				podsList, err := rcc.ListPods(ctx, cc.Namespace, k8s.LabelsForCassandraDCRack(cc, dcName, rackName))
+				podsList, err := rcc.ListPodsOrderByNameAscending(ctx, cc.Namespace, k8s.LabelsForCassandraDCRack(cc, dcName, rackName))
 				nb := len(podsList.Items)
 				if err != nil || nb < 1 {
 					return false
 				}
+				if nb < int(nodesPerRacks) {
+					logrus.WithFields(logrus.Fields{"cluster": cc.Name, "rack": dcRackName}).Warn(fmt.Sprintf(
+						"Although statefulSet has %d replicas, only %d matching pods found", nodesPerRacks, nb))
+					return false
+				}
 				pod := podsList.Items[nodesPerRacks-1]
+
 				//We need lastPod to be running to consider ScaleUp ended
 				if cassandraPodIsReady(&pod) {
+					if hasJoiningNodes, err := rcc.hasJoiningNodes(ctx, cc); err != nil {
+						return false
+					} else if hasJoiningNodes {
+						logrus.WithFields(logrus.Fields{"cluster": cc.Name, "dc": dcName, "rack": rackName,
+							"err": err}).Info("Cluster has joining nodes, ScaleUp not yet completed")
+						return false
+					}
+
 					logrus.WithFields(logrus.Fields{"cluster": cc.Name, "rack": dcRackName}).Info("ScaleUp is Done")
 					rackLastAction.Status = api.StatusDone
 					rackLastAction.EndTime = &now

controllers/cassandracluster/cassandra_status_test.go

@@ -97,6 +97,8 @@ func HelperInitCluster(t *testing.T, name string) (*CassandraClusterReconciler,
 	var cc api.CassandraCluster
 	yaml.Unmarshal(common.HelperLoadBytes(t, name), &cc)
 
+	cc.UID = "123456789" //We need to set a UID so PatchMaker does not fail when comparing owner references
+
 	ccList := api.CassandraClusterList{}
 	//Create Fake client
 	//Objects to track in the Fake client
@@ -203,33 +205,11 @@ func helperCreateCassandraCluster(ctx context.Context, t *testing.T, cassandraCl
 			rcc.Client.Status().Update(ctx, sts)
 
 			//Create Statefulsets associated fake Pods
-			podTemplate := v1.Pod{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      "template",
-					Namespace: namespace,
-					Labels: map[string]string{
-						"cluster":                              cc.Labels["cluster"],
-						"dc-rack":                              dcRackName,
-						"cassandraclusters.db.orange.com.dc":   dc.Name,
-						"cassandraclusters.db.orange.com.rack": rack.Name,
-						"app":                                  "cassandracluster",
-						"cassandracluster":                     cc.Name,
-					},
-				},
-				Status: v1.PodStatus{
-					Phase: v1.PodRunning,
-					ContainerStatuses: []v1.ContainerStatus{
-						{
-							Name:  "cassandra",
-							Ready: true,
-						},
-					},
-				},
-			}
+			podTemplate := fakePodTemplate(cc, dc.Name, rack.Name)
 
 			for i := 0; i < int(sts.Status.Replicas); i++ {
 				pod := podTemplate.DeepCopy()
-				pod.Name = sts.Name + strconv.Itoa(i)
+				pod.Name = sts.Name + "-" + strconv.Itoa(i)
 				pod.Spec.Hostname = pod.Name
 				pod.Spec.Subdomain = cc.Name
 				if err = rcc.CreatePod(ctx, pod); err != nil {
@@ -265,6 +245,33 @@ func helperCreateCassandraCluster(ctx context.Context, t *testing.T, cassandraCl
 	return rcc, &req
 }
 
+func fakePodTemplate(cc *api.CassandraCluster, dcName, rackName string) v1.Pod {
+	dcRackName := cc.GetDCRackName(dcName, rackName)
+	return v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "template",
+			Namespace: namespace,
+			Labels: map[string]string{
+				"cluster":                              cc.Labels["cluster"],
+				"dc-rack":                              dcRackName,
+				"cassandraclusters.db.orange.com.dc":   dcName,
+				"cassandraclusters.db.orange.com.rack": rackName,
+				"app":                                  "cassandracluster",
+				"cassandracluster":                     cc.Name,
+			},
+		},
+		Status: v1.PodStatus{
+			Phase: v1.PodRunning,
+			ContainerStatuses: []v1.ContainerStatus{
+				{
+					Name:  "cassandra",
+					Ready: true,
+				},
+			},
+		},
+	}
+}
+
 func TestCassandraClusterReconciler(t *testing.T) {
 	// tests speed-up
 	httpmock.Activate()
@@ -476,6 +483,24 @@ func TestUpdateStatusIfDockerImageHasChanged(t *testing.T) {
 
 }
 
+func assertRackStatusPhase(assert *assert.Assertions, rcc *CassandraClusterReconciler, dcRackName string, expectedPhase api.ClusterStateInfo) {
+	assert.Equal(expectedPhase.Name, rcc.cc.Status.CassandraRackStatus[dcRackName].Phase, dcRackName + " phase")
+}
+
+func assertClusterStatusPhase(assert *assert.Assertions, rcc *CassandraClusterReconciler, expectedPhase api.ClusterStateInfo) {
+	assert.Equal(expectedPhase.Name, rcc.cc.Status.Phase, "cluster phase")
+}
+
+func assertRackStatusLastAction(assert *assert.Assertions, rcc *CassandraClusterReconciler, dcRackName string, expectedActionType api.ClusterStateInfo, expectedActionStatus string) {
+	assert.Equal(expectedActionType.Name, rcc.cc.Status.CassandraRackStatus[dcRackName].CassandraLastAction.Name, "dc1-rack1 last action type")
+	assert.Equal(expectedActionStatus, rcc.cc.Status.CassandraRackStatus[dcRackName].CassandraLastAction.Status, "dc1-rack1 last action status")
+}
+
+func assertClusterStatusLastAction(assert *assert.Assertions, rcc *CassandraClusterReconciler, expectedActionType api.ClusterStateInfo, expectedActionStatus string) {
+	assert.Equal(expectedActionType.Name, rcc.cc.Status.LastClusterAction, "cluster last action type")
+	assert.Equal(expectedActionStatus, rcc.cc.Status.LastClusterActionStatus, "cluster last action status")
+}
+
 func overrideDelayWaitWithNoDelay() {
 	delayWait = func() time.Duration {
 		return 0

controllers/cassandracluster/decommission_test.go

@@ -69,7 +69,7 @@ type podName struct {
 }
 
 func podHost(stfsName string, id int8, rcc *CassandraClusterReconciler) podName {
-	name := stfsName + strconv.Itoa(int(id))
+	name := stfsName + "-" + strconv.Itoa(int(id))
 	return podName{name, name + "." + rcc.cc.Name}
 }
 

controllers/cassandracluster/pod.go

@@ -88,12 +88,7 @@ func GetLastOrFirstPodItem(podsList []v1.Pod, last bool) (*v1.Pod, error) {
 
 	items := podsList[:]
 
-	// Sort pod list using ending number in field ObjectMeta.Name
-	sort.Slice(items, func(i, j int) bool {
-		id1, _ := strconv.Atoi(reEndingNumber.FindString(items[i].ObjectMeta.Name))
-		id2, _ := strconv.Atoi(reEndingNumber.FindString(items[j].ObjectMeta.Name))
-		return id1 < id2
-	})
+	sortPodsList(items)
 
 	idx := 0
 	if last {
@@ -105,6 +100,15 @@ func GetLastOrFirstPodItem(podsList []v1.Pod, last bool) (*v1.Pod, error) {
 	return &pod, nil
 }
 
+// sortPodsList sorts pod list using ending number in field ObjectMeta.Name
+func sortPodsList(items []v1.Pod) {
+	sort.Slice(items, func(i, j int) bool {
+		id1, _ := strconv.Atoi(reEndingNumber.FindString(items[i].ObjectMeta.Name))
+		id2, _ := strconv.Atoi(reEndingNumber.FindString(items[j].ObjectMeta.Name))
+		return id1 < id2
+	})
+}
+
 // GetFirstPod returns the first pod satisfying the selector and being in the namespace
 func (rcc *CassandraClusterReconciler) GetFirstPod(ctx context.Context, namespace string, selector map[string]string) (*v1.Pod, error) {
 	podsList, err := rcc.ListPods(ctx, namespace, selector)
@@ -182,7 +186,6 @@ func (rcc *CassandraClusterReconciler) hasUnschedulablePod(ctx context.Context,
 }
 
 func (rcc *CassandraClusterReconciler) ListPods(ctx context.Context, namespace string, selector map[string]string) (*v1.PodList, error) {
-
 	clientOpt := &client.ListOptions{
 		Namespace:     namespace,
 		LabelSelector: labels.SelectorFromSet(selector),
@@ -196,6 +199,14 @@ func (rcc *CassandraClusterReconciler) ListPods(ctx context.Context, namespace s
 	return pl, rcc.Client.List(ctx, pl, opt...)
 }
 
+func (rcc *CassandraClusterReconciler) ListPodsOrderByNameAscending(ctx context.Context, namespace string, selector map[string]string) (*v1.PodList, error) {
+	pods, err := rcc.ListPods(ctx, namespace, selector)
+	if pods != nil {
+		sortPodsList(pods.Items)
+	}
+	return pods, err
+}
+
 func (rcc *CassandraClusterReconciler) CreatePod(ctx context.Context, pod *v1.Pod) error {
 	err := rcc.Client.Create(ctx, pod)
 	if err != nil {

controllers/cassandracluster/pod_operation.go

@@ -112,21 +112,9 @@ func (rcc *CassandraClusterReconciler) handlePodOperation(ctx context.Context, c
 		return breakResyncLoopSwitch, err
 	}
 
-	podsList, err := rcc.ListCassandraClusterPods(ctx, cc)
-	if err != nil {
-		return true, err
-	}
-	firstPod, err := GetLastOrFirstPodReady(podsList, false)
-	if err != nil {
-		return true, err
-	}
-
-	hostName := k8s.PodHostname(*firstPod)
-	jolokiaClient, _ := NewJolokiaClient(ctx, hostName, JolokiaPort, rcc, cc.Spec.ImageJolokiaSecret, cc.Namespace)
-
-	hasJoiningNodes, err := jolokiaClient.hasJoiningNodes()
+	hasJoiningNodes, err := rcc.hasJoiningNodes(ctx, cc)
 	if err != nil {
-		return true, err
+		return breakResyncLoop, err
 	}
 	if hasJoiningNodes {
 		logrus.WithFields(logrus.Fields{"cluster": cc.Name, "dc": dcName, "rack": rackName,
@@ -151,6 +139,24 @@ func (rcc *CassandraClusterReconciler) handlePodOperation(ctx context.Context, c
 	return breakResyncLoopSwitch, err
 }
 
+func (rcc *CassandraClusterReconciler) hasJoiningNodes(ctx context.Context, cc *api.CassandraCluster) (bool, error) {
+	podsList, err := rcc.ListCassandraClusterPods(ctx, cc)
+	if err != nil {
+		return false, err
+	}
+	firstPod, err := GetLastOrFirstPodReady(podsList, false)
+	if err != nil {
+		return false, err
+	}
+
+	hostName := k8s.PodHostname(*firstPod)
+	jolokiaClient, err := NewJolokiaClient(ctx, hostName, JolokiaPort, rcc, cc.Spec.ImageJolokiaSecret, cc.Namespace)
+	if err != nil {
+		return false, err
+	}
+	return jolokiaClient.hasJoiningNodes()
+}
+
 // addPodOperationLabels will add Pod Labels labels on all Pod in the Current dcRackName
 func (rcc *CassandraClusterReconciler) addPodOperationLabels(ctx context.Context, cc *api.CassandraCluster, dcName string,
 	rackName string, labels map[string]string) {

controllers/cassandracluster/scaleup_test.go

@@ -3,9 +3,11 @@ package cassandracluster
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"strings"
 	"testing"
 
+	api "github.com/cscetbon/casskop/api/v2"
 	"github.com/jarcoal/httpmock"
 	"github.com/stretchr/testify/assert"
 )
@@ -33,6 +35,31 @@ func registerJolokiaOperationJoiningNodes(host podName, numberOfJoiningNodes int
 										"status": 200}`, stringOfSlice(joiningNodes))))
 }
 
+func simulateNewPodsReady(t *testing.T, rcc *CassandraClusterReconciler, stfsName string, dc api.DC, scaleFrom, scaleTo int) {
+	assert := assert.New(t)
+
+	sts, err := rcc.GetStatefulSet(ctx, rcc.cc.Namespace, stfsName)
+	assert.NoError(err, "get sts")
+
+	//Now simulate sts to be ready for CassKop
+	sts.Status.Replicas = *sts.Spec.Replicas
+	sts.Status.ReadyReplicas = *sts.Spec.Replicas
+	err = rcc.Client.Status().Update(ctx, sts)
+	assert.NoError(err, "update sts status")
+
+	// create new fake Pods (consequence of scale-out) so action may finish
+	podTemplate := fakePodTemplate(rcc.cc, dc.Name, dc.Rack[0].Name)
+	for i := scaleFrom; i < scaleTo; i++ {
+		pod := podTemplate.DeepCopy()
+		pod.Name = sts.Name + "-" + strconv.Itoa(i)
+		pod.Spec.Hostname = pod.Name
+		pod.Spec.Subdomain = rcc.cc.Name
+		if err = rcc.CreatePod(ctx, pod); err != nil {
+			t.Fatalf("can't create pod: (%v)", err)
+		}
+	}
+}
+
 func TestAddTwoNodes(t *testing.T) {
 	overrideDelayWaitWithNoDelay()
 	defer restoreDefaultDelayWait()
@@ -78,4 +105,30 @@ func TestAddTwoNodes(t *testing.T) {
 		assert.GreaterOrEqual(jolokiaCallsCount(firstPod), 1)
 		assertStatefulsetReplicas(ctx, t, rcc, expectedReplicas+1, cassandraCluster.Namespace, stfsName)
 	}
+
+	assertClusterStatusPhase(assert, rcc, api.ClusterPhasePending)
+	assertRackStatusPhase(assert, rcc, "dc1-rack1", api.ClusterPhasePending)
+	assertClusterStatusLastAction(assert, rcc, api.ActionScaleUp, api.StatusOngoing)
+	assertRackStatusLastAction(assert, rcc, "dc1-rack1", api.ActionScaleUp, api.StatusOngoing)
+
+	//Reconcile does not end the action even when sts and all new pods are ready, because there are joining nodes
+	simulateNewPodsReady(t, rcc, stfsName, dc, 3, 5)
+	registerJolokiaOperationJoiningNodes(firstPod, 1)
+	for reconcileIteration := 0; reconcileIteration <= 2; reconcileIteration++ {
+		reconcileValidation(t, rcc, *req)
+		assert.GreaterOrEqual(jolokiaCallsCount(firstPod), 1)
+		assertClusterStatusPhase(assert, rcc, api.ClusterPhasePending)
+		assertRackStatusPhase(assert, rcc, "dc1-rack1", api.ClusterPhaseRunning)
+		assertClusterStatusLastAction(assert, rcc, api.ActionScaleUp, api.StatusOngoing)
+		assertRackStatusLastAction(assert, rcc, "dc1-rack1", api.ActionScaleUp, api.StatusOngoing)
+	}
+
+	//Reconcile ends the action when sts and all new pods are ready and there are no joining nodes
+	registerJolokiaOperationJoiningNodes(firstPod, 0)
+	reconcileValidation(t, rcc, *req)
+	assert.GreaterOrEqual(jolokiaCallsCount(firstPod), 1)
+	assertClusterStatusPhase(assert, rcc, api.ClusterPhaseRunning)
+	assertRackStatusPhase(assert, rcc, "dc1-rack1", api.ClusterPhaseRunning)
+	assertClusterStatusLastAction(assert, rcc, api.ActionScaleUp, api.StatusDone)
+	assertRackStatusLastAction(assert, rcc, "dc1-rack1", api.ActionScaleUp, api.StatusDone)
 }