refactor(flytek8s): export the failure anchor a fault is judged against

Working out when a pod's trouble actually happened is the other half of deciding
whether a GPU fault explains a failure, and it was still private to the executor.
The leaseworker in unionai/cloud is now adopting the relevance check, so without
this it would carry a second copy that has to move in lockstep with the first.

PodFailureTime moves to flytek8s, which is where pod inspection already lives, and
lands next to GetLastTransitionOccurredAt on purpose: that function is the reason
this one exists. It reports the latest transition across every container including
the init containers, so for a pod that failed while its containers were still
running it hands back the time the container started. Anchoring a six-hour task on
its own start puts every real fault outside the window. PodFailureTime takes the
latest terminated main container instead, then the deletion timestamp an eviction
leaves behind, and only then whatever time the caller already had.

It stays out of gpufault, which has no Kubernetes imports and should keep none.
phaseInfoOccurredAt stays in the executor: it reads a PhaseInfo, not a pod, and
nothing outside the plugin machinery has one.

The table tests move with the function and gained the cases that were awkward to
reach through the executor, including one asserting the divergence from
GetLastTransitionOccurredAt that motivates the whole helper. The executor keeps
its end-to-end test for the evicted pod, which is what catches the anchor being
wired up wrong rather than computed wrong.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Signed-off-by: Samhita Alla <aallasamhita@gmail.com>

Author: samhita-alla

executor/pkg/plugin/k8s/plugin_manager.go

