From 33d1408211dfe5b9e27122d592ca7ce7eaa46ecc Mon Sep 17 00:00:00 2001 From: Rory Z <16801068+Rory-Z@users.noreply.github.com> Date: Wed, 10 Jan 2024 14:52:05 +0800 Subject: [PATCH] fix: fix when scale replicant pod to 0, core pod can not ready Signed-off-by: Rory Z <16801068+Rory-Z@users.noreply.github.com> --- apis/apps/v2beta1/status.go | 4 +- apis/apps/v2beta1/zz_generated.deepcopy.go | 6 +- .../apps/v2beta1/add_emqx_repl_suite_test.go | 2 +- .../apps/v2beta1/add_emqx_repl_test.go | 2 +- controllers/apps/v2beta1/status_machine.go | 4 +- controllers/apps/v2beta1/sync_pods.go | 1 - .../apps/v2beta1/sync_pods_suite_test.go | 4 +- .../apps/v2beta1/update_emqx_status.go | 80 ++++++------ .../apps/v2beta1/update_pod_conditions.go | 58 ++++++--- controllers/apps/v2beta1/util.go | 7 - e2e/v2beta1/e2e_test.go | 121 +++++++++++++++--- 11 files changed, 190 insertions(+), 99 deletions(-) diff --git a/apis/apps/v2beta1/status.go b/apis/apps/v2beta1/status.go index 33f950cb9..c71993311 100644 --- a/apis/apps/v2beta1/status.go +++ b/apis/apps/v2beta1/status.go @@ -31,8 +31,8 @@ type EMQXStatus struct { CoreNodes []EMQXNode `json:"coreNodes,omitempty"` CoreNodesStatus EMQXNodesStatus `json:"coreNodesStatus,omitempty"` - ReplicantNodes []EMQXNode `json:"replicantNodes,omitempty"` - ReplicantNodesStatus *EMQXNodesStatus `json:"replicantNodesStatus,omitempty"` + ReplicantNodes []EMQXNode `json:"replicantNodes,omitempty"` + ReplicantNodesStatus EMQXNodesStatus `json:"replicantNodesStatus,omitempty"` NodeEvacuationsStatus []NodeEvacuationStatus `json:"nodEvacuationsStatus,omitempty"` } diff --git a/apis/apps/v2beta1/zz_generated.deepcopy.go b/apis/apps/v2beta1/zz_generated.deepcopy.go index 524169422..fa7f984aa 100644 --- a/apis/apps/v2beta1/zz_generated.deepcopy.go +++ b/apis/apps/v2beta1/zz_generated.deepcopy.go @@ -412,11 +412,7 @@ func (in *EMQXStatus) DeepCopyInto(out *EMQXStatus) { *out = make([]EMQXNode, len(*in)) copy(*out, *in) } - if in.ReplicantNodesStatus != nil { - in, out := &in.ReplicantNodesStatus, &out.ReplicantNodesStatus - *out = new(EMQXNodesStatus) - (*in).DeepCopyInto(*out) - } + in.ReplicantNodesStatus.DeepCopyInto(&out.ReplicantNodesStatus) if in.NodeEvacuationsStatus != nil { in, out := &in.NodeEvacuationsStatus, &out.NodeEvacuationsStatus *out = make([]NodeEvacuationStatus, len(*in)) diff --git a/controllers/apps/v2beta1/add_emqx_repl_suite_test.go b/controllers/apps/v2beta1/add_emqx_repl_suite_test.go index 79a13b952..ee5e60d6c 100644 --- a/controllers/apps/v2beta1/add_emqx_repl_suite_test.go +++ b/controllers/apps/v2beta1/add_emqx_repl_suite_test.go @@ -39,7 +39,7 @@ var _ = Describe("Check add repl controller", Ordered, Label("repl"), func() { }, } instance.Status = appsv2beta1.EMQXStatus{ - ReplicantNodesStatus: &appsv2beta1.EMQXNodesStatus{ + ReplicantNodesStatus: appsv2beta1.EMQXNodesStatus{ Replicas: 3, }, Conditions: []metav1.Condition{ diff --git a/controllers/apps/v2beta1/add_emqx_repl_test.go b/controllers/apps/v2beta1/add_emqx_repl_test.go index 0a807715d..300b9960c 100644 --- a/controllers/apps/v2beta1/add_emqx_repl_test.go +++ b/controllers/apps/v2beta1/add_emqx_repl_test.go @@ -40,7 +40,7 @@ func TestGetNewReplicaSet(t *testing.T) { Replicas: pointer.Int32(3), }, } - instance.Status.ReplicantNodesStatus = &appsv2beta1.EMQXNodesStatus{ + instance.Status.ReplicantNodesStatus = appsv2beta1.EMQXNodesStatus{ CollisionCount: pointer.Int32(0), } diff --git a/controllers/apps/v2beta1/status_machine.go b/controllers/apps/v2beta1/status_machine.go index 048304678..c4c9b63fe 100644 --- a/controllers/apps/v2beta1/status_machine.go +++ b/controllers/apps/v2beta1/status_machine.go @@ -139,7 +139,7 @@ func (s *coreNodesProgressingStatus) nextStatus(ctx context.Context) { emqx := s.emqxStatusMachine.GetEMQX() updateSts, _, _ := getStateFulSetList(ctx, s.emqxStatusMachine.client, emqx) - if updateSts != nil && updateSts.Status.ReadyReplicas == emqx.Status.CoreNodesStatus.Replicas { + if updateSts != nil && updateSts.Status.ReadyReplicas != 0 && updateSts.Status.ReadyReplicas == emqx.Status.CoreNodesStatus.Replicas { emqx.Status.SetCondition(metav1.Condition{ Type: appsv2beta1.CoreNodesReady, Status: metav1.ConditionTrue, @@ -191,7 +191,7 @@ func (s *replicantNodesProgressingStatus) nextStatus(ctx context.Context) { } updateRs, _, _ := getReplicaSetList(ctx, s.emqxStatusMachine.client, emqx) - if updateRs != nil && updateRs.Status.ReadyReplicas == emqx.Status.ReplicantNodesStatus.Replicas { + if updateRs != nil && updateRs.Status.ReadyReplicas != 0 && updateRs.Status.ReadyReplicas == emqx.Status.ReplicantNodesStatus.Replicas { emqx.Status.SetCondition(metav1.Condition{ Type: appsv2beta1.ReplicantNodesReady, Status: metav1.ConditionTrue, diff --git a/controllers/apps/v2beta1/sync_pods.go b/controllers/apps/v2beta1/sync_pods.go index da4526775..c60461f88 100644 --- a/controllers/apps/v2beta1/sync_pods.go +++ b/controllers/apps/v2beta1/sync_pods.go @@ -44,7 +44,6 @@ func (s *syncPods) reconcile(ctx context.Context, logger logr.Logger, instance * } } } - } else { if updateSts != nil { for _, node := range instance.Status.CoreNodes { diff --git a/controllers/apps/v2beta1/sync_pods_suite_test.go b/controllers/apps/v2beta1/sync_pods_suite_test.go index ff83bffb6..1884394b5 100644 --- a/controllers/apps/v2beta1/sync_pods_suite_test.go +++ b/controllers/apps/v2beta1/sync_pods_suite_test.go @@ -72,7 +72,7 @@ var _ = Describe("Check sync pods controller", Ordered, Label("node"), func() { ReadyReplicas: 2, Replicas: 1, }, - ReplicantNodesStatus: &appsv2beta1.EMQXNodesStatus{ + ReplicantNodesStatus: appsv2beta1.EMQXNodesStatus{ UpdateRevision: "update", UpdateReplicas: 1, CurrentRevision: "current", @@ -334,7 +334,7 @@ var _ = Describe("check can be scale down", func() { Replicas: pointer.Int32Ptr(3), }, } - instance.Status.ReplicantNodesStatus = &appsv2beta1.EMQXNodesStatus{ + instance.Status.ReplicantNodesStatus = appsv2beta1.EMQXNodesStatus{ UpdateRevision: "update", CurrentRevision: "current", } diff --git a/controllers/apps/v2beta1/update_emqx_status.go b/controllers/apps/v2beta1/update_emqx_status.go index 276a56db9..467be1f4f 100644 --- a/controllers/apps/v2beta1/update_emqx_status.go +++ b/controllers/apps/v2beta1/update_emqx_status.go @@ -21,26 +21,32 @@ type updateStatus struct { } func (u *updateStatus) reconcile(ctx context.Context, logger logr.Logger, instance *appsv2beta1.EMQX, r innerReq.RequesterInterface) subResult { - if instance.Spec.ReplicantTemplate != nil && instance.Status.ReplicantNodesStatus == nil { - instance.Status.ReplicantNodesStatus = &appsv2beta1.EMQXNodesStatus{ - Replicas: *instance.Spec.ReplicantTemplate.Spec.Replicas, - } + instance.Status.CoreNodesStatus.Replicas = *instance.Spec.CoreTemplate.Spec.Replicas + if instance.Spec.ReplicantTemplate != nil { + instance.Status.ReplicantNodesStatus.Replicas = *instance.Spec.ReplicantTemplate.Spec.Replicas } - updateRs, currentRs, oldRsList := getReplicaSetList(ctx, u.Client, instance) - if updateRs != nil { - if currentRs == nil || currentRs.Status.Replicas == 0 { + if instance.Status.CoreNodesStatus.UpdateRevision != "" && instance.Status.CoreNodesStatus.CurrentRevision == "" { + instance.Status.CoreNodesStatus.CurrentRevision = instance.Status.CoreNodesStatus.UpdateRevision + } + if instance.Status.ReplicantNodesStatus.UpdateRevision != "" && instance.Status.ReplicantNodesStatus.CurrentRevision == "" { + instance.Status.ReplicantNodesStatus.CurrentRevision = instance.Status.ReplicantNodesStatus.UpdateRevision + } + + updateSts, currentSts, oldStsList := getStateFulSetList(ctx, u.Client, instance) + if updateSts != nil && updateSts.UID != currentSts.UID { + if currentSts.Status.Replicas == 0 { var i int - for i = 0; i < len(oldRsList); i++ { - if oldRsList[i].Status.Replicas > 0 { - currentRs = oldRsList[i] + for i = 0; i < len(oldStsList); i++ { + if oldStsList[i].Status.Replicas > 0 { + currentSts = oldStsList[i] break } } - if i == len(oldRsList) { - currentRs = updateRs + if i == len(oldStsList) { + currentSts = updateSts } - instance.Status.ReplicantNodesStatus.CurrentRevision = currentRs.Labels[appsv2beta1.LabelsPodTemplateHashKey] + instance.Status.CoreNodesStatus.CurrentRevision = currentSts.Labels[appsv2beta1.LabelsPodTemplateHashKey] if err := u.Client.Status().Update(ctx, instance); err != nil { return subResult{err: emperror.Wrap(err, "failed to update status")} } @@ -48,20 +54,20 @@ func (u *updateStatus) reconcile(ctx context.Context, logger logr.Logger, instan } } - updateSts, currentSts, oldStsList := getStateFulSetList(ctx, u.Client, instance) - if updateSts != nil { - if currentSts == nil || currentSts.Status.Replicas == 0 { + updateRs, currentRs, oldRsList := getReplicaSetList(ctx, u.Client, instance) + if updateRs != nil && updateRs.UID != currentRs.UID { + if currentRs.Status.Replicas == 0 { var i int - for i = 0; i < len(oldStsList); i++ { - if oldStsList[i].Status.Replicas > 0 { - currentSts = oldStsList[i] + for i = 0; i < len(oldRsList); i++ { + if oldRsList[i].Status.Replicas > 0 { + currentRs = oldRsList[i] break } } - if i == len(oldStsList) { - currentSts = updateSts + if i == len(oldRsList) { + currentRs = updateRs } - instance.Status.CoreNodesStatus.CurrentRevision = currentSts.Labels[appsv2beta1.LabelsPodTemplateHashKey] + instance.Status.ReplicantNodesStatus.CurrentRevision = currentRs.Labels[appsv2beta1.LabelsPodTemplateHashKey] if err := u.Client.Status().Update(ctx, instance); err != nil { return subResult{err: emperror.Wrap(err, "failed to update status")} } @@ -80,7 +86,6 @@ func (u *updateStatus) reconcile(ctx context.Context, logger logr.Logger, instan } instance.Status.CoreNodes = coreNodes - instance.Status.CoreNodesStatus.Replicas = *instance.Spec.CoreTemplate.Spec.Replicas instance.Status.CoreNodesStatus.ReadyReplicas = 0 instance.Status.CoreNodesStatus.CurrentReplicas = 0 instance.Status.CoreNodesStatus.UpdateReplicas = 0 @@ -96,22 +101,19 @@ func (u *updateStatus) reconcile(ctx context.Context, logger logr.Logger, instan } } - if len(replNodes) > 0 { - instance.Status.ReplicantNodes = replNodes - instance.Status.ReplicantNodesStatus.Replicas = *instance.Spec.ReplicantTemplate.Spec.Replicas - instance.Status.ReplicantNodesStatus.ReadyReplicas = 0 - instance.Status.ReplicantNodesStatus.CurrentReplicas = 0 - instance.Status.ReplicantNodesStatus.UpdateReplicas = 0 - for _, node := range replNodes { - if node.NodeStatus == "running" { - instance.Status.ReplicantNodesStatus.ReadyReplicas++ - } - if currentRs != nil && node.ControllerUID == currentRs.UID { - instance.Status.ReplicantNodesStatus.CurrentReplicas++ - } - if updateRs != nil && node.ControllerUID == updateRs.UID { - instance.Status.ReplicantNodesStatus.UpdateReplicas++ - } + instance.Status.ReplicantNodes = replNodes + instance.Status.ReplicantNodesStatus.ReadyReplicas = 0 + instance.Status.ReplicantNodesStatus.CurrentReplicas = 0 + instance.Status.ReplicantNodesStatus.UpdateReplicas = 0 + for _, node := range replNodes { + if node.NodeStatus == "running" { + instance.Status.ReplicantNodesStatus.ReadyReplicas++ + } + if currentRs != nil && node.ControllerUID == currentRs.UID { + instance.Status.ReplicantNodesStatus.CurrentReplicas++ + } + if updateRs != nil && node.ControllerUID == updateRs.UID { + instance.Status.ReplicantNodesStatus.UpdateReplicas++ } } diff --git a/controllers/apps/v2beta1/update_pod_conditions.go b/controllers/apps/v2beta1/update_pod_conditions.go index 3bab73c6e..4530e198d 100644 --- a/controllers/apps/v2beta1/update_pod_conditions.go +++ b/controllers/apps/v2beta1/update_pod_conditions.go @@ -21,8 +21,21 @@ type updatePodConditions struct { } func (u *updatePodConditions) reconcile(ctx context.Context, logger logr.Logger, instance *appsv2beta1.EMQX, r innerReq.RequesterInterface) subResult { + var updateStsUID, currentStsUID, updateRsUID, currentRsUID types.UID updateRs, currentRs, _ := getReplicaSetList(ctx, u.Client, instance) + if updateRs != nil { + updateRsUID = updateRs.UID + } + if currentRs != nil { + currentRsUID = currentRs.UID + } updateSts, currentSts, _ := getStateFulSetList(ctx, u.Client, instance) + if updateSts != nil { + updateStsUID = updateSts.UID + } + if currentSts != nil { + currentStsUID = currentSts.UID + } pods := &corev1.PodList{} _ = u.Client.List(ctx, pods, @@ -38,40 +51,43 @@ func (u *updatePodConditions) reconcile(ctx context.Context, logger logr.Logger, } onServingCondition := corev1.PodCondition{ - Type: appsv2beta1.PodOnServing, - Status: corev1.ConditionFalse, - LastTransitionTime: metav1.Now(), + Type: appsv2beta1.PodOnServing, + } + for _, condition := range pod.Status.Conditions { + if condition.Type == appsv2beta1.PodOnServing { + onServingCondition.Status = condition.Status + onServingCondition.LastTransitionTime = condition.LastTransitionTime + } } - if instance.Status.IsConditionTrue(appsv2beta1.Available) { - if (updateSts != nil && controllerRef.UID == updateSts.UID) || - (updateRs != nil && controllerRef.UID == updateRs.UID) { - for _, condition := range pod.Status.Conditions { - if condition.Type == corev1.ContainersReady && condition.Status == corev1.ConditionTrue { - onServingCondition.Status = u.checkInCluster(instance, r, pod) - break + switch controllerRef.UID { + case updateStsUID, updateRsUID: + for _, condition := range pod.Status.Conditions { + if condition.Type == corev1.ContainersReady && condition.Status == corev1.ConditionTrue { + status := u.checkInCluster(instance, r, pod) + if status != onServingCondition.Status { + onServingCondition.Status = status + onServingCondition.LastTransitionTime = metav1.Now() } + break } } - } else { - if (currentSts != nil && controllerRef.UID == currentSts.UID) || - (currentRs != nil && controllerRef.UID == currentRs.UID) || - (updateSts != nil && controllerRef.UID == updateSts.UID) || - (updateRs != nil && controllerRef.UID == updateRs.UID) { + case currentStsUID, currentRsUID: + // When available condition is true, need clean currentSts / currentRs pod + if instance.Status.IsConditionTrue(appsv2beta1.Available) { for _, condition := range pod.Status.Conditions { if condition.Type == corev1.ContainersReady && condition.Status == corev1.ConditionTrue { - onServingCondition.Status = u.checkInCluster(instance, r, pod) + status := corev1.ConditionFalse + if status != onServingCondition.Status { + onServingCondition.Status = status + onServingCondition.LastTransitionTime = metav1.Now() + } break } } } } - for _, condition := range pod.Status.Conditions { - if condition.Type == appsv2beta1.PodOnServing && condition.Status == onServingCondition.Status { - onServingCondition.LastTransitionTime = condition.LastTransitionTime - } - } patchBytes, _ := json.Marshal(corev1.Pod{ Status: corev1.PodStatus{ Conditions: []corev1.PodCondition{onServingCondition}, diff --git a/controllers/apps/v2beta1/util.go b/controllers/apps/v2beta1/util.go index 10427378c..0fee0ef3e 100644 --- a/controllers/apps/v2beta1/util.go +++ b/controllers/apps/v2beta1/util.go @@ -93,13 +93,6 @@ func getReplicaSetList(ctx context.Context, k8sClient client.Client, instance *a client.InNamespace(instance.Namespace), client.MatchingLabels(labels), ) - if instance.Spec.ReplicantTemplate == nil { - for _, rs := range list.Items { - oldRsList = append(oldRsList, rs.DeepCopy()) - } - sort.Sort(ReplicaSetsByCreationTimestamp(oldRsList)) - return - } for _, rs := range list.Items { if hash, ok := rs.Labels[appsv2beta1.LabelsPodTemplateHashKey]; ok { diff --git a/e2e/v2beta1/e2e_test.go b/e2e/v2beta1/e2e_test.go index b098d0512..e301d842b 100644 --- a/e2e/v2beta1/e2e_test.go +++ b/e2e/v2beta1/e2e_test.go @@ -72,17 +72,24 @@ var _ = Describe("E2E Test", Label("base"), Ordered, func() { }, And( HaveField("Replicas", Equal(int32(*instance.Spec.CoreTemplate.Spec.Replicas))), HaveField("ReadyReplicas", Equal(int32(*instance.Spec.CoreTemplate.Spec.Replicas))), - HaveField("CurrentRevision", Not(BeEmpty())), + HaveField("CurrentRevision", Not(Equal(""))), HaveField("CurrentReplicas", Equal(int32(*instance.Spec.CoreTemplate.Spec.Replicas))), - HaveField("UpdateRevision", Not(BeEmpty())), + HaveField("UpdateRevision", Not(Equal(""))), HaveField("UpdateReplicas", Equal(int32(*instance.Spec.CoreTemplate.Spec.Replicas))), )), WithTransform(func(instance *appsv2beta1.EMQX) []appsv2beta1.EMQXNode { return instance.Status.ReplicantNodes }, BeNil()), - WithTransform(func(instance *appsv2beta1.EMQX) *appsv2beta1.EMQXNodesStatus { + WithTransform(func(instance *appsv2beta1.EMQX) appsv2beta1.EMQXNodesStatus { return instance.Status.ReplicantNodesStatus - }, BeNil()), + }, And( + HaveField("Replicas", Equal(int32(0))), + HaveField("ReadyReplicas", Equal(int32(0))), + HaveField("CurrentRevision", Equal("")), + HaveField("CurrentReplicas", Equal(int32(0))), + HaveField("UpdateRevision", Equal("")), + HaveField("UpdateReplicas", Equal(int32(0))), + )), ), ) @@ -124,15 +131,22 @@ var _ = Describe("E2E Test", Label("base"), Ordered, func() { HaveField("ReadyReplicas", Equal(int32(*instance.Spec.CoreTemplate.Spec.Replicas))), HaveField("CurrentRevision", Equal(storage.Status.CoreNodesStatus.CurrentRevision)), HaveField("CurrentReplicas", Equal(int32(*instance.Spec.CoreTemplate.Spec.Replicas))), - HaveField("UpdateRevision", Not(BeEmpty())), + HaveField("UpdateRevision", Not(Equal(""))), HaveField("UpdateReplicas", Equal(int32(*instance.Spec.CoreTemplate.Spec.Replicas))), )), WithTransform(func(instance *appsv2beta1.EMQX) []appsv2beta1.EMQXNode { return instance.Status.ReplicantNodes }, BeNil()), - WithTransform(func(instance *appsv2beta1.EMQX) *appsv2beta1.EMQXNodesStatus { + WithTransform(func(instance *appsv2beta1.EMQX) appsv2beta1.EMQXNodesStatus { return instance.Status.ReplicantNodesStatus - }, BeNil()), + }, And( + HaveField("Replicas", Equal(int32(0))), + HaveField("ReadyReplicas", Equal(int32(0))), + HaveField("CurrentRevision", Equal("")), + HaveField("CurrentReplicas", Equal(int32(0))), + HaveField("UpdateRevision", Equal("")), + HaveField("UpdateReplicas", Equal(int32(0))), + )), ), ) @@ -180,9 +194,16 @@ var _ = Describe("E2E Test", Label("base"), Ordered, func() { WithTransform(func(instance *appsv2beta1.EMQX) []appsv2beta1.EMQXNode { return instance.Status.ReplicantNodes }, BeNil()), - WithTransform(func(instance *appsv2beta1.EMQX) *appsv2beta1.EMQXNodesStatus { + WithTransform(func(instance *appsv2beta1.EMQX) appsv2beta1.EMQXNodesStatus { return instance.Status.ReplicantNodesStatus - }, BeNil()), + }, And( + HaveField("Replicas", Equal(int32(0))), + HaveField("ReadyReplicas", Equal(int32(0))), + HaveField("CurrentRevision", Equal("")), + HaveField("CurrentReplicas", Equal(int32(0))), + HaveField("UpdateRevision", Equal("")), + HaveField("UpdateReplicas", Equal(int32(0))), + )), ), ) @@ -220,9 +241,16 @@ var _ = Describe("E2E Test", Label("base"), Ordered, func() { HaveField("CurrentRevision", Not(Equal(storage.Status.CoreNodesStatus.CurrentRevision))), HaveField("UpdateRevision", Not(Equal(storage.Status.CoreNodesStatus.CurrentRevision))), )), - WithTransform(func(instance *appsv2beta1.EMQX) *appsv2beta1.EMQXNodesStatus { + WithTransform(func(instance *appsv2beta1.EMQX) appsv2beta1.EMQXNodesStatus { return instance.Status.ReplicantNodesStatus - }, BeNil()), + }, And( + HaveField("Replicas", Equal(int32(0))), + HaveField("ReadyReplicas", Equal(int32(0))), + HaveField("CurrentRevision", Equal("")), + HaveField("CurrentReplicas", Equal(int32(0))), + HaveField("UpdateRevision", Equal("")), + HaveField("UpdateReplicas", Equal(int32(0))), + )), ), ) @@ -327,19 +355,19 @@ var _ = Describe("E2E Test", Label("base"), Ordered, func() { }, And( HaveField("Replicas", Equal(int32(*instance.Spec.CoreTemplate.Spec.Replicas))), HaveField("ReadyReplicas", Equal(int32(*instance.Spec.CoreTemplate.Spec.Replicas))), - HaveField("CurrentRevision", Not(BeEmpty())), + HaveField("CurrentRevision", Not(Equal(""))), HaveField("CurrentReplicas", Equal(int32(*instance.Spec.CoreTemplate.Spec.Replicas))), - HaveField("UpdateRevision", Not(BeEmpty())), + HaveField("UpdateRevision", Not(Equal(""))), HaveField("UpdateReplicas", Equal(int32(*instance.Spec.CoreTemplate.Spec.Replicas))), )), - WithTransform(func(instance *appsv2beta1.EMQX) *appsv2beta1.EMQXNodesStatus { + WithTransform(func(instance *appsv2beta1.EMQX) appsv2beta1.EMQXNodesStatus { return instance.Status.ReplicantNodesStatus }, And( HaveField("Replicas", Equal(int32(*instance.Spec.ReplicantTemplate.Spec.Replicas))), HaveField("ReadyReplicas", Equal(int32(*instance.Spec.ReplicantTemplate.Spec.Replicas))), - HaveField("CurrentRevision", Not(BeEmpty())), + HaveField("CurrentRevision", Not(Equal(""))), HaveField("CurrentReplicas", Equal(int32(*instance.Spec.ReplicantTemplate.Spec.Replicas))), - HaveField("UpdateRevision", Not(BeEmpty())), + HaveField("UpdateRevision", Not(Equal(""))), HaveField("UpdateReplicas", Equal(int32(*instance.Spec.ReplicantTemplate.Spec.Replicas))), )), ), @@ -389,7 +417,64 @@ var _ = Describe("E2E Test", Label("base"), Ordered, func() { WithTransform(func(instance *appsv2beta1.EMQX) []appsv2beta1.EMQXNode { return instance.Status.ReplicantNodes }, HaveLen(int(*instance.Spec.ReplicantTemplate.Spec.Replicas))), - WithTransform(func(instance *appsv2beta1.EMQX) *appsv2beta1.EMQXNodesStatus { + WithTransform(func(instance *appsv2beta1.EMQX) appsv2beta1.EMQXNodesStatus { + return instance.Status.ReplicantNodesStatus + }, And( + HaveField("Replicas", Equal(int32(*instance.Spec.ReplicantTemplate.Spec.Replicas))), + HaveField("ReadyReplicas", Equal(int32(*instance.Spec.ReplicantTemplate.Spec.Replicas))), + HaveField("CurrentRevision", Equal(storage.Status.ReplicantNodesStatus.CurrentRevision)), + HaveField("CurrentReplicas", Equal(int32(*instance.Spec.ReplicantTemplate.Spec.Replicas))), + HaveField("UpdateRevision", Equal(storage.Status.ReplicantNodesStatus.CurrentRevision)), + HaveField("UpdateReplicas", Equal(int32(*instance.Spec.ReplicantTemplate.Spec.Replicas))), + )), + ), + ) + + checkServices(instance) + checkPods(instance) + checkEndpoints(instance, appsv2beta1.CloneAndAddLabel( + appsv2beta1.DefaultReplicantLabels(instance), + appsv2beta1.LabelsPodTemplateHashKey, + instance.Status.ReplicantNodesStatus.CurrentRevision, + )) + }) + + It("scale down EMQX replicant nodes to 0", func() { + var storage *appsv2beta1.EMQX + Expect(retry.RetryOnConflict(retry.DefaultRetry, func() error { + if err := k8sClient.Get(ctx, client.ObjectKeyFromObject(instance), instance); err != nil { + return err + } + storage = instance.DeepCopy() + instance.Spec.ReplicantTemplate.Spec.Replicas = pointer.Int32Ptr(0) + return k8sClient.Update(ctx, instance) + })).Should(Succeed()) + + Eventually(func() *appsv2beta1.EMQX { + _ = k8sClient.Get(ctx, client.ObjectKeyFromObject(instance), instance) + return instance + }).WithTimeout(timeout).WithPolling(interval).Should( + And( + WithTransform(func(instance *appsv2beta1.EMQX) bool { + return instance.Status.IsConditionTrue(appsv2beta1.Ready) + }, BeTrue()), + WithTransform(func(instance *appsv2beta1.EMQX) []appsv2beta1.EMQXNode { + return instance.Status.CoreNodes + }, HaveLen(int(*instance.Spec.CoreTemplate.Spec.Replicas))), + WithTransform(func(instance *appsv2beta1.EMQX) appsv2beta1.EMQXNodesStatus { + return instance.Status.CoreNodesStatus + }, And( + HaveField("Replicas", Equal(int32(*instance.Spec.CoreTemplate.Spec.Replicas))), + HaveField("ReadyReplicas", Equal(int32(*instance.Spec.CoreTemplate.Spec.Replicas))), + HaveField("CurrentRevision", Equal(storage.Status.CoreNodesStatus.CurrentRevision)), + HaveField("CurrentReplicas", Equal(int32(*instance.Spec.CoreTemplate.Spec.Replicas))), + HaveField("UpdateRevision", Equal(storage.Status.CoreNodesStatus.CurrentRevision)), + HaveField("UpdateReplicas", Equal(int32(*instance.Spec.CoreTemplate.Spec.Replicas))), + )), + WithTransform(func(instance *appsv2beta1.EMQX) []appsv2beta1.EMQXNode { + return instance.Status.ReplicantNodes + }, HaveLen(int(*instance.Spec.ReplicantTemplate.Spec.Replicas))), + WithTransform(func(instance *appsv2beta1.EMQX) appsv2beta1.EMQXNodesStatus { return instance.Status.ReplicantNodesStatus }, And( HaveField("Replicas", Equal(int32(*instance.Spec.ReplicantTemplate.Spec.Replicas))), @@ -436,7 +521,7 @@ var _ = Describe("E2E Test", Label("base"), Ordered, func() { HaveField("CurrentRevision", Not(Equal(storage.Status.CoreNodesStatus.CurrentRevision))), HaveField("UpdateRevision", Not(Equal(storage.Status.CoreNodesStatus.CurrentRevision))), )), - WithTransform(func(instance *appsv2beta1.EMQX) *appsv2beta1.EMQXNodesStatus { + WithTransform(func(instance *appsv2beta1.EMQX) appsv2beta1.EMQXNodesStatus { return instance.Status.ReplicantNodesStatus }, And( HaveField("CurrentRevision", Not(Equal(storage.Status.ReplicantNodesStatus.CurrentRevision))),