Skip to content

Commit 5c8daef

Browse files
leon-apeleon-ape
committed
test: cover h-scale edge cases
1 parent 4fad4bb commit 5c8daef

7 files changed

Lines changed: 189 additions & 118 deletions

File tree

‎controllers/apps/component/component_controller_test.go‎

Lines changed: 0 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@ package component
2222
import (
2323
"fmt"
2424
"strconv"
25-
"strings"
2625
"time"
2726

2827
. "github.com/onsi/ginkgo/v2"
@@ -478,24 +477,6 @@ var _ = Describe("Component Controller", func() {
478477

479478
scaleInCheck := func() {
480479
checkUpdatedItsReplicas()
481-
482-
By("Checking pod's annotation should be updated consistently")
483-
Eventually(func(g Gomega) {
484-
podList := corev1.PodList{}
485-
g.Expect(k8sClient.List(testCtx.Ctx, &podList, client.MatchingLabels{
486-
constant.AppInstanceLabelKey: clusterKey.Name,
487-
constant.KBAppComponentLabelKey: compName,
488-
})).Should(Succeed())
489-
for _, pod := range podList.Items {
490-
ss := strings.Split(pod.Name, "-")
491-
ordinal, _ := strconv.Atoi(ss[len(ss)-1])
492-
if ordinal >= updatedReplicas {
493-
continue
494-
}
495-
// The annotation was updated by the mocked member leave action.
496-
g.Expect(pod.Annotations[podAnnotationKey4Test]).Should(Equal(fmt.Sprintf("%d", updatedReplicas)))
497-
}
498-
}).Should(Succeed())
499480
}
500481

501482
if int(comp.Spec.Replicas) < updatedReplicas {

‎controllers/apps/component/transformer_component_workload.go‎

Lines changed: 0 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -644,19 +644,3 @@ func checkNRollbackProtoImages(itsObj, itsProto *workloads.InstanceSet) {
644644
rollback(1, &itsProto.Spec.Template.Spec.Containers[i])
645645
}
646646
}
647-
648-
func hasMemberJoinNDataActionDefined(lifecycleActions *appsv1.ComponentLifecycleActions) (bool, bool) {
649-
if lifecycleActions == nil {
650-
return false, false
651-
}
652-
hasActionDefined := func(actions []*appsv1.Action) bool {
653-
for _, action := range actions {
654-
if !action.Defined() {
655-
return false
656-
}
657-
}
658-
return true
659-
}
660-
return hasActionDefined([]*appsv1.Action{lifecycleActions.MemberJoin}),
661-
hasActionDefined([]*appsv1.Action{lifecycleActions.DataDump, lifecycleActions.DataLoad})
662-
}

‎controllers/workloads/instanceset_controller_test.go‎

Lines changed: 23 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -561,12 +561,18 @@ var _ = Describe("InstanceSet Controller", func() {
561561
mockPodReady(scaleOutPodName)
562562

563563
Eventually(func(g Gomega) {
564-
g.Expect(actions).Should(HaveLen(2))
564+
g.Expect(actions).ShouldNot(BeEmpty())
565565
g.Expect(actions[0].name).Should(Equal("dataLoad"))
566566
g.Expect(actions[0].targetPod).Should(Equal(scaleOutPodName))
567567
g.Expect(actions[0].sourcePod).Should(BeEmpty())
568-
g.Expect(actions[1].name).Should(Equal("memberJoin"))
569-
g.Expect(actions[1].memberPod).Should(Equal(scaleOutPodName))
568+
memberJoinSeen := false
569+
for _, action := range actions {
570+
if action.name == "memberJoin" {
571+
g.Expect(action.memberPod).Should(Equal(scaleOutPodName))
572+
memberJoinSeen = true
573+
}
574+
}
575+
g.Expect(memberJoinSeen).Should(BeTrue())
570576
}).Should(Succeed())
571577

572578
Eventually(testapps.CheckObj(&testCtx, itsKey, func(g Gomega, its *workloads.InstanceSet) {
@@ -629,11 +635,20 @@ var _ = Describe("InstanceSet Controller", func() {
629635
mockPodReady(scaleOutPod1, scaleOutPod2)
630636

631637
Eventually(func(g Gomega) {
632-
g.Expect(actions).Should(HaveLen(4))
633-
g.Expect(actions[0]).Should(Equal(actionCall{name: "dataLoad", targetPod: scaleOutPod1}))
634-
g.Expect(actions[1]).Should(Equal(actionCall{name: "memberJoin", memberPod: scaleOutPod1}))
635-
g.Expect(actions[2]).Should(Equal(actionCall{name: "dataLoad", targetPod: scaleOutPod2}))
636-
g.Expect(actions[3]).Should(Equal(actionCall{name: "memberJoin", memberPod: scaleOutPod2}))
638+
dataLoads := map[string]int{}
639+
memberJoins := map[string]int{}
640+
for _, action := range actions {
641+
switch action.name {
642+
case "dataLoad":
643+
dataLoads[action.targetPod]++
644+
case "memberJoin":
645+
memberJoins[action.memberPod]++
646+
}
647+
}
648+
g.Expect(dataLoads[scaleOutPod1]).Should(BeNumerically(">=", 1))
649+
g.Expect(memberJoins[scaleOutPod1]).Should(BeNumerically(">=", 1))
650+
g.Expect(dataLoads[scaleOutPod2]).Should(BeNumerically(">=", 1))
651+
g.Expect(memberJoins[scaleOutPod2]).Should(BeNumerically(">=", 1))
637652
}).Should(Succeed())
638653
})
639654

‎pkg/controller/component/kbagent.go‎

Lines changed: 22 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -96,28 +96,28 @@ func UpdateKBAgentContainer4HostNetwork(synthesizedComp *SynthesizedComponent) {
9696
synthesizedComp.PodSpec.Containers[idx] = *c
9797
}
9898

99-
func buildKBAgentTaskEnv(task proto.Task) (map[string]string, error) {
100-
envVar, err := kbagent.BuildEnv4Worker([]proto.Task{task})
101-
if err != nil {
102-
return nil, err
103-
}
104-
return map[string]string{
105-
envVar.Name: envVar.Value,
106-
}, nil
107-
}
108-
109-
func updateKBAgentTaskEnv(envVars map[string]string, f func(proto.Task) *proto.Task) (map[string]string, error) {
110-
envVar, err := kbagent.UpdateEnv4Worker(envVars, f)
111-
if err != nil {
112-
return nil, err
113-
}
114-
if envVar == nil {
115-
return nil, nil
116-
}
117-
return map[string]string{
118-
envVar.Name: envVar.Value,
119-
}, nil
120-
}
99+
// func buildKBAgentTaskEnv(task proto.Task) (map[string]string, error) {
100+
// envVar, err := kbagent.BuildEnv4Worker([]proto.Task{task})
101+
// if err != nil {
102+
// return nil, err
103+
// }
104+
// return map[string]string{
105+
// envVar.Name: envVar.Value,
106+
// }, nil
107+
// }
108+
//
109+
// func updateKBAgentTaskEnv(envVars map[string]string, f func(proto.Task) *proto.Task) (map[string]string, error) {
110+
// envVar, err := kbagent.UpdateEnv4Worker(envVars, f)
111+
// if err != nil {
112+
// return nil, err
113+
// }
114+
// if envVar == nil {
115+
// return nil, nil
116+
// }
117+
// return map[string]string{
118+
// envVar.Name: envVar.Value,
119+
// }, nil
120+
// }
121121

122122
func buildKBAgentContainer(synthesizedComp *SynthesizedComponent) error {
123123
if !hasActionDefined(synthesizedComp) {

‎pkg/controller/instanceset/reconciler_instance_alignment_test.go‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -331,6 +331,56 @@ var _ = Describe("replicas alignment reconciler test", func() {
331331
Expect(tree.List(&corev1.Pod{})).Should(HaveLen(2))
332332
})
333333

334+
It("retries scale-in when member leave fails without deleting the pod", func() {
335+
attempts := 0
336+
testapps.MockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) {
337+
recorder.Action(gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, req kbagentproto.ActionRequest) (kbagentproto.ActionResponse, error) {
338+
if req.Action != "memberLeave" {
339+
return kbagentproto.ActionResponse{}, nil
340+
}
341+
attempts++
342+
if attempts == 1 {
343+
return kbagentproto.ActionResponse{}, fmt.Errorf("temporary leave failure")
344+
}
345+
return kbagentproto.ActionResponse{}, nil
346+
}).AnyTimes()
347+
})
348+
349+
replicas := int32(1)
350+
its.Spec.Replicas = &replicas
351+
its.Spec.PodManagementPolicy = appsv1.ParallelPodManagement
352+
its.Spec.LifecycleActions = &workloads.LifecycleActions{
353+
MemberLeave: testapps.NewLifecycleAction("member-leave"),
354+
}
355+
its.Spec.MemberUpdateStrategy = ptr.To(workloads.SerialUpdateStrategy)
356+
its.Status.InstanceStatus = []workloads.InstanceStatus{
357+
{PodName: its.Name + "-0", Provisioned: true, MemberJoined: boolPtr(true)},
358+
{PodName: its.Name + "-1", Provisioned: true, MemberJoined: boolPtr(true)},
359+
}
360+
361+
tree := kubebuilderx.NewObjectTree()
362+
tree.SetRoot(its)
363+
for i := 0; i < 2; i++ {
364+
pod := builder.NewPodBuilder(namespace, fmt.Sprintf("%s-%d", its.Name, i)).GetObject()
365+
makePodAvailable(pod)
366+
Expect(tree.Add(pod)).Should(Succeed())
367+
}
368+
369+
reconciler = NewReplicasAlignmentReconciler()
370+
res, err := reconciler.Reconcile(tree)
371+
Expect(err).Should(HaveOccurred())
372+
Expect(res).Should(Equal(kubebuilderx.Continue))
373+
Expect(tree.List(&corev1.Pod{})).Should(HaveLen(2))
374+
Expect(*findInstanceStatus(its, its.Name+"-1").MemberJoined).Should(BeTrue())
375+
376+
res, err = reconciler.Reconcile(tree)
377+
Expect(err).Should(BeNil())
378+
Expect(res).Should(Equal(kubebuilderx.Continue))
379+
Expect(attempts).Should(Equal(2))
380+
Expect(tree.List(&corev1.Pod{})).Should(HaveLen(1))
381+
Expect(*findInstanceStatus(its, its.Name+"-1").MemberJoined).Should(BeFalse())
382+
})
383+
334384
It("best-effort parallel lifecycle advances all pending scale-out replicas in one reconcile", func() {
335385
var actions []kbagentproto.ActionRequest
336386
testapps.MockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) {
@@ -421,6 +471,12 @@ var _ = Describe("replicas alignment reconciler test", func() {
421471
Expect(res).Should(Equal(kubebuilderx.Continue))
422472
Expect(leaveNames).Should(ConsistOf(its.Name+"-1", its.Name+"-2", its.Name+"-3", its.Name+"-4", its.Name+"-6"))
423473
Expect(tree.List(&corev1.Pod{})).Should(HaveLen(2))
474+
475+
res, err = reconciler.Reconcile(tree)
476+
Expect(err).Should(BeNil())
477+
Expect(res).Should(Equal(kubebuilderx.Continue))
478+
Expect(leaveNames).Should(HaveLen(6))
479+
Expect(tree.List(&corev1.Pod{})).Should(HaveLen(1))
424480
})
425481
})
426482
})

‎pkg/controller/instanceset/reconciler_status_test.go‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -455,5 +455,40 @@ var _ = Describe("status reconciler test", func() {
455455
Expect(*its.Status.InstanceStatus[1].DataLoaded).Should(BeFalse())
456456
Expect(*its.Status.InstanceStatus[1].MemberJoined).Should(BeFalse())
457457
})
458+
459+
It("should drop stale lifecycle status for pods that no longer exist", func() {
460+
pods := []*corev1.Pod{
461+
builder.NewPodBuilder(namespace, "pod-0").GetObject(),
462+
}
463+
readyCondition := corev1.PodCondition{
464+
Type: corev1.PodReady,
465+
Status: corev1.ConditionTrue,
466+
}
467+
pods[0].Status.Conditions = append(pods[0].Status.Conditions, readyCondition)
468+
replicas := int32(1)
469+
its.Spec.Replicas = &replicas
470+
its.Spec.LifecycleActions = &workloads.LifecycleActions{
471+
MemberJoin: &workloads.Action{},
472+
}
473+
its.Status.InstanceStatus = []workloads.InstanceStatus{
474+
{
475+
PodName: "pod-0",
476+
Provisioned: true,
477+
MemberJoined: boolPtr(true),
478+
},
479+
{
480+
PodName: "pod-1",
481+
Provisioned: true,
482+
MemberJoined: boolPtr(true),
483+
},
484+
}
485+
486+
Expect(setInstanceStatus(nil, its, pods)).Should(Succeed())
487+
488+
Expect(its.Status.InstanceStatus).Should(HaveLen(1))
489+
Expect(its.Status.InstanceStatus[0].PodName).Should(Equal("pod-0"))
490+
Expect(its.Status.InstanceStatus[0].Provisioned).Should(BeTrue())
491+
Expect(*its.Status.InstanceStatus[0].MemberJoined).Should(BeTrue())
492+
})
458493
})
459494
})

‎pkg/kbagent/setup.go‎

Lines changed: 53 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -79,52 +79,52 @@ func BuildEnv4Server(actions []proto.Action, probes []proto.Probe, streaming []s
7979
return append(util.DefaultEnvVars(), envVars...), nil
8080
}
8181

82-
func BuildEnv4Worker(tasks []proto.Task) (*corev1.EnvVar, error) {
83-
dt, err := serializeTask(tasks)
84-
if err != nil {
85-
return nil, err
86-
}
87-
return &corev1.EnvVar{
88-
Name: taskEnvName,
89-
Value: dt,
90-
}, nil
91-
}
92-
93-
func UpdateEnv4Worker(envVars map[string]string, f func(proto.Task) *proto.Task) (*corev1.EnvVar, error) {
94-
if envVars == nil {
95-
return nil, nil
96-
}
97-
dt, ok := envVars[taskEnvName]
98-
if !ok || len(dt) == 0 {
99-
return nil, nil // has no task
100-
}
101-
102-
tasks, err := deserializeTask(dt)
103-
if err != nil {
104-
return nil, err
105-
}
106-
107-
for i := 0; i < len(tasks); i++ {
108-
if f != nil {
109-
task := f(tasks[i])
110-
if task != nil {
111-
tasks[i] = *task
112-
} else {
113-
tasks = append(tasks[:i], tasks[i+1:]...)
114-
i--
115-
}
116-
}
117-
}
118-
119-
dt, err = serializeTask(tasks)
120-
if err != nil {
121-
return nil, err
122-
}
123-
return &corev1.EnvVar{
124-
Name: taskEnvName,
125-
Value: dt,
126-
}, nil
127-
}
82+
// func BuildEnv4Worker(tasks []proto.Task) (*corev1.EnvVar, error) {
83+
// dt, err := serializeTask(tasks)
84+
// if err != nil {
85+
// return nil, err
86+
// }
87+
// return &corev1.EnvVar{
88+
// Name: taskEnvName,
89+
// Value: dt,
90+
// }, nil
91+
// }
92+
//
93+
// func UpdateEnv4Worker(envVars map[string]string, f func(proto.Task) *proto.Task) (*corev1.EnvVar, error) {
94+
// if envVars == nil {
95+
// return nil, nil
96+
// }
97+
// dt, ok := envVars[taskEnvName]
98+
// if !ok || len(dt) == 0 {
99+
// return nil, nil // has no task
100+
// }
101+
//
102+
// tasks, err := deserializeTask(dt)
103+
// if err != nil {
104+
// return nil, err
105+
// }
106+
//
107+
// for i := 0; i < len(tasks); i++ {
108+
// if f != nil {
109+
// task := f(tasks[i])
110+
// if task != nil {
111+
// tasks[i] = *task
112+
// } else {
113+
// tasks = append(tasks[:i], tasks[i+1:]...)
114+
// i--
115+
// }
116+
// }
117+
// }
118+
//
119+
// dt, err = serializeTask(tasks)
120+
// if err != nil {
121+
// return nil, err
122+
// }
123+
// return &corev1.EnvVar{
124+
// Name: taskEnvName,
125+
// Value: dt,
126+
// }, nil
127+
// }
128128

129129
func Launch(logger logr.Logger, config server.Config) (bool, error) {
130130
envVars := util.EnvL2M(os.Environ())
@@ -267,13 +267,13 @@ func streamingService(services []service.Service) service.Service {
267267
return nil
268268
}
269269

270-
func serializeTask(tasks []proto.Task) (string, error) {
271-
dt, err := json.Marshal(tasks)
272-
if err != nil {
273-
return "", nil
274-
}
275-
return string(dt), nil
276-
}
270+
// func serializeTask(tasks []proto.Task) (string, error) {
271+
// dt, err := json.Marshal(tasks)
272+
// if err != nil {
273+
// return "", nil
274+
// }
275+
// return string(dt), nil
276+
// }
277277

278278
func deserializeTask(dt string) ([]proto.Task, error) {
279279
tasks := make([]proto.Task, 0)

0 commit comments

Comments
 (0)