From 4313cf9cbb4fca4153287438c350023bf4cf8f5e Mon Sep 17 00:00:00 2001 From: Kun Zhao Date: Tue, 26 May 2026 13:37:12 -0700 Subject: [PATCH] feat(flow): task report and status-only messages Tighten task.message to status semantics; description stays user-provided. Add task.report (JSONB) carrying a versioned progress document that mirrors the operation rule: every rule stage maps to a StageRecord and every SequenceStep within it maps to a StepRecord at the same index. Steps whose ComponentType has no targets in the task start as StatusSkipped; the rest advance pending -> running -> completed|failed via report.Tracker at stage boundaries. Under the fail-fast activity contract a stage failure marks every still-running sibling step failed because their results are unobserved and reporting them as completed would be a guess. StepRecord carries TotalComponents, filled at NewInitial from task.attributes.components_by_type. The API task representation does not surface that map, so the report has to be self-describing or a client cannot size each step's work. CompletedComponents and FailedComponents are reserved on the wire (omitempty) for a future best-effort activity contract that reports per-component outcomes; they are not written under the current fail-fast contract and so do not appear in the JSON payload today. Pending and processed counts are derivable on the client (Pending = TotalComponents - Completed - Failed; processed = Completed + Failed) and are not stored. Workflows emit full Tracker snapshots on every stage transition, so the store treats writes as authoritative replacements: a non-empty report payload overwrites task.report, an empty payload leaves it untouched. No read-modify-write is performed in either UpdateTaskStatus or UpdateTaskReport, which closes the lost-update race that a merge helper would have introduced. Tracker.FailStage is the canonical writer for the report's task-level error (first-failure-wins); updateFinishedTaskStatus only fills it in when the Tracker did not, so a workflow that constructs a report outside the rule-based path still records its terminal error. The status updater and the report updater are wired through the executor stack as separate parameters rather than via a runtime type assertion. Each role is therefore explicit at every call site and implementations are free to differ; today both are served by the notifying-store wrapper, which declares both interface assertions statically. Signed-off-by: Kun Zhao --- flow/internal/converter/dao/converter.go | 2 + flow/internal/converter/protobuf/converter.go | 1 + .../20260521100000_add_task_report.down.sql | 5 + .../20260521100000_add_task_report.up.sql | 5 + flow/internal/db/model/task.go | 25 ++ .../scheduler/taskschedule/dispatcher_test.go | 4 + .../scheduler/taskschedule/template.go | 63 +--- .../scheduler/taskschedule/template_test.go | 6 +- .../internal/task/conflict/store_mock_test.go | 4 + flow/internal/task/executor/executor.go | 15 +- .../task/executor/temporalworkflow/README.md | 2 +- .../temporalworkflow/activity/activity.go | 18 + .../activity/activity_test.go | 6 +- .../temporalworkflow/activity/registry.go | 17 +- .../activity/registry_test.go | 11 +- .../temporalworkflow/manager/manager.go | 3 +- .../temporalworkflow/workflow/bringup.go | 5 +- .../temporalworkflow/workflow/bringup_test.go | 16 +- .../workflow/firmwarecontrol.go | 7 +- .../workflow/firmwarecontrol_test.go | 11 +- .../temporalworkflow/workflow/helpers.go | 141 ++++++-- .../workflow/injectexpectation.go | 4 +- .../workflow/injectexpectation_test.go | 6 +- .../temporalworkflow/workflow/powercontrol.go | 5 +- .../workflow/powercontrol_action_test.go | 19 +- .../workflow/powercontrol_batching_test.go | 24 +- .../workflow/powercontrol_test.go | 11 +- .../workflow/registry_test.go | 6 +- .../workflow/test_helpers_test.go | 39 +++ flow/internal/task/manager/manager.go | 9 +- flow/internal/task/manager/notifying_store.go | 10 + flow/internal/task/message/message.go | 70 ++++ flow/internal/task/message/message_test.go | 30 ++ flow/internal/task/operationrules/rules.go | 8 + flow/internal/task/operations/summary.go | 138 ++++++++ flow/internal/task/operations/summary_test.go | 82 +++++ flow/internal/task/report/report.go | 329 ++++++++++++++++++ flow/internal/task/report/report_test.go | 224 ++++++++++++ flow/internal/task/store/postgres.go | 30 +- flow/internal/task/store/store.go | 3 + flow/internal/task/task/task.go | 22 +- flow/pkg/proto/v1/flow.pb.go | 30 +- flow/proto/v1/flow.proto | 4 + 43 files changed, 1279 insertions(+), 191 deletions(-) create mode 100644 flow/internal/db/migrations/20260521100000_add_task_report.down.sql create mode 100644 flow/internal/db/migrations/20260521100000_add_task_report.up.sql create mode 100644 flow/internal/task/executor/temporalworkflow/workflow/test_helpers_test.go create mode 100644 flow/internal/task/message/message.go create mode 100644 flow/internal/task/message/message_test.go create mode 100644 flow/internal/task/operations/summary.go create mode 100644 flow/internal/task/operations/summary_test.go create mode 100644 flow/internal/task/report/report.go create mode 100644 flow/internal/task/report/report_test.go diff --git a/flow/internal/converter/dao/converter.go b/flow/internal/converter/dao/converter.go index 03882fc4a..ef26a1b61 100644 --- a/flow/internal/converter/dao/converter.go +++ b/flow/internal/converter/dao/converter.go @@ -191,6 +191,7 @@ func TaskFrom(dao *model.Task) *taskdef.Task { ExecutionID: dao.ExecutionID, Status: dao.Status, Message: dao.Message, + Report: dao.Report, AppliedRuleID: dao.AppliedRuleID, CreatedAt: dao.CreatedAt, UpdatedAt: dao.UpdatedAt, @@ -326,6 +327,7 @@ func TaskTo(task *taskdef.Task) *model.Task { ExecutionID: task.ExecutionID, Status: task.Status, Message: task.Message, + Report: task.Report, AppliedRuleID: task.AppliedRuleID, QueueExpiresAt: task.QueueExpiresAt, } diff --git a/flow/internal/converter/protobuf/converter.go b/flow/internal/converter/protobuf/converter.go index d34f67d30..16e4d8101 100644 --- a/flow/internal/converter/protobuf/converter.go +++ b/flow/internal/converter/protobuf/converter.go @@ -480,6 +480,7 @@ func TaskTo(task *taskdef.Task) *pb.Task { ExecutionId: task.ExecutionID, Status: TaskStatusTo(task.Status), Message: task.Message, + Report: string(task.Report), CreatedAt: timestamppb.New(task.CreatedAt), UpdatedAt: timestamppb.New(task.UpdatedAt), } diff --git a/flow/internal/db/migrations/20260521100000_add_task_report.down.sql b/flow/internal/db/migrations/20260521100000_add_task_report.down.sql new file mode 100644 index 000000000..45d2010b4 --- /dev/null +++ b/flow/internal/db/migrations/20260521100000_add_task_report.down.sql @@ -0,0 +1,5 @@ +-- SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +-- SPDX-License-Identifier: Apache-2.0 + +ALTER TABLE public.task + DROP COLUMN IF EXISTS report; diff --git a/flow/internal/db/migrations/20260521100000_add_task_report.up.sql b/flow/internal/db/migrations/20260521100000_add_task_report.up.sql new file mode 100644 index 000000000..cfcbe3c2b --- /dev/null +++ b/flow/internal/db/migrations/20260521100000_add_task_report.up.sql @@ -0,0 +1,5 @@ +-- SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +-- SPDX-License-Identifier: Apache-2.0 + +ALTER TABLE public.task + ADD COLUMN report jsonb; diff --git a/flow/internal/db/model/task.go b/flow/internal/db/model/task.go index 206f899dc..eb0c62717 100644 --- a/flow/internal/db/model/task.go +++ b/flow/internal/db/model/task.go @@ -50,6 +50,7 @@ type Task struct { ExecutionID string `bun:"execution_id,notnull"` Status taskcommon.TaskStatus `bun:"status,type:varchar(32),notnull"` Message string `bun:"message,nullzero"` + Report json.RawMessage `bun:"report,type:jsonb"` AppliedRuleID *uuid.UUID `bun:"applied_rule_id,type:uuid"` // Which operation rule was applied CreatedAt time.Time `bun:"created_at,nullzero,notnull,default:current_timestamp"` UpdatedAt time.Time `bun:"updated_at,nullzero,notnull,default:current_timestamp"` @@ -92,17 +93,23 @@ func (t *Task) UpdateScheduledTask( } // UpdateTaskStatus updates the status of the task. +// report, when non-nil, replaces the stored report column. func (t *Task) UpdateTaskStatus( ctx context.Context, idb bun.IDB, status taskcommon.TaskStatus, message string, + report json.RawMessage, ) error { t.Status = status t.Message = message t.UpdatedAt = time.Now().UTC() columns := []string{"status", "message", "updated_at", "finished_at"} + if report != nil { + t.Report = report + columns = append(columns, "report") + } if status == taskcommon.TaskStatusRunning && t.StartedAt == nil { t.StartedAt = &t.UpdatedAt @@ -123,6 +130,24 @@ func (t *Task) UpdateTaskStatus( return err } +// UpdateTaskReport updates the structured report without changing status or message. +func (t *Task) UpdateTaskReport( + ctx context.Context, + idb bun.IDB, + report json.RawMessage, +) error { + t.Report = report + t.UpdatedAt = time.Now().UTC() + + _, err := idb.NewUpdate(). + Model(t). + Column("report", "updated_at"). + Where("id = ?", t.ID). + Exec(ctx) + + return err +} + // taskComponentIDFilter restricts a query to tasks whose attributes // reference a specific component UUID. The UUID may live under any bucket // of attributes.components_by_type, which rules out a fixed @> check diff --git a/flow/internal/scheduler/taskschedule/dispatcher_test.go b/flow/internal/scheduler/taskschedule/dispatcher_test.go index e5ac9cfb0..fb79d0376 100644 --- a/flow/internal/scheduler/taskschedule/dispatcher_test.go +++ b/flow/internal/scheduler/taskschedule/dispatcher_test.go @@ -189,6 +189,10 @@ func (m *mockTaskStore) UpdateTaskStatus(_ context.Context, _ *taskdef.TaskStatu panic("mockTaskStore.UpdateTaskStatus: not implemented") } +func (m *mockTaskStore) UpdateTaskReport(_ context.Context, _ *taskdef.TaskReportUpdate) error { + panic("mockTaskStore.UpdateTaskReport: not implemented") +} + func (m *mockTaskStore) ListActiveTasksForRack(_ context.Context, _ uuid.UUID) ([]*taskdef.Task, error) { panic("mockTaskStore.ListActiveTasksForRack: not implemented") } diff --git a/flow/internal/scheduler/taskschedule/template.go b/flow/internal/scheduler/taskschedule/template.go index 463c9a223..22a95b2e1 100644 --- a/flow/internal/scheduler/taskschedule/template.go +++ b/flow/internal/scheduler/taskschedule/template.go @@ -23,6 +23,7 @@ import ( "github.com/NVIDIA/infra-controller-rest/flow/internal/operation" taskcommon "github.com/NVIDIA/infra-controller-rest/flow/internal/task/common" + "github.com/NVIDIA/infra-controller-rest/flow/internal/task/operations" ) // TaskTemplate is the JSON-serialized operation stored in task_schedule.operation_template. @@ -111,61 +112,15 @@ func OptionsFromTemplate(raw json.RawMessage) (TemplateOptions, error) { // (e.g. "POWER_ON", "BRING_UP"). description is a short English phrase for // display (e.g. "Power Off (forced)", "Upgrade Firmware to v2.1.0"). func SummaryFromTemplate(templateJSON json.RawMessage) (opType, description string, err error) { - var tmpl TaskTemplate - if err = json.Unmarshal(templateJSON, &tmpl); err != nil { - return "", "", fmt.Errorf("unmarshal operation_template: %w", err) + w, err := WrapperFromTemplate(templateJSON) + if err != nil { + return "", "", err } - switch tmpl.Type { - case taskcommon.TaskTypePowerControl: - switch tmpl.Code { - case taskcommon.OpCodePowerControlPowerOn, taskcommon.OpCodePowerControlForcePowerOn: - return "POWER_ON", "Power On", nil - case taskcommon.OpCodePowerControlPowerOff: - return "POWER_OFF", "Power Off", nil - case taskcommon.OpCodePowerControlForcePowerOff: - return "POWER_OFF", "Power Off (forced)", nil - case taskcommon.OpCodePowerControlRestart, taskcommon.OpCodePowerControlWarmReset: - return "POWER_RESET", "Power Reset", nil - case taskcommon.OpCodePowerControlForceRestart, taskcommon.OpCodePowerControlColdReset: - return "POWER_RESET", "Power Reset (forced)", nil - default: - return "POWER_CONTROL", tmpl.Code, nil - } - - case taskcommon.TaskTypeBringUp: - if tmpl.Code == taskcommon.OpCodeIngest { - return "INGEST", "Ingest", nil - } - return "BRING_UP", "Bring Up", nil - - case taskcommon.TaskTypeFirmwareControl: - var info struct { - TargetVersion string `json:"target_version"` - } - if len(tmpl.Info) > 0 { - if err = json.Unmarshal(tmpl.Info, &info); err != nil { - return "", "", fmt.Errorf("unmarshal firmware info: %w", err) - } - } - switch tmpl.Code { - case taskcommon.OpCodeFirmwareControlUpgrade: - if info.TargetVersion != "" { - return "UPGRADE_FIRMWARE", "Upgrade Firmware to " + info.TargetVersion, nil - } - return "UPGRADE_FIRMWARE", "Upgrade Firmware", nil - case taskcommon.OpCodeFirmwareControlDowngrade: - if info.TargetVersion != "" { - return "DOWNGRADE_FIRMWARE", "Downgrade Firmware to " + info.TargetVersion, nil - } - return "DOWNGRADE_FIRMWARE", "Downgrade Firmware", nil - case taskcommon.OpCodeFirmwareControlRollback: - return "ROLLBACK_FIRMWARE", "Rollback Firmware", nil - default: // unrecognized - return "FIRMWARE_CONTROL", tmpl.Code, nil - } - - default: - return string(tmpl.Type), tmpl.Code, nil + description, err = operations.SummaryFromWrapper(w) + if err != nil { + return "", "", err } + + return operations.OperationTypeFromWrapper(w), description, nil } diff --git a/flow/internal/scheduler/taskschedule/template_test.go b/flow/internal/scheduler/taskschedule/template_test.go index af73e42ff..306944cc1 100644 --- a/flow/internal/scheduler/taskschedule/template_test.go +++ b/flow/internal/scheduler/taskschedule/template_test.go @@ -228,7 +228,7 @@ func TestSummaryFromTemplate(t *testing.T) { "unknown power control code": { input: mustMarshalTemplate(taskcommon.TaskTypePowerControl, "bmc_cycle", emptyInfo), wantOpType: "POWER_CONTROL", - wantDesc: "bmc_cycle", + wantDesc: "bmc cycle", }, // ── bring-up / ingest ──────────────────────────────────────────────── @@ -281,8 +281,8 @@ func TestSummaryFromTemplate(t *testing.T) { // ── unknown type ────────────────────────────────────────────────────── "unknown operation type": { input: mustMarshalTemplate(taskcommon.TaskTypeInjectExpectation, "inject", emptyInfo), - wantOpType: string(taskcommon.TaskTypeInjectExpectation), - wantDesc: "inject", + wantOpType: "INJECT_EXPECTATION", + wantDesc: "Inject Expectation", }, } diff --git a/flow/internal/task/conflict/store_mock_test.go b/flow/internal/task/conflict/store_mock_test.go index 66db0dd5f..9edbc2ce9 100644 --- a/flow/internal/task/conflict/store_mock_test.go +++ b/flow/internal/task/conflict/store_mock_test.go @@ -79,6 +79,10 @@ func (m *mockStore) ListRacksWithWaitingTasks(ctx context.Context) ([]uuid.UUID, return m.racksWithWaiting, nil } +func (m *mockStore) UpdateTaskReport(_ context.Context, _ *taskdef.TaskReportUpdate) error { + return nil +} + func (m *mockStore) UpdateTaskStatus(ctx context.Context, arg *taskdef.TaskStatusUpdate) error { if m.updateStatusErr != nil { return m.updateStatusErr diff --git a/flow/internal/task/executor/executor.go b/flow/internal/task/executor/executor.go index 70671904b..b3a47c6d2 100644 --- a/flow/internal/task/executor/executor.go +++ b/flow/internal/task/executor/executor.go @@ -44,16 +44,25 @@ type Executor interface { type ExecutorConfig interface { Validate() error // Build constructs the Executor. updater receives task status transitions - // from the execution engine back to the store. It must not be nil. - Build(ctx context.Context, updater task.TaskStatusUpdater) (Executor, error) + // from the execution engine back to the store and must not be nil. + // reportUpdater receives in-flight report snapshots and may be nil; when + // nil, activities that need it return an error at invocation time. + Build( + ctx context.Context, + updater task.TaskStatusUpdater, + reportUpdater task.TaskReportUpdater, + ) (Executor, error) } // New validates the config and builds the Executor, wiring updater so the // engine can report task status changes without importing store packages. +// reportUpdater is plumbed through the same call to persist progress +// snapshots; nil is allowed and disables in-flight reporting. func New( ctx context.Context, executorConfig ExecutorConfig, updater task.TaskStatusUpdater, + reportUpdater task.TaskReportUpdater, ) (Executor, error) { if executorConfig == nil { return nil, fmt.Errorf("executor config is required") @@ -67,5 +76,5 @@ func New( return nil, fmt.Errorf("task status updater is required") } - return executorConfig.Build(ctx, updater) + return executorConfig.Build(ctx, updater, reportUpdater) } diff --git a/flow/internal/task/executor/temporalworkflow/README.md b/flow/internal/task/executor/temporalworkflow/README.md index b1395f480..0458a27cd 100644 --- a/flow/internal/task/executor/temporalworkflow/README.md +++ b/flow/internal/task/executor/temporalworkflow/README.md @@ -50,7 +50,7 @@ ExecutionRequest{OperationType, OperationInfo} **Workflow registry** (`workflow/registry.go`): uses `init()` self-registration. Task-dispatched workflow files call `registerTaskWorkflow[T, *T](taskType, name, fn)`, which derives the timeout and builds the `Unmarshal` closure automatically. Internal workflows (those without a `TaskType`) call `register(WorkflowDescriptor{...})` directly. Nothing needs to be added to a central list — the registry is populated automatically at startup. -**Activity registry** (`activity/registry.go`): uses per-instance dependency injection. `Build()` creates an `*Activities` value via `activity.New(updater, registry)` and calls `acts.All()` to obtain the name → bound-method map, then registers each entry with the Temporal worker via `RegisterActivityWithOptions(fn, {Name: name})`. Because activities are methods on `*Activities`, each manager instance holds its own isolated copy of the dependencies — no shared mutable globals. +**Activity registry** (`activity/registry.go`): uses per-instance dependency injection. `Build()` creates an `*Activities` value via `activity.New(updater, reportUpdater, registry)` and calls `acts.All()` to obtain the name → bound-method map, then registers each entry with the Temporal worker via `RegisterActivityWithOptions(fn, {Name: name})`. Status and report updaters are wired as separate parameters so each role is explicit at the call site. Because activities are methods on `*Activities`, each manager instance holds its own isolated copy of the dependencies — no shared mutable globals. ## Adding a New Operation diff --git a/flow/internal/task/executor/temporalworkflow/activity/activity.go b/flow/internal/task/executor/temporalworkflow/activity/activity.go index dd49f5a17..eb0e51794 100644 --- a/flow/internal/task/executor/temporalworkflow/activity/activity.go +++ b/flow/internal/task/executor/temporalworkflow/activity/activity.go @@ -38,6 +38,7 @@ const ( NamePowerControl = "PowerControl" NameGetPowerStatus = "GetPowerStatus" NameUpdateTaskStatus = "UpdateTaskStatus" + NameUpdateTaskReport = "UpdateTaskReport" NameFirmwareControl = "FirmwareControl" NameGetFirmwareStatus = "GetFirmwareStatus" NameBringUpControl = "BringUpControl" @@ -114,6 +115,23 @@ func (a *Activities) UpdateTaskStatus( return a.updater.UpdateTaskStatus(ctx, arg) } +// UpdateTaskReport is a Temporal activity that merges a structured report +// snapshot without changing status or message. +func (a *Activities) UpdateTaskReport( + ctx context.Context, + arg *task.TaskReportUpdate, +) error { + if a.reportUpdater == nil { + return fmt.Errorf("task report updater is not configured") + } + + if arg == nil || arg.ID == uuid.Nil { + return fmt.Errorf("invalid task identifier") + } + + return a.reportUpdater.UpdateTaskReport(ctx, arg) +} + // FirmwareControl initiates firmware update without waiting for completion. // This activity returns immediately after the update request is accepted. func (a *Activities) FirmwareControl( diff --git a/flow/internal/task/executor/temporalworkflow/activity/activity_test.go b/flow/internal/task/executor/temporalworkflow/activity/activity_test.go index f6a08b29f..02eaa5067 100644 --- a/flow/internal/task/executor/temporalworkflow/activity/activity_test.go +++ b/flow/internal/task/executor/temporalworkflow/activity/activity_test.go @@ -35,7 +35,7 @@ import ( ) func TestActivitiesReturnErrorWhenComponentManagerRegistryIsMissing(t *testing.T) { - acts := New(nil, nil) + acts := New(nil, nil, nil) for name, call := range activityCallsForMissingManagerTest(t, acts) { t.Run(name, func(t *testing.T) { @@ -54,7 +54,7 @@ func TestActivitiesReturnErrorWhenComponentManagerIsMissing(t *testing.T) { ) require.NoError(t, err) - acts := New(nil, registry) + acts := New(nil, nil, registry) for name, call := range activityCallsForMissingManagerTest(t, acts) { t.Run(name, func(t *testing.T) { @@ -292,7 +292,7 @@ func newCapabilityTestActivities( ) require.NoError(t, err) - return New(nil, registry), manager + return New(nil, nil, registry), manager } func newActivityTestTarget() common.Target { diff --git a/flow/internal/task/executor/temporalworkflow/activity/registry.go b/flow/internal/task/executor/temporalworkflow/activity/registry.go index 0eafd79e7..9067f0fa8 100644 --- a/flow/internal/task/executor/temporalworkflow/activity/registry.go +++ b/flow/internal/task/executor/temporalworkflow/activity/registry.go @@ -28,20 +28,22 @@ import ( // independent, multiple managers can coexist in the same process without // sharing mutable state. type Activities struct { - updater task.TaskStatusUpdater - registry *componentmanager.Registry + updater task.TaskStatusUpdater + reportUpdater task.TaskReportUpdater + registry *componentmanager.Registry } -// New creates an Activities instance bound to the given status updater and -// component manager registry. Either argument may be nil; activity calls that -// require the missing dependency will return an error at invocation time. +// New creates an Activities instance. Any argument may be nil; activity +// calls that need a missing dependency return an error at invocation time. func New( updater task.TaskStatusUpdater, + reportUpdater task.TaskReportUpdater, registry *componentmanager.Registry, ) *Activities { return &Activities{ - updater: updater, - registry: registry, + updater: updater, + reportUpdater: reportUpdater, + registry: registry, } } @@ -55,6 +57,7 @@ func (a *Activities) All() map[string]any { NamePowerControl: a.PowerControl, NameGetPowerStatus: a.GetPowerStatus, NameUpdateTaskStatus: a.UpdateTaskStatus, + NameUpdateTaskReport: a.UpdateTaskReport, NameFirmwareControl: a.FirmwareControl, NameGetFirmwareStatus: a.GetFirmwareStatus, NameBringUpControl: a.BringUpControl, diff --git a/flow/internal/task/executor/temporalworkflow/activity/registry_test.go b/flow/internal/task/executor/temporalworkflow/activity/registry_test.go index 8b0816220..5cb2a2242 100644 --- a/flow/internal/task/executor/temporalworkflow/activity/registry_test.go +++ b/flow/internal/task/executor/temporalworkflow/activity/registry_test.go @@ -31,7 +31,7 @@ import ( // TestActivities_All_ContainsAllActivities verifies that All() returns every // expected activity name with a non-nil function value. func TestActivities_All_ContainsAllActivities(t *testing.T) { - acts := New(nil, nil) + acts := New(nil, nil, nil) all := acts.All() expectedNames := []string{ @@ -39,6 +39,7 @@ func TestActivities_All_ContainsAllActivities(t *testing.T) { NamePowerControl, NameGetPowerStatus, NameUpdateTaskStatus, + NameUpdateTaskReport, NameFirmwareControl, NameGetFirmwareStatus, NameBringUpControl, @@ -56,7 +57,7 @@ func TestActivities_All_ContainsAllActivities(t *testing.T) { // TestActivities_All_ReturnsCopy verifies that mutating the returned map does // not affect subsequent calls — each call produces an independent map. func TestActivities_All_ReturnsCopy(t *testing.T) { - acts := New(nil, nil) + acts := New(nil, nil, nil) first := acts.All() firstLen := len(first) @@ -70,8 +71,8 @@ func TestActivities_All_ReturnsCopy(t *testing.T) { // TestActivities_Isolation verifies that two Activities instances do not share // state: mutations to one instance's map must not affect the other. func TestActivities_Isolation(t *testing.T) { - a1 := New(nil, nil) - a2 := New(nil, nil) + a1 := New(nil, nil, nil) + a2 := New(nil, nil, nil) m1 := a1.All() m1["isolation-sentinel"] = func() {} @@ -83,7 +84,7 @@ func TestActivities_Isolation(t *testing.T) { // TestActivities_ValidAndGetComponentManager_NilRegistry verifies that manager // lookup returns a clear configuration error when no registry is configured. func TestActivities_ValidAndGetComponentManager_NilRegistry(t *testing.T) { - acts := New(nil, nil) + acts := New(nil, nil, nil) _, err := acts.validAndGetComponentManager( newActivityTestTarget(), capability.CapabilityPowerControl, diff --git a/flow/internal/task/executor/temporalworkflow/manager/manager.go b/flow/internal/task/executor/temporalworkflow/manager/manager.go index d26b55d67..57f85889d 100644 --- a/flow/internal/task/executor/temporalworkflow/manager/manager.go +++ b/flow/internal/task/executor/temporalworkflow/manager/manager.go @@ -98,6 +98,7 @@ type Manager struct { func (c *Config) Build( ctx context.Context, updater task.TaskStatusUpdater, + reportUpdater task.TaskReportUpdater, ) (executor.Executor, error) { if err := c.Validate(); err != nil { return nil, err @@ -113,7 +114,7 @@ func (c *Config) Build( // Bind dependencies into an Activities instance so each manager has its // own isolated copy — no shared mutable globals between managers. - acts := activity.New(updater, c.ComponentManagerRegistry) + acts := activity.New(updater, reportUpdater, c.ComponentManagerRegistry) publisherClient, err := temporal.New(c.ClientConf) if err != nil { diff --git a/flow/internal/task/executor/temporalworkflow/workflow/bringup.go b/flow/internal/task/executor/temporalworkflow/workflow/bringup.go index e7b8f4723..af92e6c06 100644 --- a/flow/internal/task/executor/temporalworkflow/workflow/bringup.go +++ b/flow/internal/task/executor/temporalworkflow/workflow/bringup.go @@ -64,12 +64,13 @@ func bringUp( typeToTargets := buildTargets(&reqInfo) - err := executeRuleBasedOperation( + report, err := executeRuleBasedOperation( ctx, + reqInfo.TaskID, typeToTargets, info, reqInfo.RuleDefinition, ) - return updateFinishedTaskStatus(ctx, reqInfo.TaskID, err) + return updateFinishedTaskStatus(ctx, reqInfo.TaskID, err, report) } diff --git a/flow/internal/task/executor/temporalworkflow/workflow/bringup_test.go b/flow/internal/task/executor/temporalworkflow/workflow/bringup_test.go index 0722dff1d..63925fb76 100644 --- a/flow/internal/task/executor/temporalworkflow/workflow/bringup_test.go +++ b/flow/internal/task/executor/temporalworkflow/workflow/bringup_test.go @@ -38,10 +38,6 @@ import ( "github.com/NVIDIA/infra-controller-rest/flow/pkg/common/devicetypes" ) -func mockUpdateTaskStatusForBringUp(ctx context.Context, arg *task.TaskStatusUpdate) error { - return nil -} - func mockBringUpControl(ctx context.Context, target common.Target) error { return nil } @@ -125,8 +121,7 @@ func createBringUpTestComponents() []task.WorkflowComponent { func registerBringUpActivities(env *testsuite.TestWorkflowEnvironment) { env.RegisterWorkflowWithOptions(bringUp, temporalworkflow.RegisterOptions{Name: "BringUp"}) env.RegisterWorkflowWithOptions(genericComponentStepWorkflow, temporalworkflow.RegisterOptions{Name: nameGenericComponentStepWorkflow}) - env.RegisterActivityWithOptions(mockUpdateTaskStatusForBringUp, - activity.RegisterOptions{Name: activitypkg.NameUpdateTaskStatus}) + registerTaskUpdateActivities(env) env.RegisterActivityWithOptions(mockPowerControl, activity.RegisterOptions{Name: activitypkg.NamePowerControl}) env.RegisterActivityWithOptions(mockGetPowerStatus, @@ -135,6 +130,7 @@ func registerBringUpActivities(env *testsuite.TestWorkflowEnvironment) { activity.RegisterOptions{Name: activitypkg.NameBringUpControl}) env.RegisterActivityWithOptions(mockGetBringUpStatus, activity.RegisterOptions{Name: activitypkg.NameGetBringUpStatus}) + expectTaskUpdateActivities(env) } func TestBringUpWorkflow(t *testing.T) { @@ -144,7 +140,6 @@ func TestBringUpWorkflow(t *testing.T) { }{ "success": { setupMocks: func(env *testsuite.TestWorkflowEnvironment) { - env.OnActivity(activitypkg.NameUpdateTaskStatus, mock.Anything, mock.Anything).Return(nil) env.OnActivity(activitypkg.NamePowerControl, mock.Anything, mock.Anything, mock.Anything).Return(nil) env.OnActivity(activitypkg.NameGetPowerStatus, mock.Anything, mock.Anything).Return( map[string]operations.PowerStatus{ @@ -162,7 +157,6 @@ func TestBringUpWorkflow(t *testing.T) { }, "power control failure": { setupMocks: func(env *testsuite.TestWorkflowEnvironment) { - env.OnActivity(activitypkg.NameUpdateTaskStatus, mock.Anything, mock.Anything).Return(nil) env.OnActivity(activitypkg.NameGetPowerStatus, mock.Anything, mock.Anything).Return( map[string]operations.PowerStatus{ "ps-1": operations.PowerStatusOff, @@ -217,12 +211,10 @@ func TestBringUpWorkflowWithIngestion(t *testing.T) { env.RegisterWorkflowWithOptions(bringUp, temporalworkflow.RegisterOptions{Name: "BringUp"}) env.RegisterWorkflowWithOptions(genericComponentStepWorkflow, temporalworkflow.RegisterOptions{Name: nameGenericComponentStepWorkflow}) - env.RegisterActivityWithOptions(mockUpdateTaskStatusForBringUp, - activity.RegisterOptions{Name: activitypkg.NameUpdateTaskStatus}) + registerTaskUpdateActivities(env) env.RegisterActivityWithOptions(mockInjectExpectation, activity.RegisterOptions{Name: activitypkg.NameInjectExpectation}) - - env.OnActivity(activitypkg.NameUpdateTaskStatus, mock.Anything, mock.Anything).Return(nil) + expectTaskUpdateActivities(env) env.OnActivity(activitypkg.NameInjectExpectation, mock.Anything, mock.Anything, mock.Anything).Return(nil) testComponents := []task.WorkflowComponent{ diff --git a/flow/internal/task/executor/temporalworkflow/workflow/firmwarecontrol.go b/flow/internal/task/executor/temporalworkflow/workflow/firmwarecontrol.go index 84577cd1d..ce95fc746 100644 --- a/flow/internal/task/executor/temporalworkflow/workflow/firmwarecontrol.go +++ b/flow/internal/task/executor/temporalworkflow/workflow/firmwarecontrol.go @@ -64,19 +64,20 @@ func firmwareControl( } if err := checkFirmwareUpdatePrerequisites(ctx, &reqInfo); err != nil { - return updateFinishedTaskStatus(ctx, reqInfo.TaskID, err) + return updateFinishedTaskStatus(ctx, reqInfo.TaskID, err, nil) } typeToTargets := buildTargets(&reqInfo) - err := executeRuleBasedOperation( + report, err := executeRuleBasedOperation( ctx, + reqInfo.TaskID, typeToTargets, info, reqInfo.RuleDefinition, ) - return updateFinishedTaskStatus(ctx, reqInfo.TaskID, err) + return updateFinishedTaskStatus(ctx, reqInfo.TaskID, err, report) } // checkFirmwareUpdatePrerequisites validates that firmware update can proceed. diff --git a/flow/internal/task/executor/temporalworkflow/workflow/firmwarecontrol_test.go b/flow/internal/task/executor/temporalworkflow/workflow/firmwarecontrol_test.go index e8dc4ada1..2d01dbddc 100644 --- a/flow/internal/task/executor/temporalworkflow/workflow/firmwarecontrol_test.go +++ b/flow/internal/task/executor/temporalworkflow/workflow/firmwarecontrol_test.go @@ -38,11 +38,6 @@ import ( "github.com/NVIDIA/infra-controller-rest/flow/pkg/common/devicetypes" ) -// mockUpdateTaskStatusForFirmwareControl is a mock activity for updating task status -func mockUpdateTaskStatusForFirmwareControl(ctx context.Context, arg *task.TaskStatusUpdate) error { - return nil -} - // mockFirmwareControl is a mock activity for starting firmware update func mockFirmwareControl(ctx context.Context, target common.Target, info operations.FirmwareControlTaskInfo) error { return nil @@ -180,9 +175,7 @@ func TestFirmwareControlWorkflow(t *testing.T) { env.RegisterWorkflowWithOptions(genericComponentStepWorkflow, temporalworkflow.RegisterOptions{Name: nameGenericComponentStepWorkflow}) - env.RegisterActivityWithOptions(mockUpdateTaskStatusForFirmwareControl, activity.RegisterOptions{ - Name: activitypkg.NameUpdateTaskStatus, - }) + registerTaskUpdateActivities(env) env.RegisterActivityWithOptions(mockFirmwareControl, activity.RegisterOptions{ Name: activitypkg.NameFirmwareControl, }) @@ -196,7 +189,6 @@ func TestFirmwareControlWorkflow(t *testing.T) { Name: activitypkg.NameGetPowerStatus, }) - env.OnActivity(mockUpdateTaskStatusForFirmwareControl, mock.Anything, mock.Anything).Return(nil) env.OnActivity(mockFirmwareControl, mock.Anything, mock.Anything, mock.Anything).Return(tc.activityError) env.OnActivity(mockGetFirmwareStatus, mock.Anything, mock.Anything).Return( &activitypkg.GetFirmwareStatusResult{ @@ -209,6 +201,7 @@ func TestFirmwareControlWorkflow(t *testing.T) { env.OnActivity(mockGetPowerStatus, mock.Anything, mock.Anything).Return( map[string]operations.PowerStatus{"comp1": operations.PowerStatusOn, "comp2": operations.PowerStatusOn}, nil) + expectTaskUpdateActivities(env) env.ExecuteWorkflow(firmwareControl, tc.reqInfo, tc.info) assert.True(t, env.IsWorkflowCompleted()) diff --git a/flow/internal/task/executor/temporalworkflow/workflow/helpers.go b/flow/internal/task/executor/temporalworkflow/workflow/helpers.go index 7a6d94346..6baaf67eb 100644 --- a/flow/internal/task/executor/temporalworkflow/workflow/helpers.go +++ b/flow/internal/task/executor/temporalworkflow/workflow/helpers.go @@ -19,6 +19,7 @@ package workflow import ( "context" + "encoding/json" "errors" "fmt" "time" @@ -32,8 +33,10 @@ import ( taskcommon "github.com/NVIDIA/infra-controller-rest/flow/internal/task/common" "github.com/NVIDIA/infra-controller-rest/flow/internal/task/executor/temporalworkflow/activity" "github.com/NVIDIA/infra-controller-rest/flow/internal/task/executor/temporalworkflow/common" + "github.com/NVIDIA/infra-controller-rest/flow/internal/task/message" "github.com/NVIDIA/infra-controller-rest/flow/internal/task/operationrules" - "github.com/NVIDIA/infra-controller-rest/flow/internal/task/task" + "github.com/NVIDIA/infra-controller-rest/flow/internal/task/report" + taskdef "github.com/NVIDIA/infra-controller-rest/flow/internal/task/task" "github.com/NVIDIA/infra-controller-rest/flow/pkg/common/devicetypes" ) @@ -53,15 +56,49 @@ func updateRunningTaskStatus( return fmt.Errorf("task ID is not specified") } - arg := &task.TaskStatusUpdate{ + arg := &taskdef.TaskStatusUpdate{ ID: taskID, Status: taskcommon.TaskStatusRunning, - Message: "Running", + Message: message.ForStatus(taskcommon.TaskStatusRunning), } return workflow.ExecuteActivity(ctx, activity.NameUpdateTaskStatus, arg).Get(ctx, nil) } +// updateTaskReportBestEffort persists a report snapshot without failing the workflow. +func updateTaskReportBestEffort( + ctx workflow.Context, + taskID uuid.UUID, + rep *report.Report, +) { + if taskID == uuid.Nil || rep == nil { + return + } + + raw, err := rep.MarshalRaw() + if err != nil { + log.Warn().Err(err).Msg("failed to marshal task report") + return + } + + actx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{ + StartToCloseTimeout: 30 * time.Second, + RetryPolicy: &temporal.RetryPolicy{ + MaximumAttempts: 1, + }, + }) + + arg := &taskdef.TaskReportUpdate{ + ID: taskID, + Report: raw, + } + + derr := workflow.ExecuteActivity(actx, activity.NameUpdateTaskReport, arg).Get(ctx, nil) + if derr != nil { + log.Warn().Err(derr).Str("task_id", taskID.String()).Msg("failed to update task report") + } +} + // updateFinishedTaskStatus records the terminal task status (Completed or Failed) // via the UpdateTaskStatus activity. If both the operation error and the status // update fail, the errors are joined. The operation error is always returned so @@ -70,28 +107,56 @@ func updateFinishedTaskStatus( ctx workflow.Context, taskID uuid.UUID, err error, + rep *report.Report, ) error { if taskID == uuid.Nil { return fmt.Errorf("task ID is not specified") } - var arg *task.TaskStatusUpdate + // A failure to marshal the report must not mask err: the original + // operation error is what determines the terminal task status. Drop + // the report from this status update and proceed; raw stays nil so + // the activity records the status without touching task.report. + var raw json.RawMessage + if rep != nil { + // rep.Error is the canonical task-level error and the Tracker is + // the authoritative writer (first failure wins; see + // Tracker.FailStage). Only fill it in here if the Tracker did + // not — e.g. a workflow that constructed a report outside the + // rule-based path and never called FailStage. + if err != nil && rep.Error == "" { + rep.Error = message.ForFailure(err) + } + r, merr := rep.MarshalRaw() + if merr != nil { + log.Warn().Err(merr). + Str("task_id", taskID.String()). + Msg("failed to marshal task report; recording status without report") + } else { + raw = r + } + } + + var arg *taskdef.TaskStatusUpdate if err != nil { - arg = &task.TaskStatusUpdate{ + arg = &taskdef.TaskStatusUpdate{ ID: taskID, Status: taskcommon.TaskStatusFailed, - Message: err.Error(), + Message: message.ForFailure(err), + Report: raw, } } else { - arg = &task.TaskStatusUpdate{ + arg = &taskdef.TaskStatusUpdate{ ID: taskID, Status: taskcommon.TaskStatusCompleted, - Message: "Completed successfully", + Message: message.ForStatus(taskcommon.TaskStatusCompleted), + Report: raw, } } - if lerr := workflow.ExecuteActivity(ctx, activity.NameUpdateTaskStatus, arg).Get(ctx, nil); lerr != nil { //nolint + lerr := workflow.ExecuteActivity(ctx, activity.NameUpdateTaskStatus, arg).Get(ctx, nil) + if lerr != nil { return errors.Join(err, fmt.Errorf("failed to update task status: %w", lerr)) } @@ -101,7 +166,7 @@ func updateFinishedTaskStatus( // buildTargets groups the components in ExecutionInfo by type, returning a map // of ComponentType to Target. A nil info produces an empty (non-nil) map. func buildTargets( - info *task.ExecutionInfo, + info *taskdef.ExecutionInfo, ) map[devicetypes.ComponentType]common.Target { if info == nil { // This is unreachable code, but just in case, handle it anyway. @@ -129,6 +194,20 @@ func buildTargets( return results } +// componentTotalsByType returns a per-ComponentType count of targeted +// components, suitable for report.NewInitial. ComponentTypes absent from +// typeToTargets do not appear in the result; NewInitial then sees them +// as zero-count via map lookup and marks their steps skipped. +func componentTotalsByType( + typeToTargets map[devicetypes.ComponentType]common.Target, +) map[devicetypes.ComponentType]int { + out := make(map[devicetypes.ComponentType]int, len(typeToTargets)) + for ct, target := range typeToTargets { + out[ct] = len(target.ComponentIDs) + } + return out +} + // buildActivityOptions constructs activity options from a sequence step func buildActivityOptions(step operationrules.SequenceStep) workflow.ActivityOptions { opts := workflow.ActivityOptions{ @@ -304,54 +383,76 @@ func parseDurationParam(val any) time.Duration { } } -// executeRuleBasedOperation drives any operation through its RuleDefinition. -// Stages execute sequentially; steps within a stage execute in parallel via -// child workflows. +// executeRuleBasedOperation drives the rule's stages in order. At each +// stage boundary it advances the report.Tracker (begin -> complete or +// fail) and emits a best-effort report snapshot to the store. The first +// stage that fails short-circuits the loop; the partially-populated +// report is returned alongside the wrapped stage error so the caller can +// attach it to the terminal task status. func executeRuleBasedOperation( ctx workflow.Context, + taskID uuid.UUID, typeToTargets map[devicetypes.ComponentType]common.Target, operationInfo any, ruleDef *operationrules.RuleDefinition, -) error { +) (*report.Report, error) { if ruleDef == nil { - return fmt.Errorf( + return nil, fmt.Errorf( "rule definition is nil (resolver should never return nil)", ) } if len(ruleDef.Steps) == 0 { - return fmt.Errorf("rule definition has no steps") + return nil, fmt.Errorf("rule definition has no steps") } + // The report mirrors the rule's stage layout up front: every stage + // and every SequenceStep has a slot from the moment NewInitial + // returns. Tracker advances those slots as the workflow progresses. + tracker := &report.Tracker{ + Report: report.NewInitial(ruleDef, componentTotalsByType(typeToTargets)), + } + + updateTaskReportBestEffort(ctx, taskID, tracker.Report) + log.Info(). Int("step_count", len(ruleDef.Steps)). Msg("Executing operation with rule definition") iter := operationrules.NewStageIterator(ruleDef) for stage := iter.Next(); stage != nil; stage = iter.Next() { + tracker.BeginStage(stage.Number, workflow.Now(ctx)) + updateTaskReportBestEffort(ctx, taskID, tracker.Report) + log.Info(). Int("stage", stage.Number). Int("step_count", len(stage.Steps)). Msg("Executing stage") - if err := executeGenericStageParallel( + err := executeGenericStageParallel( ctx, stage.Steps, typeToTargets, operationInfo, - ); err != nil { + ) + if err != nil { + tracker.FailStage(stage.Number, err, workflow.Now(ctx)) + updateTaskReportBestEffort(ctx, taskID, tracker.Report) log.Error(). Err(err). Int("stage", stage.Number). Msg("Stage execution failed") - return fmt.Errorf("stage %d failed: %w", stage.Number, err) + return tracker.Report, fmt.Errorf("stage %d failed: %w", stage.Number, err) } + tracker.CompleteStage(stage.Number, workflow.Now(ctx)) + updateTaskReportBestEffort(ctx, taskID, tracker.Report) + log.Info(). Int("stage", stage.Number). Msg("Stage completed successfully") } log.Info().Msg("Rule-based operation completed successfully") - return nil + return tracker.Report, nil } diff --git a/flow/internal/task/executor/temporalworkflow/workflow/injectexpectation.go b/flow/internal/task/executor/temporalworkflow/workflow/injectexpectation.go index 80469a923..3eb60014c 100644 --- a/flow/internal/task/executor/temporalworkflow/workflow/injectexpectation.go +++ b/flow/internal/task/executor/temporalworkflow/workflow/injectexpectation.go @@ -90,10 +90,10 @@ func injectExpectation( typeToTargets := buildTargets(&reqInfo) if err := injectExpectationForAll(ctx, typeToTargets, info); err != nil { - return updateFinishedTaskStatus(ctx, reqInfo.TaskID, err) + return updateFinishedTaskStatus(ctx, reqInfo.TaskID, err, nil) } - return updateFinishedTaskStatus(ctx, reqInfo.TaskID, nil) + return updateFinishedTaskStatus(ctx, reqInfo.TaskID, nil, nil) } // injectExpectationForAll calls the InjectExpectation activity for each diff --git a/flow/internal/task/executor/temporalworkflow/workflow/injectexpectation_test.go b/flow/internal/task/executor/temporalworkflow/workflow/injectexpectation_test.go index 40b743c4d..b146857c1 100644 --- a/flow/internal/task/executor/temporalworkflow/workflow/injectexpectation_test.go +++ b/flow/internal/task/executor/temporalworkflow/workflow/injectexpectation_test.go @@ -100,12 +100,9 @@ func TestInjectExpectationWorkflow(t *testing.T) { env.RegisterActivityWithOptions(mockInjectExpectation, activity.RegisterOptions{ Name: activitypkg.NameInjectExpectation, }) - env.RegisterActivityWithOptions(mockUpdateTaskStatus, activity.RegisterOptions{ - Name: activitypkg.NameUpdateTaskStatus, - }) + registerTaskUpdateActivities(env) env.OnActivity(mockInjectExpectation, mock.Anything, mock.Anything, mock.Anything).Return(tc.activityError) - env.OnActivity(mockUpdateTaskStatus, mock.Anything, mock.Anything).Return(nil) info := &operations.InjectExpectationTaskInfo{} reqInfo := taskdef.ExecutionInfo{ @@ -113,6 +110,7 @@ func TestInjectExpectationWorkflow(t *testing.T) { Components: toWorkflowComponents(tc.components), } + expectTaskUpdateActivities(env) env.ExecuteWorkflow(injectExpectation, reqInfo, info) assert.True(t, env.IsWorkflowCompleted()) diff --git a/flow/internal/task/executor/temporalworkflow/workflow/powercontrol.go b/flow/internal/task/executor/temporalworkflow/workflow/powercontrol.go index c2f5ae2bc..c97646ccb 100644 --- a/flow/internal/task/executor/temporalworkflow/workflow/powercontrol.go +++ b/flow/internal/task/executor/temporalworkflow/workflow/powercontrol.go @@ -64,12 +64,13 @@ func powerControl( typeToTargets := buildTargets(&reqInfo) - err := executeRuleBasedOperation( + report, err := executeRuleBasedOperation( ctx, + reqInfo.TaskID, typeToTargets, info, reqInfo.RuleDefinition, ) - return updateFinishedTaskStatus(ctx, reqInfo.TaskID, err) + return updateFinishedTaskStatus(ctx, reqInfo.TaskID, err, report) } diff --git a/flow/internal/task/executor/temporalworkflow/workflow/powercontrol_action_test.go b/flow/internal/task/executor/temporalworkflow/workflow/powercontrol_action_test.go index fde30044d..2c58c549c 100644 --- a/flow/internal/task/executor/temporalworkflow/workflow/powercontrol_action_test.go +++ b/flow/internal/task/executor/temporalworkflow/workflow/powercontrol_action_test.go @@ -114,8 +114,7 @@ func TestPowerControlWorkflow_GracefulWithVerification(t *testing.T) { activity.RegisterOptions{Name: activitypkg.NamePowerControl}) env.RegisterActivityWithOptions(mockGetPowerStatus, activity.RegisterOptions{Name: activitypkg.NameGetPowerStatus}) - env.RegisterActivityWithOptions(mockUpdateTaskStatus, - activity.RegisterOptions{Name: activitypkg.NameUpdateTaskStatus}) + registerTaskUpdateActivities(env) // Mock activity responses env.OnActivity(mockPowerControl, mock.Anything, mock.Anything, @@ -131,9 +130,6 @@ func TestPowerControlWorkflow_GracefulWithVerification(t *testing.T) { "ext-compute-1": expectedStatus, }, nil) - env.OnActivity(mockUpdateTaskStatus, mock.Anything, - mock.Anything).Return(nil) - // Create test components components := []*component.Component{ newTestComponent(uuid.New(), "compute-1", "ext-compute-1", @@ -154,6 +150,7 @@ func TestPowerControlWorkflow_GracefulWithVerification(t *testing.T) { env.RegisterWorkflowWithOptions(genericComponentStepWorkflow, temporalworkflow.RegisterOptions{Name: nameGenericComponentStepWorkflow}) // Execute workflow + expectTaskUpdateActivities(env) env.ExecuteWorkflow(powerControl, reqInfo, info) // Verify workflow completed successfully @@ -224,8 +221,7 @@ func TestPowerControlWorkflow_ForcefulWithFinalVerification(t *testing.T) { activity.RegisterOptions{Name: activitypkg.NamePowerControl}) env.RegisterActivityWithOptions(mockGetPowerStatus, activity.RegisterOptions{Name: activitypkg.NameGetPowerStatus}) - env.RegisterActivityWithOptions(mockUpdateTaskStatus, - activity.RegisterOptions{Name: activitypkg.NameUpdateTaskStatus}) + registerTaskUpdateActivities(env) // Mock activity responses env.OnActivity(mockPowerControl, mock.Anything, mock.Anything, @@ -234,8 +230,6 @@ func TestPowerControlWorkflow_ForcefulWithFinalVerification(t *testing.T) { mock.Anything).Return(map[string]operations.PowerStatus{ "ext-compute-1": operations.PowerStatusOn, }, nil) - env.OnActivity(mockUpdateTaskStatus, mock.Anything, - mock.Anything).Return(nil) // Create test components components := []*component.Component{ @@ -257,6 +251,7 @@ func TestPowerControlWorkflow_ForcefulWithFinalVerification(t *testing.T) { env.RegisterWorkflowWithOptions(genericComponentStepWorkflow, temporalworkflow.RegisterOptions{Name: nameGenericComponentStepWorkflow}) // Execute workflow + expectTaskUpdateActivities(env) env.ExecuteWorkflow(powerControl, reqInfo, info) // Verify workflow completed successfully @@ -290,8 +285,7 @@ func TestPowerControlWorkflow_CompositeVerification(t *testing.T) { activity.RegisterOptions{Name: activitypkg.NameGetPowerStatus}) env.RegisterActivityWithOptions(mockVerifyReachability, activity.RegisterOptions{Name: "VerifyReachability"}) - env.RegisterActivityWithOptions(mockUpdateTaskStatus, - activity.RegisterOptions{Name: activitypkg.NameUpdateTaskStatus}) + registerTaskUpdateActivities(env) // Mock activity responses env.OnActivity(mockPowerControl, mock.Anything, mock.Anything, @@ -302,8 +296,6 @@ func TestPowerControlWorkflow_CompositeVerification(t *testing.T) { }, nil) env.OnActivity(mockVerifyReachability, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) - env.OnActivity(mockUpdateTaskStatus, mock.Anything, - mock.Anything).Return(nil) // Rule with composite verification (power status + reachability) ruleDef := &operationrules.RuleDefinition{ @@ -362,6 +354,7 @@ func TestPowerControlWorkflow_CompositeVerification(t *testing.T) { env.RegisterWorkflowWithOptions(genericComponentStepWorkflow, temporalworkflow.RegisterOptions{Name: nameGenericComponentStepWorkflow}) // Execute workflow + expectTaskUpdateActivities(env) env.ExecuteWorkflow(powerControl, reqInfo, info) // Verify workflow completed successfully diff --git a/flow/internal/task/executor/temporalworkflow/workflow/powercontrol_batching_test.go b/flow/internal/task/executor/temporalworkflow/workflow/powercontrol_batching_test.go index c74282940..8422320f4 100644 --- a/flow/internal/task/executor/temporalworkflow/workflow/powercontrol_batching_test.go +++ b/flow/internal/task/executor/temporalworkflow/workflow/powercontrol_batching_test.go @@ -94,15 +94,12 @@ func TestPowerControlWorkflowWithBatching(t *testing.T) { env.RegisterActivityWithOptions(mockPowerControlWithTracking, activity.RegisterOptions{ Name: activitypkg.NamePowerControl, }) - env.RegisterActivityWithOptions(mockUpdateTaskStatus, activity.RegisterOptions{ - Name: activitypkg.NameUpdateTaskStatus, - }) + registerTaskUpdateActivities(env) env.RegisterActivityWithOptions(mockGetPowerStatus, activity.RegisterOptions{ Name: activitypkg.NameGetPowerStatus, }) env.RegisterWorkflowWithOptions(genericComponentStepWorkflow, temporalworkflow.RegisterOptions{Name: nameGenericComponentStepWorkflow}) - env.OnActivity(mockUpdateTaskStatus, mock.Anything, mock.Anything).Return(nil) env.OnActivity(mockGetPowerStatus, mock.Anything, mock.Anything).Return( func(ctx context.Context, target common.Target) (map[string]operations.PowerStatus, error) { // Return "On" status for all components @@ -123,6 +120,7 @@ func TestPowerControlWorkflowWithBatching(t *testing.T) { RuleDefinition: ruleDef, } + expectTaskUpdateActivities(env) env.ExecuteWorkflow(powerControl, reqInfo, info) assert.True(t, env.IsWorkflowCompleted()) @@ -181,15 +179,12 @@ func TestPowerControlWorkflowWithBatching(t *testing.T) { env.RegisterActivityWithOptions(mockPowerControlTypeTracking, activity.RegisterOptions{ Name: activitypkg.NamePowerControl, }) - env.RegisterActivityWithOptions(mockUpdateTaskStatus, activity.RegisterOptions{ - Name: activitypkg.NameUpdateTaskStatus, - }) + registerTaskUpdateActivities(env) env.RegisterActivityWithOptions(mockGetPowerStatus, activity.RegisterOptions{ Name: activitypkg.NameGetPowerStatus, }) env.RegisterWorkflowWithOptions(genericComponentStepWorkflow, temporalworkflow.RegisterOptions{Name: nameGenericComponentStepWorkflow}) - env.OnActivity(mockUpdateTaskStatus, mock.Anything, mock.Anything).Return(nil) env.OnActivity(mockGetPowerStatus, mock.Anything, mock.Anything).Return( func(ctx context.Context, target common.Target) (map[string]operations.PowerStatus, error) { // Return "On" status for all components @@ -212,6 +207,7 @@ func TestPowerControlWorkflowWithBatching(t *testing.T) { RuleDefinition: ruleDef, } + expectTaskUpdateActivities(env) env.ExecuteWorkflow(powerControl, reqInfo, info) assert.True(t, env.IsWorkflowCompleted()) @@ -258,16 +254,13 @@ func TestPowerControlWorkflowWithBatching(t *testing.T) { env.RegisterActivityWithOptions(mockPowerControl, activity.RegisterOptions{ Name: activitypkg.NamePowerControl, }) - env.RegisterActivityWithOptions(mockUpdateTaskStatus, activity.RegisterOptions{ - Name: activitypkg.NameUpdateTaskStatus, - }) + registerTaskUpdateActivities(env) env.RegisterActivityWithOptions(mockGetPowerStatus, activity.RegisterOptions{ Name: activitypkg.NameGetPowerStatus, }) env.RegisterWorkflowWithOptions(genericComponentStepWorkflow, temporalworkflow.RegisterOptions{Name: nameGenericComponentStepWorkflow}) env.OnActivity(mockPowerControl, mock.Anything, mock.Anything, mock.Anything).Return(nil) - env.OnActivity(mockUpdateTaskStatus, mock.Anything, mock.Anything).Return(nil) env.OnActivity(mockGetPowerStatus, mock.Anything, mock.Anything).Return( func(ctx context.Context, target common.Target) (map[string]operations.PowerStatus, error) { // Return "On" status for all components @@ -288,6 +281,7 @@ func TestPowerControlWorkflowWithBatching(t *testing.T) { RuleDefinition: ruleDef, } + expectTaskUpdateActivities(env) env.ExecuteWorkflow(powerControl, reqInfo, info) assert.True(t, env.IsWorkflowCompleted()) @@ -324,16 +318,13 @@ func TestPowerControlWorkflowWithBatching(t *testing.T) { env.RegisterActivityWithOptions(mockPowerControl, activity.RegisterOptions{ Name: activitypkg.NamePowerControl, }) - env.RegisterActivityWithOptions(mockUpdateTaskStatus, activity.RegisterOptions{ - Name: activitypkg.NameUpdateTaskStatus, - }) + registerTaskUpdateActivities(env) env.RegisterActivityWithOptions(mockGetPowerStatus, activity.RegisterOptions{ Name: activitypkg.NameGetPowerStatus, }) env.RegisterWorkflowWithOptions(genericComponentStepWorkflow, temporalworkflow.RegisterOptions{Name: nameGenericComponentStepWorkflow}) env.OnActivity(mockPowerControl, mock.Anything, mock.Anything, mock.Anything).Return(nil) - env.OnActivity(mockUpdateTaskStatus, mock.Anything, mock.Anything).Return(nil) env.OnActivity(mockGetPowerStatus, mock.Anything, mock.Anything).Return( func(ctx context.Context, target common.Target) (map[string]operations.PowerStatus, error) { // Return "On" status for all components @@ -352,6 +343,7 @@ func TestPowerControlWorkflowWithBatching(t *testing.T) { RuleDefinition: ruleDef, } + expectTaskUpdateActivities(env) env.ExecuteWorkflow(powerControl, reqInfo, info) assert.True(t, env.IsWorkflowCompleted()) diff --git a/flow/internal/task/executor/temporalworkflow/workflow/powercontrol_test.go b/flow/internal/task/executor/temporalworkflow/workflow/powercontrol_test.go index f66732601..dd85c26a7 100644 --- a/flow/internal/task/executor/temporalworkflow/workflow/powercontrol_test.go +++ b/flow/internal/task/executor/temporalworkflow/workflow/powercontrol_test.go @@ -50,10 +50,6 @@ func mockPowerControl( return nil } -func mockUpdateTaskStatus(ctx context.Context, arg *taskdef.TaskStatusUpdate) error { - return nil -} - // mockGetPowerStatus is a mock activity function for testing power status verification. // The actual return values are defined via env.OnActivity().Return() in each test case. func mockGetPowerStatus(ctx context.Context, target common.Target) (map[string]operations.PowerStatus, error) { @@ -280,12 +276,10 @@ func TestPowerControlWorkflow(t *testing.T) { testSuite := &testsuite.WorkflowTestSuite{} env := testSuite.NewTestWorkflowEnvironment() + registerTaskUpdateActivities(env) env.RegisterActivityWithOptions(mockPowerControl, activity.RegisterOptions{ Name: taskactivity.NamePowerControl, }) - env.RegisterActivityWithOptions(mockUpdateTaskStatus, activity.RegisterOptions{ - Name: taskactivity.NameUpdateTaskStatus, - }) env.RegisterActivityWithOptions(mockGetPowerStatus, activity.RegisterOptions{ Name: taskactivity.NameGetPowerStatus, }) @@ -293,7 +287,6 @@ func TestPowerControlWorkflow(t *testing.T) { env.RegisterWorkflowWithOptions(genericComponentStepWorkflow, temporalworkflow.RegisterOptions{Name: nameGenericComponentStepWorkflow}) env.OnActivity(taskactivity.NamePowerControl, mock.Anything, mock.Anything, mock.Anything).Return(tc.activityError) - env.OnActivity(taskactivity.NameUpdateTaskStatus, mock.Anything, mock.Anything).Return(nil) // Track call count for restart operations which need Off then On callCount := 0 @@ -336,6 +329,8 @@ func TestPowerControlWorkflow(t *testing.T) { }, ) + expectTaskUpdateActivities(env) + info := &operations.PowerControlTaskInfo{Operation: tc.op} reqInfo := taskdef.ExecutionInfo{ TaskID: uuid.New(), diff --git a/flow/internal/task/executor/temporalworkflow/workflow/registry_test.go b/flow/internal/task/executor/temporalworkflow/workflow/registry_test.go index 6a157b6fd..ead222c56 100644 --- a/flow/internal/task/executor/temporalworkflow/workflow/registry_test.go +++ b/flow/internal/task/executor/temporalworkflow/workflow/registry_test.go @@ -292,18 +292,16 @@ func TestRegistryNameDispatch(t *testing.T) { ) env.RegisterWorkflowWithOptions(genericComponentStepWorkflow, temporalworkflow.RegisterOptions{Name: nameGenericComponentStepWorkflow}) + registerTaskUpdateActivities(env) env.RegisterActivityWithOptions(mockPowerControl, sdkactivity.RegisterOptions{ Name: taskactivity.NamePowerControl, }) - env.RegisterActivityWithOptions(mockUpdateTaskStatus, sdkactivity.RegisterOptions{ - Name: taskactivity.NameUpdateTaskStatus, - }) env.RegisterActivityWithOptions(mockGetPowerStatus, sdkactivity.RegisterOptions{ Name: taskactivity.NameGetPowerStatus, }) env.OnActivity(taskactivity.NamePowerControl, mock.Anything, mock.Anything, mock.Anything).Return(nil) - env.OnActivity(taskactivity.NameUpdateTaskStatus, mock.Anything, mock.Anything).Return(nil) + expectTaskUpdateActivities(env) env.OnActivity(taskactivity.NameGetPowerStatus, mock.Anything, mock.Anything).Return( map[string]operations.PowerStatus{"ext-compute-1": operations.PowerStatusOn}, nil, ) diff --git a/flow/internal/task/executor/temporalworkflow/workflow/test_helpers_test.go b/flow/internal/task/executor/temporalworkflow/workflow/test_helpers_test.go new file mode 100644 index 000000000..7f01058ba --- /dev/null +++ b/flow/internal/task/executor/temporalworkflow/workflow/test_helpers_test.go @@ -0,0 +1,39 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +package workflow + +import ( + "context" + + "github.com/stretchr/testify/mock" + "go.temporal.io/sdk/activity" + "go.temporal.io/sdk/testsuite" + + taskactivity "github.com/NVIDIA/infra-controller-rest/flow/internal/task/executor/temporalworkflow/activity" + taskdef "github.com/NVIDIA/infra-controller-rest/flow/internal/task/task" +) + +func mockUpdateTaskStatus(_ context.Context, _ *taskdef.TaskStatusUpdate) error { + return nil +} + +func mockUpdateTaskReport(_ context.Context, _ *taskdef.TaskReportUpdate) error { + return nil +} + +func registerTaskUpdateActivities(env *testsuite.TestWorkflowEnvironment) { + env.RegisterActivityWithOptions(mockUpdateTaskStatus, activity.RegisterOptions{ + Name: taskactivity.NameUpdateTaskStatus, + }) + env.RegisterActivityWithOptions(mockUpdateTaskReport, activity.RegisterOptions{ + Name: taskactivity.NameUpdateTaskReport, + }) +} + +func expectTaskUpdateActivities(env *testsuite.TestWorkflowEnvironment) { + env.OnActivity(taskactivity.NameUpdateTaskStatus, mock.Anything, mock.Anything).Return(nil) + env.OnActivity(taskactivity.NameUpdateTaskReport, mock.Anything, mock.Anything).Return(nil) +} diff --git a/flow/internal/task/manager/manager.go b/flow/internal/task/manager/manager.go index d72d694db..63dfc6de3 100644 --- a/flow/internal/task/manager/manager.go +++ b/flow/internal/task/manager/manager.go @@ -31,6 +31,7 @@ import ( taskcommon "github.com/NVIDIA/infra-controller-rest/flow/internal/task/common" "github.com/NVIDIA/infra-controller-rest/flow/internal/task/conflict" "github.com/NVIDIA/infra-controller-rest/flow/internal/task/executor" + "github.com/NVIDIA/infra-controller-rest/flow/internal/task/message" "github.com/NVIDIA/infra-controller-rest/flow/internal/task/operationrules" "github.com/NVIDIA/infra-controller-rest/flow/internal/task/operations" taskstore "github.com/NVIDIA/infra-controller-rest/flow/internal/task/store" @@ -145,7 +146,7 @@ func New(ctx context.Context, conf *Config) (*ManagerImpl, error) { wrappedStore := newNotifyingTaskStore(conf.TaskStore, promoter) // Build executor — updater is passed explicitly, no global involved at this layer. - exec, err := executor.New(ctx, conf.ExecutorConfig, wrappedStore) + exec, err := executor.New(ctx, conf.ExecutorConfig, wrappedStore, wrappedStore) if err != nil { return nil, err } @@ -346,11 +347,11 @@ func (m *ManagerImpl) createAndExecuteTask( } task.Status = taskcommon.TaskStatusWaiting - task.Message = "Queued: waiting for rack to become available" + task.Message = message.ForStatus(taskcommon.TaskStatusWaiting) task.QueueExpiresAt = m.getReqExpiresAt(req) } else { task.Status = taskcommon.TaskStatusPending - task.Message = "Created" + task.Message = message.ForStatus(taskcommon.TaskStatusPending) } return m.taskStore.CreateTask(txCtx, &task) @@ -436,7 +437,7 @@ func (m *ManagerImpl) resolveAndExecuteTask( if uerr := m.taskStore.UpdateTaskStatus(ctx, &taskdef.TaskStatusUpdate{ ID: task.ID, Status: taskcommon.TaskStatusFailed, - Message: err.Error(), + Message: message.ForFailure(err), }); uerr != nil { log.Error().Err(uerr). Msgf("failed to mark task %s failed", task.ID) diff --git a/flow/internal/task/manager/notifying_store.go b/flow/internal/task/manager/notifying_store.go index fc73c377a..08b5b3318 100644 --- a/flow/internal/task/manager/notifying_store.go +++ b/flow/internal/task/manager/notifying_store.go @@ -73,6 +73,16 @@ func (s *notifyingTaskStore) UpdateTaskStatus( return nil } +// UpdateTaskReport delegates to the underlying store. +func (s *notifyingTaskStore) UpdateTaskReport( + ctx context.Context, + u *taskdef.TaskReportUpdate, +) error { + return s.Store.UpdateTaskReport(ctx, u) +} + // Ensure notifyingTaskStore satisfies taskdef.TaskStatusUpdater so it can // be passed to activity.SetTaskStatusUpdater. var _ taskdef.TaskStatusUpdater = (*notifyingTaskStore)(nil) //nolint + +var _ taskdef.TaskReportUpdater = (*notifyingTaskStore)(nil) //nolint diff --git a/flow/internal/task/message/message.go b/flow/internal/task/message/message.go new file mode 100644 index 000000000..4fdbb9855 --- /dev/null +++ b/flow/internal/task/message/message.go @@ -0,0 +1,70 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package message + +import ( + "strings" + + taskcommon "github.com/NVIDIA/infra-controller-rest/flow/internal/task/common" +) + +const maxLen = 512 + +// ForStatus returns the default message for a status transition. Callers may +// override Waiting / Terminated with more specific text before persisting. +func ForStatus(status taskcommon.TaskStatus) string { + switch status { + case taskcommon.TaskStatusWaiting: + return "Queued: waiting for rack to become available" + case taskcommon.TaskStatusPending: + return "Pending" + case taskcommon.TaskStatusRunning: + return "Running" + case taskcommon.TaskStatusCompleted: + return "Succeeded" + case taskcommon.TaskStatusFailed: + return "Failed" + case taskcommon.TaskStatusTerminated: + return "Terminated" + default: + return "" + } +} + +// ForFailure returns a one-line failure summary suitable for the message field. +func ForFailure(err error) string { + if err == nil { + return ForStatus(taskcommon.TaskStatusFailed) + } + msg := strings.TrimSpace(err.Error()) + if msg == "" { + return ForStatus(taskcommon.TaskStatusFailed) + } + idx := strings.IndexByte(msg, '\n') + if idx >= 0 { + msg = strings.TrimSpace(msg[:idx]) + } + return truncate(msg) +} + +func truncate(msg string) string { + if len(msg) <= maxLen { + return msg + } + return msg[:maxLen] +} diff --git a/flow/internal/task/message/message_test.go b/flow/internal/task/message/message_test.go new file mode 100644 index 000000000..3a6f2fbad --- /dev/null +++ b/flow/internal/task/message/message_test.go @@ -0,0 +1,30 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +package message + +import ( + "errors" + "testing" + + "github.com/stretchr/testify/assert" + + taskcommon "github.com/NVIDIA/infra-controller-rest/flow/internal/task/common" +) + +func TestForStatus(t *testing.T) { + t.Parallel() + assert.Equal(t, "Pending", ForStatus(taskcommon.TaskStatusPending)) + assert.Equal(t, "Running", ForStatus(taskcommon.TaskStatusRunning)) + assert.Equal(t, "Succeeded", ForStatus(taskcommon.TaskStatusCompleted)) +} + +func TestForFailure(t *testing.T) { + t.Parallel() + assert.Equal(t, "disk full", ForFailure(errors.New("disk full"))) + assert.Equal(t, "stage failed", ForFailure( + errors.New("stage failed\ncomponent type Compute failed"), + )) +} diff --git a/flow/internal/task/operationrules/rules.go b/flow/internal/task/operationrules/rules.go index 2cefe27de..ee457bd0b 100644 --- a/flow/internal/task/operationrules/rules.go +++ b/flow/internal/task/operationrules/rules.go @@ -158,6 +158,14 @@ func (si *StageIterator) HasNext() bool { return si != nil && si.position < len(si.stages) } +// Total returns the number of stages in the rule definition. +func (si *StageIterator) Total() int { + if si == nil { + return 0 + } + return len(si.stages) +} + // Reset resets the iterator to the beginning // Allows re-iteration without creating a new iterator func (si *StageIterator) Reset() { diff --git a/flow/internal/task/operations/summary.go b/flow/internal/task/operations/summary.go new file mode 100644 index 000000000..eb9e833d1 --- /dev/null +++ b/flow/internal/task/operations/summary.go @@ -0,0 +1,138 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package operations + +import ( + "encoding/json" + "fmt" + "strings" + + "github.com/NVIDIA/infra-controller-rest/flow/internal/operation" + taskcommon "github.com/NVIDIA/infra-controller-rest/flow/internal/task/common" +) + +// SummaryFromWrapper returns a short English phrase describing the operation +// and its key parameters (e.g. "Power Off (forced)", "Upgrade Firmware to v2.1.0"). +func SummaryFromWrapper(w operation.Wrapper) (string, error) { + switch w.Type { + case taskcommon.TaskTypePowerControl: + switch w.Code { + case taskcommon.OpCodePowerControlPowerOn, taskcommon.OpCodePowerControlForcePowerOn: + return "Power On", nil + case taskcommon.OpCodePowerControlPowerOff: + return "Power Off", nil + case taskcommon.OpCodePowerControlForcePowerOff: + return "Power Off (forced)", nil + case taskcommon.OpCodePowerControlRestart, taskcommon.OpCodePowerControlWarmReset: + return "Power Reset", nil + case taskcommon.OpCodePowerControlForceRestart, taskcommon.OpCodePowerControlColdReset: + return "Power Reset (forced)", nil + default: + if w.Code != "" { + return humanizeCode(w.Code), nil + } + return "Power Control", nil + } + + case taskcommon.TaskTypeBringUp: + if w.Code == taskcommon.OpCodeIngest { + return "Ingest", nil + } + return "Bring Up", nil + + case taskcommon.TaskTypeFirmwareControl: + var info struct { + TargetVersion string `json:"target_version"` + } + if len(w.Info) > 0 { + err := json.Unmarshal(w.Info, &info) + if err != nil { + return "", fmt.Errorf("unmarshal firmware info: %w", err) + } + } + switch w.Code { + case taskcommon.OpCodeFirmwareControlUpgrade: + if info.TargetVersion != "" { + return "Upgrade Firmware to " + info.TargetVersion, nil + } + return "Upgrade Firmware", nil + case taskcommon.OpCodeFirmwareControlDowngrade: + if info.TargetVersion != "" { + return "Downgrade Firmware to " + info.TargetVersion, nil + } + return "Downgrade Firmware", nil + case taskcommon.OpCodeFirmwareControlRollback: + return "Rollback Firmware", nil + default: + if w.Code != "" { + return humanizeCode(w.Code), nil + } + return "Firmware Control", nil + } + + case taskcommon.TaskTypeInjectExpectation: + return "Inject Expectation", nil + + default: + if w.Code != "" { + return humanizeCode(w.Code), nil + } + return string(w.Type), nil + } +} + +// OperationTypeFromWrapper returns a stable SCREAMING_SNAKE_CASE operation type +// string suitable for schedule filtering (e.g. "POWER_ON", "BRING_UP"). +func OperationTypeFromWrapper(w operation.Wrapper) string { + switch w.Type { + case taskcommon.TaskTypePowerControl: + switch w.Code { + case taskcommon.OpCodePowerControlPowerOn, taskcommon.OpCodePowerControlForcePowerOn: + return "POWER_ON" + case taskcommon.OpCodePowerControlPowerOff, taskcommon.OpCodePowerControlForcePowerOff: + return "POWER_OFF" + case taskcommon.OpCodePowerControlRestart, taskcommon.OpCodePowerControlWarmReset, + taskcommon.OpCodePowerControlForceRestart, taskcommon.OpCodePowerControlColdReset: + return "POWER_RESET" + default: + return "POWER_CONTROL" + } + case taskcommon.TaskTypeBringUp: + if w.Code == taskcommon.OpCodeIngest { + return "INGEST" + } + return "BRING_UP" + case taskcommon.TaskTypeFirmwareControl: + switch w.Code { + case taskcommon.OpCodeFirmwareControlUpgrade: + return "UPGRADE_FIRMWARE" + case taskcommon.OpCodeFirmwareControlDowngrade: + return "DOWNGRADE_FIRMWARE" + case taskcommon.OpCodeFirmwareControlRollback: + return "ROLLBACK_FIRMWARE" + default: + return "FIRMWARE_CONTROL" + } + default: + return strings.ToUpper(string(w.Type)) + } +} + +func humanizeCode(code string) string { + return strings.ReplaceAll(strings.ReplaceAll(code, "_", " "), "-", " ") +} diff --git a/flow/internal/task/operations/summary_test.go b/flow/internal/task/operations/summary_test.go new file mode 100644 index 000000000..a55427b5c --- /dev/null +++ b/flow/internal/task/operations/summary_test.go @@ -0,0 +1,82 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package operations + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/NVIDIA/infra-controller-rest/flow/internal/operation" + taskcommon "github.com/NVIDIA/infra-controller-rest/flow/internal/task/common" +) + +func TestSummaryFromWrapper(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + wrapper operation.Wrapper + want string + }{ + { + name: "power on", + wrapper: operation.Wrapper{ + Type: taskcommon.TaskTypePowerControl, + Code: taskcommon.OpCodePowerControlPowerOn, + }, + want: "Power On", + }, + { + name: "forced power off", + wrapper: operation.Wrapper{ + Type: taskcommon.TaskTypePowerControl, + Code: taskcommon.OpCodePowerControlForcePowerOff, + }, + want: "Power Off (forced)", + }, + { + name: "firmware upgrade with version", + wrapper: operation.Wrapper{ + Type: taskcommon.TaskTypeFirmwareControl, + Code: taskcommon.OpCodeFirmwareControlUpgrade, + Info: json.RawMessage(`{"target_version":"2.3.1"}`), + }, + want: "Upgrade Firmware to 2.3.1", + }, + { + name: "ingest", + wrapper: operation.Wrapper{ + Type: taskcommon.TaskTypeBringUp, + Code: taskcommon.OpCodeIngest, + }, + want: "Ingest", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + got, err := SummaryFromWrapper(tt.wrapper) + require.NoError(t, err) + assert.Equal(t, tt.want, got) + }) + } +} diff --git a/flow/internal/task/report/report.go b/flow/internal/task/report/report.go new file mode 100644 index 000000000..baff5b991 --- /dev/null +++ b/flow/internal/task/report/report.go @@ -0,0 +1,329 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +// Package report defines the JSON document persisted in task.report and +// the Tracker that mutates it as a rule-based workflow advances. +// +// The document mirrors the structure of the operation rule that drives +// the workflow. NewInitial expands a RuleDefinition through +// operationrules.NewStageIterator: every emitted operationrules.Stage +// becomes one StageRecord, and every SequenceStep within that stage +// becomes one StepRecord at the same index. StageRecord.Number equals the +// rule stage number and is the canonical key for joining a record back to +// its rule entry. +package report + +import ( + "encoding/json" + "fmt" + "time" + + "github.com/NVIDIA/infra-controller-rest/flow/internal/task/operationrules" + "github.com/NVIDIA/infra-controller-rest/flow/pkg/common/devicetypes" +) + +// Version is the report document schema version. +const Version = 1 + +// maxErrLen caps the length of error strings written into the report so +// a single failure cannot inflate the JSONB payload. +const maxErrLen = 512 + +// Status enumerates the per-stage and per-step state values. It is the +// only set of strings that appears in Status fields on the wire. +type Status string + +const ( + // StatusPending - the workflow has not yet reached this stage/step. + StatusPending Status = "pending" + // StatusRunning - execution is in progress. + StatusRunning Status = "running" + // StatusCompleted - execution finished successfully. + StatusCompleted Status = "completed" + // StatusFailed - execution finished with an error. + StatusFailed Status = "failed" + // StatusSkipped - the rule lists this step's ComponentType but the + // task targets no components of that type, so the workflow will not + // invoke it. + StatusSkipped Status = "skipped" +) + +// Report is the JSON document stored in task.report. +type Report struct { + Version int `json:"version"` + Stages []StageRecord `json:"stages"` + // Error is set to the failure summary of the first stage that fails + // in this report. It is the canonical task-level error text and is + // not overwritten by subsequent stage failures. + Error string `json:"error,omitempty"` +} + +// StageRecord captures the execution state of one rule stage (i.e. one +// operationrules.Stage produced by NewStageIterator). Number matches the +// rule stage number; Steps preserves the rule's SequenceStep order +// within the stage. +type StageRecord struct { + Number int `json:"number"` + Status Status `json:"status"` + Steps []StepRecord `json:"steps"` + StartedAt string `json:"started_at,omitempty"` + FinishedAt string `json:"finished_at,omitempty"` + // Error is set when Status == StatusFailed. Empty otherwise. + Error string `json:"error,omitempty"` +} + +// StepRecord captures the execution state of one rule SequenceStep. It +// pairs 1:1 with operationrules.SequenceStep within the containing stage +// and shares its index. +// +// Per-component counters describe components of ComponentType targeted +// by this step in the current task. Pending and processed counts are +// derivable on the client (Pending = TotalComponents - +// CompletedComponents - FailedComponents; processed = CompletedComponents +// + FailedComponents) and are not stored. +type StepRecord struct { + // ComponentType mirrors operationrules.SequenceStep.ComponentType. + ComponentType string `json:"component_type"` + Status Status `json:"status"` + + // TotalComponents is the count of components of ComponentType this + // step targets. Populated once at NewInitial from + // task.attributes.components_by_type. Carried in the report because + // the API task representation does not surface that map; without + // this field a client cannot size the work the step performs. + TotalComponents int `json:"total_components,omitempty"` + + // CompletedComponents and FailedComponents are reserved on the wire + // for a future best-effort activity contract that reports + // per-component outcomes. The current fail-fast contract surfaces + // only stage-level success or failure, so neither field is written + // today and both are omitted from the JSON payload. + CompletedComponents int `json:"completed_components,omitempty"` + FailedComponents int `json:"failed_components,omitempty"` + + // StartedAt is set when the step transitions out of StatusPending; + // FinishedAt is set when it transitions to a terminal state + // (StatusCompleted or StatusFailed). StatusSkipped steps carry + // neither timestamp. + StartedAt string `json:"started_at,omitempty"` + FinishedAt string `json:"finished_at,omitempty"` + + // Error carries the failure summary when Status == StatusFailed. + Error string `json:"error,omitempty"` +} + +// NewInitial builds a v1 report whose Stages mirror the rule definition +// stage-by-stage and step-by-step. +// +// Steps whose ComponentType has zero entries in totalByType (or is +// absent) are pre-marked StatusSkipped because the workflow will not +// invoke them. All other steps and every stage start as StatusPending. +// +// Returns a Version-stamped report with empty Stages when ruleDef is nil +// or has no steps. +func NewInitial( + ruleDef *operationrules.RuleDefinition, + totalByType map[devicetypes.ComponentType]int, +) *Report { + rep := &Report{Version: Version} + if ruleDef == nil { + return rep + } + + iter := operationrules.NewStageIterator(ruleDef) + rep.Stages = make([]StageRecord, 0, iter.Total()) + for stage := iter.Next(); stage != nil; stage = iter.Next() { + stageRec := StageRecord{ + Number: stage.Number, + Status: StatusPending, + Steps: make([]StepRecord, 0, len(stage.Steps)), + } + for _, step := range stage.Steps { + total := totalByType[step.ComponentType] + status := StatusPending + if total == 0 { + status = StatusSkipped + } + stageRec.Steps = append(stageRec.Steps, StepRecord{ + ComponentType: devicetypes.ComponentTypeToString(step.ComponentType), + Status: status, + TotalComponents: total, + }) + } + rep.Stages = append(rep.Stages, stageRec) + } + + return rep +} + +// MarshalJSON returns the JSON encoding for persistence. A nil receiver +// encodes as JSON null. +func (r *Report) MarshalJSON() ([]byte, error) { + if r == nil { + return []byte("null"), nil + } + if r.Version == 0 { + r.Version = Version + } + type alias Report + return json.Marshal((*alias)(r)) +} + +// MarshalRaw returns json.RawMessage suitable for direct insertion into +// the task.report JSONB column. +func (r *Report) MarshalRaw() (json.RawMessage, error) { + if r == nil { + return nil, nil + } + b, err := r.MarshalJSON() + if err != nil { + return nil, err + } + return json.RawMessage(b), nil +} + +// Unmarshal decodes a stored report document. Empty input returns a nil +// report without error. +func Unmarshal(data json.RawMessage) (*Report, error) { + if len(data) == 0 { + return nil, nil + } + var r Report + err := json.Unmarshal(data, &r) + if err != nil { + return nil, fmt.Errorf("unmarshal task report: %w", err) + } + return &r, nil +} + +// Tracker mutates a Report through stage execution. All methods are +// workflow-safe: pure in-memory updates, no I/O, deterministic given +// identical inputs. Callers serialize the Report after each transition +// and persist the snapshot via the UpdateTaskReport activity. +type Tracker struct { + Report *Report +} + +// BeginStage transitions the stage at stageNum to StatusRunning. Each of +// the stage's StatusPending steps is advanced to StatusRunning at the +// same time; StatusSkipped steps are left untouched. Calls on an unknown +// stageNum or a stage already past StatusPending are a no-op so +// double-fired transitions stay idempotent. +func (t *Tracker) BeginStage(stageNum int, now time.Time) { + if t == nil || t.Report == nil { + return + } + stage := t.findStage(stageNum) + if stage == nil || stage.Status != StatusPending { + return + } + + started := formatTime(now) + stage.Status = StatusRunning + stage.StartedAt = started + for i := range stage.Steps { + if stage.Steps[i].Status == StatusPending { + stage.Steps[i].Status = StatusRunning + stage.Steps[i].StartedAt = started + } + } +} + +// CompleteStage transitions the stage at stageNum to StatusCompleted. +// Every step still in StatusRunning is finalized to StatusCompleted with +// the same FinishedAt; under the fail-fast activity contract a completed +// stage implies all its non-skipped steps succeeded. StatusSkipped steps +// and steps already in a terminal state are left untouched. +func (t *Tracker) CompleteStage(stageNum int, now time.Time) { + if t == nil || t.Report == nil { + return + } + stage := t.findStage(stageNum) + if stage == nil { + return + } + + finished := formatTime(now) + stage.Status = StatusCompleted + stage.FinishedAt = finished + for i := range stage.Steps { + if stage.Steps[i].Status == StatusRunning { + stage.Steps[i].Status = StatusCompleted + stage.Steps[i].FinishedAt = finished + } + } +} + +// FailStage transitions the stage at stageNum to StatusFailed and +// propagates the failure to every step still in StatusRunning. The +// fail-fast activity contract aborts the stage on the first failing +// child and discards any sibling outcomes, so all running steps are +// marked StatusFailed with the same Error: their actual results are +// unobserved and reporting any of them as completed would be a guess. +// +// Report.Error is set to the failure summary the first time a stage +// fails in this report; subsequent FailStage calls leave it unchanged so +// the first failure remains the canonical task-level error. +func (t *Tracker) FailStage(stageNum int, stageErr error, now time.Time) { + if t == nil || t.Report == nil { + return + } + stage := t.findStage(stageNum) + if stage == nil { + return + } + + msg := truncateErr(stageErr) + finished := formatTime(now) + stage.Status = StatusFailed + stage.FinishedAt = finished + stage.Error = msg + for i := range stage.Steps { + if stage.Steps[i].Status == StatusRunning { + stage.Steps[i].Status = StatusFailed + stage.Steps[i].FinishedAt = finished + stage.Steps[i].Error = msg + } + } + if t.Report.Error == "" { + t.Report.Error = msg + } +} + +func (t *Tracker) findStage(stageNum int) *StageRecord { + for i := range t.Report.Stages { + if t.Report.Stages[i].Number == stageNum { + return &t.Report.Stages[i] + } + } + return nil +} + +func formatTime(now time.Time) string { + return now.UTC().Format(time.RFC3339) +} + +func truncateErr(err error) string { + if err == nil { + return "" + } + msg := err.Error() + if len(msg) > maxErrLen { + msg = msg[:maxErrLen] + } + return msg +} diff --git a/flow/internal/task/report/report_test.go b/flow/internal/task/report/report_test.go new file mode 100644 index 000000000..8c6d1a0d2 --- /dev/null +++ b/flow/internal/task/report/report_test.go @@ -0,0 +1,224 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +package report + +import ( + "errors" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/NVIDIA/infra-controller-rest/flow/internal/task/operationrules" + "github.com/NVIDIA/infra-controller-rest/flow/pkg/common/devicetypes" +) + +// twoStageRule mirrors a force-power-on-style plan: stage 1 powers up +// PowerShelves, stage 2 brings up Compute. It exists to anchor tests +// against a stable rule shape without pulling in the full resolver. +func twoStageRule() *operationrules.RuleDefinition { + return &operationrules.RuleDefinition{ + Version: "v1", + Steps: []operationrules.SequenceStep{ + {ComponentType: devicetypes.ComponentTypePowerShelf, Stage: 1}, + {ComponentType: devicetypes.ComponentTypeCompute, Stage: 2}, + }, + } +} + +func TestNewInitial_mirrorsRuleAndMarksSkipped(t *testing.T) { + t.Parallel() + + // totalByType lists Compute only; PowerShelf step has no targets and + // must be marked skipped. + totals := map[devicetypes.ComponentType]int{ + devicetypes.ComponentTypeCompute: 4, + } + + rep := NewInitial(twoStageRule(), totals) + + require.Equal(t, Version, rep.Version) + require.Len(t, rep.Stages, 2) + + stage1 := rep.Stages[0] + assert.Equal(t, 1, stage1.Number) + assert.Equal(t, StatusPending, stage1.Status) + require.Len(t, stage1.Steps, 1) + assert.Equal(t, "PowerShelf", stage1.Steps[0].ComponentType) + assert.Equal(t, StatusSkipped, stage1.Steps[0].Status, + "step with zero targets must start skipped") + assert.Zero(t, stage1.Steps[0].TotalComponents) + + stage2 := rep.Stages[1] + assert.Equal(t, 2, stage2.Number) + require.Len(t, stage2.Steps, 1) + assert.Equal(t, "Compute", stage2.Steps[0].ComponentType) + assert.Equal(t, StatusPending, stage2.Steps[0].Status) + assert.Equal(t, 4, stage2.Steps[0].TotalComponents) +} + +func TestNewInitial_nilRule(t *testing.T) { + t.Parallel() + + rep := NewInitial(nil, nil) + + require.Equal(t, Version, rep.Version) + assert.Empty(t, rep.Stages) +} + +func TestTracker_BeginStage_advancesPendingOnly(t *testing.T) { + t.Parallel() + + totals := map[devicetypes.ComponentType]int{ + devicetypes.ComponentTypeCompute: 2, + } + tr := &Tracker{Report: NewInitial(twoStageRule(), totals)} + now := time.Date(2026, 5, 28, 12, 0, 0, 0, time.UTC) + + tr.BeginStage(1, now) + stage1 := tr.Report.Stages[0] + assert.Equal(t, StatusRunning, stage1.Status) + assert.Equal(t, "2026-05-28T12:00:00Z", stage1.StartedAt) + // Stage 1's only step targets PowerShelf which has zero components, + // so it must remain skipped after BeginStage. + assert.Equal(t, StatusSkipped, stage1.Steps[0].Status) + assert.Empty(t, stage1.Steps[0].StartedAt) + + // BeginStage on a stage that is already running is a no-op so a + // retried Temporal activity does not rewrite timestamps. + tr.BeginStage(1, now.Add(time.Hour)) + assert.Equal(t, "2026-05-28T12:00:00Z", tr.Report.Stages[0].StartedAt) +} + +func TestTracker_CompleteStage_finalizesRunningSteps(t *testing.T) { + t.Parallel() + + totals := map[devicetypes.ComponentType]int{ + devicetypes.ComponentTypeCompute: 3, + } + tr := &Tracker{Report: NewInitial(twoStageRule(), totals)} + startedAt := time.Date(2026, 5, 28, 12, 0, 0, 0, time.UTC) + finishedAt := startedAt.Add(30 * time.Second) + + tr.BeginStage(2, startedAt) + tr.CompleteStage(2, finishedAt) + + stage2 := tr.Report.Stages[1] + assert.Equal(t, StatusCompleted, stage2.Status) + assert.Equal(t, "2026-05-28T12:00:30Z", stage2.FinishedAt) + + step := stage2.Steps[0] + assert.Equal(t, StatusCompleted, step.Status) + assert.Equal(t, "2026-05-28T12:00:00Z", step.StartedAt) + assert.Equal(t, "2026-05-28T12:00:30Z", step.FinishedAt) + assert.Empty(t, step.Error) +} + +func TestTracker_FailStage_propagatesToRunningStepsAndReportError(t *testing.T) { + t.Parallel() + + totals := map[devicetypes.ComponentType]int{ + devicetypes.ComponentTypeCompute: 2, + devicetypes.ComponentTypePowerShelf: 1, + } + tr := &Tracker{Report: NewInitial(twoStageRule(), totals)} + now := time.Date(2026, 5, 28, 12, 0, 0, 0, time.UTC) + + tr.BeginStage(1, now) + failErr := errors.New("BMC unreachable") + tr.FailStage(1, failErr, now.Add(15*time.Second)) + + stage1 := tr.Report.Stages[0] + assert.Equal(t, StatusFailed, stage1.Status) + assert.Equal(t, "BMC unreachable", stage1.Error) + assert.Equal(t, "2026-05-28T12:00:15Z", stage1.FinishedAt) + + // Under fail-fast the running PowerShelf step must inherit the + // failure so the wire shape never reports it as still in flight. + step := stage1.Steps[0] + assert.Equal(t, StatusFailed, step.Status) + assert.Equal(t, "BMC unreachable", step.Error) + assert.Equal(t, "2026-05-28T12:00:15Z", step.FinishedAt) + + assert.Equal(t, "BMC unreachable", tr.Report.Error, + "first stage failure becomes the canonical task-level error") + + // A second failure must not overwrite the canonical task-level + // error: the first cause is what callers diagnose against. + tr.BeginStage(2, now.Add(time.Minute)) + tr.FailStage(2, errors.New("compute boot timed out"), now.Add(2*time.Minute)) + assert.Equal(t, "BMC unreachable", tr.Report.Error) +} + +func TestTracker_FailStage_truncatesLongError(t *testing.T) { + t.Parallel() + + totals := map[devicetypes.ComponentType]int{ + devicetypes.ComponentTypeCompute: 1, + } + tr := &Tracker{Report: NewInitial(twoStageRule(), totals)} + now := time.Date(2026, 5, 28, 12, 0, 0, 0, time.UTC) + + huge := strings.Repeat("x", maxErrLen+50) + tr.BeginStage(2, now) + tr.FailStage(2, errors.New(huge), now.Add(time.Second)) + + assert.Len(t, tr.Report.Stages[1].Error, maxErrLen) + assert.Len(t, tr.Report.Error, maxErrLen) +} + +func TestTracker_skippedStepsAreNotMutated(t *testing.T) { + t.Parallel() + + // PowerShelf has no targets so its step starts skipped; the test + // asserts every Tracker transition leaves it skipped. + totals := map[devicetypes.ComponentType]int{ + devicetypes.ComponentTypeCompute: 1, + } + tr := &Tracker{Report: NewInitial(twoStageRule(), totals)} + now := time.Date(2026, 5, 28, 12, 0, 0, 0, time.UTC) + + tr.BeginStage(1, now) + tr.CompleteStage(1, now.Add(time.Second)) + + stage1 := tr.Report.Stages[0] + assert.Equal(t, StatusCompleted, stage1.Status) + assert.Equal(t, StatusSkipped, stage1.Steps[0].Status) + assert.Empty(t, stage1.Steps[0].StartedAt) + assert.Empty(t, stage1.Steps[0].FinishedAt) +} + +func TestReport_RoundTrip(t *testing.T) { + t.Parallel() + + totals := map[devicetypes.ComponentType]int{ + devicetypes.ComponentTypeCompute: 2, + devicetypes.ComponentTypePowerShelf: 0, + } + original := NewInitial(twoStageRule(), totals) + tr := &Tracker{Report: original} + tr.BeginStage(1, time.Unix(100, 0)) + tr.CompleteStage(1, time.Unix(150, 0)) + tr.BeginStage(2, time.Unix(160, 0)) + tr.FailStage(2, errors.New("driver crashed"), time.Unix(200, 0)) + + raw, err := original.MarshalRaw() + require.NoError(t, err) + + got, err := Unmarshal(raw) + require.NoError(t, err) + require.NotNil(t, got) + + assert.Equal(t, original.Version, got.Version) + assert.Equal(t, original.Error, got.Error) + assert.Equal(t, len(original.Stages), len(got.Stages)) + for i := range original.Stages { + assert.Equal(t, original.Stages[i].Status, got.Stages[i].Status) + assert.Equal(t, len(original.Stages[i].Steps), len(got.Stages[i].Steps)) + } +} diff --git a/flow/internal/task/store/postgres.go b/flow/internal/task/store/postgres.go index d1403b485..53e7acad0 100644 --- a/flow/internal/task/store/postgres.go +++ b/flow/internal/task/store/postgres.go @@ -152,16 +152,38 @@ func (s *PostgresStore) UpdateScheduledTask( return nil } -// UpdateTaskStatus updates the status and message of a task. +// UpdateTaskStatus persists status, message, and (optionally) the report +// snapshot. The report carried in arg is treated as authoritative: when +// non-empty it replaces the stored document, when empty the stored +// document is left untouched (the underlying model omits the report +// column from the UPDATE in that case). No read-modify-write is performed, +// so concurrent transitions cannot lose updates. func (s *PostgresStore) UpdateTaskStatus( ctx context.Context, arg *taskdef.TaskStatusUpdate, ) error { - taskDao := &model.Task{ - ID: arg.ID, + taskDao := &model.Task{ID: arg.ID} + err := taskDao.UpdateTaskStatus(ctx, s.idb(ctx), arg.Status, arg.Message, arg.Report) + if err != nil { + return errors.GRPCErrorInternal(err.Error()) + } + + return nil +} + +// UpdateTaskReport replaces the stored report with the snapshot in arg. +// Empty snapshots are dropped rather than written so a malformed caller +// cannot clear the stored report by accident. +func (s *PostgresStore) UpdateTaskReport( + ctx context.Context, + arg *taskdef.TaskReportUpdate, +) error { + if len(arg.Report) == 0 { + return nil } - err := taskDao.UpdateTaskStatus(ctx, s.pg.DB, arg.Status, arg.Message) + taskDao := &model.Task{ID: arg.ID} + err := taskDao.UpdateTaskReport(ctx, s.idb(ctx), arg.Report) if err != nil { return errors.GRPCErrorInternal(err.Error()) } diff --git a/flow/internal/task/store/store.go b/flow/internal/task/store/store.go index 0b931f61a..db7f477d6 100644 --- a/flow/internal/task/store/store.go +++ b/flow/internal/task/store/store.go @@ -59,6 +59,9 @@ type Store interface { // UpdateTaskStatus updates the status and message of a task. UpdateTaskStatus(ctx context.Context, arg *taskdef.TaskStatusUpdate) error + // UpdateTaskReport merges a report snapshot without a status change. + UpdateTaskReport(ctx context.Context, arg *taskdef.TaskReportUpdate) error + // ListActiveTasksForRack returns non-finished, non-waiting tasks for a rack // (i.e. tasks with status pending or running). ListActiveTasksForRack(ctx context.Context, rackID uuid.UUID) ([]*taskdef.Task, error) diff --git a/flow/internal/task/task/task.go b/flow/internal/task/task/task.go index c630ecf9f..431bc630d 100644 --- a/flow/internal/task/task/task.go +++ b/flow/internal/task/task/task.go @@ -36,11 +36,12 @@ import ( // -- Operation: The operation to be performed by the task. // -- RackID: The rack this task operates on (1 task = 1 rack). // -- Attributes: Flexible metadata including targeted components by type. -// -- Description: The description of the task provided by the user. +// -- Description: User-provided description at task creation. // -- ExecutorType: The type of executor to be used for the task. // -- ExecutionID: The identifier of the execution of the task. // -- Status: The status of the task. -// -- Message: Status message or error details. +// -- Message: Brief text tied to status (not execution progress). +// -- Report: Structured JSON progress document (task.report). // -- AppliedRuleID: The ID of the operation rule that was applied (if any). type Task struct { ID uuid.UUID @@ -52,6 +53,7 @@ type Task struct { ExecutionID string Status taskcommon.TaskStatus Message string + Report json.RawMessage AppliedRuleID *uuid.UUID // The ID of the operation rule that was applied CreatedAt time.Time UpdatedAt time.Time @@ -149,6 +151,17 @@ type TaskStatusUpdate struct { ID uuid.UUID Status taskcommon.TaskStatus Message string + // Report, when non-empty, replaces the stored report document. An + // empty value leaves the stored report untouched. + Report json.RawMessage +} + +// TaskReportUpdate replaces the stored report with the supplied snapshot +// without changing status or message. Empty snapshots are dropped to +// avoid clearing the stored report by accident. +type TaskReportUpdate struct { + ID uuid.UUID + Report json.RawMessage } // TaskStatusUpdater is implemented by any store that can persist task status changes. @@ -156,3 +169,8 @@ type TaskStatusUpdater interface { // UpdateTaskStatus persists the status change described by arg. UpdateTaskStatus(ctx context.Context, arg *TaskStatusUpdate) error } + +// TaskReportUpdater persists in-flight report snapshots (best-effort). +type TaskReportUpdater interface { + UpdateTaskReport(ctx context.Context, arg *TaskReportUpdate) error +} diff --git a/flow/pkg/proto/v1/flow.pb.go b/flow/pkg/proto/v1/flow.pb.go index f1c340f72..2dd08765e 100644 --- a/flow/pkg/proto/v1/flow.pb.go +++ b/flow/pkg/proto/v1/flow.pb.go @@ -2183,11 +2183,13 @@ type Task struct { Operation string `protobuf:"bytes,2,opt,name=operation,proto3" json:"operation,omitempty"` RackId *UUID `protobuf:"bytes,3,opt,name=rack_id,json=rackId,proto3" json:"rack_id,omitempty"` ComponentUuids []*UUID `protobuf:"bytes,4,rep,name=component_uuids,json=componentUuids,proto3" json:"component_uuids,omitempty"` - Description string `protobuf:"bytes,5,opt,name=description,proto3" json:"description,omitempty"` - ExecutorType TaskExecutorType `protobuf:"varint,6,opt,name=executor_type,json=executorType,proto3,enum=v1.TaskExecutorType" json:"executor_type,omitempty"` - ExecutionId string `protobuf:"bytes,7,opt,name=execution_id,json=executionId,proto3" json:"execution_id,omitempty"` - Status TaskStatus `protobuf:"varint,8,opt,name=status,proto3,enum=v1.TaskStatus" json:"status,omitempty"` - Message string `protobuf:"bytes,9,opt,name=message,proto3" json:"message,omitempty"` + // description is provided by the client when the task is created. + Description string `protobuf:"bytes,5,opt,name=description,proto3" json:"description,omitempty"` + ExecutorType TaskExecutorType `protobuf:"varint,6,opt,name=executor_type,json=executorType,proto3,enum=v1.TaskExecutorType" json:"executor_type,omitempty"` + ExecutionId string `protobuf:"bytes,7,opt,name=execution_id,json=executionId,proto3" json:"execution_id,omitempty"` + Status TaskStatus `protobuf:"varint,8,opt,name=status,proto3,enum=v1.TaskStatus" json:"status,omitempty"` + // message is brief text tied to status (not execution progress). + Message string `protobuf:"bytes,9,opt,name=message,proto3" json:"message,omitempty"` // queue_expires_at is set only for waiting tasks; absent for all other statuses. QueueExpiresAt *timestamppb.Timestamp `protobuf:"bytes,10,opt,name=queue_expires_at,json=queueExpiresAt,proto3,oneof" json:"queue_expires_at,omitempty"` CreatedAt *timestamppb.Timestamp `protobuf:"bytes,11,opt,name=created_at,json=createdAt,proto3" json:"created_at,omitempty"` @@ -2195,8 +2197,10 @@ type Task struct { AppliedRuleId *UUID `protobuf:"bytes,13,opt,name=applied_rule_id,json=appliedRuleId,proto3,oneof" json:"applied_rule_id,omitempty"` UpdatedAt *timestamppb.Timestamp `protobuf:"bytes,14,opt,name=updated_at,json=updatedAt,proto3" json:"updated_at,omitempty"` StartedAt *timestamppb.Timestamp `protobuf:"bytes,15,opt,name=started_at,json=startedAt,proto3,oneof" json:"started_at,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // report is a versioned JSON document with structured execution progress. + Report string `protobuf:"bytes,16,opt,name=report,proto3" json:"report,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *Task) Reset() { @@ -2334,6 +2338,13 @@ func (x *Task) GetStartedAt() *timestamppb.Timestamp { return nil } +func (x *Task) GetReport() string { + if x != nil { + return x.Report + } + return "" +} + type CreateExpectedRackRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Rack *Rack `protobuf:"bytes,1,opt,name=rack,proto3" json:"rack,omitempty"` @@ -7784,7 +7795,7 @@ const file_flow_proto_rawDesc = "" + "rack_field\x18\x01 \x01(\x0e2\x14.v1.RackOrderByFieldH\x00R\trackField\x12D\n" + "\x0fcomponent_field\x18\x02 \x01(\x0e2\x19.v1.ComponentOrderByFieldH\x00R\x0ecomponentField\x12\x1c\n" + "\tdirection\x18\x03 \x01(\tR\tdirectionB\a\n" + - "\x05field\"\x98\x06\n" + + "\x05field\"\xb0\x06\n" + "\x04Task\x12\x18\n" + "\x02id\x18\x01 \x01(\v2\b.v1.UUIDR\x02id\x12\x1c\n" + "\toperation\x18\x02 \x01(\tR\toperation\x12!\n" + @@ -7805,7 +7816,8 @@ const file_flow_proto_rawDesc = "" + "\n" + "updated_at\x18\x0e \x01(\v2\x1a.google.protobuf.TimestampR\tupdatedAt\x12>\n" + "\n" + - "started_at\x18\x0f \x01(\v2\x1a.google.protobuf.TimestampH\x03R\tstartedAt\x88\x01\x01B\x13\n" + + "started_at\x18\x0f \x01(\v2\x1a.google.protobuf.TimestampH\x03R\tstartedAt\x88\x01\x01\x12\x16\n" + + "\x06report\x18\x10 \x01(\tR\x06reportB\x13\n" + "\x11_queue_expires_atB\x0e\n" + "\f_finished_atB\x12\n" + "\x10_applied_rule_idB\r\n" + diff --git a/flow/proto/v1/flow.proto b/flow/proto/v1/flow.proto index 597b94a4f..376ee4ba7 100644 --- a/flow/proto/v1/flow.proto +++ b/flow/proto/v1/flow.proto @@ -330,10 +330,12 @@ message Task { string operation = 2; UUID rack_id = 3; repeated UUID component_uuids = 4; + // description is provided by the client when the task is created. string description = 5; TaskExecutorType executor_type = 6; string execution_id = 7; TaskStatus status = 8; + // message is brief text tied to status (not execution progress). string message = 9; // queue_expires_at is set only for waiting tasks; absent for all other statuses. optional google.protobuf.Timestamp queue_expires_at = 10; @@ -342,6 +344,8 @@ message Task { optional UUID applied_rule_id = 13; google.protobuf.Timestamp updated_at = 14; optional google.protobuf.Timestamp started_at = 15; + // report is a versioned JSON document with structured execution progress. + string report = 16; } message CreateExpectedRackRequest {