Skip to content

Commit 51236eb

Browse files
weicaoweicao
authored
chore: execute lifecycle actions on all selected pods (#10629)
1 parent f808f13 commit 51236eb

3 files changed

Lines changed: 204 additions & 10 deletions

File tree

‎pkg/controller/lifecycle/errors.go‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,3 +40,35 @@ func IgnoreNotDefined(err error) error {
4040
}
4141
return err
4242
}
43+
44+
type actionAggregateError struct {
45+
errs []error
46+
}
47+
48+
func newActionAggregateError(errs []error) error {
49+
if len(errs) == 0 {
50+
return nil
51+
}
52+
return &actionAggregateError{errs: append([]error(nil), errs...)}
53+
}
54+
55+
func (e *actionAggregateError) Error() string {
56+
return errors.Join(e.errs...).Error()
57+
}
58+
59+
func (e *actionAggregateError) Is(target error) bool {
60+
if target == ErrActionNotDefined {
61+
for _, err := range e.errs {
62+
if !errors.Is(err, target) {
63+
return false
64+
}
65+
}
66+
return true
67+
}
68+
for _, err := range e.errs {
69+
if errors.Is(err, target) {
70+
return true
71+
}
72+
}
73+
return false
74+
}

‎pkg/controller/lifecycle/kbagent.go‎