@@ -18,6 +18,7 @@ import (
 
 	"github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/errors"
 	pluginsCore "github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/core"
+	"github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/flytek8s"
 	"github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/flytek8s/config"
 	"github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/gpufault"
 	"github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/k8s"
@@ -381,7 +382,7 @@ func (pm *PluginManager) classifyGpuFailure(
 	// Xid that killed the task is usually recorded rounds before the pod's status
 	// catches up with it, and by then the watermark has moved past it. What bounds the
 	// search is the identity and the recency of each event, checked below.
-	failureAt := podFailureTime(resource.(*v1.Pod), phaseInfoOccurredAt(phaseInfo))
+	failureAt := flytek8s.PodFailureTime(resource.(*v1.Pod), phaseInfoOccurredAt(phaseInfo))
 
 	events := pm.eventWatcher.List(objectKeyFor(resource), time.Time{}, time.Time{})
 	if len(events) == 0 {
@@ -433,47 +434,6 @@ func phaseInfoOccurredAt(phaseInfo pluginsCore.PhaseInfo) time.Time {
 	return time.Time{}
 }
 
-// podFailureTime is the time a pod's own trouble is anchored on, which is what the fault
-// relevance interval is centred on.
-//
-// A container's termination is stamped by the kubelet on the same node and clock as the
-// fault events, so it is the closest thing to the moment a fault would have to explain. A
-// pod on its way out without a terminated container is anchored on its deletion, which is
-// what an eviction leaves behind.
-//
-// Only then does the plugin's own reported time stand in, and it is the last resort on
-// purpose. It comes from GetLastTransitionOccurredAt, which for a pod that failed while
-// its containers were still running is the time the container started, not the time
-// anything went wrong. Anchoring a long-running task on its own start would put every real
-// fault outside the window and quietly classify nothing.
-//
-// Init containers are not eligible. They finish before the workload starts, and a native
-// sidecar declared among them is reaped after everything else, so either would anchor on a
-// moment that has nothing to do with when the work died.
-func podFailureTime(pod *v1.Pod, occurredAt time.Time) time.Time {
-	latest := time.Time{}
-	for _, status := range pod.Status.ContainerStatuses {
-		terminated := status.State.Terminated
-		if terminated == nil || terminated.FinishedAt.IsZero() {
-			continue
-		}
-		if terminated.FinishedAt.After(latest) {
-			latest = terminated.FinishedAt.Time
-		}
-	}
-
-	switch {
-	case !latest.IsZero():
-		return latest
-	case pod.DeletionTimestamp != nil && !pod.DeletionTimestamp.IsZero():
-		return pod.DeletionTimestamp.Time
-	case !occurredAt.IsZero():
-		return occurredAt
-	default:
-		return time.Now()
-	}
-}
-
 // Abort implements pluginsCore.Plugin. Called when the task should be killed/aborted.
 func (pm *PluginManager) Abort(ctx context.Context, tCtx pluginsCore.TaskExecutionContext) error {
 	logger.Infof(ctx, "KillTask invoked. We will attempt to delete object [%v].",

executor/pkg/plugin/k8s/plugin_manager_test.go

@@ -567,48 +567,6 @@ func TestClassifyGpuFailureRelevanceIsAnInterval(t *testing.T) {
 	})
 }
 
-func TestPodFailureTime(t *testing.T) {
-	occurredAt := time.Date(2026, 8, 25, 12, 0, 0, 0, time.UTC)
-
-	t.Run("prefers the latest container termination", func(t *testing.T) {
-		first := occurredAt.Add(-10 * time.Minute)
-		last := occurredAt.Add(-2 * time.Minute)
-		pod := &v1.Pod{Status: v1.PodStatus{ContainerStatuses: []v1.ContainerStatus{
-			{State: v1.ContainerState{Terminated: &v1.ContainerStateTerminated{FinishedAt: metav1.NewTime(first)}}},
-			{State: v1.ContainerState{Terminated: &v1.ContainerStateTerminated{FinishedAt: metav1.NewTime(last)}}},
-		}}}
-		assert.Equal(t, last, podFailureTime(pod, occurredAt))
-	})
-
-	t.Run("ignores init containers", func(t *testing.T) {
-		// An init container finished long before the work started, and a native sidecar
-		// declared among the init containers is reaped after everything else. Anchoring on
-		// either would put every real fault outside the window.
-		initFinished := occurredAt.Add(-3 * time.Hour)
-		pod := &v1.Pod{Status: v1.PodStatus{
-			InitContainerStatuses: []v1.ContainerStatus{
-				{State: v1.ContainerState{Terminated: &v1.ContainerStateTerminated{FinishedAt: metav1.NewTime(initFinished)}}},
-			},
-			ContainerStatuses: []v1.ContainerStatus{
-				{State: v1.ContainerState{Running: &v1.ContainerStateRunning{}}},
-			},
-		}}
-		assert.Equal(t, occurredAt, podFailureTime(pod, occurredAt))
-	})
-
-	t.Run("falls back to the deletion timestamp", func(t *testing.T) {
-		deletedAt := occurredAt.Add(-time.Minute)
-		deletion := metav1.NewTime(deletedAt)
-		pod := &v1.Pod{ObjectMeta: metav1.ObjectMeta{DeletionTimestamp: &deletion}}
-		assert.Equal(t, deletedAt, podFailureTime(pod, occurredAt))
-	})
-
-	t.Run("falls back to the reported time, then to now", func(t *testing.T) {
-		assert.Equal(t, occurredAt, podFailureTime(&v1.Pod{}, occurredAt))
-		assert.WithinDuration(t, time.Now(), podFailureTime(&v1.Pod{}, time.Time{}), time.Minute)
-	})
-}
-
 // TestClassifyGpuFailureAnchorsOnThePodNotItsStartTime covers the pod that failed without
 // any container terminating. GetLastTransitionOccurredAt then reports the time the running
 // container started, so the failure the plugin hands over is stamped hours before anything

flyteplugins/go/tasks/pluginmachinery/flytek8s/pod_failure_time_test.go

@@ -0,0 +1,110 @@
+package flytek8s
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func terminatedAt(at time.Time) v1.ContainerStatus {
+	return v1.ContainerStatus{
+		State: v1.ContainerState{Terminated: &v1.ContainerStateTerminated{FinishedAt: metav1.NewTime(at)}},
+	}
+}
+
+func runningSince(at time.Time) v1.ContainerStatus {
+	return v1.ContainerStatus{
+		State: v1.ContainerState{Running: &v1.ContainerStateRunning{StartedAt: metav1.NewTime(at)}},
+	}
+}
+
+func TestPodFailureTime(t *testing.T) {
+	occurredAt := time.Date(2026, 8, 25, 12, 0, 0, 0, time.UTC)
+
+	t.Run("prefers the latest container termination", func(t *testing.T) {
+		first := occurredAt.Add(-10 * time.Minute)
+		last := occurredAt.Add(-2 * time.Minute)
+		pod := &v1.Pod{Status: v1.PodStatus{ContainerStatuses: []v1.ContainerStatus{
+			terminatedAt(first),
+			terminatedAt(last),
+		}}}
+		assert.Equal(t, last, PodFailureTime(pod, occurredAt))
+	})
+
+	t.Run("ignores a container that has not terminated", func(t *testing.T) {
+		died := occurredAt.Add(-2 * time.Minute)
+		pod := &v1.Pod{Status: v1.PodStatus{ContainerStatuses: []v1.ContainerStatus{
+			runningSince(occurredAt.Add(-6 * time.Hour)),
+			terminatedAt(died),
+		}}}
+		assert.Equal(t, died, PodFailureTime(pod, occurredAt))
+	})
+
+	t.Run("ignores a termination with no finish time", func(t *testing.T) {
+		pod := &v1.Pod{Status: v1.PodStatus{ContainerStatuses: []v1.ContainerStatus{
+			{State: v1.ContainerState{Terminated: &v1.ContainerStateTerminated{}}},
+		}}}
+		assert.Equal(t, occurredAt, PodFailureTime(pod, occurredAt))
+	})
+
+	t.Run("ignores init containers", func(t *testing.T) {
+		// This is where it parts company with GetLastTransitionOccurredAt. An init
+		// container finished before the work started, and a native sidecar declared among
+		// the init containers is reaped after everything else. Anchoring on either would
+		// name a moment that has nothing to do with when the work died.
+		initFinished := occurredAt.Add(-3 * time.Hour)
+		pod := &v1.Pod{Status: v1.PodStatus{
+			InitContainerStatuses: []v1.ContainerStatus{terminatedAt(initFinished)},
+			ContainerStatuses:     []v1.ContainerStatus{runningSince(occurredAt.Add(-3 * time.Hour))},
+		}}
+		assert.Equal(t, occurredAt, PodFailureTime(pod, occurredAt))
+		assert.NotEqual(t, initFinished, PodFailureTime(pod, occurredAt))
+	})
+
+	t.Run("falls back to the deletion timestamp", func(t *testing.T) {
+		// What an eviction leaves behind: nothing terminated, but the pod is on its way
+		// out and the API server stamped when.
+		deletedAt := occurredAt.Add(-time.Minute)
+		deletion := metav1.NewTime(deletedAt)
+		pod := &v1.Pod{
+			ObjectMeta: metav1.ObjectMeta{DeletionTimestamp: &deletion},
+			Status:     v1.PodStatus{ContainerStatuses: []v1.ContainerStatus{runningSince(occurredAt.Add(-6 * time.Hour))}},
+		}
+		assert.Equal(t, deletedAt, PodFailureTime(pod, occurredAt))
+	})
+
+	t.Run("prefers a termination over the deletion timestamp", func(t *testing.T) {
+		died := occurredAt.Add(-5 * time.Minute)
+		deletion := metav1.NewTime(occurredAt.Add(-time.Minute))
+		pod := &v1.Pod{
+			ObjectMeta: metav1.ObjectMeta{DeletionTimestamp: &deletion},
+			Status:     v1.PodStatus{ContainerStatuses: []v1.ContainerStatus{terminatedAt(died)}},
+		}
+		assert.Equal(t, died, PodFailureTime(pod, occurredAt))
+	})
+
+	t.Run("falls back to the time the caller offered, then to now", func(t *testing.T) {
+		assert.Equal(t, occurredAt, PodFailureTime(&v1.Pod{}, occurredAt))
+		assert.WithinDuration(t, time.Now(), PodFailureTime(&v1.Pod{}, time.Time{}), time.Minute)
+	})
+}
+
+// The anchor exists because GetLastTransitionOccurredAt cannot serve as one. For a pod
+// that failed while its containers were still running it reports the time the container
+// started, so a long-running task would be anchored hours before anything went wrong.
+func TestPodFailureTimeDoesNotAnchorOnAStartTime(t *testing.T) {
+	startedAt := time.Now().Add(-6 * time.Hour)
+	evictedAt := time.Now().Add(-2 * time.Minute)
+
+	deletion := metav1.NewTime(evictedAt)
+	pod := &v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{DeletionTimestamp: &deletion},
+		Status:     v1.PodStatus{ContainerStatuses: []v1.ContainerStatus{runningSince(startedAt)}},
+	}
+
+	assert.Equal(t, startedAt.Unix(), GetLastTransitionOccurredAt(pod).Unix())
+	assert.Equal(t, evictedAt, PodFailureTime(pod, GetLastTransitionOccurredAt(pod).Time))
+}

flyteplugins/go/tasks/pluginmachinery/flytek8s/pod_helper.go

@@ -1685,6 +1685,49 @@ func GetLastTransitionOccurredAt(pod *v1.Pod) metav1.Time {
 	return lastTransitionTime
 }
 
+// PodFailureTime is the time a pod's own trouble is anchored on, which is what a GPU
+// fault's relevance is measured against (see gpufault.RelevantToFailure).
+//
+// A container's termination is stamped by the kubelet on the same node and clock as the
+// fault events, so it is the closest thing to the moment a fault would have to explain. A
+// pod on its way out without a terminated container is anchored on its deletion, which is
+// what an eviction leaves behind.
+//
+// Only then does the time the caller already had stand in, and it is the last resort on
+// purpose. Callers typically get it from GetLastTransitionOccurredAt above, which for a
+// pod that failed while its containers were still running reports the time the container
+// started, not the time anything went wrong. Anchoring a long-running task on its own
+// start would put every real fault outside the window and quietly find nothing. Pass the
+// zero time when there is no such time to offer, and the current time is used instead.
+//
+// Init containers are not eligible, which is where this parts company with
+// GetLastTransitionOccurredAt. They finish before the workload starts, and a native
+// sidecar declared among them is reaped after everything else, so either would anchor on
+// a moment that has nothing to do with when the work died.
+func PodFailureTime(pod *v1.Pod, occurredAt time.Time) time.Time {
+	latest := time.Time{}
+	for _, status := range pod.Status.ContainerStatuses {
+		terminated := status.State.Terminated
+		if terminated == nil || terminated.FinishedAt.IsZero() {
+			continue
+		}
+		if terminated.FinishedAt.After(latest) {
+			latest = terminated.FinishedAt.Time
+		}
+	}
+
+	switch {
+	case !latest.IsZero():
+		return latest
+	case pod.DeletionTimestamp != nil && !pod.DeletionTimestamp.IsZero():
+		return pod.DeletionTimestamp.Time
+	case !occurredAt.IsZero():
+		return occurredAt
+	default:
+		return time.Now()
+	}
+}
+
 func GetReportedAt(pod *v1.Pod) metav1.Time {
 	var reportedAt metav1.Time
 	for _, condition := range pod.Status.Conditions {