diff --git a/kubernetes/internal/controller/apis.go b/kubernetes/internal/controller/apis.go index 0d5fb4d70..abb3095c1 100644 --- a/kubernetes/internal/controller/apis.go +++ b/kubernetes/internal/controller/apis.go @@ -27,6 +27,7 @@ const ( AnnoAllocStatusKey = "sandbox.opensandbox.io/alloc-status" AnnoAllocReleaseKey = "sandbox.opensandbox.io/alloc-release" AnnoAllocReleasedKey = "sandbox.opensandbox.io/alloc-released" + AnnoRestartBaselineKey = "sandbox.opensandbox.io/restart-baseline" LabelBatchSandboxPodIndexKey = "batch-sandbox.sandbox.opensandbox.io/pod-index" LabelBatchSandboxNameKey = "batch-sandbox.sandbox.opensandbox.io/name" LabelPrivilegedNodeAccess = "sandbox.opensandbox.io/privileged-node-access" diff --git a/kubernetes/internal/controller/batchsandbox_pause_resume_test.go b/kubernetes/internal/controller/batchsandbox_pause_resume_test.go index 6e5f728dc..8c7b2652f 100644 --- a/kubernetes/internal/controller/batchsandbox_pause_resume_test.go +++ b/kubernetes/internal/controller/batchsandbox_pause_resume_test.go @@ -16,6 +16,7 @@ package controller import ( "context" + "encoding/json" "fmt" "sync" "testing" @@ -1641,6 +1642,9 @@ func TestPersistRuntimeView_SkipsStatusUpdateWhenRuntimeStatusUnchanged(t *testi ObjectMeta: metav1.ObjectMeta{ Name: "test-bs-0", Namespace: "default", + Annotations: map[string]string{ + AnnoRestartBaselineKey: fmt.Sprintf(`{"batchSandboxUID":"test-uid","startedAt":%d}`, transitionTime.UnixNano()), + }, }, Status: corev1.PodStatus{ Phase: corev1.PodRunning, @@ -1655,7 +1659,7 @@ func TestPersistRuntimeView_SkipsStatusUpdateWhenRuntimeStatusUnchanged(t *testi fakeClient := fake.NewClientBuilder(). WithScheme(testscheme). WithStatusSubresource(&sandboxv1alpha1.BatchSandbox{}). - WithObjects(bs). + WithObjects(bs, pod). WithInterceptorFuncs(interceptor.Funcs{ SubResourceUpdate: func(ctx context.Context, c client.Client, subResourceName string, obj client.Object, opts ...client.SubResourceUpdateOption) error { if subResourceName == "status" { @@ -1807,6 +1811,524 @@ 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, Namespace: "default", 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 setRestartBaselineAnnotation(t *testing.T, pod *corev1.Pod, batchSandboxUID types.UID, startedAt int64) { + t.Helper() + record, err := json.Marshal(podRestartBaselineRecord{BatchSandboxUID: batchSandboxUID, StartedAt: startedAt}) + require.NoError(t, err) + if pod.Annotations == nil { + pod.Annotations = map[string]string{} + } + pod.Annotations[AnnoRestartBaselineKey] = string(record) +} + +func assertRestartHistoryIgnored(t *testing.T, view runtimeView, readyAt metav1.Time) { + t.Helper() + assert.Equal(t, sandboxv1alpha1.BatchSandboxPhaseSucceed, view.status.Phase) + for _, condition := range view.status.Conditions { + assert.NotEqual(t, sandboxv1alpha1.BatchSandboxConditionPodFailed, condition.Type) + if condition.Type == sandboxv1alpha1.BatchSandboxConditionReady { + require.NotNil(t, condition.LastTransitionTime) + assert.Equal(t, readyAt.Time, condition.LastTransitionTime.Time, + "pod membership changes must not alter the Ready transition 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) +} + +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) +} + +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) +} + +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_DetectsDelayedExistingPodRestartAfterScaleUp(t *testing.T) { + readyAt := metav1.NewTime(time.Now().Add(-3 * time.Minute)) + createdBeforeReady := metav1.NewTime(readyAt.Add(-time.Minute)) + preReadyRestart := metav1.NewTime(readyAt.Add(-30 * time.Second)) + bs := restartTestSandbox(readyAt, "10.0.0.10") + + existing := restartTestPod("existing", "10.0.0.10", createdBeforeReady, preReadyRestart) + newPod := restartTestPod("new-prewarmed", "10.0.0.11", createdBeforeReady, preReadyRestart) + firstView := buildRuntimeView(bs, []*corev1.Pod{existing, newPod}) + require.Equal(t, sandboxv1alpha1.BatchSandboxPhaseSucceed, firstView.status.Phase) + + bs.Status = *firstView.status + endpointRaw, err := json.Marshal(firstView.endpointIPs) + require.NoError(t, err) + bs.Annotations[AnnotationSandboxEndpoints] = string(endpointRaw) + setRestartBaselineAnnotation(t, newPod, bs.UID, firstView.restartDetectionBaseline[newPod.Name]) + + // The informer reports the old Pod's restart only after the scale-up + // reconcile. Its original baseline must still detect the delayed evidence. + delayedRestart := metav1.NewTime(readyAt.Add(time.Minute)) + existing = restartTestPod("existing", "10.0.0.10", createdBeforeReady, delayedRestart) + secondView := buildRuntimeView(bs, []*corev1.Pod{existing, newPod}) + assert.Equal(t, sandboxv1alpha1.BatchSandboxPhaseFailed, secondView.status.Phase) +} + +func TestBuildRuntimeView_DetectsNewPodRestartAfterEndpointExposure(t *testing.T) { + readyAt := metav1.NewTime(time.Now().Add(-3 * time.Minute)) + preExposureRestart := metav1.NewTime(readyAt.Add(time.Minute)) + bs := restartTestSandbox(readyAt, "10.0.0.10") + bs.UID = "test-bs-uid" + newPod := restartTestPod("new-prewarmed", "10.0.0.11", metav1.Time{}, preExposureRestart) + + firstView := buildRuntimeView(bs, []*corev1.Pod{newPod}) + assertRestartHistoryIgnored(t, firstView, readyAt) + pending, exists := firstView.restartDetectionBaseline[newPod.Name] + require.True(t, exists) + assert.Zero(t, pending) + setRestartBaselineAnnotation(t, newPod, bs.UID, 0) + + endpointRaw, err := json.Marshal(firstView.endpointIPs) + require.NoError(t, err) + bs.Annotations[AnnotationSandboxEndpoints] = string(endpointRaw) + publishedView := buildRuntimeView(bs, []*corev1.Pod{newPod}) + exposedAtNanos, exists := publishedView.restartDetectionBaseline[newPod.Name] + require.True(t, exists) + exposedAt := time.Unix(0, exposedAtNanos) + assert.True(t, exposedAt.After(preExposureRestart.Time)) + + setRestartBaselineAnnotation(t, newPod, bs.UID, exposedAtNanos) + + postExposureRestart := metav1.NewTime(exposedAt.Add(time.Second)) + newPod = restartTestPod("new-prewarmed", "10.0.0.11", metav1.Time{}, postExposureRestart) + setRestartBaselineAnnotation(t, newPod, bs.UID, exposedAtNanos) + secondView := buildRuntimeView(bs, []*corev1.Pod{newPod}) + assert.Equal(t, sandboxv1alpha1.BatchSandboxPhaseFailed, secondView.status.Phase) +} + +func TestBuildRuntimeView_ReassignedPodDoesNotReusePriorSandboxBaseline(t *testing.T) { + readyAt := metav1.NewTime(time.Now().Add(-3 * time.Minute)) + preExposureRestart := metav1.NewTime(readyAt.Add(time.Minute)) + bs := restartTestSandbox(readyAt, "10.0.0.10") + bs.UID = "new-sandbox-uid" + pod := restartTestPod("reassigned", "10.0.0.10", metav1.Time{}, preExposureRestart) + setRestartBaselineAnnotation(t, pod, "old-sandbox-uid", readyAt.UnixNano()) + + view := buildRuntimeView(bs, []*corev1.Pod{pod}) + assertRestartHistoryIgnored(t, view, readyAt) + desired, exists := view.restartDetectionBaseline[pod.Name] + require.True(t, exists) + assert.Zero(t, desired) +} + +func TestBuildRuntimeView_InitialPodsUseReadyTransitionWithoutPodBaseline(t *testing.T) { + bs := &sandboxv1alpha1.BatchSandbox{} + pod := restartTestPod("initial", "10.0.0.10", metav1.Time{}, metav1.Time{}) + + view := buildRuntimeView(bs, []*corev1.Pod{pod}) + assert.Empty(t, view.restartDetectionBaseline) + for _, condition := range view.status.Conditions { + if condition.Type == sandboxv1alpha1.BatchSandboxConditionReady { + require.NotNil(t, condition.LastTransitionTime) + return + } + } + t.Fatal("expected Ready condition") +} + +func TestPersistRuntimeView_PersistsPodBaselinesBeforePublishingEndpoints(t *testing.T) { + readyAt := metav1.NewTime(time.Now().Add(-3 * time.Minute)) + bs := restartTestSandbox(readyAt, "10.0.0.10") + bs.Name = "test-bs" + bs.Namespace = "default" + bs.UID = "test-bs-uid" + bs.ResourceVersion = "1" + existing := restartTestPod("existing", "10.0.0.10", metav1.Time{}, metav1.NewTime(readyAt.Add(-time.Minute))) + existing.UID = "existing-uid" + newPod := restartTestPod("new", "10.0.0.11", metav1.Time{}, metav1.NewTime(readyAt.Add(time.Minute))) + newPod.UID = "new-uid" + legacyNewBaseline := time.Now().UnixNano() + legacyBaselines, err := json.Marshal(map[string]int64{ + "existing-uid": readyAt.UnixNano(), + "new-uid": legacyNewBaseline, + }) + require.NoError(t, err) + bs.Annotations[AnnoRestartBaselineKey] = string(legacyBaselines) + r := newTestReconciler(bs.DeepCopy(), existing.DeepCopy(), newPod.DeepCopy()) + + view := buildRuntimeView(bs.DeepCopy(), []*corev1.Pod{existing, newPod}) + requeue, errs := r.persistRuntimeView(context.Background(), bs.DeepCopy(), view) + require.Empty(t, errs) + assert.Equal(t, time.Second, requeue) + + // Baselines must be observed from the Pod objects before endpoints become + // visible on the BatchSandbox. + intermediate := &sandboxv1alpha1.BatchSandbox{} + require.NoError(t, r.Get(context.Background(), types.NamespacedName{Namespace: bs.Namespace, Name: bs.Name}, intermediate)) + assert.Equal(t, `["10.0.0.10"]`, intermediate.Annotations[AnnotationSandboxEndpoints]) + persistedExisting := &corev1.Pod{} + require.NoError(t, r.Get(context.Background(), types.NamespacedName{Namespace: existing.Namespace, Name: existing.Name}, persistedExisting)) + persistedNew := &corev1.Pod{} + require.NoError(t, r.Get(context.Background(), types.NamespacedName{Namespace: newPod.Namespace, Name: newPod.Name}, persistedNew)) + + observedView := buildRuntimeView(bs.DeepCopy(), []*corev1.Pod{persistedExisting, persistedNew}) + _, errs = r.persistRuntimeView(context.Background(), bs.DeepCopy(), observedView) + require.Empty(t, errs) + + updated := &sandboxv1alpha1.BatchSandbox{} + require.NoError(t, r.Get(context.Background(), types.NamespacedName{Namespace: bs.Namespace, Name: bs.Name}, updated)) + assert.Equal(t, `["10.0.0.10","10.0.0.11"]`, updated.Annotations[AnnotationSandboxEndpoints]) + assert.NotContains(t, updated.Annotations, AnnoRestartBaselineKey) + existingBaseline, exists := restartBaselineFromPod(persistedExisting, bs.UID) + require.True(t, exists) + assert.Equal(t, readyAt.UnixNano(), existingBaseline) + newBaseline, exists := restartBaselineFromPod(persistedNew, bs.UID) + require.True(t, exists) + assert.Equal(t, legacyNewBaseline, newBaseline) + + for _, condition := range updated.Status.Conditions { + if condition.Type == sandboxv1alpha1.BatchSandboxConditionReady { + require.NotNil(t, condition.LastTransitionTime) + assert.Equal(t, readyAt.Unix(), condition.LastTransitionTime.Unix()) + } + } +} + +func TestPersistRuntimeView_DoesNotPublishEndpointsWhenPodBaselinePatchFails(t *testing.T) { + readyAt := metav1.NewTime(time.Now().Add(-3 * time.Minute)) + bs := restartTestSandbox(readyAt, "10.0.0.10") + bs.Name = "test-bs" + bs.Namespace = "default" + bs.UID = "test-bs-uid" + newPod := restartTestPod("new", "10.0.0.11", metav1.Time{}, metav1.NewTime(readyAt.Add(time.Minute))) + newPod.UID = "new-uid" + + fakeClient := fake.NewClientBuilder(). + WithScheme(testscheme). + WithStatusSubresource(&sandboxv1alpha1.BatchSandbox{}). + WithObjects(bs.DeepCopy(), newPod.DeepCopy()). + WithInterceptorFuncs(interceptor.Funcs{ + Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + if _, isPod := obj.(*corev1.Pod); isPod { + return fmt.Errorf("injected Pod patch failure") + } + return c.Patch(ctx, obj, patch, opts...) + }, + }). + Build() + r := &BatchSandboxReconciler{ + Client: fakeClient, + Scheme: testscheme, + Recorder: record.NewFakeRecorder(10), + StatusRVExpectation: expectations.NewResourceVersionExpectation(), + } + + view := buildRuntimeView(bs.DeepCopy(), []*corev1.Pod{newPod}) + _, errs := r.persistRuntimeView(context.Background(), bs.DeepCopy(), view) + require.Len(t, errs, 1) + assert.Contains(t, errs[0].Error(), "injected Pod patch failure") + + updated := &sandboxv1alpha1.BatchSandbox{} + require.NoError(t, r.Get(context.Background(), types.NamespacedName{Namespace: bs.Namespace, Name: bs.Name}, updated)) + assert.Equal(t, `["10.0.0.10"]`, updated.Annotations[AnnotationSandboxEndpoints]) + assert.Equal(t, sandboxv1alpha1.BatchSandboxPhaseSucceed, updated.Status.Phase) +} + +func TestPersistRuntimeView_KeepsPodBaselinePendingWhenEndpointPatchFails(t *testing.T) { + readyAt := metav1.NewTime(time.Now().Add(-3 * time.Minute)) + bs := restartTestSandbox(readyAt, "10.0.0.10") + bs.Name = "test-bs" + bs.Namespace = "default" + bs.UID = "test-bs-uid" + newPod := restartTestPod("new", "10.0.0.11", metav1.Time{}, metav1.NewTime(time.Now().Add(-time.Second))) + newPod.UID = "new-uid" + setRestartBaselineAnnotation(t, newPod, bs.UID, 0) + + fakeClient := fake.NewClientBuilder(). + WithScheme(testscheme). + WithStatusSubresource(&sandboxv1alpha1.BatchSandbox{}). + WithObjects(bs.DeepCopy(), newPod.DeepCopy()). + WithInterceptorFuncs(interceptor.Funcs{ + Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + if _, isBatchSandbox := obj.(*sandboxv1alpha1.BatchSandbox); isBatchSandbox { + return fmt.Errorf("injected endpoint patch failure") + } + return c.Patch(ctx, obj, patch, opts...) + }, + }). + Build() + r := &BatchSandboxReconciler{ + Client: fakeClient, + Scheme: testscheme, + Recorder: record.NewFakeRecorder(10), + StatusRVExpectation: expectations.NewResourceVersionExpectation(), + } + + view := buildRuntimeView(bs.DeepCopy(), []*corev1.Pod{newPod}) + assert.Equal(t, sandboxv1alpha1.BatchSandboxPhaseSucceed, view.status.Phase) + _, errs := r.persistRuntimeView(context.Background(), bs.DeepCopy(), view) + require.Len(t, errs, 1) + assert.Contains(t, errs[0].Error(), "injected endpoint patch failure") + + updated := &sandboxv1alpha1.BatchSandbox{} + require.NoError(t, r.Get(context.Background(), types.NamespacedName{Namespace: bs.Namespace, Name: bs.Name}, updated)) + assert.Equal(t, `["10.0.0.10"]`, updated.Annotations[AnnotationSandboxEndpoints]) + persistedPod := &corev1.Pod{} + require.NoError(t, r.Get(context.Background(), types.NamespacedName{Namespace: newPod.Namespace, Name: newPod.Name}, persistedPod)) + pendingRecord, exists := restartBaselineRecordFromPod(persistedPod, bs.UID) + require.True(t, exists) + assert.Zero(t, pendingRecord.StartedAt) +} + +func TestPersistRuntimeView_IgnoresRestartBeforeEndpointPublication(t *testing.T) { + readyAt := metav1.NewTime(time.Now().Add(-3 * time.Minute)) + bs := restartTestSandbox(readyAt, "10.0.0.10") + bs.Name = "test-bs" + bs.Namespace = "default" + bs.UID = "test-bs-uid" + bs.ResourceVersion = "1" + prePublicationRestart := metav1.NewTime(time.Now().Add(-time.Second)) + newPod := restartTestPod("new", "10.0.0.11", metav1.Time{}, prePublicationRestart) + newPod.UID = "new-uid" + newPod.Status.ContainerStatuses[0].RestartCount = 0 + newPod.Status.ContainerStatuses[0].LastTerminationState = corev1.ContainerState{} + r := newTestReconciler(bs.DeepCopy(), newPod.DeepCopy()) + + // First persist a pending marker without exposing the new endpoint. + view := buildRuntimeView(bs.DeepCopy(), []*corev1.Pod{newPod}) + requeue, errs := r.persistRuntimeView(context.Background(), bs.DeepCopy(), view) + require.Empty(t, errs) + assert.Equal(t, time.Second, requeue) + pendingPod := &corev1.Pod{} + require.NoError(t, r.Get(context.Background(), types.NamespacedName{Namespace: newPod.Namespace, Name: newPod.Name}, pendingPod)) + pendingRecord, exists := restartBaselineRecordFromPod(pendingPod, bs.UID) + require.True(t, exists) + assert.Zero(t, pendingRecord.StartedAt) + + intermediate := &sandboxv1alpha1.BatchSandbox{} + require.NoError(t, r.Get(context.Background(), types.NamespacedName{Namespace: bs.Namespace, Name: bs.Name}, intermediate)) + assert.Equal(t, `["10.0.0.10"]`, intermediate.Annotations[AnnotationSandboxEndpoints]) + + // A restart observed while pending is ignored. Endpoint publication then + // activates the baseline using a timestamp captured after the patch succeeds. + restartedPod := restartTestPod("new", "10.0.0.11", metav1.Time{}, prePublicationRestart) + pendingPod.Status = restartedPod.Status + view = buildRuntimeView(bs.DeepCopy(), []*corev1.Pod{pendingPod}) + assert.Equal(t, sandboxv1alpha1.BatchSandboxPhaseSucceed, view.status.Phase) + requeue, errs = r.persistRuntimeView(context.Background(), bs.DeepCopy(), view) + require.Empty(t, errs) + assert.Equal(t, time.Second, requeue) + + updated := &sandboxv1alpha1.BatchSandbox{} + require.NoError(t, r.Get(context.Background(), types.NamespacedName{Namespace: bs.Namespace, Name: bs.Name}, updated)) + assert.Equal(t, `["10.0.0.11"]`, updated.Annotations[AnnotationSandboxEndpoints]) + activePod := &corev1.Pod{} + require.NoError(t, r.Get(context.Background(), types.NamespacedName{Namespace: newPod.Namespace, Name: newPod.Name}, activePod)) + activeBaseline, exists := restartBaselineFromPod(activePod, bs.UID) + require.True(t, exists) + assert.Greater(t, activeBaseline, prePublicationRestart.UnixNano()) + + activePod.Status = restartedPod.Status + postPublicationRestart := metav1.NewTime(time.Unix(0, activeBaseline).Add(time.Second)) + activePod.Status.ContainerStatuses[0].LastTerminationState.Terminated.FinishedAt = postPublicationRestart + postPublicationView := buildRuntimeView(updated, []*corev1.Pod{activePod}) + assert.Equal(t, sandboxv1alpha1.BatchSandboxPhaseFailed, postPublicationView.status.Phase) +} + +func TestPersistRuntimeView_DoesNotExposeEndpointsFromStaleRuntimeView(t *testing.T) { + readyAt := metav1.NewTime(time.Now().Add(-3 * time.Minute)) + bs := restartTestSandbox(readyAt, "10.0.0.10") + bs.Name = "test-bs" + bs.Namespace = "default" + bs.UID = "test-bs-uid" + bs.ResourceVersion = "1" + bs.Annotations[AnnoRestartBaselineKey] = fmt.Sprintf(`{"existing-uid":%d}`, readyAt.UnixNano()) + r := newTestReconciler(bs.DeepCopy()) + r.StatusRVExpectation.Expect(&sandboxv1alpha1.BatchSandbox{ObjectMeta: metav1.ObjectMeta{UID: bs.UID, ResourceVersion: "2"}}) + + view := runtimeView{ + status: bs.Status.DeepCopy(), + endpointIPs: []string{"10.0.0.10", "10.0.0.11"}, + pods: []*corev1.Pod{}, + restartDetectionBaseline: map[string]int64{"existing-uid": readyAt.UnixNano(), "new-uid": time.Now().UnixNano()}, + } + requeue, errs := r.persistRuntimeView(context.Background(), bs.DeepCopy(), view) + require.Empty(t, errs) + assert.Equal(t, time.Second, requeue) + + updated := &sandboxv1alpha1.BatchSandbox{} + require.NoError(t, r.Get(context.Background(), types.NamespacedName{Namespace: bs.Namespace, Name: bs.Name}, updated)) + assert.Equal(t, `["10.0.0.10"]`, updated.Annotations[AnnotationSandboxEndpoints]) + assert.NotContains(t, updated.Annotations[AnnoRestartBaselineKey], "new-uid") +} + +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 +2404,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, diff --git a/kubernetes/internal/controller/batchsandbox_status.go b/kubernetes/internal/controller/batchsandbox_status.go index 0427df935..78abb17b7 100644 --- a/kubernetes/internal/controller/batchsandbox_status.go +++ b/kubernetes/internal/controller/batchsandbox_status.go @@ -33,9 +33,17 @@ import ( ) type runtimeView struct { - status *sandboxv1alpha1.BatchSandboxStatus - endpointIPs []string - resumeCompleted bool + status *sandboxv1alpha1.BatchSandboxStatus + endpointIPs []string + pods []*corev1.Pod + restartDetectionBaseline map[string]int64 + resumeCompleted bool +} + +type podRestartBaselineRecord struct { + BatchSandboxUID types.UID `json:"batchSandboxUID"` + // StartedAt is zero while endpoint publication is pending. + StartedAt int64 `json:"startedAt"` } func setConditionInStatus( @@ -107,7 +115,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,9 +125,177 @@ 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 { + batchSandboxUID types.UID + readySince *metav1.Time + previousEndpointIPs map[string]struct{} + legacyPerPod map[string]int64 + detectRestarts bool + desiredPerPod map[string]int64 +} + +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.detectRestarts { + return nil + } + if baseline, exists := restartBaselineFromPod(pod, b.batchSandboxUID); exists { + persisted := metav1.NewTime(time.Unix(0, baseline)) + return &persisted + } + if baseline, exists := b.legacyPerPod[podRestartBaselineKey(pod)]; exists && baseline > 0 { + persisted := metav1.NewTime(time.Unix(0, baseline)) + return &persisted + } + // A baseline owned by another BatchSandbox proves this pooled Pod was + // reassigned, even when it reuses an endpoint IP from the previous member. + if pod.Annotations != nil && pod.Annotations[AnnoRestartBaselineKey] != "" { + return nil + } + 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 podRestartBaselineKey(pod *corev1.Pod) string { + if pod.UID != "" { + return string(pod.UID) + } + // Pods returned by the API server always have a UID. Falling back to the + // name keeps direct unit fixtures useful without weakening production keys. + return pod.Name +} + +func restartBaselineFromPod(pod *corev1.Pod, batchSandboxUID types.UID) (int64, bool) { + record, exists := restartBaselineRecordFromPod(pod, batchSandboxUID) + if !exists || record.StartedAt <= 0 { + return 0, false + } + return record.StartedAt, true +} + +func restartBaselineRecordFromPod(pod *corev1.Pod, batchSandboxUID types.UID) (podRestartBaselineRecord, bool) { + if pod == nil || pod.Annotations == nil { + return podRestartBaselineRecord{}, false + } + record := podRestartBaselineRecord{} + if json.Unmarshal([]byte(pod.Annotations[AnnoRestartBaselineKey]), &record) != nil || + record.BatchSandboxUID != batchSandboxUID || record.StartedAt < 0 { + return podRestartBaselineRecord{}, false + } + return record, true +} + +func buildRestartDetectionBaseline(batchSbx *sandboxv1alpha1.BatchSandbox, pods []*corev1.Pod, endpointIPs []string) restartDetectionBaseline { + status := batchSbx.Status + baseline := restartDetectionBaseline{ + batchSandboxUID: batchSbx.UID, + detectRestarts: status.Phase == sandboxv1alpha1.BatchSandboxPhaseSucceed, + desiredPerPod: make(map[string]int64, len(pods)), + } + + if baseline.detectRestarts { + 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 batchSbx.Annotations != nil { + if raw := batchSbx.Annotations[AnnoRestartBaselineKey]; raw != "" { + _ = json.Unmarshal([]byte(raw), &baseline.legacyPerPod) + } + } + + var previousIPs []string + if batchSbx.Annotations != nil && json.Unmarshal([]byte(batchSbx.Annotations[AnnotationSandboxEndpoints]), &previousIPs) == nil { + baseline.previousEndpointIPs = endpointMembership(previousIPs) + } + + now := time.Now().UnixNano() + for i, pod := range pods { + if i >= len(endpointIPs) || endpointIPs[i] == "" { + continue + } + key := podRestartBaselineKey(pod) + if key == "" { + continue + } + if record, exists := restartBaselineRecordFromPod(pod, batchSbx.UID); exists { + if record.StartedAt == 0 { + // A zero baseline marks a new or replacement Pod whose endpoint has + // not been published yet. Once its endpoint is visible, use the current + // reconcile time so all observed pre-publication restarts are ignored. + if _, published := baseline.previousEndpointIPs[pod.Status.PodIP]; published { + baseline.desiredPerPod[key] = now + } + } + continue + } + if persisted, exists := baseline.legacyPerPod[key]; exists && persisted > 0 { + baseline.desiredPerPod[key] = persisted + continue + } + // Pods already covered by the Ready transition need no annotation. New or + // replacement Pods first get a pending marker; persistRuntimeView activates + // it with a timestamp only after endpoint publication succeeds. + if baseline.forPod(pod) != nil || status.Phase != sandboxv1alpha1.BatchSandboxPhaseSucceed { + continue + } + baseline.desiredPerPod[key] = 0 + } + return baseline +} + type podFailureSummary struct { observed int failed int @@ -127,14 +303,18 @@ type podFailureSummary struct { 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 } @@ -163,6 +343,7 @@ func (s podFailureSummary) message(duringResume bool) string { } func buildRuntimeView(batchSbx *sandboxv1alpha1.BatchSandbox, pods []*corev1.Pod) runtimeView { + view := runtimeView{} newStatus := batchSbx.Status.DeepCopy() newStatus.ObservedGeneration = batchSbx.Generation newStatus.Replicas = 0 @@ -181,26 +362,28 @@ func buildRuntimeView(batchSbx *sandboxv1alpha1.BatchSandbox, pods []*corev1.Pod } } + baseline := buildRestartDetectionBaseline(batchSbx, pods, ipList) switch batchSbx.Status.Phase { case sandboxv1alpha1.BatchSandboxPhasePausing, sandboxv1alpha1.BatchSandboxPhasePaused: // Keep lifecycle-owned stable phases unchanged. case sandboxv1alpha1.BatchSandboxPhaseResuming: applyResumingRuntimePhase(newStatus, pods) default: - applySteadyRuntimePhase(batchSbx, newStatus, pods) + applySteadyRuntimePhase(batchSbx, newStatus, pods, &baseline) } applyBatchSandboxPhaseConditions(newStatus) - return runtimeView{ - status: newStatus, - endpointIPs: ipList, - resumeCompleted: batchSbx.Status.Phase == sandboxv1alpha1.BatchSandboxPhaseResuming && newStatus.Phase == sandboxv1alpha1.BatchSandboxPhaseSucceed, - } + view.status = newStatus + view.endpointIPs = ipList + view.pods = pods + view.restartDetectionBaseline = baseline.desiredPerPod + view.resumeCompleted = batchSbx.Status.Phase == sandboxv1alpha1.BatchSandboxPhaseResuming && newStatus.Phase == sandboxv1alpha1.BatchSandboxPhaseSucceed + return view } 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 +395,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 @@ -248,29 +431,64 @@ func (r *BatchSandboxReconciler) persistRuntimeView( ) (time.Duration, []error) { var aggErrors []error log := logf.FromContext(ctx) - if err := r.patchBatchSandboxEndpoints(ctx, batchSbx, view.endpointIPs); err != nil { - aggErrors = append(aggErrors, err) + if isInitialUnallocatedSandbox(batchSbx, view) { + return 0, aggErrors } - if !equality.Semantic.DeepEqual(*view.status, batchSbx.Status) { - if isInitialUnallocatedSandbox(batchSbx, view) { - return 0, aggErrors - } + statusChanged := !equality.Semantic.DeepEqual(*view.status, batchSbx.Status) + endpointsChanged := endpointsNeedPatch(batchSbx, view.endpointIPs) + baselinesChanged := podBaselinesNeedPatch(batchSbx.UID, view) + if statusChanged || endpointsChanged || baselinesChanged { // Skip redundant status writes caused by informer cache lag: if we recently - // patched status but the informer hasn't seen the new RV yet, the diff is a - // false positive. Allow a 10s safety valve in case the cache never catches up. + // patched status but the informer hasn't seen the new RV yet, the runtime + // view may also contain stale endpoint baselines. Allow a 10s safety valve + // in case the cache never catches up. if satisfied, dur := r.StatusRVExpectation.IsSatisfied(batchSbx); !satisfied { if dur < 10*time.Second { - log.Info("Skipping status update: informer cache is stale", "unsatisfiedDuration", dur.String()) + log.Info("Skipping runtime view update: informer cache is stale", "unsatisfiedDuration", dur.String()) return time.Second, aggErrors } - log.Info("Proceeding with status update despite stale cache (timeout exceeded)", "unsatisfiedDuration", dur.String()) + log.Info("Proceeding with runtime view update despite stale cache (timeout exceeded)", "unsatisfiedDuration", dur.String()) // Fetch the latest object so lifecycle conditions (PauseFailed/ResumeFailed) - // written by pause/resume handlers are not overwritten by the stale cache. + // are not overwritten by the stale cache. latest := &sandboxv1alpha1.BatchSandbox{} if err := r.Get(ctx, types.NamespacedName{Namespace: batchSbx.Namespace, Name: batchSbx.Name}, latest); err == nil { batchSbx = latest + statusChanged = !equality.Semantic.DeepEqual(*view.status, batchSbx.Status) + endpointsChanged = endpointsNeedPatch(batchSbx, view.endpointIPs) } } + } + if baselinesChanged { + patched, err := r.persistPodRestartBaselines(ctx, batchSbx.UID, view) + if err != nil { + aggErrors = append(aggErrors, err) + return 0, aggErrors + } + if patched { + // Wait until the informer observes every Pod baseline before exposing + // the corresponding endpoints. A stale Pod view will retry with an + // optimistic-lock conflict instead of advancing the baseline. + return time.Second, aggErrors + } + } + if endpointsChanged { + if err := r.patchBatchSandboxEndpoints(ctx, batchSbx, view.endpointIPs); err != nil { + aggErrors = append(aggErrors, err) + return 0, aggErrors + } + publishedAt := time.Now().UnixNano() + patched, err := r.activatePublishedPodRestartBaselines(ctx, batchSbx.UID, publishedAt, view) + if err != nil { + aggErrors = append(aggErrors, err) + return 0, aggErrors + } + if patched { + // The activation timestamp was captured after endpoint publication. + // Wait for the informer to observe it before evaluating restart history. + return time.Second, aggErrors + } + } + if statusChanged { if err := r.updateStatus(ctx, batchSbx, view.status); err != nil { aggErrors = append(aggErrors, err) return 0, aggErrors @@ -286,28 +504,98 @@ func (r *BatchSandboxReconciler) persistRuntimeView( return 0, aggErrors } -func (r *BatchSandboxReconciler) patchBatchSandboxEndpoints(ctx context.Context, batchSbx *sandboxv1alpha1.BatchSandbox, endpointIPs []string) error { - raw, _ := json.Marshal(endpointIPs) - if batchSbx.Annotations[AnnotationSandboxEndpoints] == string(raw) { - return nil +func endpointsNeedPatch(batchSbx *sandboxv1alpha1.BatchSandbox, endpointIPs []string) bool { + endpointRaw, _ := json.Marshal(endpointIPs) + _, endpointExists := batchSbx.Annotations[AnnotationSandboxEndpoints] + endpointChanged := batchSbx.Annotations[AnnotationSandboxEndpoints] != string(endpointRaw) + if !endpointExists && string(endpointRaw) == "[]" { + endpointChanged = false } - // Skip writing empty endpoints when annotation doesn't exist yet (e.g. sandbox just created, no pods assigned). - // Still allow clearing endpoints when annotation was previously set (e.g. pause scenario). - _, annotationExists := batchSbx.Annotations[AnnotationSandboxEndpoints] - if !annotationExists && string(raw) == "[]" { - return nil + _, legacyBaselineExists := batchSbx.Annotations[AnnoRestartBaselineKey] + return endpointChanged || legacyBaselineExists +} + +func podBaselinesNeedPatch(batchSandboxUID types.UID, view runtimeView) bool { + for _, pod := range view.pods { + desired, exists := view.restartDetectionBaseline[podRestartBaselineKey(pod)] + if !exists { + continue + } + if record, exists := restartBaselineRecordFromPod(pod, batchSandboxUID); !exists || record.StartedAt != desired { + return true + } + } + return false +} + +func (r *BatchSandboxReconciler) persistPodRestartBaselines(ctx context.Context, batchSandboxUID types.UID, view runtimeView) (bool, error) { + patched := false + for _, pod := range view.pods { + desired, exists := view.restartDetectionBaseline[podRestartBaselineKey(pod)] + if !exists { + continue + } + if record, exists := restartBaselineRecordFromPod(pod, batchSandboxUID); exists && record.StartedAt == desired { + continue + } + record, err := json.Marshal(podRestartBaselineRecord{BatchSandboxUID: batchSandboxUID, StartedAt: desired}) + if err != nil { + return patched, fmt.Errorf("failed to marshal restart baseline for Pod %s/%s: %w", pod.Namespace, pod.Name, err) + } + original := pod.DeepCopy() + updated := pod.DeepCopy() + if updated.Annotations == nil { + updated.Annotations = map[string]string{} + } + updated.Annotations[AnnoRestartBaselineKey] = string(record) + patch := client.MergeFrom(original) + if pod.ResourceVersion != "" { + patch = client.MergeFromWithOptions(original, client.MergeFromWithOptimisticLock{}) + } + if err := r.Patch(ctx, updated, patch); err != nil { + return patched, fmt.Errorf("failed to persist restart baseline for Pod %s/%s: %w", pod.Namespace, pod.Name, err) + } + patched = true } + return patched, nil +} + +func (r *BatchSandboxReconciler) activatePublishedPodRestartBaselines(ctx context.Context, batchSandboxUID types.UID, publishedAt int64, view runtimeView) (bool, error) { + desired := make(map[string]int64) + for i, pod := range view.pods { + if i >= len(view.endpointIPs) || view.endpointIPs[i] == "" { + continue + } + if record, exists := restartBaselineRecordFromPod(pod, batchSandboxUID); exists && record.StartedAt == 0 { + desired[podRestartBaselineKey(pod)] = publishedAt + } + } + if len(desired) == 0 { + return false, nil + } + view.restartDetectionBaseline = desired + return r.persistPodRestartBaselines(ctx, batchSandboxUID, view) +} + +func (r *BatchSandboxReconciler) patchBatchSandboxEndpoints(ctx context.Context, batchSbx *sandboxv1alpha1.BatchSandbox, endpointIPs []string) error { + endpointRaw, _ := json.Marshal(endpointIPs) log := logf.FromContext(ctx) patchData, _ := json.Marshal(map[string]any{ "metadata": map[string]any{ - "annotations": map[string]string{ - AnnotationSandboxEndpoints: string(raw), + "annotations": map[string]any{ + AnnotationSandboxEndpoints: string(endpointRaw), + AnnoRestartBaselineKey: nil, }, }, }) log.Info("Patching BatchSandbox endpoints", "resourceVersion", batchSbx.ResourceVersion, "patchData", string(patchData)) obj := &sandboxv1alpha1.BatchSandbox{ObjectMeta: metav1.ObjectMeta{Namespace: batchSbx.Namespace, Name: batchSbx.Name}} - return r.Patch(ctx, obj, client.RawPatch(types.MergePatchType, patchData)) + if err := r.Patch(ctx, obj, client.RawPatch(types.MergePatchType, patchData)); err != nil { + return err + } + // Prevent a stale informer view from republishing an obsolete endpoint set. + r.StatusRVExpectation.Expect(obj) + return nil } func (r *BatchSandboxReconciler) updateStatus(ctx context.Context, batchSandbox *sandboxv1alpha1.BatchSandbox, newStatus *sandboxv1alpha1.BatchSandboxStatus) error {