Lines changed: 35 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -364,11 +364,14 @@ func (a *kbagent) callActionWithSelector(ctx context.Context, spec *appsv1.Actio
364364
if len(pods) == 0 {
365365
return nil, fmt.Errorf("no available pod to execute action %s", lfa.name())
366366
}
367+
selector, _ := resolveTargetPodSelector(spec)
368+
aggregateErrors := selector == appsv1.AllReplicas
367369

368370
// TODO: impl
369371
// - back-off to retry
370372
// - timeout
371373
var output []byte
374+
var actionErrors []error
372375
for _, pod := range pods {
373376
endpoint := func() (string, int32, error) {
374377
host, port, err := a.serverEndpoint(pod)
@@ -387,7 +390,11 @@ func (a *kbagent) callActionWithSelector(ctx context.Context, spec *appsv1.Actio
387390
cli, err = kbacli.NewClient(endpoint)
388391
}
389392
if err != nil {
390-
return nil, err // mock client error
393+
if !aggregateErrors {
394+
return nil, err // mock client error
395+
}
396+
actionErrors = append(actionErrors, errors.Wrapf(err, "error creating client to execute action %s at pod %s", lfa.name(), pod.Name))
397+
continue
391398
}
392399
if cli == nil {
393400
continue // not kb-agent container and port defined, for test only
@@ -397,16 +404,29 @@ func (a *kbagent) callActionWithSelector(ctx context.Context, spec *appsv1.Actio
397404
_ = cli.Close()
398405

399406
if err != nil {
400-
return nil, errors.Wrapf(err, "http error occurred when executing action %s at pod %s", lfa.name(), pod.Name)
407+
actionErr := errors.Wrapf(err, "http error occurred when executing action %s at pod %s", lfa.name(), pod.Name)
408+
if !aggregateErrors {
409+
return nil, actionErr
410+
}
411+
actionErrors = append(actionErrors, actionErr)
412+
continue
401413
}
402414
if len(rsp.Error) > 0 {
403-
return nil, a.formatError(lfa, rsp, pod.Name)
415+
actionErr := a.formatError(lfa, rsp, pod.Name)
416+
if !aggregateErrors {
417+
return nil, actionErr
418+
}
419+
actionErrors = append(actionErrors, actionErr)
420+
continue
404421
}
405422
// take first non-nil output
406423
if output == nil && rsp.Output != nil {
407424
output = rsp.Output
408425
}
409426
}
427+
if len(actionErrors) > 0 {
428+
return nil, newActionAggregateError(actionErrors)
429+
}
410430
return output, nil
411431
}
412432

@@ -459,13 +479,7 @@ func (a *kbagent) formatError(lfa lifecycleAction, rsp proto.ActionResponse, pod
459479
}
460480

461481
func SelectTargetPods(pods []*corev1.Pod, pod *corev1.Pod, spec *appsv1.Action) ([]*corev1.Pod, error) {
462-
selector := spec.TargetPodSelector
463-
matchingKey := spec.MatchingKey
464-
if len(selector) == 0 && spec.Exec != nil && len(spec.Exec.TargetPodSelector) > 0 {
465-
// back-off to use spec.Exec
466-
selector = spec.Exec.TargetPodSelector
467-
matchingKey = spec.Exec.MatchingKey
468-
}
482+
selector, matchingKey := resolveTargetPodSelector(spec)
469483
if len(selector) == 0 {
470484
return []*corev1.Pod{pod}, nil
471485
}
@@ -536,3 +550,14 @@ func SelectTargetPods(pods []*corev1.Pod, pod *corev1.Pod, spec *appsv1.Action)
536550
return nil, fmt.Errorf("unknown pod selector: %s", selector)
537551
}
538552
}
553+
554+
func resolveTargetPodSelector(spec *appsv1.Action) (appsv1.TargetPodSelector, string) {
555+
selector := spec.TargetPodSelector
556+
matchingKey := spec.MatchingKey
557+
if len(selector) == 0 && spec.Exec != nil && len(spec.Exec.TargetPodSelector) > 0 {
558+
// back-off to use spec.Exec
559+
selector = spec.Exec.TargetPodSelector
560+
matchingKey = spec.Exec.MatchingKey
561+
}
562+
return selector, matchingKey
563+
}

‎pkg/controller/lifecycle/lifecycle_test.go‎

Lines changed: 137 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -679,6 +679,143 @@ var _ = Describe("lifecycle", func() {
679679
Expect(err).Should(BeNil())
680680
})
681681

682+
It("pod selector - all lets a later replica unblock an earlier primary", func() {
683+
lifecycleActions.PostProvision.Exec.TargetPodSelector = appsv1.AllReplicas
684+
pods = []*corev1.Pod{
685+
{ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: "primary-0"}},
686+
{ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: "replica-0"}},
687+
}
688+
689+
lifecycle, err := New(namespace, clusterName, compName, lifecycleActions, nil, nil, pods)
690+
Expect(err).Should(BeNil())
691+
Expect(lifecycle).ShouldNot(BeNil())
692+
693+
totalCallCount := 0
694+
roundCallCount := 0
695+
replicaAttached := false
696+
mockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) {
697+
recorder.Action(gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, req proto.ActionRequest) (proto.ActionResponse, error) {
698+
totalCallCount++
699+
roundCallCount++
700+
// The stable selection order is primary first, replica second. The
701+
// primary cannot finish until the replica has executed its own action.
702+
if roundCallCount == 1 && !replicaAttached {
703+
return proto.ActionResponse{Error: proto.Error2Type(proto.ErrFailed)}, nil
704+
}
705+
if roundCallCount == 2 {
706+
replicaAttached = true
707+
}
708+
return proto.ActionResponse{}, nil
709+
}).AnyTimes()
710+
})
711+
712+
err = lifecycle.PostProvision(ctx, k8sClient, nil)
713+
Expect(errors.Is(err, ErrActionFailed)).Should(BeTrue())
714+
Expect(roundCallCount).Should(Equal(2))
715+
Expect(replicaAttached).Should(BeTrue())
716+
717+
roundCallCount = 0
718+
err = lifecycle.PostProvision(ctx, k8sClient, nil)
719+
Expect(err).Should(BeNil())
720+
Expect(roundCallCount).Should(Equal(2))
721+
Expect(totalCallCount).Should(Equal(4))
722+
})
723+
724+
It("pod selector - all aggregates every pod error with pod identity", func() {
725+
lifecycleActions.PostProvision.Exec.TargetPodSelector = appsv1.AllReplicas
726+
pods = []*corev1.Pod{
727+
{ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: "pod-0"}},
728+
{ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: "pod-1"}},
729+
}
730+
731+
lifecycle, err := New(namespace, clusterName, compName, lifecycleActions, nil, nil, pods)
732+
Expect(err).Should(BeNil())
733+
Expect(lifecycle).ShouldNot(BeNil())
734+
735+
callCount := 0
736+
mockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) {
737+
recorder.Action(gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, req proto.ActionRequest) (proto.ActionResponse, error) {
738+
callCount++
739+
if callCount == 1 {
740+
return proto.ActionResponse{Error: proto.Error2Type(proto.ErrInProgress)}, nil
741+
}
742+
return proto.ActionResponse{Error: proto.Error2Type(proto.ErrBusy)}, nil
743+
}).Times(2)
744+
})
745+
746+
err = lifecycle.PostProvision(ctx, k8sClient, nil)
747+
Expect(err).ShouldNot(BeNil())
748+
Expect(errors.Is(err, ErrActionInProgress)).Should(BeTrue())
749+
Expect(errors.Is(err, ErrActionBusy)).Should(BeTrue())
750+
Expect(err.Error()).Should(ContainSubstring("pod-0"))
751+
Expect(err.Error()).Should(ContainSubstring("pod-1"))
752+
Expect(callCount).Should(Equal(2))
753+
})
754+
755+
It("pod selector - all ignores NotDefined only when every pod is NotDefined", func() {
756+
lifecycleActions.PostProvision.Exec.TargetPodSelector = appsv1.AllReplicas
757+
pods = []*corev1.Pod{
758+
{ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: "pod-0"}},
759+
{ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: "pod-1"}},
760+
}
761+
762+
lifecycle, err := New(namespace, clusterName, compName, lifecycleActions, nil, nil, pods)
763+
Expect(err).Should(BeNil())
764+
Expect(lifecycle).ShouldNot(BeNil())
765+
766+
callCount := 0
767+
mockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) {
768+
recorder.Action(gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, req proto.ActionRequest) (proto.ActionResponse, error) {
769+
callCount++
770+
if callCount == 1 {
771+
return proto.ActionResponse{Error: proto.Error2Type(proto.ErrNotDefined)}, nil
772+
}
773+
return proto.ActionResponse{Error: proto.Error2Type(proto.ErrFailed)}, nil
774+
}).Times(2)
775+
})
776+
777+
err = lifecycle.PostProvision(ctx, k8sClient, nil)
778+
Expect(err).ShouldNot(BeNil())
779+
Expect(errors.Is(err, ErrActionNotDefined)).Should(BeFalse())
780+
Expect(errors.Is(err, ErrActionFailed)).Should(BeTrue())
781+
Expect(IgnoreNotDefined(err)).ShouldNot(BeNil())
782+
Expect(err.Error()).Should(ContainSubstring("pod-0"))
783+
Expect(err.Error()).Should(ContainSubstring("pod-1"))
784+
785+
mockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) {
786+
recorder.Action(gomock.Any(), gomock.Any()).Return(proto.ActionResponse{Error: proto.Error2Type(proto.ErrNotDefined)}, nil).Times(2)
787+
})
788+
789+
err = lifecycle.PostProvision(ctx, k8sClient, nil)
790+
Expect(errors.Is(err, ErrActionNotDefined)).Should(BeTrue())
791+
Expect(IgnoreNotDefined(err)).Should(BeNil())
792+
})
793+
794+
It("pod selector - role keeps first-error behavior", func() {
795+
lifecycleActions.PostProvision.Exec.TargetPodSelector = appsv1.RoleSelector
796+
lifecycleActions.PostProvision.Exec.MatchingKey = "leader"
797+
pods = []*corev1.Pod{
798+
{ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: "pod-0", Labels: map[string]string{constant.RoleLabelKey: "leader"}}},
799+
{ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: "pod-1", Labels: map[string]string{constant.RoleLabelKey: "leader"}}},
800+
}
801+
802+
lifecycle, err := New(namespace, clusterName, compName, lifecycleActions, nil, nil, pods)
803+
Expect(err).Should(BeNil())
804+
Expect(lifecycle).ShouldNot(BeNil())
805+
806+
callCount := 0
807+
mockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) {
808+
recorder.Action(gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, req proto.ActionRequest) (proto.ActionResponse, error) {
809+
callCount++
810+
return proto.ActionResponse{Error: proto.Error2Type(proto.ErrFailed)}, nil
811+
}).Times(1)
812+
})
813+
814+
err = lifecycle.PostProvision(ctx, k8sClient, nil)
815+
Expect(errors.Is(err, ErrActionFailed)).Should(BeTrue())
816+
Expect(callCount).Should(Equal(1))
817+
})
818+
682819
It("pod selector - role", func() {
683820
lifecycleActions.PostProvision.Exec.TargetPodSelector = appsv1.RoleSelector
684821
lifecycleActions.PostProvision.Exec.MatchingKey = "leader"

0 commit comments

Comments
 (0)