From d905c99d32d8eaf29062db84ff5a60e4e23639d5 Mon Sep 17 00:00:00 2001 From: mrhid6 Date: Mon, 20 Jul 2026 15:18:35 +0100 Subject: [PATCH] feat(server): stream step logs to files, drop log bodies from run docs --- server/internal/grpc/server.go | 7 ++++ server/internal/models/workflow.go | 3 +- server/internal/services/workflow_runner.go | 41 ++++++++++++++------- 3 files changed, 36 insertions(+), 15 deletions(-) diff --git a/server/internal/grpc/server.go b/server/internal/grpc/server.go index 6b7e8d5..9a96d4f 100644 --- a/server/internal/grpc/server.go +++ b/server/internal/grpc/server.go @@ -132,6 +132,13 @@ func (s *vantageServer) CommandStream(stream pb.Vantage_CommandStreamServer) err if m.StepResult != nil { services.StepResults.Deliver(m.StepResult) } + if m.StepOutput != nil { + if m.StepOutput.Eof { + services.StepLogs.Close(m.StepOutput.CommandId) + } else { + services.StepLogs.Append(m.StepOutput.CommandId, m.StepOutput.Data) + } + } } }() diff --git a/server/internal/models/workflow.go b/server/internal/models/workflow.go index 7b9afd5..3c03019 100644 --- a/server/internal/models/workflow.go +++ b/server/internal/models/workflow.go @@ -68,8 +68,7 @@ type StepRun struct { Status string `bson:"status" json:"status"` // queued|running|success|failed|skipped Attempts int `bson:"attempts" json:"attempts"` ExitCode int `bson:"exit_code" json:"exit_code"` - Stdout string `bson:"stdout" json:"stdout"` - Stderr string `bson:"stderr" json:"stderr"` + LogOffset int64 `bson:"log_offset" json:"log_offset"` OutputEnv map[string]string `bson:"output_env" json:"output_env"` StartedAt *time.Time `bson:"started_at,omitempty" json:"started_at,omitempty"` FinishedAt *time.Time `bson:"finished_at,omitempty" json:"finished_at,omitempty"` diff --git a/server/internal/services/workflow_runner.go b/server/internal/services/workflow_runner.go index c5f0358..710052e 100644 --- a/server/internal/services/workflow_runner.go +++ b/server/internal/services/workflow_runner.go @@ -195,40 +195,48 @@ func runServer(runID string, srvIdx int, steps []models.ResolvedStep, serverID s cmdEnv[k] = v } + // Write the step marker to the server-run log and remember the offset so + // the UI can slice this step's output later. + marker := fmt.Sprintf("\n===== step %d: %s =====\n", step.Order, step.Name) + offset, _ := AppendMarker(runID, serverID, marker) + logPath := ServerRunLogPath(runID, serverID) + secretsSlice := secretValues(secretVals) + + commandID := uuid.New().String() for attempts < maxAttempts { attempts++ - res = dispatchAndWait(serverID, &pb.RunStepCmd{ + // Open a fresh writer per attempt; the agent's eof closes it, and the + // defensive Close below covers a missing result. + _ = StepLogs.Open(commandID, logPath, secretsSlice) + res = dispatchAndWait(serverID, commandID, &pb.RunStepCmd{ Interpreter: step.Interpreter, Script: step.Script, Env: cmdEnv, TimeoutSeconds: 0, }) + StepLogs.Close(commandID) // idempotent; no-op if eof already closed it if res != nil && res.ExitCode == 0 { break } } - // Mask secret values before persisting. - stdout, stderr := "", "" exit := 1 - outEnv := map[string]string{} // masked copy, safe to persist + outEnv := map[string]string{} // masked copy, safe to persist if res != nil { - stdout = maskSecrets(res.Stdout, allSecrets) - stderr = maskSecrets(res.Stderr, allSecrets) exit = res.ExitCode for k, v := range res.OutputEnv { runEnv[k] = v // real, unmasked value threads forward to later steps outEnv[k] = maskSecrets(v, allSecrets) } } else { - stderr = "[vantage] agent did not return a result" + _, _ = AppendMarker(runID, serverID, "[vantage] agent did not return a result\n") } status := "success" if exit != 0 { status = "failed" } - finishStep(runID, serverID, i, status, attempts, exit, stdout, stderr, outEnv) + finishStep(runID, serverID, i, status, attempts, exit, offset, outEnv) if exit != 0 { switch step.OnFailure { @@ -264,8 +272,7 @@ func runServer(runID string, srvIdx int, steps []models.ResolvedStep, serverID s // dispatchAndWait registers a waiter, dispatches the step, and blocks for the // result or a timeout. -func dispatchAndWait(serverID string, cmd *pb.RunStepCmd) *pb.StepResult { - commandID := uuid.New().String() +func dispatchAndWait(serverID, commandID string, cmd *pb.RunStepCmd) *pb.StepResult { ch := StepResults.Await(commandID) if err := DispatchRunStep(serverID, commandID, cmd); err != nil { StepResults.Cancel(commandID) @@ -336,19 +343,27 @@ func startStep(runID, serverID string, order int, status string) { }) } -func finishStep(runID, serverID string, order int, status string, attempts, exit int, stdout, stderr string, outEnv map[string]string) { +func finishStep(runID, serverID string, order int, status string, attempts, exit int, logOffset int64, outEnv map[string]string) { now := time.Now() updateStep(runID, serverID, order, bson.M{ "server_runs.$[s].steps.$[t].status": status, "server_runs.$[s].steps.$[t].attempts": attempts, "server_runs.$[s].steps.$[t].exit_code": exit, - "server_runs.$[s].steps.$[t].stdout": stdout, - "server_runs.$[s].steps.$[t].stderr": stderr, + "server_runs.$[s].steps.$[t].log_offset": logOffset, "server_runs.$[s].steps.$[t].output_env": outEnv, "server_runs.$[s].steps.$[t].finished_at": now, }) } +// secretValues returns just the values of a secret map, for masking log output. +func secretValues(m map[string]string) []string { + out := make([]string, 0, len(m)) + for _, v := range m { + out = append(out, v) + } + return out +} + func markRemainingSkipped(runID, serverID string, fromOrder int) { ctx, cancel := wfCtx() defer cancel()