Skip to content

Commit aabcdf3

Browse files
qj0r9j0vc2qj0r9j0vc2
authored
Merge pull request #43 from b-harvest/feat/pod-ready-rollout-check
fix: check K8s pod ready status before counting pods for rollout
2 parents 250365f + 9ec33fa commit aabcdf3

2 files changed

Lines changed: 306 additions & 4 deletions

File tree

‎internal/fullnode/pod_control.go‎

Lines changed: 44 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,41 @@ type CacheInvalidator interface {
2828
Invalidate(controller client.ObjectKey, pods []string)
2929
}
3030

31+
// isPodReadyForRollout returns true if the pod is fully ready for rollout purposes.
32+
// This is stricter than kube.IsPodReady() which only checks the Ready condition.
33+
//
34+
// For rollout calculations, we need to ensure:
35+
// 1. Pod is in Running phase (not Pending/Failed/Succeeded/Unknown)
36+
// 2. All init containers have completed successfully (prevents counting pods where
37+
// version-check or other init containers are still running)
38+
// 3. K8s Ready condition is True (readiness probe passing)
39+
//
40+
// Note: Pods without init containers will pass check #2 (empty slice iteration).
41+
// This is correct behavior - those pods don't have init container prerequisites.
42+
func isPodReadyForRollout(pod *corev1.Pod) bool {
43+
// Check if pod is running
44+
if pod.Status.Phase != corev1.PodRunning {
45+
return false
46+
}
47+
48+
// Check if all init containers have completed successfully.
49+
// Pods without init containers pass this check (empty InitContainerStatuses).
50+
for _, cs := range pod.Status.InitContainerStatuses {
51+
if cs.State.Terminated == nil || cs.State.Terminated.ExitCode != 0 {
52+
return false
53+
}
54+
}
55+
56+
// Check K8s Ready condition
57+
for _, cond := range pod.Status.Conditions {
58+
if cond.Type == corev1.PodReady {
59+
return cond.Status == corev1.ConditionTrue
60+
}
61+
}
62+
63+
return false
64+
}
65+
3166
// PodControl reconciles pods for a CosmosFullNode.
3267
type PodControl struct {
3368
client Client
@@ -119,13 +154,20 @@ func (pc PodControl) Reconcile(
119154
continue
120155
}
121156

157+
// Check if pod is actually ready for rollout (init containers completed, readiness probe passing)
158+
podReady := isPodReadyForRollout(&existing)
159+
122160
var rpcReachable bool
123161
if ps, ok := syncInfo[podName]; ok {
124-
if ps.InSync != nil && *ps.InSync {
162+
// Only count as in-sync if both K8s says ready AND RPC reports in-sync
163+
if podReady && ps.InSync != nil && *ps.InSync {
125164
inSyncPods++
165+
} else {
166+
reporter.Debug("Pod not counted as in-sync", "pod", podName, "podReady", podReady, "inSync", ps.InSync)
126167
}
127168
rpcReachable = ps.Error == nil
128-
if rpcReachable {
169+
// Only count as RPC reachable if K8s pod is also ready
170+
if podReady && rpcReachable {
129171
rpcReachablePods++
130172
}
131173
}

‎internal/fullnode/pod_control_test.go‎

Lines changed: 262 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,32 @@ import (
1818

1919
type mockPodClient struct{ mockClient[*corev1.Pod] }
2020

21+
// setPodsReady sets pods to K8s Ready state (Running phase, init containers completed, Ready condition true)
22+
func setPodsReady(pods []*corev1.Pod) {
23+
for _, pod := range pods {
24+
pod.Status = corev1.PodStatus{
25+
Phase: corev1.PodRunning,
26+
InitContainerStatuses: []corev1.ContainerStatus{
27+
{
28+
Name: "chain-init",
29+
State: corev1.ContainerState{
30+
Terminated: &corev1.ContainerStateTerminated{ExitCode: 0},
31+
},
32+
},
33+
{
34+
Name: "version-check",
35+
State: corev1.ContainerState{
36+
Terminated: &corev1.ContainerStateTerminated{ExitCode: 0},
37+
},
38+
},
39+
},
40+
Conditions: []corev1.PodCondition{
41+
{Type: corev1.PodReady, Status: corev1.ConditionTrue},
42+
},
43+
}
44+
}
45+
}
46+
2147
func newMockPodClient(pods []*corev1.Pod) *mockPodClient {
2248
return &mockPodClient{
2349
mockClient: mockClient[*corev1.Pod]{
@@ -134,7 +160,9 @@ func TestPodControl_Reconcile(t *testing.T) {
134160
pods, err := BuildPods(&crd, nil)
135161
require.NoError(t, err)
136162

137-
mClient := newMockPodClient(diff.New(nil, pods).Creates())
163+
existing := diff.New(nil, pods).Creates()
164+
setPodsReady(existing)
165+
mClient := newMockPodClient(existing)
138166

139167
syncInfo := map[string]*cosmosv1.SyncInfoPodStatus{
140168
"hub-0": {InSync: ptr(true)},
@@ -226,6 +254,7 @@ func TestPodControl_Reconcile(t *testing.T) {
226254
pods, err := BuildPods(&crd, nil)
227255
require.NoError(t, err)
228256
existing := diff.New(nil, pods).Creates()
257+
setPodsReady(existing)
229258

230259
mClient := newMockPodClient(existing)
231260

@@ -419,6 +448,7 @@ func TestPodControl_Reconcile(t *testing.T) {
419448
pods, err := BuildPods(&crd, nil)
420449
require.NoError(t, err)
421450
existing := diff.New(nil, pods).Creates()
451+
setPodsReady(existing)
422452

423453
mClient := newMockPodClient(existing)
424454

@@ -470,11 +500,16 @@ func TestPodControl_Reconcile(t *testing.T) {
470500
})
471501
}
472502

473-
// revision hash must be taken without the revision label and the ordinal annotation.
503+
// revision hash must be taken without the revision label, the ordinal annotation,
504+
// and Status (since BuildPods creates pods with empty Status).
474505
func recalculatePodRevision(pod *corev1.Pod, ordinal int) {
475506
delete(pod.Labels, "app.kubernetes.io/revision")
476507
delete(pod.Annotations, "app.kubernetes.io/ordinal")
508+
// Temporarily clear status to match what BuildPods produces
509+
savedStatus := pod.Status
510+
pod.Status = corev1.PodStatus{}
477511
rev1 := diff.Adapt(pod, ordinal).Revision()
512+
pod.Status = savedStatus
478513
pod.Labels["app.kubernetes.io/revision"] = rev1
479514
pod.Annotations["app.kubernetes.io/ordinal"] = fmt.Sprintf("%d", ordinal)
480515
}
@@ -483,6 +518,27 @@ func newPodWithNewImage(pod *corev1.Pod) {
483518
pod.DeletionTimestamp = nil
484519
pod.Spec.Containers[0].Image = "new-image"
485520
pod.Spec.InitContainers[1].Image = "new-image"
521+
// Set pod to ready state after upgrade
522+
pod.Status = corev1.PodStatus{
523+
Phase: corev1.PodRunning,
524+
InitContainerStatuses: []corev1.ContainerStatus{
525+
{
526+
Name: "chain-init",
527+
State: corev1.ContainerState{
528+
Terminated: &corev1.ContainerStateTerminated{ExitCode: 0},
529+
},
530+
},
531+
{
532+
Name: "version-check",
533+
State: corev1.ContainerState{
534+
Terminated: &corev1.ContainerStateTerminated{ExitCode: 0},
535+
},
536+
},
537+
},
538+
Conditions: []corev1.PodCondition{
539+
{Type: corev1.PodReady, Status: corev1.ConditionTrue},
540+
},
541+
}
486542
}
487543

488544
func deletedPod(pod *corev1.Pod) {
@@ -503,3 +559,207 @@ func updatePod(t *testing.T, crdName string, ordinal int, pods []*corev1.Pod, up
503559

504560
require.FailNow(t, "pod not found", podName)
505561
}
562+
563+
func TestIsPodReadyForRollout(t *testing.T) {
564+
t.Parallel()
565+
566+
t.Run("pod is ready when all conditions met", func(t *testing.T) {
567+
pod := &corev1.Pod{
568+
Status: corev1.PodStatus{
569+
Phase: corev1.PodRunning,
570+
InitContainerStatuses: []corev1.ContainerStatus{
571+
{
572+
Name: "version-check",
573+
State: corev1.ContainerState{
574+
Terminated: &corev1.ContainerStateTerminated{ExitCode: 0},
575+
},
576+
},
577+
},
578+
Conditions: []corev1.PodCondition{
579+
{Type: corev1.PodReady, Status: corev1.ConditionTrue},
580+
},
581+
},
582+
}
583+
require.True(t, isPodReadyForRollout(pod))
584+
})
585+
586+
t.Run("pod is not ready when init container running", func(t *testing.T) {
587+
pod := &corev1.Pod{
588+
Status: corev1.PodStatus{
589+
Phase: corev1.PodRunning,
590+
InitContainerStatuses: []corev1.ContainerStatus{
591+
{
592+
Name: "version-check",
593+
State: corev1.ContainerState{
594+
Running: &corev1.ContainerStateRunning{},
595+
},
596+
},
597+
},
598+
Conditions: []corev1.PodCondition{
599+
{Type: corev1.PodReady, Status: corev1.ConditionFalse},
600+
},
601+
},
602+
}
603+
require.False(t, isPodReadyForRollout(pod))
604+
})
605+
606+
t.Run("pod is not ready when init container failed", func(t *testing.T) {
607+
pod := &corev1.Pod{
608+
Status: corev1.PodStatus{
609+
Phase: corev1.PodRunning,
610+
InitContainerStatuses: []corev1.ContainerStatus{
611+
{
612+
Name: "version-check",
613+
State: corev1.ContainerState{
614+
Terminated: &corev1.ContainerStateTerminated{ExitCode: 1},
615+
},
616+
},
617+
},
618+
Conditions: []corev1.PodCondition{
619+
{Type: corev1.PodReady, Status: corev1.ConditionFalse},
620+
},
621+
},
622+
}
623+
require.False(t, isPodReadyForRollout(pod))
624+
})
625+
626+
t.Run("pod is not ready when phase is pending", func(t *testing.T) {
627+
pod := &corev1.Pod{
628+
Status: corev1.PodStatus{
629+
Phase: corev1.PodPending,
630+
},
631+
}
632+
require.False(t, isPodReadyForRollout(pod))
633+
})
634+
635+
t.Run("pod is not ready when Ready condition is false", func(t *testing.T) {
636+
pod := &corev1.Pod{
637+
Status: corev1.PodStatus{
638+
Phase: corev1.PodRunning,
639+
InitContainerStatuses: []corev1.ContainerStatus{
640+
{
641+
Name: "version-check",
642+
State: corev1.ContainerState{
643+
Terminated: &corev1.ContainerStateTerminated{ExitCode: 0},
644+
},
645+
},
646+
},
647+
Conditions: []corev1.PodCondition{
648+
{Type: corev1.PodReady, Status: corev1.ConditionFalse},
649+
},
650+
},
651+
}
652+
require.False(t, isPodReadyForRollout(pod))
653+
})
654+
655+
t.Run("pod is not ready when no Ready condition exists", func(t *testing.T) {
656+
pod := &corev1.Pod{
657+
Status: corev1.PodStatus{
658+
Phase: corev1.PodRunning,
659+
InitContainerStatuses: []corev1.ContainerStatus{
660+
{
661+
Name: "version-check",
662+
State: corev1.ContainerState{
663+
Terminated: &corev1.ContainerStateTerminated{ExitCode: 0},
664+
},
665+
},
666+
},
667+
Conditions: []corev1.PodCondition{},
668+
},
669+
}
670+
require.False(t, isPodReadyForRollout(pod))
671+
})
672+
673+
t.Run("pod with no init containers is ready when phase and condition are correct", func(t *testing.T) {
674+
pod := &corev1.Pod{
675+
Status: corev1.PodStatus{
676+
Phase: corev1.PodRunning,
677+
InitContainerStatuses: []corev1.ContainerStatus{}, // No init containers
678+
Conditions: []corev1.PodCondition{
679+
{Type: corev1.PodReady, Status: corev1.ConditionTrue},
680+
},
681+
},
682+
}
683+
require.True(t, isPodReadyForRollout(pod))
684+
})
685+
}
686+
687+
func TestPodControl_Reconcile_InitContainerRunning(t *testing.T) {
688+
t.Parallel()
689+
690+
ctx := context.Background()
691+
const namespace = "test"
692+
693+
t.Run("does not count pod as ready when init container is running", func(t *testing.T) {
694+
crd := defaultCRD()
695+
crd.Name = "hub"
696+
crd.Namespace = namespace
697+
crd.Spec.Replicas = 3
698+
crd.Spec.RolloutStrategy = cosmosv1.RolloutStrategy{
699+
MaxUnavailable: ptr(intstr.FromInt(1)),
700+
}
701+
702+
pods, err := BuildPods(&crd, nil)
703+
require.NoError(t, err)
704+
existing := diff.New(nil, pods).Creates()
705+
706+
// Set pod-0 as not ready (init container running)
707+
existing[0].Status = corev1.PodStatus{
708+
Phase: corev1.PodPending,
709+
InitContainerStatuses: []corev1.ContainerStatus{
710+
{
711+
Name: "version-check",
712+
State: corev1.ContainerState{
713+
Running: &corev1.ContainerStateRunning{},
714+
},
715+
},
716+
},
717+
}
718+
719+
// Set pod-1 and pod-2 as ready
720+
for i := 1; i < 3; i++ {
721+
existing[i].Status = corev1.PodStatus{
722+
Phase: corev1.PodRunning,
723+
InitContainerStatuses: []corev1.ContainerStatus{
724+
{
725+
Name: "version-check",
726+
State: corev1.ContainerState{
727+
Terminated: &corev1.ContainerStateTerminated{ExitCode: 0},
728+
},
729+
},
730+
},
731+
Conditions: []corev1.PodCondition{
732+
{Type: corev1.PodReady, Status: corev1.ConditionTrue},
733+
},
734+
}
735+
}
736+
737+
mClient := newMockPodClient(existing)
738+
739+
syncInfo := map[string]*cosmosv1.SyncInfoPodStatus{
740+
"hub-0": {InSync: ptr(true)}, // RPC says in sync, but K8s pod not ready
741+
"hub-1": {InSync: ptr(true)},
742+
"hub-2": {InSync: ptr(true)},
743+
}
744+
745+
control := NewPodControl(mClient, nil)
746+
747+
control.computeRollout = func(maxUnavail *intstr.IntOrString, desired, ready int) int {
748+
require.EqualValues(t, crd.Spec.Replicas, desired)
749+
// Only 2 pods should be ready (pod-0 has init container running)
750+
require.Equal(t, 2, ready)
751+
return kube.ComputeRollout(maxUnavail, desired, ready)
752+
}
753+
754+
// Trigger updates by changing the image
755+
crd.Spec.PodTemplate.Image = "new-image"
756+
requeue, err := control.Reconcile(ctx, nopReporter, &crd, nil, syncInfo)
757+
require.NoError(t, err)
758+
require.True(t, requeue)
759+
760+
// With maxUnavailable=1 and ready=2, we can delete 0 pods
761+
// because 2 - 1 = 1 minAvail, ready(2) > minAvail(1), target = 1 - (3-2) = 0
762+
// Wait, let's recalculate: unavail=1, minAvail=3-1=2, ready=2, ready<=minAvail, so target=0
763+
require.Zero(t, mClient.DeleteCount)
764+
})
765+
}

0 commit comments

Comments
 (0)