diff --git a/actions/k8s/client.go b/actions/k8s/client.go index e2669595e2..059af6c646 100644 --- a/actions/k8s/client.go +++ b/actions/k8s/client.go @@ -327,7 +327,7 @@ func (c *ActionsClient) Signal(ctx context.Context, actionID *common.ActionIdent return connect.NewError(connect.CodeFailedPrecondition, fmt.Errorf("condition %s already signalled with a different value", taskActionName)) } - if isTerminalPhase(GetPhaseFromConditions(taskAction)) { + if IsTerminalPhase(GetPhaseFromConditions(taskAction)) { return connect.NewError(connect.CodeFailedPrecondition, fmt.Errorf("condition %s already completed", taskActionName)) } @@ -668,7 +668,7 @@ func buildActionUpdate(ctx context.Context, taskAction *executorv1.TaskAction, e } phase := GetPhaseFromConditions(taskAction) - if eventType == watch.Deleted && !isTerminalPhase(phase) { + if eventType == watch.Deleted && !IsTerminalPhase(phase) { // Only force ABORTED if the action wasn't already in a terminal phase. // Otherwise a missed-delete tombstone or post-terminal CR cleanup would // overwrite a recorded Succeeded/Failed status with Aborted. @@ -845,7 +845,7 @@ func (c *ActionsClient) notifyRunService(ctx context.Context, taskAction *execut } else if code := resp.Msg.GetStatus().GetCode(); code != 0 { logger.Warnf(ctx, "Run service rejected action status update for %s: code=%d message=%s", update.ActionID.Name, code, resp.Msg.GetStatus().GetMessage()) - } else if isTerminalPhase(update.Phase) && !update.IsDeleted { + } else if IsTerminalPhase(update.Phase) && !update.IsDeleted { // Skip label patching for deleted CRs — the patch would always fail // with "not found" since the object is already gone. if err := c.markTerminalStatusRecorded(ctx, taskAction); err != nil { @@ -899,7 +899,8 @@ func GetPhaseFromConditions(taskAction *executorv1.TaskAction) common.ActionPhas return common.ActionPhase_ACTION_PHASE_UNSPECIFIED } -func isTerminalPhase(phase common.ActionPhase) bool { +// IsTerminalPhase reports whether phase ends an action attempt. +func IsTerminalPhase(phase common.ActionPhase) bool { return phase == common.ActionPhase_ACTION_PHASE_SUCCEEDED || phase == common.ActionPhase_ACTION_PHASE_FAILED || phase == common.ActionPhase_ACTION_PHASE_ABORTED || diff --git a/executor/pkg/controller/k8s_event.go b/executor/pkg/controller/k8s_event.go new file mode 100644 index 0000000000..6be848c1e1 --- /dev/null +++ b/executor/pkg/controller/k8s_event.go @@ -0,0 +1,211 @@ +package controller + +import ( + "fmt" + "strconv" + "strings" + + "google.golang.org/protobuf/encoding/protojson" + corev1 "k8s.io/api/core/v1" + eventsv1 "k8s.io/api/events/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + actionsk8s "github.com/flyteorg/flyte/v2/actions/k8s" + flyteorgv1 "github.com/flyteorg/flyte/v2/executor/api/v1" + "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/common" + "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/workflow" +) + +// EventLevel controls which action events are also emitted as Kubernetes Events. +// Levels are ordered: an event is emitted when its level is at or below the +// configured level. +type EventLevel uint8 + +const ( + EventLevelOff EventLevel = iota + EventLevelTerminal + EventLevelInfo + EventLevelDebug +) + +const ( + annPrefix = "flyte.org/" + annProject = annPrefix + "project" + annDomain = annPrefix + "domain" + annRunName = annPrefix + "run-name" + annActionName = annPrefix + "action-name" + annAttempt = annPrefix + "attempt" + annPhase = annPrefix + "phase" + annVersion = annPrefix + "version" + annErrorKind = annPrefix + "error-kind" + annErrorCode = annPrefix + "error-code" + annInfo = annPrefix + "info" // ActionEvent as protojson, default options + annLaunchPlan = annPrefix + "launch-plan" + annPrincipal = annPrefix + "principal" + annCluster = annPrefix + "cluster" +) + +// noteLimit is the apiserver's NoteLengthLimit. Exceeding it rejects the event. +const noteLimit = 1024 + +const ( + ReasonActionQueued K8sEventType = "ActionQueued" + ReasonActionWaitingForResources K8sEventType = "ActionWaitingForResources" + ReasonActionInitializing K8sEventType = "ActionInitializing" + ReasonActionRunning K8sEventType = "ActionRunning" + ReasonActionSucceeded K8sEventType = "ActionSucceeded" + ReasonActionFailed K8sEventType = "ActionFailed" + ReasonActionAborted K8sEventType = "ActionAborted" + ReasonActionTimedOut K8sEventType = "ActionTimedOut" + ReasonActionPaused K8sEventType = "ActionPaused" + ReasonActionRecovered K8sEventType = "ActionRecovered" + ReasonActionSystemRetry K8sEventType = "ActionSystemRetry" + ReasonActionPhaseUnknown K8sEventType = "ActionPhaseUnknown" +) + +func buildActionEventK8s( + taskAction *flyteorgv1.TaskAction, + event *workflow.ActionEvent, + instance string, +) (*eventsv1.Event, error) { + info, err := protojson.Marshal(event) + if err != nil { + return nil, err + } + + ann := map[string]string{ + annProject: event.GetId().GetRun().GetProject(), + annDomain: event.GetId().GetRun().GetDomain(), + annRunName: event.GetId().GetRun().GetName(), + annActionName: event.GetId().GetName(), + annAttempt: strconv.FormatUint(uint64(event.GetAttempt()), 10), + annPhase: event.GetPhase().String(), + annVersion: strconv.FormatUint(uint64(event.GetVersion()), 10), + annCluster: event.GetCluster(), + // TODO: should we limit the size of this? apiserver limits all annotations + // on one object to 256 kiB. + annInfo: string(info), + } + if e := event.GetErrorInfo(); e != nil { + ann[annErrorKind] = e.GetKind().String() + ann[annErrorCode] = e.GetCode() + } + // TODO: implement this later + // for k, v := range controlPlaneContext(taskAction) { + // ann[k] = v + // } + + return &eventsv1.Event{ + ObjectMeta: metav1.ObjectMeta{ + GenerateName: taskAction.Name + ".", + Namespace: taskAction.Namespace, // must equal Regarding.Namespace + Annotations: ann, + }, + // TODO: need to fallback when event.GetReportedTime() is nil + EventTime: metav1.NewMicroTime(event.GetReportedTime().AsTime()), // required + ReportingController: "taskaction-controller", + ReportingInstance: instance, // required, <= 128 chars + Type: eventType(event), // Normal | Warning + Reason: eventReason(event), // ActionSucceeded | ActionFailed | SystemRetry | ... + Action: "Reconciling", + Note: humanSummary(event), + Regarding: corev1.ObjectReference{ + APIVersion: flyteorgv1.GroupVersion.String(), + Kind: "TaskAction", + Namespace: taskAction.Namespace, + Name: taskAction.Name, + UID: taskAction.UID, + }, + }, nil +} + +// eventType returns Warning for events that need attention and Normal for the rest. +func eventType(event *workflow.ActionEvent) string { + if isSystemRetryEvent(event) { + return corev1.EventTypeWarning + } + switch event.GetPhase() { + case common.ActionPhase_ACTION_PHASE_FAILED, common.ActionPhase_ACTION_PHASE_TIMED_OUT: + return corev1.EventTypeWarning + default: + return corev1.EventTypeNormal + } +} + +// eventReason returns the reason of the k8s event for an action event. +func eventReason(event *workflow.ActionEvent) string { + if isSystemRetryEvent(event) { + return string(ReasonActionSystemRetry) + } + switch event.GetPhase() { + case common.ActionPhase_ACTION_PHASE_QUEUED: + return string(ReasonActionQueued) + case common.ActionPhase_ACTION_PHASE_WAITING_FOR_RESOURCES: + return string(ReasonActionWaitingForResources) + case common.ActionPhase_ACTION_PHASE_INITIALIZING: + return string(ReasonActionInitializing) + case common.ActionPhase_ACTION_PHASE_RUNNING: + return string(ReasonActionRunning) + case common.ActionPhase_ACTION_PHASE_SUCCEEDED: + return string(ReasonActionSucceeded) + case common.ActionPhase_ACTION_PHASE_FAILED: + return string(ReasonActionFailed) + case common.ActionPhase_ACTION_PHASE_ABORTED: + return string(ReasonActionAborted) + case common.ActionPhase_ACTION_PHASE_TIMED_OUT: + return string(ReasonActionTimedOut) + case common.ActionPhase_ACTION_PHASE_PAUSED: + return string(ReasonActionPaused) + case common.ActionPhase_ACTION_PHASE_RECOVERED: + return string(ReasonActionRecovered) + default: + return string(ReasonActionPhaseUnknown) + } +} + +// humanSummary returns a one-line description of event for human reading the k8s event. +// the result fits in the note of a k8s event +func humanSummary(event *workflow.ActionEvent) string { + var state string + if isSystemRetryEvent(event) { + state = "restarted after a system failure" + } else { + phase := strings.TrimPrefix(event.GetPhase().String(), "ACTION_PHASE_") + state = strings.ToLower(strings.ReplaceAll(phase, "_", " ")) + } + summary := fmt.Sprintf("action %s attempt %d %s", event.GetId().GetName(), event.GetAttempt(), state) + if e := event.GetErrorInfo(); e != nil { + kind := strings.TrimPrefix(e.GetKind().String(), "KIND_") + summary += fmt.Sprintf(": %s error %q: %s", kind, e.GetCode(), e.GetMessage()) + } + return truncateUTF8(summary, noteLimit) +} + +// eventLevelOf classifies an action event. Terminal phases and system retries are what an +// external consumer acts on; a version bump is detail. +func eventLevelOf(event *workflow.ActionEvent, prevPhase common.ActionPhase) EventLevel { + switch { + case actionsk8s.IsTerminalPhase(event.GetPhase()), isSystemRetryEvent(event): + return EventLevelTerminal + case event.GetPhase() != prevPhase: + return EventLevelInfo + default: + return EventLevelDebug + } +} + +// isSystemRetryEvent reports whether event is the Queued event recordSystemRetry publishes. +// ActionEvent has no reason field, so the reserved version range identifies it. +func isSystemRetryEvent(event *workflow.ActionEvent) bool { + return event.GetPhase() == common.ActionPhase_ACTION_PHASE_QUEUED && + event.GetVersion() >= systemRetryEventVersionBase +} + +// truncateUTF8 cuts s to at most limit bytes, with any invalid UTF-8 +// removed. limit must be non-negative. +func truncateUTF8(s string, limit int) string { + if len(s) > limit { + s = s[:limit] + } + return strings.ToValidUTF8(s, "") +} diff --git a/executor/pkg/controller/k8s_event_test.go b/executor/pkg/controller/k8s_event_test.go new file mode 100644 index 0000000000..88982fbf5a --- /dev/null +++ b/executor/pkg/controller/k8s_event_test.go @@ -0,0 +1,237 @@ +package controller + +import ( + "strings" + "testing" + "unicode/utf8" + + "github.com/stretchr/testify/assert" + "google.golang.org/protobuf/encoding/protojson" + "google.golang.org/protobuf/proto" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + + flyteorgv1 "github.com/flyteorg/flyte/v2/executor/api/v1" + "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/common" + "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/workflow" +) + +func TestTruncateUTF8(t *testing.T) { + longMessage := strings.Repeat("task failed ", 100) + + tests := []struct { + name string + s string + max int + expected string + }{ + {name: "empty input", s: "", max: 5, expected: ""}, + {name: "shorter than max", s: "task failed", max: 20, expected: "task failed"}, + {name: "exactly max", s: "task failed", max: 11, expected: "task failed"}, + {name: "word over max", s: "failed", max: 4, expected: "fail"}, + {name: "sentence over max", s: "the task failed after three retries", max: 15, expected: "the task failed"}, + {name: "zero max", s: "task failed", max: 0, expected: ""}, + {name: "message over note limit", s: longMessage, max: 1024, expected: longMessage[:1024]}, + // "task " is 5 bytes and each euro sign is 3, so byte 9 is inside the second euro sign. + {name: "cut inside a multi-byte character", s: "task \u20ac\u20ac!", max: 9, expected: "task \u20ac"}, + {name: "invalid byte inside the limit", s: "ab\xffcd", max: 10, expected: "abcd"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := truncateUTF8(tt.s, tt.max) + assert.Equal(t, tt.expected, got) + assert.LessOrEqual(t, len(got), tt.max) + assert.True(t, utf8.ValidString(got)) + }) + } +} + +// testTaskAction returns the TaskAction that the k8s event of an action event refers to. +func testTaskAction() *flyteorgv1.TaskAction { + return &flyteorgv1.TaskAction{ + ObjectMeta: metav1.ObjectMeta{ + Name: "run1-a0", + Namespace: "flytesnacks-development", + UID: types.UID("uid-1"), + }, + } +} + +func TestBuildActionEventK8s(t *testing.T) { + taskAction := testTaskAction() + + t.Run("fields the apiserver checks", func(t *testing.T) { + event := testActionEvent(common.ActionPhase_ACTION_PHASE_SUCCEEDED, 0) + + ev, err := buildActionEventK8s(taskAction, event, "taskaction-controller-host1") + assert.NoError(t, err) + + assert.Equal(t, taskAction.Namespace, ev.Namespace) + assert.Equal(t, ev.Namespace, ev.Regarding.Namespace) + assert.Equal(t, taskAction.Name+".", ev.GenerateName) + assert.Equal(t, flyteorgv1.GroupVersion.String(), ev.Regarding.APIVersion) + assert.Equal(t, "TaskAction", ev.Regarding.Kind) + assert.Equal(t, taskAction.Name, ev.Regarding.Name) + assert.Equal(t, taskAction.UID, ev.Regarding.UID) + + assert.NotEmpty(t, ev.Type) + assert.NotEmpty(t, ev.Reason) + assert.NotEmpty(t, ev.Action) + assert.NotEmpty(t, ev.ReportingController) + assert.Equal(t, "taskaction-controller-host1", ev.ReportingInstance) + }) + + t.Run("no error annotations without error info", func(t *testing.T) { + event := testActionEvent(common.ActionPhase_ACTION_PHASE_SUCCEEDED, 0) + + ev, err := buildActionEventK8s(taskAction, event, "taskaction-controller-host1") + assert.NoError(t, err) + + _, ok := ev.Annotations[annErrorKind] + assert.False(t, ok) + _, ok = ev.Annotations[annErrorCode] + assert.False(t, ok) + }) + + t.Run("error annotations with error info", func(t *testing.T) { + event := testActionEvent(common.ActionPhase_ACTION_PHASE_FAILED, 0) + event.ErrorInfo = &workflow.ErrorInfo{ + Kind: workflow.ErrorInfo_KIND_USER, + Code: "OOMKilled", + Message: "container exceeded its memory limit", + } + + ev, err := buildActionEventK8s(taskAction, event, "taskaction-controller-host1") + assert.NoError(t, err) + + assert.Equal(t, "KIND_USER", ev.Annotations[annErrorKind]) + assert.Equal(t, "OOMKilled", ev.Annotations[annErrorCode]) + }) + + t.Run("info annotation decodes to the original event", func(t *testing.T) { + event := testActionEvent(common.ActionPhase_ACTION_PHASE_FAILED, 3) + event.ErrorInfo = &workflow.ErrorInfo{Kind: workflow.ErrorInfo_KIND_SYSTEM, Code: "Evicted"} + + ev, err := buildActionEventK8s(taskAction, event, "taskaction-controller-host1") + assert.NoError(t, err) + + decoded := &workflow.ActionEvent{} + assert.NoError(t, protojson.Unmarshal([]byte(ev.Annotations[annInfo]), decoded)) + assert.True(t, proto.Equal(event, decoded)) + }) +} + +// testActionEvent returns an event for action a0 of run run1, attempt 1. +func testActionEvent(phase common.ActionPhase, version uint32) *workflow.ActionEvent { + return &workflow.ActionEvent{ + Id: &common.ActionIdentifier{ + Run: &common.RunIdentifier{Project: "flytesnacks", Domain: "development", Name: "run1"}, + Name: "a0", + }, + Attempt: 1, + Phase: phase, + Version: version, + Cluster: "cluster1", + } +} + +func TestEventLevelOf(t *testing.T) { + tests := []struct { + name string + phase common.ActionPhase + version uint32 + prevPhase common.ActionPhase + expected EventLevel + }{ + {"succeeded", common.ActionPhase_ACTION_PHASE_SUCCEEDED, 0, common.ActionPhase_ACTION_PHASE_RUNNING, EventLevelTerminal}, + {"failed", common.ActionPhase_ACTION_PHASE_FAILED, 0, common.ActionPhase_ACTION_PHASE_RUNNING, EventLevelTerminal}, + {"aborted", common.ActionPhase_ACTION_PHASE_ABORTED, 0, common.ActionPhase_ACTION_PHASE_RUNNING, EventLevelTerminal}, + {"timed out", common.ActionPhase_ACTION_PHASE_TIMED_OUT, 0, common.ActionPhase_ACTION_PHASE_RUNNING, EventLevelTerminal}, + {"system retry", common.ActionPhase_ACTION_PHASE_QUEUED, systemRetryEventVersionBase, common.ActionPhase_ACTION_PHASE_RUNNING, EventLevelTerminal}, + {"just below system retry range", common.ActionPhase_ACTION_PHASE_QUEUED, systemRetryEventVersionBase - 1, common.ActionPhase_ACTION_PHASE_QUEUED, EventLevelDebug}, + {"first event", common.ActionPhase_ACTION_PHASE_QUEUED, 0, common.ActionPhase_ACTION_PHASE_UNSPECIFIED, EventLevelInfo}, + {"phase change", common.ActionPhase_ACTION_PHASE_RUNNING, 0, common.ActionPhase_ACTION_PHASE_QUEUED, EventLevelInfo}, + {"same phase, higher version", common.ActionPhase_ACTION_PHASE_RUNNING, 2, common.ActionPhase_ACTION_PHASE_RUNNING, EventLevelDebug}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := eventLevelOf(testActionEvent(tt.phase, tt.version), tt.prevPhase) + assert.Equal(t, tt.expected, got) + }) + } +} + +func TestEventType(t *testing.T) { + tests := []struct { + name string + phase common.ActionPhase + version uint32 + expected string + }{ + {"running", common.ActionPhase_ACTION_PHASE_RUNNING, 0, corev1.EventTypeNormal}, + {"succeeded", common.ActionPhase_ACTION_PHASE_SUCCEEDED, 0, corev1.EventTypeNormal}, + {"aborted is requested, not a fault", common.ActionPhase_ACTION_PHASE_ABORTED, 0, corev1.EventTypeNormal}, + {"failed", common.ActionPhase_ACTION_PHASE_FAILED, 0, corev1.EventTypeWarning}, + {"timed out", common.ActionPhase_ACTION_PHASE_TIMED_OUT, 0, corev1.EventTypeWarning}, + {"system retry", common.ActionPhase_ACTION_PHASE_QUEUED, systemRetryEventVersionBase, corev1.EventTypeWarning}, + {"queued", common.ActionPhase_ACTION_PHASE_QUEUED, 0, corev1.EventTypeNormal}} + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.expected, eventType(testActionEvent(tt.phase, tt.version))) + }) + } +} + +func TestEventReason(t *testing.T) { + t.Run("system retry is checked before the phase", func(t *testing.T) { + event := testActionEvent(common.ActionPhase_ACTION_PHASE_QUEUED, systemRetryEventVersionBase) + assert.Equal(t, string(ReasonActionSystemRetry), eventReason(event)) + }) + + t.Run("failed", func(t *testing.T) { + event := testActionEvent(common.ActionPhase_ACTION_PHASE_FAILED, 0) + assert.Equal(t, string(ReasonActionFailed), eventReason(event)) + }) + + t.Run("every phase has a valid reason", func(t *testing.T) { + for value, name := range common.ActionPhase_name { + event := testActionEvent(common.ActionPhase(value), 0) + reason := eventReason(event) + assert.NotEmpty(t, reason, name) + assert.LessOrEqual(t, len(reason), 128, name) + } + }) +} + +func TestHumanSummary(t *testing.T) { + t.Run("no error", func(t *testing.T) { + event := testActionEvent(common.ActionPhase_ACTION_PHASE_TIMED_OUT, 0) + assert.Equal(t, "action a0 attempt 1 timed out", humanSummary(event)) + }) + + t.Run("system retry", func(t *testing.T) { + event := testActionEvent(common.ActionPhase_ACTION_PHASE_QUEUED, systemRetryEventVersionBase) + assert.Equal(t, "action a0 attempt 1 restarted after a system failure", humanSummary(event)) + }) + + t.Run("with error", func(t *testing.T) { + event := testActionEvent(common.ActionPhase_ACTION_PHASE_FAILED, 0) + event.ErrorInfo = &workflow.ErrorInfo{ + Kind: workflow.ErrorInfo_KIND_USER, + Code: "OOMKilled", + Message: "container exceeded its memory limit", + } + assert.Equal(t, + `action a0 attempt 1 failed: USER error "OOMKilled": container exceeded its memory limit`, + humanSummary(event)) + }) + + t.Run("long message fits in the note", func(t *testing.T) { + event := testActionEvent(common.ActionPhase_ACTION_PHASE_FAILED, 0) + event.ErrorInfo = &workflow.ErrorInfo{Message: strings.Repeat("stack frame ", 200)} + assert.LessOrEqual(t, len(humanSummary(event)), noteLimit) + }) +}