feat(api): allow turning off node evacuations for rolling updates

Author: keynslug

api/v3alpha1/emqx_types_spec.go

@@ -150,7 +150,19 @@ type ReplicantsUpdateStrategy struct {
 	MaxSurge *intstr.IntOrString `json:"maxSurge,omitempty"`
 }
 
+type EvacuationStrategyType string
+
+const (
+	NodeEvacuationStrategy     EvacuationStrategyType = "NodeEvacuation"
+	DisabledEvacuationStrategy EvacuationStrategyType = "Disabled"
+)
+
 type EvacuationStrategy struct {
+	// Type of the evacuation policy.
+	// +kubebuilder:validation:Enum=NodeEvacuation;Disabled
+	// +kubebuilder:default=NodeEvacuation
+	Type EvacuationStrategyType `json:"type"`
+
 	// Client disconnect rate (number per second).
 	// Same as `conn-evict-rate` in [EMQX Node Evacuation](https://docs.emqx.com/en/emqx/v5.10/deploy/cluster/rebalancing.html#node-evacuation).
 	// +kubebuilder:validation:Minimum=1
@@ -342,6 +354,10 @@ func (spec *EMQXSpec) HasReplicants() bool {
 	return spec.ReplicantTemplate != nil && spec.ReplicantTemplate.Spec.Replicas != nil && *spec.ReplicantTemplate.Spec.Replicas > 0
 }
 
+func (spec *EMQXSpec) IsEvacuationEnabled() bool {
+	return spec.UpdateStrategy.EvacuationStrategy.Type != DisabledEvacuationStrategy
+}
+
 func (s *ServiceTemplate) IsEnabled() bool {
 	return s.Enabled != nil && *s.Enabled
 }

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

@@ -15986,6 +15986,13 @@ spec:
                         format: int32
                         minimum: 1
                         type: integer
+                      type:
+                        default: NodeEvacuation
+                        description: Type of the evacuation policy.
+                        enum:
+                        - NodeEvacuation
+                        - Disabled
+                        type: string
                       waitHealthCheck:
                         default: 60
                         description: |-
@@ -16002,6 +16009,8 @@ spec:
                         format: int32
                         minimum: 0
                         type: integer
+                    required:
+                    - type
                     type: object
                   replicants:
                     description: |-

internal/controller/suite_test.go

@@ -77,6 +77,12 @@ var emqx *crd.EMQX = &crd.EMQX{
 	},
 	Spec: crd.EMQXSpec{
 		Image: "emqx",
+		UpdateStrategy: crd.UpdateStrategy{
+			Type: "RollingUpdate",
+			EvacuationStrategy: crd.EvacuationStrategy{
+				Type: "NodeEvacuation",
+			},
+		},
 	},
 }
 

internal/controller/sync_core_set.go

@@ -81,6 +81,9 @@ func (s *syncCoreSet) reconcile(r *reconcileRound, instance *crd.EMQX) subResult
 		return s.scaleDown(r, instance, currentReplicas)
 	}
 
+	// Stop evacuations of any updated pods.
+	// This is done irrespective of whether Node Evacuation is enabled or not,
+	// to avoid ending up in transient state if it was disabled mid-update.
 	err := s.updateEvacuationState(r, instance)
 	if err != nil {
 		return reconcileError(emperror.Wrap(err, "failed to update evacuation state"))
@@ -270,7 +273,7 @@ func checkCorePodRemoval(
 		return coreAdmission{Action: admissionWait, Reason: "node evacuation is still in progress"}
 	}
 
-	if nodeInfo.Sessions > 0 {
+	if nodeInfo.Sessions > 0 && instance.Spec.IsEvacuationEnabled() {
 		if instance.Spec.NumCoreReplicas() == 1 && !instance.Spec.HasReplicants() {
 			return coreAdmission{
 				Action: admissionRemove,

internal/controller/sync_core_set_suite_test.go

@@ -170,6 +170,15 @@ var _ = Describe("Reconciler syncCoreSet", Ordered, func() {
 		))
 	})
 
+	It("node session > 0 & node evacuation is disabled", func() {
+		instance.Spec.UpdateStrategy.EvacuationStrategy.Type = crd.DisabledEvacuationStrategy
+		instance.Status.CoreNodes[1].Sessions = 99999
+		admission := checkCorePodRemoval(round, instance, pod1, false)
+		Expect(admission).Should(And(
+			HaveField("Action", Equal(admissionRemove)),
+		))
+	})
+
 	It("node session is 0", func() {
 		instance.Status.CoreNodes[1].Sessions = 0
 		admission := checkCorePodRemoval(round, instance, pod1, false)

internal/controller/sync_replicant_sets.go

@@ -153,6 +153,8 @@ func (s *syncReplicantSets) ensureConsistency(
 		}
 
 		// 2. Stop any ongoing node evacuation.
+		// This is attempted irrespective of whether Node Evacuation is enabled or not,
+		// to avoid ending up in transient state if it was disabled mid-update.
 		stopped, err := s.stopStaleReplicantEvacuation(r, instance, pod)
 		if err != nil {
 			return err
@@ -474,7 +476,7 @@ func checkReplicantPodRemoval(instance *crd.EMQX, pod *corev1.Pod) replicantAdmi
 		}
 	}
 
-	if nodeInfo.Sessions > 0 {
+	if nodeInfo.Sessions > 0 && instance.Spec.IsEvacuationEnabled() {
 		if instance.Spec.NumReplicantReplicas() == 1 && instance.Spec.NumMaxSurgeReplicantReplicas() == 0 {
 			return replicantAdmission{
 				Action: admissionRemove,

internal/controller/sync_replicant_sets_suite_test.go

@@ -824,6 +824,16 @@ var _ = Describe("Reconciler syncReplicantSets admission", Ordered, func() {
 		))
 	})
 
+	It("node session > 0 & evacuation is disabled", func() {
+		instance.Spec.UpdateStrategy.EvacuationStrategy.Type = crd.DisabledEvacuationStrategy
+		instance.Status.ReplicantNodes[0].Sessions = 99999
+		admission := checkReplicantPodRemoval(instance, currentPod)
+		Expect(admission).Should(And(
+			HaveField("Action", Equal(admissionRemove)),
+			HaveField("Reason", ContainSubstring("safe to stop")),
+		))
+	})
+
 	It("node session is 0", func() {
 		instance.Status.ReplicantNodes[0].Sessions = 0
 		admission := checkReplicantPodRemoval(instance, currentPod)