GPU fault classification: typed failure codes from Xid events

executor/pkg/controller/taskaction_controller.go

@@ -786,9 +786,14 @@ func toActionErrorInfo(err *core.ExecutionError) *workflow.ErrorInfo {
 	if err == nil {
 		return nil
 	}
+	// Code and GpuFault are carried through as they are. They are the parts of the
+	// failure the user can act on, and this event is the only place they reach the
+	// console and the SDK from.
 	out := &workflow.ErrorInfo{
-		Message: err.GetMessage(),
-		Kind:    workflow.ErrorInfo_KIND_UNSPECIFIED,
+		Message:  err.GetMessage(),
+		Kind:     workflow.ErrorInfo_KIND_UNSPECIFIED,
+		Code:     err.GetCode(),
+		GpuFault: err.GetGpuFault(),
 	}
 	switch err.GetKind() {
 	case core.ExecutionError_USER:

executor/pkg/controller/taskaction_error_info_test.go

@@ -0,0 +1,118 @@
+package controller
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/gpufault"
+	"github.com/flyteorg/flyte/v2/gen/go/flyteidl2/core"
+	"github.com/flyteorg/flyte/v2/gen/go/flyteidl2/workflow"
+)
+
+func TestToActionErrorInfo(t *testing.T) {
+	fault := gpufault.ToProto(
+		gpufault.Fault{
+			Kind: gpufault.KindXid, Code: 79, Name: gpufault.NameFor(gpufault.KindXid, 79),
+			Severity: gpufault.SeverityCritical, PCI: "0000:3b:00.0",
+		},
+		gpufault.Attribution{NodeName: "ip-10-0-0-1", GPUIndex: 0},
+	)
+
+	tests := []struct {
+		name string
+		err  *core.ExecutionError
+		want *workflow.ErrorInfo
+	}{
+		{
+			name: "no error",
+			err:  nil,
+			want: nil,
+		},
+		{
+			name: "user error keeps its code",
+			err:  &core.ExecutionError{Code: "OOMKilled", Message: "container was oom killed", Kind: core.ExecutionError_USER},
+			want: &workflow.ErrorInfo{Code: "OOMKilled", Message: "container was oom killed", Kind: workflow.ErrorInfo_KIND_USER},
+		},
+		{
+			name: "system error keeps its code",
+			err:  &core.ExecutionError{Code: "NodeShutdown", Message: "node is shutting down", Kind: core.ExecutionError_SYSTEM},
+			want: &workflow.ErrorInfo{Code: "NodeShutdown", Message: "node is shutting down", Kind: workflow.ErrorInfo_KIND_SYSTEM},
+		},
+		{
+			name: "unknown kind and no code",
+			err:  &core.ExecutionError{Message: "something went wrong"},
+			want: &workflow.ErrorInfo{Message: "something went wrong", Kind: workflow.ErrorInfo_KIND_UNSPECIFIED},
+		},
+		{
+			name: "gpu fault travels with the error",
+			err: &core.ExecutionError{
+				Code: gpufault.CodeGpuFallenOffBus, Message: "the gpu fell off the bus",
+				Kind: core.ExecutionError_SYSTEM, GpuFault: fault,
+			},
+			want: &workflow.ErrorInfo{
+				Code: gpufault.CodeGpuFallenOffBus, Message: "the gpu fell off the bus",
+				Kind: workflow.ErrorInfo_KIND_SYSTEM, GpuFault: fault,
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := toActionErrorInfo(tt.err)
+			if tt.want == nil {
+				assert.Nil(t, got)
+				return
+			}
+			require.NotNil(t, got)
+			assert.Equal(t, tt.want.GetCode(), got.GetCode())
+			assert.Equal(t, tt.want.GetMessage(), got.GetMessage())
+			assert.Equal(t, tt.want.GetKind(), got.GetKind())
+			assert.Equal(t, tt.want.GetGpuFault(), got.GetGpuFault())
+		})
+	}
+}
+
+// The CR-persisted ErrorState is the other route a failure takes to the user, when the
+// executor reports the action from the stored status rather than from the live phase.
+func TestErrorStateFromExecError(t *testing.T) {
+	tests := []struct {
+		name        string
+		err         *core.ExecutionError
+		wantCode    string
+		wantKind    string
+		wantMessage string
+		wantNil     bool
+	}{
+		{name: "no error", err: nil, wantNil: true},
+		{
+			name:        "user error",
+			err:         &core.ExecutionError{Code: "OOMKilled", Message: "oom", Kind: core.ExecutionError_USER},
+			wantCode:    "OOMKilled",
+			wantKind:    "USER",
+			wantMessage: "oom",
+		},
+		{
+			name:        "gpu fault code survives the round trip through the CR",
+			err:         &core.ExecutionError{Code: gpufault.CodeGpuFallenOffBus, Message: "gpu gone", Kind: core.ExecutionError_SYSTEM},
+			wantCode:    gpufault.CodeGpuFallenOffBus,
+			wantKind:    "SYSTEM",
+			wantMessage: "gpu gone",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := errorStateFromExecError(tt.err)
+			if tt.wantNil {
+				assert.Nil(t, got)
+				return
+			}
+			require.NotNil(t, got)
+			assert.Equal(t, tt.wantCode, got.Code)
+			assert.Equal(t, tt.wantKind, got.Kind)
+			assert.Equal(t, tt.wantMessage, got.Message)
+		})
+	}
+}

