refactor(gpufault): the fault relevance window lives with the contract

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"
@@ -363,32 +364,6 @@ func objectKeyFor(resource client.Object) watchedObjectKey {
 	}
 }
 
-// gpuFaultRelevanceWindow bounds how long before a failure a fault can still explain
-// it. Which pod a fault belongs to is settled by the UID when the pod's UID is known
-// (see classifyGpuFailure for the one case it is not); the window only separates the
-// fault that explains this failure from one the node saw much earlier. It is measured
-// from the failure's own time, not from when classification runs, so a slow reconcile
-// cannot age a fault out. Thirty minutes spans the slow paths between a fault and the
-// failure it causes: a container left wedged after a bus fault until the kubelet gives
-// up on it, and a node going NotReady with its pods evicted only after the
-// node-monitor grace period and eviction timeout.
-//
-// A fault that was still firing inside the window counts even if it started before it,
-// because what the window bounds is how stale a fault's last sign of life may be, not how
-// old the fault is. See faultOverlapsFailure.
-const gpuFaultRelevanceWindow = 30 * time.Minute
-
-// gpuFaultAfterFailureSlack is how far past the failure a fault may first be recorded and
-// still count. The kernel line and the container's termination are stamped by different
-// processes on the same node and the daemon reads the kernel log with a small lag, so a
-// fault can first be recorded moments after the failure it caused; a fault that only
-// started later than that cannot have caused it.
-//
-// It bounds when a fault started, not when it stopped. Hardware that keeps faulting after
-// the container died goes on being observed for as long as it goes on faulting, and that
-// says nothing about whether it caused the failure. See faultOverlapsFailure.
-const gpuFaultAfterFailureSlack = 2 * time.Minute
-
 // classifyGpuFailure folds the GPU faults recorded against a failed attempt's pod into
 // the failure the plugin reported, so that a fault the node saw becomes the code and
 // the message the user reads. Anything that is not a failed pod is left alone.
@@ -407,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 {
@@ -436,7 +411,7 @@ func (pm *PluginManager) classifyGpuFailure(
 		if resource.GetUID() != "" && event.RegardingUID != resource.GetUID() {
 			continue
 		}
-		if !faultOverlapsFailure(event, failureAt) {
+		if !gpufault.RelevantToFailure(event.CreatedAt, event.LastObservedAt, failureAt) {
 			logger.Debugf(context.TODO(),
 				"ignoring GPU fault event %q on %s: active %s to %s, which does not reach the failure at %s",
 				event.Reason, objectKeyFor(resource).Name, event.CreatedAt, event.LastObservedAt, failureAt)
@@ -450,43 +425,6 @@ func (pm *PluginManager) classifyGpuFailure(
 	return gpufault.ClassifyFailure(phaseInfo, faults)
 }
 
-// faultOverlapsFailure reports whether a fault event was active close enough to the
-// failure to explain it.
-//
-// A fault that keeps repeating is aggregated into a single event whose last observation
-// moves with every repeat, so an event describes an interval and not a moment: it was
-// first recorded at CreatedAt and was still firing at LastObservedAt. The failure has an
-// interval of its own, the window before it in which a fault could have caused it and the
-// small slack after it in which a fault it caused could still be recorded. The event
-// counts when those two intervals overlap.
-//
-// Testing the last observation alone, as this used to, drops the fault that matters most:
-// hardware that keeps faulting after the container died has a last observation well past
-// the failure, so the longer it goes on the more certainly it was discarded. Testing the
-// creation alone drops the opposite case, a fault that started before the window opened
-// and was still firing when the task died. Overlap keeps both and still rejects a fault
-// that only started after the failure, or one that had stopped firing before the window
-// opened.
-func faultOverlapsFailure(event *eventInfo, failureAt time.Time) bool {
-	activeFrom, activeUntil := event.CreatedAt, event.LastObservedAt
-	if activeFrom.IsZero() {
-		activeFrom = activeUntil
-	}
-	if activeUntil.IsZero() {
-		activeUntil = activeFrom
-	}
-	if activeFrom.IsZero() || activeUntil.Before(activeFrom) {
-		// No usable time at all, or a last observation older than the creation, which no
-		// honest recorder produces. Nothing can be concluded, so it does not explain.
-		return false
-	}
-
-	relevantFrom := failureAt.Add(-gpuFaultRelevanceWindow)
-	relevantUntil := failureAt.Add(gpuFaultAfterFailureSlack)
-
-	return !activeFrom.After(relevantUntil) && !activeUntil.Before(relevantFrom)
-}
-
 // phaseInfoOccurredAt is the time the plugin put on the failure, or the zero time when it
 // put none there.
 func phaseInfoOccurredAt(phaseInfo pluginsCore.PhaseInfo) time.Time {
@@ -496,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

@@ -342,7 +342,7 @@ func TestClassifyGpuFailure(t *testing.T) {
 	// Recency is measured against the clock now, so the fixtures have to sit relative to
 	// it: base is inside the relevance window, stale is well outside it.
 	base := time.Now().Add(-time.Minute)
-	stale := time.Now().Add(-2 * gpuFaultRelevanceWindow)
+	stale := time.Now().Add(-2 * gpufault.RelevanceWindow)
 
 	tests := []struct {
 		name        string
@@ -554,7 +554,7 @@ func TestClassifyGpuFailureRelevanceIsAnInterval(t *testing.T) {
 	})
 
 	t.Run("a fault that only started after the failure does not explain it", func(t *testing.T) {
-		started := failedAt.Add(gpuFaultAfterFailureSlack + time.Minute)
+		started := failedAt.Add(gpufault.AfterFailureSlack + time.Minute)
 		got := classify(t, started, started.Add(5*time.Minute))
 		assert.Equal(t, "UnknownError", got.Err().GetCode())
 		assert.Nil(t, got.Err().GetGpuFault())
@@ -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 {