ws solution B

Author: pchmieli

api/v2/cassandracluster_types.go

@@ -55,6 +55,11 @@ var (
 	ClusterPhaseRunning = ClusterStateInfo{2, "Running"}
 	ClusterPhasePending = ClusterStateInfo{3, "Pending"}
 
+	//Indicates whether all racks has at least one node Ready
+	ClusterFirstLayerInitial = ClusterStateInfo{1, "Initializing"}
+	ClusterFirstLayerRunning = ClusterStateInfo{2, "Running"}
+	ClusterFirstLayerSkipped = ClusterStateInfo{2, "Skipped"}
+
 	//Available actions
 	ActionUpdateConfigMap   = ClusterStateInfo{1, "UpdateConfigMap"}
 	ActionUpdateDockerImage = ClusterStateInfo{2, "UpdateDockerImage"}
@@ -179,6 +184,10 @@ func (cc *CassandraCluster) SetDefaults() bool {
 		}
 		changed = true
 	}
+	if len(cc.Status.FirstLayerPhase) == 0 {
+		cc.Status.FirstLayerPhase = ClusterFirstLayerInitial.Name
+		changed = true
+	}
 	if ccs.MaxPodUnavailable == 0 {
 		ccs.MaxPodUnavailable = defaultMaxPodUnavailable
 		changed = true
@@ -306,7 +315,8 @@ func (cc *CassandraCluster) initTopology(dcName string, rackName string) {
 func (cc *CassandraCluster) InitCassandraRackStatus(status *CassandraClusterStatus, dcName string, rackName string) {
 	dcRackName := cc.GetDCRackName(dcName, rackName)
 	rackStatus := CassandraRackStatus{
-		Phase: ClusterPhaseInitial.Name,
+		Phase:           ClusterPhaseInitial.Name,
+		FirstLayerPhase: ClusterFirstLayerInitial.Name,
 		CassandraLastAction: CassandraLastAction{
 			Name:   ClusterPhaseInitial.Name,
 			Status: StatusOngoing,
@@ -655,6 +665,10 @@ func (rack *RackSlice) Remove(idx int) {
 	*rack = append((*rack)[:idx], (*rack)[idx+1:]...)
 }
 
+func (in *CassandraClusterStatus) IsFirstLayerDuringInitialization() bool {
+	return in.FirstLayerPhase == ClusterFirstLayerInitial.Name
+}
+
 // CassandraClusterSpec defines the configuration of CassandraCluster
 
 type CassandraClusterSpec struct {
@@ -938,6 +952,12 @@ type CassandraRackStatus struct {
 	//   Initial -> Running <-> updating
 	Phase string `json:"phase,omitempty"`
 
+	// FirstLayerPhase indicates whether the rack has at least one node Ready so further initial scale-out might be allowed
+	// Needed to correctly handle `allocate_tokens_for_local_replication_factor` introduced in Cassandra 4.0
+	// FirstLayerPhase goes as one way as below:
+	//   Initial -> Running
+	FirstLayerPhase string `json:"firstLayerPhase,omitempty"`
+
 	// CassandraLastAction is the set of Cassandra State & Actions: Active, Standby..
 	CassandraLastAction CassandraLastAction `json:"cassandraLastAction,omitempty"`
 
@@ -952,6 +972,12 @@ type CassandraClusterStatus struct {
 	//   Initial -> Running <-> updating
 	Phase string `json:"phase,omitempty"`
 
+	// FirstLayerPhase indicates whether all racks has at least one node Ready so further initial scale-out might be allowed
+	// Needed to correctly handle `allocate_tokens_for_local_replication_factor` introduced in Cassandra 4.0
+	// FirstLayerPhase goes as one way as below:
+	//   Initial -> Running
+	FirstLayerPhase string `json:"firstLayerPhase,omitempty"`
+
 	// Store last action at cluster level
 	LastClusterAction       string `json:"lastClusterAction,omitempty"`
 	LastClusterActionStatus string `json:"lastClusterActionStatus,omitempty"`

charts/casskop/crds/db.orange.com_cassandraclusters.yaml

@@ -2270,6 +2270,9 @@ spec:
                             type: array
                             items:
                               type: string
+                      firstLayerPhase:
+                        description: 'FirstLayerPhase indicates whether the rack has at least one node Ready so further initial scale-out might be allowed Needed to correctly handle `allocate_tokens_for_local_replication_factor` introduced in Cassandra 4.0 FirstLayerPhase goes as one way as below:   Initial -> Running'
+                        type: string
                       phase:
                         description: |-
                           Phase indicates the state this Cassandra cluster jumps in.
@@ -2313,6 +2316,9 @@ spec:
                   type: string
                 lastClusterActionStatus:
                   type: string
+                firstLayerPhase:
+                  description: 'FirstLayerPhase indicates whether all racks has at least one node Ready so further initial scale-out might be allowed Needed to correctly handle `allocate_tokens_for_local_replication_factor` introduced in Cassandra 4.0 FirstLayerPhase goes as one way as below:   Initial -> Running'
+                  type: string
                 phase:
                   description: |-
                     Phase indicates the state this Cassandra cluster jumps in.

config/crd/bases/db.orange.com_cassandraclusters.yaml

@@ -2270,6 +2270,9 @@ spec:
                             type: array
                             items:
                               type: string
+                      firstLayerPhase:
+                        description: 'FirstLayerPhase indicates whether the rack has at least one node Ready so further initial scale-out might be allowed Needed to correctly handle `allocate_tokens_for_local_replication_factor` introduced in Cassandra 4.0 FirstLayerPhase goes as one way as below:   Initial -> Running'
+                        type: string
                       phase:
                         description: |-
                           Phase indicates the state this Cassandra cluster jumps in.
@@ -2308,6 +2311,9 @@ spec:
                             format: date-time
                           status:
                             type: string
+                firstLayerPhase:
+                  description: 'FirstLayerPhase indicates whether all racks has at least one node Ready so further initial scale-out might be allowed Needed to correctly handle `allocate_tokens_for_local_replication_factor` introduced in Cassandra 4.0 FirstLayerPhase goes as one way as below:   Initial -> Running'
+                  type: string
                 lastClusterAction:
                   description: Store last action at cluster level
                   type: string

controllers/cassandracluster/cassandra_status.go

@@ -459,6 +459,7 @@ func (rcc *CassandraClusterReconciler) UpdateCassandraRackStatusPhase(ctx contex
 		}
 		if len(podsList.Items) < int(nodesPerRacks) {
 			logrus.WithFields(logrusFields).Infof("StatefulSet is scaling up")
+			return
 		}
 		pod := podsList.Items[nodesPerRacks-1]
 		if cassandraPodIsReady(&pod) {
@@ -483,6 +484,39 @@ func (rcc *CassandraClusterReconciler) UpdateCassandraRackStatusPhase(ctx contex
 	}
 }
 
+// UpdateCassandraRackStatusFirstLayerPhase goal is to calculate the Cluster FirstLayerPhase according to StatefulSet Status.
+// The FirstLayerPhase is going one way: Initializing -> Running
+// The FirstLayerPhase is used to correctly setup needed replicas before scaling the cluster out
+func (rcc *CassandraClusterReconciler) UpdateCassandraRackStatusFirstLayerPhase(ctx context.Context, cc *api.CassandraCluster, dcName string,
+	rackName string, storedStatefulSet *appsv1.StatefulSet, status *api.CassandraClusterStatus) {
+	dcRackName := cc.GetDCRackName(dcName, rackName)
+
+	logrusFields := logrus.Fields{"cluster": cc.Name, "rack": dcRackName, "firstLayerPhase": status.CassandraRackStatus[dcRackName].FirstLayerPhase,
+		"ReadyReplicas": storedStatefulSet.Status.ReadyReplicas, "RequestedReplicas": *storedStatefulSet.Spec.Replicas}
+
+	ClusterPhaseMetric.set(api.ClusterPhaseInitial, cc.Name)
+
+	if isStatefulSetNotReady(storedStatefulSet) {
+		logrus.WithFields(logrusFields).Infof("Initializing StatefulSet: Replicas count is not okay")
+		return
+	}
+	//If yes, just check that lastPod is running
+	podsList, err := rcc.ListPods(ctx, cc.Namespace, k8s.LabelsForCassandraDCRack(cc, dcName, rackName))
+	if err != nil || len(podsList.Items) < 1 {
+		return
+	}
+	nodesPerRacks := cc.GetNodesPerRacks(dcRackName)
+	if len(podsList.Items) < int(nodesPerRacks) {
+		logrus.WithFields(logrusFields).Infof("StatefulSet is scaling up")
+		return
+	}
+	pod := podsList.Items[len(podsList.Items)-1]
+	if cassandraPodIsReady(&pod) {
+		status.CassandraRackStatus[dcRackName].FirstLayerPhase = api.ClusterFirstLayerRunning.Name
+		logrus.WithFields(logrusFields).Infof("StatefulSet: Replicas count is okay")
+	}
+}
+
 func setDecommissionStatus(status *api.CassandraClusterStatus, dcRackName string) {
 	status.CassandraRackStatus[dcRackName].Phase = api.ClusterPhasePending.Name
 	now := metav1.Now()

controllers/cassandracluster/cassandracluster_controller.go

@@ -128,6 +128,12 @@ func (rcc *CassandraClusterReconciler) Reconcile(ctx context.Context, request re
 		logrus.WithFields(logrus.Fields{"cluster": cc.Name}).Errorf("CheckPodsState Error: %v", err)
 	}
 
+	if shouldBreak, err := rcc.ReconcileFirstLayer(ctx, cc, status); err != nil {
+		return requeue5, err
+	} else if shouldBreak == breakResyncLoop {
+		return requeue5, nil
+	}
+
 	//ReconcileRack will also add and initiate new racks, we must not go through racks before this method
 	if err = rcc.ReconcileRack(ctx, cc, status); err != nil {
 		return requeue5, err

controllers/cassandracluster/deploy_cassandra.go

@@ -17,7 +17,9 @@ package cassandracluster
 import (
 	"context"
 	"fmt"
+
 	api "github.com/cscetbon/casskop/api/v2"
+	appsv1 "k8s.io/api/apps/v1"
 	policyv1 "k8s.io/api/policy/v1"
 
 	"github.com/cscetbon/casskop/pkg/k8s"
@@ -86,10 +88,11 @@ func (rcc *CassandraClusterReconciler) podDisruptionBudgetEnvelope(cc *api.Cassa
 }
 
 // ensureCassandraStatefulSet generate and apply the statefulset
-// take dcRackName to accordingly named the statefulset
-// take dc and rack index of dc and rack in conf to retrieve according  nodeselectors labels
+// take dcRackName to accordingly name the statefulset
+// take dc and rack index of dc and rack in conf to retrieve according nodeselectors labels
 func (rcc *CassandraClusterReconciler) ensureCassandraStatefulSet(ctx context.Context, cc *api.CassandraCluster,
-	status *api.CassandraClusterStatus, dcName string, dcRackName string, dc int, rack int) (bool, error) {
+	status *api.CassandraClusterStatus, dcName string, dcRackName string, dc int, rack int,
+	cassandraStatefulSetOptions ...cassandraStatefulSetOption) (bool, error) {
 
 	labels, nodeSelector := k8s.DCRackLabelsAndNodeSelectorForStatefulSet(cc, dc, rack)
 
@@ -99,10 +102,16 @@ func (rcc *CassandraClusterReconciler) ensureCassandraStatefulSet(ctx context.Co
 	}
 	k8s.AddOwnerRefToObject(ss, k8s.AsOwner(cc))
 
+	for _, option := range cassandraStatefulSetOptions {
+		ss.Spec = option(ss.Spec)
+	}
+
 	breakResyncloop, err := rcc.CreateOrUpdateStatefulSet(ctx, ss, status, dcRackName)
 	if err != nil && !apierrors.IsAlreadyExists(err) {
 		return breakResyncloop, fmt.Errorf("failed to create cassandra statefulset: %v", err)
 	}
 
 	return breakResyncloop, nil
 }
+
+type cassandraStatefulSetOption func(appsv1.StatefulSetSpec) appsv1.StatefulSetSpec

controllers/cassandracluster/reconcile.go

@@ -28,9 +28,11 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/r3labs/diff"
 	"github.com/sirupsen/logrus"
+	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/ptr"
 )
 
 type gaugeVec struct {
@@ -566,7 +568,8 @@ func (rcc *CassandraClusterReconciler) initiateRackStatusIfNeeded(status *api.Ca
 }
 
 func (rcc *CassandraClusterReconciler) ensureCassandraObjectsDeployed(ctx context.Context,
-	cc *api.CassandraCluster, status *api.CassandraClusterStatus, dc int, rack int) bool {
+	cc *api.CassandraCluster, status *api.CassandraClusterStatus, dc int, rack int,
+	cassandraStatefulSetOptions ...cassandraStatefulSetOption) bool {
 
 	dcName := cc.GetDCName(dc)
 	rackName := cc.GetRackName(dc, rack)
@@ -581,14 +584,119 @@ func (rcc *CassandraClusterReconciler) ensureCassandraObjectsDeployed(ctx contex
 			"dc-rack": dcRackName}).Errorf("ensureCassandraServiceMonitoring Error: %v", err)
 	}
 
-	breakLoop, err := rcc.ensureCassandraStatefulSet(ctx, cc, status, dcName, dcRackName, dc, rack)
+	breakLoop, err := rcc.ensureCassandraStatefulSet(ctx, cc, status, dcName, dcRackName, dc, rack, cassandraStatefulSetOptions...)
 	if err != nil {
 		logrus.WithFields(logrus.Fields{"cluster": cc.Name,
 			"dc-rack": dcRackName}).Errorf("ensureCassandraStatefulSet Error: %v", err)
 	}
 	return breakLoop
 }
 
+func (rcc *CassandraClusterReconciler) ReconcileFirstLayer(ctx context.Context, cc *api.CassandraCluster,
+	status *api.CassandraClusterStatus) (bool, error) {
+
+	featureDisabled := strings.ToLower(cc.Annotations["cassandraclusters.db.orange.com/disable-first-layer-init-flow"]) == "true"
+	if featureDisabled {
+		status.FirstLayerPhase = api.ClusterFirstLayerSkipped.Name
+		return continueResyncLoop, nil
+	}
+
+	if status.IsFirstLayerDuringInitialization() {
+		return breakResyncLoop, rcc.ReconcileRackFirstLayer(ctx, cc, status)
+	}
+	return continueResyncLoop, nil
+}
+
+// ReconcileRackFirstLayer will try to create one node for each of the couple DC/Rack defined in the topology
+func (rcc *CassandraClusterReconciler) ReconcileRackFirstLayer(ctx context.Context, cc *api.CassandraCluster,
+	status *api.CassandraClusterStatus) (err error) {
+
+	newStatus := false
+	for dc := 0; dc < cc.GetDCSize(); dc++ {
+		dcName := cc.GetDCName(dc)
+		for rack := 0; rack < cc.GetRackSize(dc); rack++ {
+
+			rackName := cc.GetRackName(dc, rack)
+			dcRackName := cc.GetDCRackName(dcName, rackName)
+			if dcRackName == "" {
+				return fmt.Errorf("name used for DC and/or Rack are not good")
+			}
+
+			if rcc.initiateRackStatusIfNeeded(status, dcRackName, cc, dcName, rackName) {
+				newStatus = true
+				continue
+			}
+			dcRackStatus := status.CassandraRackStatus[dcRackName]
+
+			if cc.DeletionTimestamp != nil && cc.Spec.DeletePVC {
+				rcc.DeletePVCs(ctx, cc, dcName, rackName)
+				//Go to next rack
+				continue
+			}
+
+			Name := cc.Name + "-" + dcRackName
+			storedStatefulSet, err := rcc.GetStatefulSet(ctx, cc.Namespace, Name)
+			if err != nil {
+				logrus.WithFields(logrus.Fields{"cluster": cc.Name,
+					"dc-rack": dcRackName}).Infof("failed to get cassandra's statefulset (%s) %v", Name, err)
+			} else {
+				rcc.UpdateCassandraRackStatusFirstLayerPhase(ctx, cc, dcName, rackName, storedStatefulSet, status)
+			}
+
+			_ = rcc.ensureCassandraObjectsDeployed(ctx, cc, status, dc, rack, overrideReplicasCountForFirstLayer)
+
+			// in first layer (pre-initial) phase move to next rack as soon as current rack has 1 ready replica
+			if dcRackStatus.FirstLayerPhase != api.ClusterFirstLayerRunning.Name {
+				logrus.WithFields(logrus.Fields{"cluster": cc.Name,
+					"dc-rack": dcRackName}).Infof("Waiting Rack to be running before continuing, " +
+					"we break ReconcileRackFirstLayer after updated statefulset")
+				return nil
+			}
+		}
+
+	}
+
+	newStatus = newStatus || rcc.updateClusterFirstLayerStatus(cc, status)
+
+	if newStatus {
+		return nil
+	}
+
+	//If cluster is deleted and DeletePVC is set, we can now stop preventing the cluster from being deleted
+	//cause PVCs have been deleted
+	if cc.DeletionTimestamp != nil && cc.Spec.DeletePVC {
+		preventClusterDeletion(cc, false)
+		return rcc.Client.Update(ctx, cc)
+	}
+
+	return nil
+}
+
+func (rcc *CassandraClusterReconciler) updateClusterFirstLayerStatus(cc *api.CassandraCluster, status *api.CassandraClusterStatus) bool {
+	for dc := 0; dc < cc.GetDCSize(); dc++ {
+		dcName := cc.GetDCName(dc)
+		for rack := 0; rack < cc.GetRackSize(dc); rack++ {
+			rackName := cc.GetRackName(dc, rack)
+			dcRackName := cc.GetDCRackName(dcName, rackName)
+			dcRackStatus := status.CassandraRackStatus[dcRackName]
+			if dcRackStatus.FirstLayerPhase != api.ClusterFirstLayerRunning.Name {
+				return false
+			}
+		}
+	}
+
+	// assuming first layer on all racks is ready
+	oldStatus := status.FirstLayerPhase
+	logrus.WithFields(logrus.Fields{"cluster": cc.Name}).Info("FirstLayer is now Ready on all racks")
+	status.FirstLayerPhase = api.ClusterFirstLayerRunning.Name
+	return oldStatus != status.FirstLayerPhase
+}
+
+func overrideReplicasCountForFirstLayer(stsSpec appsv1.StatefulSetSpec) appsv1.StatefulSetSpec {
+	stsSpec.Replicas = ptr.To(int32(1))
+	return stsSpec
+}
+
 func (rcc *CassandraClusterReconciler) waitForStatefulSetToBeUpdated(ctx context.Context, cc *api.CassandraCluster, dcRackName string,
 	err error) {
 	logrus.WithFields(logrus.Fields{