scaling down by removing a nodepool

src/go/k8s/api/vectorized/v1alpha1/cluster_types.go

@@ -206,6 +206,9 @@ type NodePoolSpec struct {
 	// assigned to these containers will be required on the cluster on top of
 	// the resources defined here
 	Resources RedpandaResourceRequirements `json:"resources"`
+	// Removed is set internally to signal the reconciliation loop that this nodepool
+	// has been removed from spec
+	Removed bool `json:"removed,omitempty"`
 }
 
 // RestartConfig contains strategies to configure how the cluster behaves when restarting, because of upgrades
@@ -433,6 +436,9 @@ type ClusterStatus struct {
 	// Indicates that a node is currently being decommissioned from the cluster and provides its ordinal number
 	// +optional
 	DecommissioningNode *int32 `json:"decommissioningNode,omitempty"`
+	// Indicates the pod that hosts the node we are currently decommissioning
+	// +optional
+	DecommissioningPod string `json:"decommissioningPod,omitempty"`
 	// Current version of the cluster.
 	// +optional
 	Version string `json:"version"`
@@ -687,9 +693,8 @@ type LoadBalancerStatus struct {
 }
 
 type NodePoolStatus struct {
-	CurrentReplicas int32    `json:"currentReplicas,omitempty"`
-	Replicas        int32    `json:"replicas,omitempty"`
-	Pods            []string `json:"pods,omitempty"`
+	CurrentReplicas int32 `json:"currentReplicas,omitempty"`
+	Replicas        int32 `json:"replicas,omitempty"`
 }
 
 // KafkaAPITLS configures TLS for redpanda Kafka API
@@ -1191,11 +1196,17 @@ func (r *Cluster) GetCurrentReplicas() int32 {
 	if r == nil {
 		return 0
 	}
-	if r.Status.CurrentReplicas <= 0 {
+	var currentReplicas int32
+
+	for _, np := range r.Status.NodePools {
+		currentReplicas += np.CurrentReplicas
+	}
+
+	if currentReplicas <= 0 {
 		// Not initialized, let's give the computed value
 		return r.ComputeInitialCurrentReplicasField()
 	}
-	return r.Status.CurrentReplicas
+	return currentReplicas
 }
 
 func (r *Cluster) GetReplicas() int32 {
@@ -1419,3 +1430,11 @@ func (r *Cluster) GetDecommissionBrokerID() *int32 {
 func (r *Cluster) SetDecommissionBrokerID(id *int32) {
 	r.Status.DecommissioningNode = id
 }
+
+func (r *Cluster) GetDecomissionningPod() string {
+	return r.Status.DecommissioningPod
+}
+
+func (r *Cluster) SetDecommissioningPod(pod string) {
+	r.Status.DecommissioningPod = pod
+}

src/go/k8s/api/vectorized/v1alpha1/cluster_types_test.go

@@ -224,7 +224,12 @@ func TestInitialReplicas(t *testing.T) {
 	// backward compatibility. Remove when v22.2 is no longer supported.
 	cluster := v1alpha1.Cluster{}
 	cluster.Spec.Version = featuregates.V22_2_1.String()
-	cluster.Spec.Replicas = ptr.To(int32(3))
+	cluster.Spec.NodePools = []*v1alpha1.NodePoolSpec{
+		{
+			Name:     "test",
+			Replicas: ptr.To(int32(3)),
+		},
+	}
 	assert.Equal(t, int32(1), cluster.GetCurrentReplicas())
 	cluster.Status.Replicas = 2
 	assert.Equal(t, int32(3), cluster.GetCurrentReplicas())
@@ -239,7 +244,12 @@ func TestInitialReplicas(t *testing.T) {
 
 	// test with latest version
 	cluster = v1alpha1.Cluster{}
-	cluster.Spec.Replicas = ptr.To(int32(3))
+	cluster.Spec.NodePools = []*v1alpha1.NodePoolSpec{
+		{
+			Name:     "test",
+			Replicas: ptr.To(int32(3)),
+		},
+	}
 	assert.Equal(t, int32(3), cluster.GetCurrentReplicas())
 	cluster.Status.Replicas = 2
 	assert.Equal(t, int32(3), cluster.GetCurrentReplicas())

src/go/k8s/api/vectorized/v1alpha1/cluster_webhook_test.go

@@ -316,7 +316,7 @@ func TestNilReplicasIsNotAllowed(t *testing.T) {
 	rpCluster := validRedpandaCluster()
 	_, err := rpCluster.ValidateCreate()
 	require.Nil(t, err, "Initial cluster is not valid")
-	rpCluster.Spec.Replicas = nil
+	rpCluster.Spec.NodePools = nil
 	_, err = rpCluster.ValidateCreate()
 	assert.Error(t, err)
 }
@@ -1960,6 +1960,12 @@ func validRedpandaCluster() *v1alpha1.Cluster {
 			Namespace: "",
 		},
 		Spec: v1alpha1.ClusterSpec{
+			NodePools: []*v1alpha1.NodePoolSpec{
+				{
+					Name:     "test",
+					Replicas: ptr.To(int32(3)),
+				},
+			},
 			Replicas: ptr.To(int32(1)),
 			Configuration: v1alpha1.RedpandaConfig{
 				KafkaAPI:       []v1alpha1.KafkaAPI{{Port: 124, AuthenticationMethod: "none"}},

src/go/k8s/config/crd/bases/redpanda.vectorized.io_clusters.yaml

@@ -1042,6 +1042,11 @@ spec:
                         If specified, Redpanda Pod node selectors. For reference please visit
                         https://kubernetes.io/docs/concepts/scheduling-eviction/assign-pod-node
                       type: object
+                    removed:
+                      description: |-
+                        Removed is set internally to signal the reconciliation loop that this nodepool
+                        has been removed from spec
+                      type: boolean
                     replicas:
                       description: Replicas determine how big the node pool will be.
                       format: int32
@@ -1522,6 +1527,10 @@ spec:
                   from the cluster and provides its ordinal number
                 format: int32
                 type: integer
+              decommissioningPod:
+                description: Indicates the pod that hosts the node we are currently
+                  decommissioning
+                type: string
               nodePools:
                 additionalProperties:
                   properties:

src/go/k8s/internal/controller/redpanda/cluster_controller.go

@@ -194,7 +194,8 @@ func (r *ClusterReconciler) Reconcile(
 		return ctrl.Result{}, fmt.Errorf("creating statefulset: %w", err)
 	}
 
-	if vectorizedCluster.Status.CurrentReplicas >= 1 {
+	// KO: CR
+	if vectorizedCluster.GetCurrentReplicas() >= 1 {
 		if err = r.setPodNodeIDAnnotation(ctx, &vectorizedCluster, log, ar); err != nil {
 			log.Error(err, "setting pod node_id annotation")
 		}
@@ -228,12 +229,6 @@ func (r *ClusterReconciler) Reconcile(
 		return ctrl.Result{}, err
 	}
 
-	npStatus := make(map[string]string)
-	for _, sts := range stSets {
-		rep := strconv.FormatInt(int64(sts.GetReplicas()), 10)
-		npStatus[sts.Key().Name] = rep
-	}
-
 	err = r.reportStatus(
 		ctx,
 		&vectorizedCluster,
@@ -665,19 +660,9 @@ func (r *ClusterReconciler) reportStatus(
 		readyReplicas += sts.LastObservedState.Status.ReadyReplicas
 		replicas += sts.LastObservedState.Status.Replicas
 
-		pods, err := sts.GetNodePoolPods(ctx)
-		if err != nil {
-			return fmt.Errorf("while retrieving STS %s list of pods: %w", sts.LastObservedState.Name, err)
-		}
-		podNames := make([]string, 0)
-		for _, pod := range pods.Items {
-			podNames = append(podNames, pod.Name)
-		}
-
 		nodePoolStatus[sts.Key().Name] = vectorizedv1alpha1.NodePoolStatus{
 			CurrentReplicas: sts.LastObservedState.Status.CurrentReplicas,
 			Replicas:        sts.LastObservedState.Status.Replicas,
-			Pods:            podNames,
 		}
 	}
 
@@ -1084,6 +1069,7 @@ func (r *ClusterReconciler) doDecommissionGhostBrokers(c context.Context, vClust
 		if _, ok := actualBrokerIDs[broker.NodeID]; ok {
 			continue
 		}
+		log.Info("decommissioning ghost broker", "nodeid", broker.NodeID)
 		if err := adminClient.DecommissionBroker(ctx, broker.NodeID); err != nil {
 			return fmt.Errorf("failed to decommission ghost broker: %w", err)
 		}

src/go/k8s/internal/controller/redpanda/cluster_controller_attached_resources.go

@@ -414,8 +414,12 @@ func (a *attachedResources) statefulSet() error {
 			continue
 		}
 
+		replicas := sts.Spec.Replicas
+		if st, ok := a.cluster.Status.NodePools[sts.Name]; ok {
+			replicas = &st.CurrentReplicas
+		}
+
 		// Add the sts again in the items map in order for the reconciliation process to take place
-		// Since the np spec was removed, the replicas number hardcoded to 0.
 		a.items[stsKey] = resources.NewStatefulSet(
 			a.reconciler.Client,
 			a.cluster,
@@ -434,7 +438,8 @@ func (a *attachedResources) statefulSet() error {
 			a.reconciler.MetricsTimeout,
 			vectorizedv1alpha1.NodePoolSpec{
 				Name:     npName,
-				Replicas: sts.Spec.Replicas,
+				Replicas: replicas,
+				Removed:  true,
 			})
 	}
 

src/go/k8s/pkg/resources/statefulset.go

@@ -167,8 +167,6 @@ func NewStatefulSet(
 // Ensure will manage kubernetes v1.StatefulSet for redpanda.vectorized.io custom resource
 func (r *StatefulSetResource) Ensure(ctx context.Context) error {
 	log := r.logger.WithName("StatefulSetResource.Ensure").WithValues("nodepool", r.nodePool.Name)
-	var sts appsv1.StatefulSet
-
 	if r.pandaCluster.ExternalListener() != nil {
 		err := r.Get(ctx, r.nodePortName, &r.nodePortSvc)
 		if err != nil {
@@ -196,6 +194,7 @@ func (r *StatefulSetResource) Ensure(ctx context.Context) error {
 		return nil
 	}
 
+	var sts appsv1.StatefulSet
 	err = r.Get(ctx, r.Key(), &sts)
 	if err != nil {
 		return fmt.Errorf("error while fetching StatefulSet resource: %w", err)