Merge pull request #1185 from emqx/feat/EMQX-14563/hpa-compat

feat(scale): fix subresource compatibility + HPA integration

Author: keynslug

api/v3alpha1/emqx_types.go

@@ -23,7 +23,7 @@ import (
 // +kubebuilder:object:root=true
 // +kubebuilder:subresource:status
 // +kubebuilder:resource:shortName=emqx,path=emqxes
-// +kubebuilder:subresource:scale:specpath=.spec.replicantTemplate.spec.replicas,statuspath=.status.replicantNodeReplicas
+// +kubebuilder:subresource:scale:specpath=.spec.replicantTemplate.spec.replicas,statuspath=.status.replicantReplicas,selectorpath=.status.replicantSelector
 // +kubebuilder:printcolumn:name="Status",type="string",JSONPath=".status.conditions[?(@.status==\"True\")].type"
 // +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp"
 

api/v3alpha1/emqx_types_status.go

@@ -29,6 +29,20 @@ type EMQXStatus struct {
 	// +listMapKey=type
 	Conditions []metav1.Condition `json:"conditions,omitempty"`
 
+	// Observed replica count for the core set.
+	CoreReplicas int32 `json:"coreReplicas"`
+
+	// Serialized label selector matching core pods.
+	CoreSelector string `json:"coreSelector,omitempty"`
+
+	// Observed replica count for the replicant set pods.
+	// Used by the scale subresource.
+	ReplicantReplicas int32 `json:"replicantReplicas"`
+
+	// Serialized label selector matching replicant pods.
+	// Used by the scale subresource for HPA pod discovery.
+	ReplicantSelector string `json:"replicantSelector,omitempty"`
+
 	// Status of each core node in the cluster.
 	CoreNodes []EMQXNode `json:"coreNodes,omitempty"`
 	// Summary status of the set of core nodes.

config/crd/bases/apps.emqx.io_emqxes.yaml

@@ -16193,6 +16193,13 @@ spec:
                 - readyReplicas
                 - updatedReplicas
                 type: object
+              coreReplicas:
+                description: Observed replica count for the core set.
+                format: int32
+                type: integer
+              coreSelector:
+                description: Serialized label selector matching core pods.
+                type: string
               dsReplication:
                 description: Status of EMQX Durable Storage replication.
                 properties:
@@ -16351,12 +16358,27 @@ spec:
                 - readyReplicas
                 - updateReplicas
                 type: object
+              replicantReplicas:
+                description: |-
+                  Observed replica count for the replicant set pods.
+                  Used by the scale subresource.
+                format: int32
+                type: integer
+              replicantSelector:
+                description: |-
+                  Serialized label selector matching replicant pods.
+                  Used by the scale subresource for HPA pod discovery.
+                type: string
+            required:
+            - coreReplicas
+            - replicantReplicas
             type: object
         type: object
     served: true
     storage: true
     subresources:
       scale:
+        labelSelectorPath: .status.replicantSelector
         specReplicasPath: .spec.replicantTemplate.spec.replicas
-        statusReplicasPath: .status.replicantNodeReplicas
+        statusReplicasPath: .status.replicantReplicas
       status: {}

internal/controller/add_replicant_set.go

@@ -155,8 +155,6 @@ func generateReplicaSet(instance *crd.EMQX) *appsv1.ReplicaSet {
 	template := instance.Spec.ReplicantTemplate
 
 	// Add a PreStop hook to leave the cluster when the pod is asked to stop.
-	// This is especially important when DS Raft is enabled, otherwise there will be a
-	// lot of leftover records in the DS cluster metadata.
 	lifecycle := &corev1.Lifecycle{}
 	if template.Spec.Lifecycle != nil {
 		lifecycle = template.Spec.Lifecycle.DeepCopy()
@@ -167,9 +165,8 @@ func generateReplicaSet(instance *crd.EMQX) *appsv1.ReplicaSet {
 		},
 	}
 
-	readinessProbe := resources.EvacuationReadinessProbe()
-
 	// Prefer evacuation-aware probe over older-version defaults.
+	readinessProbe := resources.EvacuationReadinessProbe()
 	if template.Spec.ReadinessProbe != nil {
 		if template.Spec.ReadinessProbe.HTTPGet != nil &&
 			template.Spec.ReadinessProbe.HTTPGet.Path != "/status" {

internal/controller/load_state.go

@@ -2,6 +2,7 @@ package controller
 
 import (
 	"context"
+	"reflect"
 
 	emperror "emperror.dev/errors"
 	crd "github.com/emqx/emqx-operator/api/v3alpha1"
@@ -28,11 +29,14 @@ type reconcileStatePodFilter interface {
 }
 
 type podsManagedBy struct {
-	metav1.Object
+	manager metav1.Object
 }
 
 func (self podsManagedBy) passes(pod *corev1.Pod) bool {
-	return self.Object != nil && util.IsPodManagedBy(pod, self.Object)
+	if self.manager == nil && reflect.ValueOf(self.manager).IsNil() {
+		return false
+	}
+	return util.IsPodManagedBy(pod, self.manager)
 }
 
 type podsWithRole struct {
@@ -191,6 +195,14 @@ func (r *reconcileState) outdatedReplicantPods(instance *crd.EMQX) []*corev1.Pod
 	return out
 }
 
+func (r *reconcileState) numReplicants() int32 {
+	out := int32(0)
+	for _, rs := range r.replicantSets {
+		out += rs.Status.Replicas
+	}
+	return out
+}
+
 func (r *reconcileState) numReadyReplicants() int32 {
 	out := int32(0)
 	for _, rs := range r.replicantSets {

internal/controller/sync_replicant_sets.go

@@ -45,31 +45,34 @@ func (s *syncReplicantSets) reconcile(r *reconcileRound, instance *crd.EMQX) sub
 	}
 
 	// Steady state: handle scale-up and scale-down.
+	specReplicas := ptr.Deref(updateRs.Spec.Replicas, 0)
 	desiredReplicas := instance.Spec.NumReplicantReplicas()
-	currentReplicas := ptr.Deref(updateRs.Spec.Replicas, 0)
+	currentReplicas := updateRs.Status.Replicas
 
-	if currentReplicas < desiredReplicas {
+	if specReplicas < desiredReplicas {
 		r.log.V(1).Info("scaling up replicantSet",
 			"replicaSet", klog.KObj(updateRs),
-			"from", currentReplicas,
+			"from", specReplicas,
 			"to", desiredReplicas,
 		)
 		return s.scaleUp(r, updateRs, desiredReplicas)
 	}
 
-	if currentReplicas > desiredReplicas {
+	if specReplicas > desiredReplicas {
 		r.log.V(1).Info("scaling down replicantSet",
 			"replicaSet", klog.KObj(updateRs),
-			"from", currentReplicas,
+			"from", specReplicas,
 			"to", desiredReplicas,
 		)
-		return s.scaleDown(r, instance, updateRs, currentReplicas, desiredReplicas)
+		return s.scaleDown(r, instance, updateRs, specReplicas, desiredReplicas)
 	}
 
 	// Steady state: clean up stale artifacts.
-	err := s.ensureConsistency(r, instance, updateRs)
-	if err != nil {
-		return reconcileError(emperror.Wrap(err, "failed to restore replicant consistency"))
+	if currentReplicas == specReplicas {
+		err := s.ensureConsistency(r, instance, updateRs)
+		if err != nil {
+			return reconcileError(emperror.Wrap(err, "failed to restore replicant consistency"))
+		}
 	}
 
 	return subResult{}
@@ -142,7 +145,7 @@ func (s *syncReplicantSets) ensureConsistency(
 	instance *crd.EMQX,
 	updateRs *appsv1.ReplicaSet,
 ) error {
-	for _, pod := range r.state.podsManagedBy(updateRs) {
+	for _, pod := range r.state.listPods(podsManagedBy{updateRs}, podsAlive{}) {
 		// 1. Check if pod has stale scale-down annotations.
 		dirty := s.removeStaleReplicantAnnotations(pod)
 		if !dirty {

internal/controller/update_emqx_status.go

@@ -11,6 +11,7 @@ import (
 	"github.com/emqx/emqx-operator/internal/emqx/api"
 	appsv1 "k8s.io/api/apps/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/klog/v2"
 )
 
@@ -23,17 +24,28 @@ func (u *updateStatus) reconcile(r *reconcileRound, instance *crd.EMQX) subResul
 
 	// Core: count pods on each revision for rolling update progress.
 	coreSet := r.state.coreSet()
+	status.CoreReplicas = 0
+	status.CoreSelector = labels.Set(instance.DefaultLabelsWith(crd.CoreLabels())).String()
 	status.CoreNodesStatus.UpdatedReplicas = 0
 	status.CoreNodesStatus.CurrentReplicas = 0
-	for _, pod := range r.state.podsManagedBy(r.state.coreSet()) {
-		if r.state.partOfCoreSetRevision(pod, coreSet.Status.UpdateRevision) {
-			status.CoreNodesStatus.UpdatedReplicas++
-		}
-		if r.state.partOfCoreSetRevision(pod, coreSet.Status.CurrentRevision) {
-			status.CoreNodesStatus.CurrentReplicas++
+	if coreSet != nil {
+		for _, pod := range r.state.podsManagedBy(coreSet) {
+			if r.state.partOfCoreSetRevision(pod, coreSet.Status.UpdateRevision) {
+				status.CoreNodesStatus.UpdatedReplicas++
+			}
+			if r.state.partOfCoreSetRevision(pod, coreSet.Status.CurrentRevision) {
+				status.CoreNodesStatus.CurrentReplicas++
+			}
 		}
 	}
 
+	status.ReplicantReplicas = 0
+	status.ReplicantSelector = ""
+	if instance.Spec.HasReplicants() {
+		status.ReplicantReplicas = r.state.numReplicants()
+		status.ReplicantSelector = labels.Set(instance.DefaultLabelsWith(crd.ReplicantLabels())).String()
+	}
+
 	// Replicant: multi-ReplicaSet pattern retained.
 	currentReplicantSet, updateReplicantSet := switchReplicantSet(r, instance)
 

test/e2e/e2e_suite_test.go

@@ -85,32 +85,26 @@ var _ = BeforeSuite(func() {
 	By("load emqx-operator docker image into kind cluster")
 	Expect(util.LoadImageToKindClusterWithName(projectImage)).To(Succeed())
 
+	By("install Metrics Server")
+	Expect(util.InstallMetricsServer()).To(Succeed())
+
 	// The tests-e2e are intended to run on a temporary cluster that is created and destroyed for testing.
 	// To prevent errors when tests run in environments with Prometheus or CertManager already installed,
 	// we check for their presence before execution.
-	// Setup Prometheus and CertManager before the suite if not skipped and if not already installed
 	if !skipPrometheusInstall {
-		if !util.IsPrometheusCRDsInstalled() {
-			By("install Prometheus Operator")
-			Expect(util.InstallPrometheusOperator()).To(Succeed())
-			isPrometheusInstalled = true
-		} else {
-			GinkgoWriter.Println("WARNING: Prometheus Operator is already installed, skipping installation.")
-		}
+		By("install Prometheus Operator")
+		Expect(util.InstallPrometheusOperator()).To(Succeed())
+		isPrometheusInstalled = true
 	}
 	if !skipCertManagerInstall {
-		if !util.IsCertManagerCRDsInstalled() {
-			By("install CertManager")
-			Expect(util.InstallCertManager()).To(Succeed())
-			isCertManagerInstalled = true
-		} else {
-			GinkgoWriter.Println("WARNING: CertManager is already installed, skipping installation.")
-		}
+		By("install CertManager")
+		Expect(util.InstallCertManager()).To(Succeed())
+		isCertManagerInstalled = true
 	}
 })
 
 var _ = AfterSuite(func() {
-	// Teardown Prometheus and CertManager after the suite if not skipped and if they were not already installed
+	util.UnnstallMetricsServer()
 	if !skipPrometheusInstall && isPrometheusInstalled {
 		By("uninstall Prometheus Operator")
 		util.UninstallPrometheusOperator()