executor/pkg/plugin/k8s/event_watcher.go

@@ -23,8 +23,19 @@ type eventInfo struct {
 	CreatedAt  time.Time
 	RecordedAt time.Time
 	Reason     string
+	// RegardingUID is the UID of the object the event was recorded against. Events are
+	// cached by namespace, name and kind, all of which a recreated object reuses, so this
+	// is what tells one incarnation of an object from the next.
+	RegardingUID k8stypes.UID
+	// LastObservedAt is the freshest time the source saw this event. Kubernetes aggregates
+	// a repeating event by updating the same object rather than creating a new one, so
+	// CreatedAt only says when the first occurrence was seen.
+	LastObservedAt time.Time
 }
 
+// objectEventWatcher lists the events cached for an object. The bounds filter on the time
+// the event was first created and recorded; a caller that cares about a repeating event
+// still happening now filters on LastObservedAt itself.
 type objectEventWatcher interface {
 	List(objectKey watchedObjectKey, createdAfter time.Time, recordedAfter time.Time) []*eventInfo
 }
@@ -53,6 +64,20 @@ func newControllerRuntimeEventWatcher(ctx context.Context, cache ctrlcache.Cache
 }
 
 func (w *controllerRuntimeEventWatcher) OnAdd(obj interface{}, _ bool) {
+	w.store(obj)
+}
+
+func (w *controllerRuntimeEventWatcher) OnUpdate(_, newObj interface{}) {
+	// A repeating event is aggregated into the object that already exists, so its later
+	// occurrences reach us as updates rather than as adds.
+	w.store(newObj)
+}
+
+// store records an event, or refreshes the entry an earlier occurrence of the same event
+// left behind. A refresh only moves what the newer occurrence actually tells us, the
+// identity and the last observed time; the times the entry was first created and first
+// recorded stay put, so a caller listing by watermark does not see the event again.
+func (w *controllerRuntimeEventWatcher) store(obj interface{}) {
 	event, ok := obj.(*eventsv1.Event)
 	if !ok || event == nil {
 		return
@@ -74,18 +99,49 @@ func (w *controllerRuntimeEventWatcher) OnAdd(obj interface{}, _ bool) {
 	})
 	eventInfos := value.(*eventObjects)
 
+	info := &eventInfo{
+		Message:        event.Note,
+		CreatedAt:      event.CreationTimestamp.Time,
+		RecordedAt:     time.Now(),
+		Reason:         event.Reason,
+		RegardingUID:   event.Regarding.UID,
+		LastObservedAt: lastObservedTime(event),
+	}
+
 	eventInfos.mu.Lock()
-	eventInfos.eventInfos[eventKey] = &eventInfo{
-		Message:    event.Note,
-		CreatedAt:  event.CreationTimestamp.Time,
-		RecordedAt: time.Now(),
-		Reason:     event.Reason,
+	defer eventInfos.mu.Unlock()
+
+	if existing, ok := eventInfos.eventInfos[eventKey]; ok {
+		info.CreatedAt = existing.CreatedAt
+		info.RecordedAt = existing.RecordedAt
+		if existing.LastObservedAt.After(info.LastObservedAt) {
+			info.LastObservedAt = existing.LastObservedAt
+		}
 	}
-	eventInfos.mu.Unlock()
+	// The entry is replaced rather than mutated: List hands out these pointers, and a
+	// reader may still be looking at the one it got.
+	eventInfos.eventInfos[eventKey] = info
 }
 
-func (w *controllerRuntimeEventWatcher) OnUpdate(_, _ interface{}) {
-	// Ignore updates; we only need newly observed object events.
+// lastObservedTime is the freshest occurrence the event reports. An aggregated event
+// carries a series whose last observed time moves with every repeat; a plain one only
+// has the time it was recorded at, and older recorders fill in the deprecated field.
+func lastObservedTime(event *eventsv1.Event) time.Time {
+	candidates := []time.Time{event.EventTime.Time, event.DeprecatedLastTimestamp.Time}
+	if event.Series != nil {
+		candidates = append(candidates, event.Series.LastObservedTime.Time)
+	}
+
+	latest := time.Time{}
+	for _, candidate := range candidates {
+		if candidate.After(latest) {
+			latest = candidate
+		}
+	}
+	if latest.IsZero() {
+		return event.CreationTimestamp.Time
+	}
+	return latest
 }
 
 func (w *controllerRuntimeEventWatcher) OnDelete(obj interface{}) {
@@ -126,6 +182,9 @@ func (w *controllerRuntimeEventWatcher) OnDelete(obj interface{}) {
 	// a new event is being added to the bucket while the top-level map entry is concurrently removed.
 }
 
+// List returns the cached events for an object, ordered by when they were created. The
+// bounds are on first creation and first recording, which a later occurrence of the same
+// event does not move, so an event already reported is not reported again.
 func (w *controllerRuntimeEventWatcher) List(objectKey watchedObjectKey, createdAfter time.Time, recordedAfter time.Time) []*eventInfo {
 	value, ok := w.objectCache.Load(objectKey)
 	if !ok {

executor/pkg/plugin/k8s/event_watcher_test.go

@@ -0,0 +1,104 @@
+package k8s
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	eventsv1 "k8s.io/api/events/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	k8stypes "k8s.io/apimachinery/pkg/types"
+)
+
+func testEvent(name string, regardingUID k8stypes.UID, createdAt time.Time) *eventsv1.Event {
+	return &eventsv1.Event{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace:         "ns",
+			Name:              name,
+			CreationTimestamp: metav1.NewTime(createdAt),
+		},
+		Regarding: corev1.ObjectReference{
+			Namespace: "ns",
+			Name:      "pod",
+			Kind:      "Pod",
+			UID:       regardingUID,
+		},
+		Reason: "GPUXidError",
+		Note:   "Xid 79",
+	}
+}
+
+func TestEventWatcherOnAddRecordsIdentityAndObservation(t *testing.T) {
+	watcher := &controllerRuntimeEventWatcher{}
+	createdAt := time.Now().Add(-time.Hour)
+	lastObserved := time.Now().Add(-time.Minute)
+
+	event := testEvent("event-1", "pod-uid", createdAt)
+	event.Series = &eventsv1.EventSeries{Count: 3, LastObservedTime: metav1.NewMicroTime(lastObserved)}
+	watcher.OnAdd(event, false)
+
+	events := watcher.List(watchedObjectKey{Namespace: "ns", Name: "pod", Kind: "Pod"}, time.Time{}, time.Time{})
+	require.Len(t, events, 1)
+	assert.Equal(t, k8stypes.UID("pod-uid"), events[0].RegardingUID)
+	assert.Equal(t, createdAt.UTC(), events[0].CreatedAt.UTC())
+	assert.WithinDuration(t, lastObserved, events[0].LastObservedAt, time.Microsecond)
+}
+
+func TestEventWatcherOnAddFallsBackToCreationTimestamp(t *testing.T) {
+	watcher := &controllerRuntimeEventWatcher{}
+	createdAt := time.Now().Add(-time.Hour)
+
+	watcher.OnAdd(testEvent("event-1", "pod-uid", createdAt), false)
+
+	events := watcher.List(watchedObjectKey{Namespace: "ns", Name: "pod", Kind: "Pod"}, time.Time{}, time.Time{})
+	require.Len(t, events, 1)
+	assert.Equal(t, createdAt.UTC(), events[0].LastObservedAt.UTC())
+}
+
+func TestEventWatcherOnUpdateRefreshesLastObservedAt(t *testing.T) {
+	watcher := &controllerRuntimeEventWatcher{}
+	createdAt := time.Now().Add(-time.Hour)
+	key := watchedObjectKey{Namespace: "ns", Name: "pod", Kind: "Pod"}
+
+	watcher.OnAdd(testEvent("event-1", "pod-uid", createdAt), false)
+
+	first := watcher.List(key, time.Time{}, time.Time{})
+	require.Len(t, first, 1)
+	recordedAt := first[0].RecordedAt
+
+	// The same event, seen again: Kubernetes aggregates the repeat into the object it
+	// already has, so it reaches the watcher as an update.
+	lastObserved := time.Now().Add(-time.Minute)
+	repeat := testEvent("event-1", "pod-uid", createdAt)
+	repeat.Series = &eventsv1.EventSeries{Count: 2, LastObservedTime: metav1.NewMicroTime(lastObserved)}
+	watcher.OnUpdate(nil, repeat)
+
+	events := watcher.List(key, time.Time{}, time.Time{})
+	require.Len(t, events, 1, "the repeat refreshes the entry instead of adding one")
+	assert.WithinDuration(t, lastObserved, events[0].LastObservedAt, time.Microsecond)
+	// The watermarks a caller consumes incrementally have to stay where they were, or
+	// the refreshed event would be handed out a second time.
+	assert.Equal(t, createdAt.UTC(), events[0].CreatedAt.UTC())
+	assert.Equal(t, recordedAt, events[0].RecordedAt)
+	assert.Empty(t, watcher.List(key, events[0].CreatedAt, events[0].RecordedAt))
+}
+
+func TestEventWatcherOnUpdateKeepsTheFreshestObservation(t *testing.T) {
+	watcher := &controllerRuntimeEventWatcher{}
+	createdAt := time.Now().Add(-time.Hour)
+	lastObserved := time.Now().Add(-time.Minute)
+	key := watchedObjectKey{Namespace: "ns", Name: "pod", Kind: "Pod"}
+
+	fresh := testEvent("event-1", "pod-uid", createdAt)
+	fresh.Series = &eventsv1.EventSeries{Count: 2, LastObservedTime: metav1.NewMicroTime(lastObserved)}
+	watcher.OnAdd(fresh, false)
+
+	// An update that carries nothing newer must not walk the observation back.
+	watcher.OnUpdate(nil, testEvent("event-1", "pod-uid", createdAt))
+
+	events := watcher.List(key, time.Time{}, time.Time{})
+	require.Len(t, events, 1)
+	assert.WithinDuration(t, lastObserved, events[0].LastObservedAt, time.Microsecond)
+}