Add NodePool support to Cluster resource

operator/api/vectorized/v1alpha1/cluster_types.go

@@ -13,15 +13,15 @@ import (
 	"fmt"
 	"net"
 	"net/url"
+	"slices"
 	"time"
 
 	cmmeta "github.com/cert-manager/cert-manager/pkg/apis/meta/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
-
-	"github.com/redpanda-data/redpanda-operator/operator/pkg/resources/featuregates"
+	"k8s.io/utils/ptr"
 )
 
 const (
@@ -31,6 +31,8 @@ const (
 	ExternalListenerName = "kafka-external"
 )
 
+const DefaultNodePoolName = "default"
+
 // RedpandaResourceRequirements extends corev1.ResourceRequirements
 // to allow specification of resources directly passed to Redpanda that
 // are different to Requests or Limits.
@@ -713,9 +715,9 @@ type LoadBalancerStatus struct {
 
 // NodePoolStatus describes the status of the NodePool.
 type NodePoolStatus struct {
-	CurrentReplicas int32 `json:"currentReplicas,omitempty"`
-	Replicas        int32 `json:"replicas,omitempty"`
-	ReadyReplicas   int32 `json:"readyReplicas,omitempty"`
+	CurrentReplicas int32 `json:"currentReplicas"`
+	Replicas        int32 `json:"replicas"`
+	ReadyReplicas   int32 `json:"readyReplicas"`
 	// Indicates that a nodePool's pods are restarting.
 	// +optional
 	Restarting bool `json:"restarting"`
@@ -1214,36 +1216,31 @@ func (s *ClusterStatus) SetRestarting(restarting bool) {
 	s.DeprecatedUpgrading = restarting
 }
 
-// GetCurrentReplicas returns the current number of replicas that the controller wants to run.
-// It returns 1 when not initialized (as fresh clusters start from 1 replica)
-func (r *Cluster) GetCurrentReplicas() int32 {
-	if r == nil {
+func (r *Cluster) GetDesiredReplicas() int32 {
+	nps := r.GetNodePoolsFromSpec()
+	if r == nil || len(nps) == 0 {
 		return 0
 	}
-	if r.Status.CurrentReplicas <= 0 {
-		// Not initialized, let's give the computed value
-		return r.ComputeInitialCurrentReplicasField()
-	}
-	return r.Status.CurrentReplicas
-}
 
-// ComputeInitialCurrentReplicasField calculates the initial value for status.currentReplicas.
-//
-// It needs to consider the following cases:
-// - EmptySeedStartCluster is supported: we use spec.replicas as the starting point
-// - Fresh cluster: we start from 1 replicas, then upscale if needed (initialization to bypass https://github.com/redpanda-data/redpanda/issues/333)
-// - Existing clusters: we keep spec.replicas as starting point
-func (r *Cluster) ComputeInitialCurrentReplicasField() int32 {
-	if r == nil {
-		return 0
+	replicas := int32(0)
+
+	var npNamesSeen []string
+
+	for i := range nps {
+		np := nps[i]
+		replicas += ptr.Deref(np.Replicas, 0)
+		npNamesSeen = append(npNamesSeen, np.Name)
 	}
-	if r.Status.Replicas > 1 || r.Status.ReadyReplicas > 1 || len(r.Status.Nodes.Internal) > 1 || featuregates.EmptySeedStartCluster(r.Spec.Version) {
-		// A cluster seems to be already running, we start from the existing amount of replicas
-		return *r.Spec.Replicas
+
+	// We may have nodepools deleted from spec - but they are not yet fully deleted (in scale-down).
+	// Source these from the status instead.
+	for npName, npStatus := range r.Status.NodePools {
+		if !slices.Contains(npNamesSeen, npName) {
+			replicas += npStatus.Replicas
+		}
 	}
 
-	// Clusters start from a single replica, then upscale
-	return 1
+	return replicas
 }
 
 // TLSConfig is a generic TLS configuration
@@ -1422,3 +1419,41 @@ func (r *Cluster) GetDecommissionBrokerID() *int32 {
 func (r *Cluster) SetDecommissionBrokerID(id *int32) {
 	r.Status.DecommissioningNode = id
 }
+
+// getNodePoolsFromSpec returns the NodePools defined in the spec.
+// This contains the default NodePool (driven by spec.replicas for example), and the
+// NodePools driven by the nodePools field.
+func (r *Cluster) GetNodePoolsFromSpec() []NodePoolSpec {
+	out := make([]NodePoolSpec, 0)
+	if r.Spec.Replicas != nil {
+		defaultNodePool := NodePoolSpec{
+			Name:                           DefaultNodePoolName,
+			Replicas:                       r.Spec.Replicas,
+			Tolerations:                    r.Spec.Tolerations,
+			NodeSelector:                   r.Spec.NodeSelector,
+			Storage:                        r.Spec.Storage,
+			Resources:                      r.Spec.Resources,
+			AdditionalCommandlineArguments: r.Spec.Configuration.AdditionalCommandlineArguments,
+		}
+		if r.Spec.CloudStorage.CacheStorage != nil {
+			defaultNodePool.CloudCacheStorage = *r.Spec.CloudStorage.CacheStorage
+		}
+		out = append(out, defaultNodePool)
+	}
+	out = append(out, r.Spec.NodePools...)
+
+	return out
+}
+
+func (r *Cluster) CalculateCurrentReplicas() int32 {
+	var result int32
+	for _, np := range r.Status.NodePools {
+		result += np.CurrentReplicas
+	}
+	return result
+}
+
+type NodePoolSpecWithDeleted struct {
+	NodePoolSpec
+	Deleted bool
+}

operator/api/vectorized/v1alpha1/cluster_types_test.go

@@ -17,10 +17,8 @@ import (
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
-	"k8s.io/utils/ptr"
 
 	"github.com/redpanda-data/redpanda-operator/operator/api/vectorized/v1alpha1"
-	"github.com/redpanda-data/redpanda-operator/operator/pkg/resources/featuregates"
 )
 
 //nolint:funlen // this is ok for a test
@@ -219,36 +217,3 @@ func TestConditions(t *testing.T) {
 		assert.Equal(t, condTime, cond2.LastTransitionTime)
 	})
 }
-
-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))
-	assert.Equal(t, int32(1), cluster.GetCurrentReplicas())
-	cluster.Status.Replicas = 2
-	assert.Equal(t, int32(3), cluster.GetCurrentReplicas())
-	cluster.Status.Replicas = 0
-	cluster.Status.ReadyReplicas = 2
-	assert.Equal(t, int32(3), cluster.GetCurrentReplicas())
-	cluster.Status.ReadyReplicas = 0
-	cluster.Status.Nodes.Internal = []string{"1", "2"}
-	assert.Equal(t, int32(3), cluster.GetCurrentReplicas())
-	cluster.Status.Nodes.Internal = nil
-	assert.Equal(t, int32(1), cluster.GetCurrentReplicas())
-
-	// test with latest version
-	cluster = v1alpha1.Cluster{}
-	cluster.Spec.Replicas = ptr.To(int32(3))
-	assert.Equal(t, int32(3), cluster.GetCurrentReplicas())
-	cluster.Status.Replicas = 2
-	assert.Equal(t, int32(3), cluster.GetCurrentReplicas())
-	cluster.Status.Replicas = 0
-	cluster.Status.ReadyReplicas = 2
-	assert.Equal(t, int32(3), cluster.GetCurrentReplicas())
-	cluster.Status.ReadyReplicas = 0
-	cluster.Status.Nodes.Internal = []string{"1", "2"}
-	assert.Equal(t, int32(3), cluster.GetCurrentReplicas())
-	cluster.Status.Nodes.Internal = nil
-	assert.Equal(t, int32(3), cluster.GetCurrentReplicas())
-}

operator/api/vectorized/v1alpha1/cluster_webhook.go

@@ -108,8 +108,19 @@ func (r *Cluster) SetupWebhookWithManager(mgr ctrl.Manager) error {
 
 var _ webhook.Defaulter = &Cluster{}
 
-func redpandaResourceFields(c *Cluster) redpandaResourceField {
-	return redpandaResourceField{&c.Spec.Resources, field.NewPath("spec").Child("resources")}
+func redpandaResourceFields(c *Cluster) []redpandaResourceField {
+	var result []redpandaResourceField
+	// Default NodePool
+	if c.Spec.Replicas != nil {
+		result = append(result, redpandaResourceField{&c.Spec.Resources, field.NewPath("spec").Child("resources")})
+	}
+
+	// Additional NodePools
+	for _, np := range c.Spec.NodePools {
+		result = append(result, redpandaResourceField{&np.Resources, field.NewPath("spec").Child("nodePools").Child("resources")})
+	}
+
+	return result
 }
 
 func sidecarResourceFields(c *Cluster) []resourceField {
@@ -259,7 +270,9 @@ func (r *Cluster) validateCommon(log logr.Logger) field.ErrorList {
 	for _, rf := range sidecarResourceFields(r) {
 		allErrs = append(allErrs, r.validateResources(rf)...)
 	}
-	allErrs = append(allErrs, r.validateRedpandaResources(redpandaResourceFields(r))...)
+	for _, rf := range redpandaResourceFields(r) {
+		allErrs = append(allErrs, r.validateRedpandaResources(rf)...)
+	}
 	allErrs = append(allErrs, r.validateArchivalStorage()...)
 	allErrs = append(allErrs, r.validatePodDisruptionBudget()...)
 	if featuregates.InternalTopicReplication(r.Spec.Version) {
@@ -270,16 +283,13 @@ func (r *Cluster) validateCommon(log logr.Logger) field.ErrorList {
 }
 
 func (r *Cluster) validateScaling() field.ErrorList {
+	replicas := r.GetDesiredReplicas()
+
 	var allErrs field.ErrorList
-	if r.Spec.Replicas == nil {
+	if replicas <= 0 {
 		allErrs = append(allErrs,
-			field.Invalid(field.NewPath("spec").Child("replicas"),
-				r.Spec.Replicas,
-				"replicas must be specified explicitly"))
-	} else if *r.Spec.Replicas <= 0 {
-		allErrs = append(allErrs,
-			field.Invalid(field.NewPath("spec").Child("replicas"),
-				r.Spec.Replicas,
+			field.Invalid(field.NewPath("spec").Child("nodepools"),
+				replicas,
 				"downscaling is not allowed to less than 1 instance"))
 	}
 
@@ -770,26 +780,54 @@ func (r *Cluster) validateRedpandaMemory() field.ErrorList {
 	}
 	var allErrs field.ErrorList
 
-	// Ensure a requested 2GB of memory per core
-	requests := r.Spec.Resources.Requests.DeepCopy()
-	requests.Cpu().RoundUp(0)
-	requestedCores := requests.Cpu().Value()
-	if r.Spec.Resources.Requests.Memory().Value() < requestedCores*MinimumMemoryPerCore {
-		allErrs = append(allErrs,
-			field.Invalid(
-				field.NewPath("spec").Child("resources").Child("requests").Child("memory"),
-				r.Spec.Resources.Requests.Memory(),
-				"requests.memory < 2Gi per core; either decrease requests.cpu or increase requests.memory"))
+	for i := range r.Spec.NodePools {
+		np := r.Spec.NodePools[i]
+
+		// Ensure a requested 2GB of memory per core
+		requests := np.Resources.Requests.DeepCopy()
+		requests.Cpu().RoundUp(0)
+		requestedCores := requests.Cpu().Value()
+		if np.Resources.Requests.Memory().Value() < requestedCores*MinimumMemoryPerCore {
+			allErrs = append(allErrs,
+				field.Invalid(
+					field.NewPath("spec").Child("nodePool").Child("resources").Child("requests").Child("memory"),
+					np.Resources.Requests.Memory(),
+					"requests.memory < 2Gi per core; either decrease requests.cpu or increase requests.memory"))
+		}
+
+		redpandaCores := np.Resources.RedpandaCPU().Value()
+		minimumMemoryPerCore := int64(math.Floor(MinimumMemoryPerCore * RedpandaMemoryAllocationRatio))
+		if !np.Resources.RedpandaMemory().IsZero() && np.Resources.RedpandaMemory().Value() < redpandaCores*minimumMemoryPerCore {
+			allErrs = append(allErrs,
+				field.Invalid(
+					field.NewPath("spec").Child("resources").Child("redpanda").Child("memory"),
+					r.Spec.Resources.Requests.Memory(),
+					"redpanda.memory < 2Gi per core; either decrease redpanda.cpu or increase redpanda.memory"))
+		}
 	}
 
-	redpandaCores := r.Spec.Resources.RedpandaCPU().Value()
-	minimumMemoryPerCore := int64(math.Floor(MinimumMemoryPerCore * RedpandaMemoryAllocationRatio))
-	if !r.Spec.Resources.RedpandaMemory().IsZero() && r.Spec.Resources.RedpandaMemory().Value() < redpandaCores*minimumMemoryPerCore {
-		allErrs = append(allErrs,
-			field.Invalid(
-				field.NewPath("spec").Child("resources").Child("redpanda").Child("memory"),
-				r.Spec.Resources.Requests.Memory(),
-				"redpanda.memory < 2Gi per core; either decrease redpanda.cpu or increase redpanda.memory"))
+	if r.Spec.Replicas != nil {
+		// Ensure a requested 2GB of memory per core
+		requests := r.Spec.Resources.Requests.DeepCopy()
+		requests.Cpu().RoundUp(0)
+		requestedCores := requests.Cpu().Value()
+		if r.Spec.Resources.Requests.Memory().Value() < requestedCores*MinimumMemoryPerCore {
+			allErrs = append(allErrs,
+				field.Invalid(
+					field.NewPath("spec").Child("resources").Child("requests").Child("memory"),
+					r.Spec.Resources.Requests.Memory(),
+					"requests.memory < 2Gi per core; either decrease requests.cpu or increase requests.memory"))
+		}
+
+		redpandaCores := r.Spec.Resources.RedpandaCPU().Value()
+		minimumMemoryPerCore := int64(math.Floor(MinimumMemoryPerCore * RedpandaMemoryAllocationRatio))
+		if !r.Spec.Resources.RedpandaMemory().IsZero() && r.Spec.Resources.RedpandaMemory().Value() < redpandaCores*minimumMemoryPerCore {
+			allErrs = append(allErrs,
+				field.Invalid(
+					field.NewPath("spec").Child("resources").Child("redpanda").Child("memory"),
+					r.Spec.Resources.Requests.Memory(),
+					"redpanda.memory < 2Gi per core; either decrease redpanda.cpu or increase redpanda.memory"))
+		}
 	}
 
 	return allErrs
@@ -804,19 +842,54 @@ func (r *Cluster) validateRedpandaCoreChanges(old *Cluster) field.ErrorList {
 	}
 	var allErrs field.ErrorList
 
-	oldCPURequest := old.Spec.Resources.RedpandaCPU()
-	newCPURequest := r.Spec.Resources.RedpandaCPU()
-	if oldCPURequest != nil && newCPURequest != nil {
-		oldCores := oldCPURequest.Value()
-		newCores := newCPURequest.Value()
+	if r.Spec.Replicas != nil {
+		oldCPURequest := old.Spec.Resources.RedpandaCPU()
+		newCPURequest := r.Spec.Resources.RedpandaCPU()
+		if oldCPURequest != nil && newCPURequest != nil {
+			oldCores := oldCPURequest.Value()
+			newCores := newCPURequest.Value()
 
-		if newCores < oldCores {
-			minAllowedCPU := (oldCores-1)*1000 + 1
-			allErrs = append(allErrs,
-				field.Invalid(
-					field.NewPath("spec").Child("resources").Child("requests").Child("cpu"),
-					r.Spec.Resources.Requests.Cpu(),
-					fmt.Sprintf("CPU request must not be decreased; increase requests.cpu or redpanda.cpu to at least %dm", minAllowedCPU)))
+			if newCores < oldCores {
+				minAllowedCPU := (oldCores-1)*1000 + 1
+				allErrs = append(allErrs,
+					field.Invalid(
+						field.NewPath("spec").Child("resources").Child("requests").Child("cpu"),
+						r.Spec.Resources.Requests.Cpu(),
+						fmt.Sprintf("CPU request must not be decreased; increase requests.cpu or redpanda.cpu to at least %dm", minAllowedCPU)))
+			}
+		}
+	}
+
+	// Same check per NP
+	for i := range r.Spec.NodePools {
+		np := r.Spec.NodePools[i]
+		var oldNp *NodePoolSpec
+		for y := range old.Spec.NodePools {
+			old := old.Spec.NodePools[y]
+			if old.Name == np.Name {
+				oldNp = &old
+			}
+		}
+
+		if oldNp == nil {
+			continue
+		}
+
+		oldCPURequest := oldNp.Resources.RedpandaCPU()
+		newCPURequest := np.Resources.RedpandaCPU()
+
+		if oldCPURequest != nil && newCPURequest != nil {
+			oldCores := oldCPURequest.Value()
+			newCores := newCPURequest.Value()
+
+			if newCores < oldCores {
+				minAllowedCPU := (oldCores-1)*1000 + 1
+				allErrs = append(allErrs,
+					field.Invalid(
+						field.NewPath("spec").Child("resources").Child("requests").Child("cpu"),
+						np.Resources.Requests.Cpu(),
+						fmt.Sprintf("CPU request must not be decreased; increase requests.cpu or redpanda.cpu to at least %dm", minAllowedCPU)))
+			}
 		}
 	}