fix(kubernetes): fail sandbox after main container restart

kubernetes/internal/controller/batchsandbox_pause_resume_test.go

@@ -16,6 +16,7 @@ package controller
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"sync"
 	"testing"
@@ -1807,6 +1808,200 @@ func TestBuildRuntimeView_AggregatesPodFailuresInSteadyState(t *testing.T) {
 	assert.Equal(t, "3/4 observed pods failed; primary reason=ErrImagePull; sample pod=err-image-0", podFailed.Message)
 }
 
+func restartTestSandbox(readyAt metav1.Time, endpoint string) *sandboxv1alpha1.BatchSandbox {
+	return &sandboxv1alpha1.BatchSandbox{
+		ObjectMeta: metav1.ObjectMeta{Annotations: map[string]string{AnnotationSandboxEndpoints: fmt.Sprintf(`[%q]`, endpoint)}},
+		Status: sandboxv1alpha1.BatchSandboxStatus{
+			Phase:    sandboxv1alpha1.BatchSandboxPhaseSucceed,
+			Replicas: 1,
+			Conditions: []sandboxv1alpha1.BatchSandboxCondition{{
+				Type:               sandboxv1alpha1.BatchSandboxConditionReady,
+				Status:             sandboxv1alpha1.ConditionTrue,
+				Reason:             "PodsReady",
+				Message:            "Sandbox is running",
+				LastTransitionTime: &readyAt,
+			}},
+		},
+	}
+}
+
+func restartTestPod(name, endpoint string, createdAt, restartedAt metav1.Time) *corev1.Pod {
+	return &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Name: name, CreationTimestamp: createdAt},
+		Spec:       corev1.PodSpec{Containers: []corev1.Container{{Name: "sandbox"}}},
+		Status: corev1.PodStatus{
+			Phase:      corev1.PodRunning,
+			PodIP:      endpoint,
+			Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue}},
+			ContainerStatuses: []corev1.ContainerStatus{{
+				Name:         "sandbox",
+				RestartCount: 1,
+				LastTerminationState: corev1.ContainerState{Terminated: &corev1.ContainerStateTerminated{
+					Reason:     "OOMKilled",
+					FinishedAt: restartedAt,
+				}},
+			}},
+		},
+	}
+}
+
+func assertRestartHistoryIgnored(t *testing.T, view runtimeView, readyAt metav1.Time, baselineReset bool) {
+	t.Helper()
+	assert.Equal(t, sandboxv1alpha1.BatchSandboxPhaseSucceed, view.status.Phase)
+	for _, condition := range view.status.Conditions {
+		assert.NotEqual(t, sandboxv1alpha1.BatchSandboxConditionPodFailed, condition.Type)
+		if baselineReset && condition.Type == sandboxv1alpha1.BatchSandboxConditionReady {
+			require.NotNil(t, condition.LastTransitionTime)
+			assert.True(t, condition.LastTransitionTime.After(readyAt.Time))
+		}
+	}
+}
+
+func TestBuildRuntimeView_FailsWhenContainerRestartsAfterReady(t *testing.T) {
+	readyAt := metav1.NewTime(time.Now().Add(-time.Minute))
+	restartedAt := metav1.NewTime(readyAt.Add(30 * time.Second))
+	view := buildRuntimeView(restartTestSandbox(readyAt, "10.0.0.10"), []*corev1.Pod{
+		restartTestPod("oom-restarted", "10.0.0.10", metav1.Time{}, restartedAt),
+	})
+
+	assert.Equal(t, sandboxv1alpha1.BatchSandboxPhaseFailed, view.status.Phase)
+	for _, condition := range view.status.Conditions {
+		if condition.Type == sandboxv1alpha1.BatchSandboxConditionPodFailed {
+			assert.Equal(t, "OOMKilled", condition.Reason)
+			return
+		}
+	}
+	t.Fatal("expected PodFailed condition")
+}
+
+func TestBuildRuntimeView_DetectsRestartAcrossSpecOnlyUpdate(t *testing.T) {
+	readyAt := metav1.NewTime(time.Now().Add(-time.Minute))
+	restartedAt := metav1.NewTime(readyAt.Add(30 * time.Second))
+	bs := restartTestSandbox(readyAt, "10.0.0.10")
+	bs.Generation = 2
+	bs.Status.ObservedGeneration = 1
+
+	view := buildRuntimeView(bs, []*corev1.Pod{
+		restartTestPod("oom-restarted", "10.0.0.10", metav1.Time{}, restartedAt),
+	})
+
+	assert.Equal(t, sandboxv1alpha1.BatchSandboxPhaseFailed, view.status.Phase)
+}
+
+func TestBuildRuntimeView_FailsWhenMainContainerTerminatesAfterReady(t *testing.T) {
+	readyAt := metav1.NewTime(time.Now().Add(-time.Minute))
+	terminatedAt := metav1.NewTime(readyAt.Add(30 * time.Second))
+	pod := restartTestPod("oom-terminated", "10.0.0.10", metav1.Time{}, terminatedAt)
+	pod.Status.ContainerStatuses[0].RestartCount = 0
+	pod.Status.ContainerStatuses[0].State.Terminated = pod.Status.ContainerStatuses[0].LastTerminationState.Terminated
+	pod.Status.ContainerStatuses[0].LastTerminationState = corev1.ContainerState{}
+
+	view := buildRuntimeView(restartTestSandbox(readyAt, "10.0.0.10"), []*corev1.Pod{pod})
+
+	assert.Equal(t, sandboxv1alpha1.BatchSandboxPhaseFailed, view.status.Phase)
+}
+
+func TestGetPodFailureReasonAndMessage_IgnoresSidecarRestart(t *testing.T) {
+	readyAt := metav1.NewTime(time.Now().Add(-time.Minute))
+	pod := restartTestPod("sidecar-restarted", "10.0.0.10", metav1.Time{}, metav1.NewTime(readyAt.Add(30*time.Second)))
+	pod.Spec.Containers = append(pod.Spec.Containers, corev1.Container{Name: "egress"})
+	pod.Status.ContainerStatuses[0].Name = "egress"
+
+	reason, message, failed := getPodFailureReasonAndMessage(pod, &readyAt)
+	assert.False(t, failed)
+	assert.Empty(t, reason)
+	assert.Empty(t, message)
+}
+
+func TestBuildRuntimeView_IgnoresContainerRestartBeforeReady(t *testing.T) {
+	restartedAt := metav1.NewTime(time.Now().Add(-2 * time.Minute))
+	readyAt := metav1.NewTime(restartedAt.Add(time.Minute))
+	view := buildRuntimeView(restartTestSandbox(readyAt, "10.0.0.10"), []*corev1.Pod{
+		restartTestPod("prewarmed", "10.0.0.10", metav1.Time{}, restartedAt),
+	})
+	assertRestartHistoryIgnored(t, view, readyAt, false)
+}
+
+func TestBuildRuntimeView_IgnoresRestartHistoryWhenPodMembershipChanges(t *testing.T) {
+	readyAt := metav1.NewTime(time.Now().Add(-2 * time.Minute))
+	restartedAt := metav1.NewTime(readyAt.Add(time.Minute))
+	view := buildRuntimeView(restartTestSandbox(readyAt, "10.0.0.9"), []*corev1.Pod{
+		restartTestPod("new-prewarmed-pod", "10.0.0.10", metav1.Time{}, restartedAt),
+	})
+	assertRestartHistoryIgnored(t, view, readyAt, true)
+}
+
+func TestBuildRuntimeView_IgnoresRestartHistoryFromNewPodReusingEndpoint(t *testing.T) {
+	readyAt := metav1.NewTime(time.Now().Add(-2 * time.Minute))
+	createdAt := metav1.NewTime(readyAt.Add(30 * time.Second))
+	restartedAt := metav1.NewTime(createdAt.Add(30 * time.Second))
+	view := buildRuntimeView(restartTestSandbox(readyAt, "10.0.0.10"), []*corev1.Pod{
+		restartTestPod("replacement-pod", "10.0.0.10", createdAt, restartedAt),
+	})
+	assertRestartHistoryIgnored(t, view, readyAt, true)
+}
+
+func TestBuildRuntimeView_DetectsExistingPodRestartDuringScaleUp(t *testing.T) {
+	readyAt := metav1.NewTime(time.Now().Add(-2 * time.Minute))
+	restartedAt := metav1.NewTime(readyAt.Add(time.Minute))
+	createdBeforeReady := metav1.NewTime(readyAt.Add(-time.Minute))
+	bs := restartTestSandbox(readyAt, "10.0.0.10")
+
+	view := buildRuntimeView(bs, []*corev1.Pod{
+		restartTestPod("existing-restarted", "10.0.0.10", createdBeforeReady, restartedAt),
+		restartTestPod("new-prewarmed", "10.0.0.11", createdBeforeReady, restartedAt),
+	})
+
+	assert.Equal(t, sandboxv1alpha1.BatchSandboxPhaseFailed, view.status.Phase)
+	for _, condition := range view.status.Conditions {
+		if condition.Type == sandboxv1alpha1.BatchSandboxConditionPodFailed {
+			assert.Equal(t, "1/2 observed pods failed; primary reason=OOMKilled; sample pod=existing-restarted", condition.Message)
+			return
+		}
+	}
+	t.Fatal("expected PodFailed condition")
+}
+
+func TestBuildRuntimeView_DetectsDelayedRestartAfterPodOrderChanges(t *testing.T) {
+	readyAt := metav1.NewTime(time.Now().Add(-2 * time.Minute))
+	createdBeforeReady := metav1.NewTime(readyAt.Add(-time.Minute))
+	preReadyRestart := metav1.NewTime(readyAt.Add(-30 * time.Second))
+	bs := restartTestSandbox(readyAt, "10.0.0.1")
+	bs.Annotations[AnnotationSandboxEndpoints] = `["10.0.0.1","10.0.0.2"]`
+	bs.Status.Replicas = 2
+
+	// A Pod list reorder must not advance the Ready baseline while the
+	// endpoint membership is unchanged and restart status is still stale.
+	firstView := buildRuntimeView(bs, []*corev1.Pod{
+		restartTestPod("pod-2", "10.0.0.2", createdBeforeReady, preReadyRestart),
+		restartTestPod("pod-1", "10.0.0.1", createdBeforeReady, preReadyRestart),
+	})
+	require.Equal(t, sandboxv1alpha1.BatchSandboxPhaseSucceed, firstView.status.Phase)
+	var preservedReadyAt *metav1.Time
+	for i := range firstView.status.Conditions {
+		condition := &firstView.status.Conditions[i]
+		if condition.Type == sandboxv1alpha1.BatchSandboxConditionReady {
+			preservedReadyAt = condition.LastTransitionTime
+			break
+		}
+	}
+	require.NotNil(t, preservedReadyAt)
+	assert.Equal(t, readyAt.Time, preservedReadyAt.Time)
+
+	// A later informer update exposes a restart that happened after the
+	// original Ready transition. It must still fail the sandbox.
+	bs.Status = *firstView.status
+	reorderedEndpoints, err := json.Marshal(firstView.endpointIPs)
+	require.NoError(t, err)
+	bs.Annotations[AnnotationSandboxEndpoints] = string(reorderedEndpoints)
+	delayedRestart := metav1.NewTime(readyAt.Add(time.Minute))
+	secondView := buildRuntimeView(bs, []*corev1.Pod{
+		restartTestPod("pod-2", "10.0.0.2", createdBeforeReady, preReadyRestart),
+		restartTestPod("pod-1", "10.0.0.1", createdBeforeReady, delayedRestart),
+	})
+	assert.Equal(t, sandboxv1alpha1.BatchSandboxPhaseFailed, secondView.status.Phase)
+}
+
 func TestBuildRuntimeView_AggregatesResumeFailures(t *testing.T) {
 	bs := &sandboxv1alpha1.BatchSandbox{
 		ObjectMeta: metav1.ObjectMeta{
@@ -1882,9 +2077,14 @@ func TestBuildRuntimeView_PreservesConditionTransitionTimeWhenUnchanged(t *testi
 			Name:       "test-bs",
 			Namespace:  "default",
 			Generation: 3,
+			Annotations: map[string]string{
+				AnnotationSandboxEndpoints: `["10.0.0.10"]`,
+			},
 		},
 		Status: sandboxv1alpha1.BatchSandboxStatus{
-			Phase: sandboxv1alpha1.BatchSandboxPhaseSucceed,
+			ObservedGeneration: 3,
+			Phase:              sandboxv1alpha1.BatchSandboxPhaseSucceed,
+			Replicas:           1,
 			Conditions: []sandboxv1alpha1.BatchSandboxCondition{
 				{
 					Type:               sandboxv1alpha1.BatchSandboxConditionReady,

kubernetes/internal/controller/batchsandbox_status.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"maps"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
@@ -107,7 +108,7 @@ func applyBatchSandboxPhaseConditions(status *sandboxv1alpha1.BatchSandboxStatus
 	}
 }
 
-func getPodFailureReasonAndMessage(pod *corev1.Pod) (string, string, bool) {
+func getPodFailureReasonAndMessage(pod *corev1.Pod, readySince *metav1.Time) (string, string, bool) {
 	for _, cs := range pod.Status.ContainerStatuses {
 		if cs.State.Waiting == nil {
 			continue
@@ -117,24 +118,120 @@ func getPodFailureReasonAndMessage(pod *corev1.Pod) (string, string, bool) {
 			return cs.State.Waiting.Reason, fmt.Sprintf("Pod %s: %s - %s", pod.Name, cs.State.Waiting.Reason, cs.State.Waiting.Message), true
 		}
 	}
+
+	if readySince == nil {
+		return "", "", false
+	}
+	if len(pod.Spec.Containers) == 0 {
+		return "", "", false
+	}
+	// OpenSandbox treats the first regular container as the stateful sandbox workload;
+	// later containers are supporting sidecars such as egress.
+	mainContainerName := pod.Spec.Containers[0].Name
+	for _, cs := range pod.Status.ContainerStatuses {
+		if cs.Name == mainContainerName {
+			terminated := cs.State.Terminated
+			if terminated == nil && cs.RestartCount > 0 {
+				terminated = cs.LastTerminationState.Terminated
+			}
+			if terminated == nil || !terminated.FinishedAt.After(readySince.Time) {
+				return "", "", false
+			}
+			reason := terminated.Reason
+			if reason == "" {
+				reason = "ContainerRestarted"
+			}
+			return reason, fmt.Sprintf("Pod %s container %s terminated after the sandbox became ready", pod.Name, cs.Name), true
+		}
+	}
 	return "", "", false
 }
 
+type restartDetectionBaseline struct {
+	readySince          *metav1.Time
+	previousEndpointIPs map[string]struct{}
+	resetReadyCondition bool
+}
+
+func endpointMembership(endpointIPs []string) map[string]struct{} {
+	membership := make(map[string]struct{}, len(endpointIPs))
+	for _, ip := range endpointIPs {
+		if ip != "" {
+			membership[ip] = struct{}{}
+		}
+	}
+	return membership
+}
+
+func (b restartDetectionBaseline) forPod(pod *corev1.Pod) *metav1.Time {
+	if b.readySince == nil || pod.Status.PodIP == "" {
+		return nil
+	}
+	if _, existedWhenReady := b.previousEndpointIPs[pod.Status.PodIP]; !existedWhenReady {
+		return nil
+	}
+	if !pod.CreationTimestamp.IsZero() && pod.CreationTimestamp.After(b.readySince.Time) {
+		return nil
+	}
+	return b.readySince
+}
+
+func buildRestartDetectionBaseline(batchSbx *sandboxv1alpha1.BatchSandbox, pods []*corev1.Pod, endpointIPs []string) restartDetectionBaseline {
+	status := batchSbx.Status
+	baseline := restartDetectionBaseline{resetReadyCondition: true}
+	if status.Phase != sandboxv1alpha1.BatchSandboxPhaseSucceed {
+		return baseline
+	}
+
+	for i := range status.Conditions {
+		condition := &status.Conditions[i]
+		if condition.Type == sandboxv1alpha1.BatchSandboxConditionReady && condition.Status == sandboxv1alpha1.ConditionTrue &&
+			condition.LastTransitionTime != nil && !condition.LastTransitionTime.IsZero() {
+			baseline.readySince = condition.LastTransitionTime
+			break
+		}
+	}
+	if baseline.readySince == nil {
+		return baseline
+	}
+
+	var previousIPs []string
+	if batchSbx.Annotations == nil || json.Unmarshal([]byte(batchSbx.Annotations[AnnotationSandboxEndpoints]), &previousIPs) != nil {
+		baseline.readySince = nil
+		return baseline
+	}
+
+	baseline.previousEndpointIPs = endpointMembership(previousIPs)
+	baseline.resetReadyCondition = status.Replicas != int32(len(pods)) ||
+		!maps.Equal(baseline.previousEndpointIPs, endpointMembership(endpointIPs))
+	for _, pod := range pods {
+		if !pod.CreationTimestamp.IsZero() && pod.CreationTimestamp.After(baseline.readySince.Time) {
+			baseline.resetReadyCondition = true
+			break
+		}
+	}
+	return baseline
+}
+
 type podFailureSummary struct {
 	observed      int
 	failed        int
 	primaryReason string
 	samplePod     string
 }
 
-func summarizePodFailures(pods []*corev1.Pod) (podFailureSummary, bool) {
+func summarizePodFailures(pods []*corev1.Pod, baseline *restartDetectionBaseline) (podFailureSummary, bool) {
 	summary := podFailureSummary{observed: len(pods)}
 	reasonCounts := make(map[string]int)
 	firstPodByReason := make(map[string]string)
 	primaryCount := 0
 
 	for _, pod := range pods {
-		reason, _, failed := getPodFailureReasonAndMessage(pod)
+		var readySince *metav1.Time
+		if baseline != nil {
+			readySince = baseline.forPod(pod)
+		}
+		reason, _, failed := getPodFailureReasonAndMessage(pod, readySince)
 		if !failed {
 			continue
 		}
@@ -187,7 +284,11 @@ func buildRuntimeView(batchSbx *sandboxv1alpha1.BatchSandbox, pods []*corev1.Pod
 	case sandboxv1alpha1.BatchSandboxPhaseResuming:
 		applyResumingRuntimePhase(newStatus, pods)
 	default:
-		applySteadyRuntimePhase(batchSbx, newStatus, pods)
+		baseline := buildRestartDetectionBaseline(batchSbx, pods, ipList)
+		applySteadyRuntimePhase(batchSbx, newStatus, pods, &baseline)
+		if baseline.resetReadyCondition && newStatus.Phase == sandboxv1alpha1.BatchSandboxPhaseSucceed {
+			setConditionInStatus(newStatus, sandboxv1alpha1.BatchSandboxConditionReady, sandboxv1alpha1.ConditionFalse, "", "")
+		}
 	}
 
 	applyBatchSandboxPhaseConditions(newStatus)
@@ -200,7 +301,7 @@ func buildRuntimeView(batchSbx *sandboxv1alpha1.BatchSandbox, pods []*corev1.Pod
 }
 
 func applyResumingRuntimePhase(status *sandboxv1alpha1.BatchSandboxStatus, pods []*corev1.Pod) {
-	if summary, hasFailures := summarizePodFailures(pods); hasFailures {
+	if summary, hasFailures := summarizePodFailures(pods, nil); hasFailures {
 		setConditionInStatus(status, sandboxv1alpha1.BatchSandboxConditionResumeFailed, sandboxv1alpha1.ConditionTrue, summary.primaryReason, summary.message(true))
 		setConditionInStatus(status, sandboxv1alpha1.BatchSandboxConditionPodFailed, sandboxv1alpha1.ConditionTrue, summary.primaryReason, summary.message(false))
 		status.Phase = sandboxv1alpha1.BatchSandboxPhaseFailed
@@ -212,8 +313,8 @@ func applyResumingRuntimePhase(status *sandboxv1alpha1.BatchSandboxStatus, pods
 	}
 }
 
-func applySteadyRuntimePhase(batchSbx *sandboxv1alpha1.BatchSandbox, status *sandboxv1alpha1.BatchSandboxStatus, pods []*corev1.Pod) {
-	if summary, hasFailures := summarizePodFailures(pods); hasFailures {
+func applySteadyRuntimePhase(batchSbx *sandboxv1alpha1.BatchSandbox, status *sandboxv1alpha1.BatchSandboxStatus, pods []*corev1.Pod, baseline *restartDetectionBaseline) {
+	if summary, hasFailures := summarizePodFailures(pods, baseline); hasFailures {
 		if batchSbx.Status.Phase != sandboxv1alpha1.BatchSandboxPhaseFailed {
 			setConditionInStatus(status, sandboxv1alpha1.BatchSandboxConditionPodFailed, sandboxv1alpha1.ConditionTrue, summary.primaryReason, summary.message(false))
 			status.Phase = sandboxv1alpha1.BatchSandboxPhaseFailed