From 160e18e7052723f83650fc1bef2f0a3400079add Mon Sep 17 00:00:00 2001 From: bircni Date: Mon, 12 Oct 2026 20:35:02 +0000 Subject: [PATCH] enhance: report the steps of a task as they run (#1293) Gitea derives the steps of a job from the workflow file, so the `pre` and `post` steps of actions only show up inside "Set up job" and "Complete job", see https://github.com/go-gitea/gitea/issues/24604. The runner now also reports the steps as it runs them (https://gitea.com/gitea/actions-proto-def/pulls/19): "Set up job", a "Pre" and "Post" step for each action that has one, the workflow steps with their evaluated names, and "Complete job", as actions/runner shows them. The per-step state Gitea reads today is unchanged, and the job log only moves "Complete job name" and "Matrix:" ahead of the pre steps, so older Gitea versions keep showing everything as before. Job hooks and containers stay part of "Set up job" and "Complete job". The `replace` of `gitea.dev/actionslib` is swapped for the release once the proto change is merged. Reviewed-on: https://gitea.com/gitea/runner/pulls/1293 Reviewed-by: Zettat123 <39446+zettat123@noreply.gitea.com> Co-authored-by: bircni --- act/runner/job_executor.go | 44 +++-- act/runner/job_executor_test.go | 33 +++- act/runner/logger.go | 10 ++ act/runner/step.go | 11 +- act/runner/step_test.go | 18 ++ go.mod | 2 +- go.sum | 4 +- internal/app/run/runner.go | 2 +- internal/pkg/report/joblog_test.go | 2 +- internal/pkg/report/reporter.go | 176 ++++++++++++++++++- internal/pkg/report/reporter_test.go | 242 +++++++++++++++++++++++++-- 11 files changed, 497 insertions(+), 47 deletions(-) diff --git a/act/runner/job_executor.go b/act/runner/job_executor.go index bd281d65..354bf246 100644 --- a/act/runner/job_executor.go +++ b/act/runner/job_executor.go @@ -99,7 +99,7 @@ func printPrepareActions(rc *RunContext, preparers []actionPreparer) common.Exec if err := preparer.prepareActionExecutor()(ctx); err != nil { // No step has run yet, so the failure belongs to the job. reportStepError(ctx, rc, err) - return err + return setupError{err} } reference, sha, ok := preparer.actionDownloadInfo() if !ok { @@ -115,6 +115,11 @@ func printPrepareActions(rc *RunContext, preparers []actionPreparer) common.Exec } } +// setupError is a failure of the job setup, such as an action that cannot be downloaded. +type setupError struct{ error } + +func (e setupError) Unwrap() error { return e.error } + // printCompleteJobName closes the setup section the way actions/runner ends its "Set up job" step. func printCompleteJobName(rc *RunContext) common.Executor { return func(ctx context.Context) error { @@ -139,20 +144,18 @@ func newJobExecutor(info jobInfo, sf stepFactory, rc *RunContext) common.Executo preparers := make([]actionPreparer, 0) var postExecutor common.Executor - steps = append(steps, func(ctx context.Context) error { - logger := common.Logger(ctx) - if len(info.matrix()) > 0 { - logger.Infof("Matrix: %v", info.matrix()) - } - return nil - }) - infoSteps := info.steps() if len(infoSteps) == 0 { return common.NewDebugExecutor("No steps found") } + preSteps = append(preSteps, func(ctx context.Context) error { + if len(info.matrix()) > 0 { + common.Logger(ctx).Infof("Matrix: %v", info.matrix()) + } + return nil + }) preSteps = append(preSteps, func(ctx context.Context) error { // Have to be skipped for some Tests if rc.Run == nil { @@ -231,14 +234,15 @@ func newJobExecutor(info jobInfo, sf stepFactory, rc *RunContext) common.Executo // The setup section of the job log. The started hook goes first, so what it sets up is // in place for the first action download and the first step. + // The pre steps come after it, as steps of their own. preSteps = append(preSteps, rc.runJobStartedHook) preSteps = append(preSteps, printPrepareActions(rc, preparers)) - preSteps = append(preSteps, stepPreSteps...) preSteps = append(preSteps, printCompleteJobName(rc)) + preSteps = append(preSteps, stepPreSteps...) // Ahead of the teardown below, while the job environment is still up. - postExecutor = postExecutor.Finally(rc.runJobCompletedHook) - postExecutor = postExecutor.Finally(func(ctx context.Context) error { + postExecutor = postExecutor.Finally(useJobStageLogger(StageComplete, rc.runJobCompletedHook)) + postExecutor = postExecutor.Finally(useJobStageLogger(StageComplete, func(ctx context.Context) error { // swallowed: a bad output fails this job, it must not abandon the rest of the plan if err := info.interpolateOutputs()(ctx); err != nil { reportStepError(ctx, rc, err) @@ -246,9 +250,9 @@ func newJobExecutor(info jobInfo, sf stepFactory, rc *RunContext) common.Executo reportStepError(ctx, rc, err) } return nil - }) + })) - postExecutor = postExecutor.Finally(func(ctx context.Context) error { + postExecutor = postExecutor.Finally(useJobStageLogger(StageComplete, func(ctx context.Context) error { jobError := common.JobError(ctx) var err error // always allow 1 min for stopping and removing the runner, even if we were cancelled @@ -265,7 +269,7 @@ func newJobExecutor(info jobInfo, sf stepFactory, rc *RunContext) common.Executo rc.removeDockerActionImages(ctx) return err - }) + })) return common.Executor(func(ctx context.Context) error { if err := info.startContainer()(ctx); err != nil { @@ -319,6 +323,9 @@ func newStepsExecutor(rc *RunContext, preSteps, steps []common.Executor) common. mainExecutor := newMainStepsExecutor(rc, steps) return func(ctx context.Context) error { preErr := preExecutor(ctx) + if errors.As(preErr, new(setupError)) { + return preErr // as on GitHub, no step runs after the job setup failed + } mainErr := mainExecutor(ctx) if preErr != nil { return preErr @@ -684,6 +691,13 @@ func archiveEntryMatchesPath(entryName, requestedPath string) bool { return entryName == requestedPath || entryName == path.Base(requestedPath) } +// useJobStageLogger tags the log entries of a job level executor with the stage they belong to. +func useJobStageLogger(stage string, executor common.Executor) common.Executor { + return func(ctx context.Context) error { + return executor(common.WithLogger(ctx, common.Logger(ctx).With("stage", stage))) + } +} + func useStepLogger(rc *RunContext, stepModel *model.Step, stage stepStage, executor common.Executor) common.Executor { return func(ctx context.Context) error { ctx = withStepLogger(ctx, stepModel.Number, stepModel.ID, rc.ExprEval.InterpolateName(ctx, stepModel.String()), stage.String()) diff --git a/act/runner/job_executor_test.go b/act/runner/job_executor_test.go index b24ba4c9..83e477d8 100644 --- a/act/runner/job_executor_test.go +++ b/act/runner/job_executor_test.go @@ -187,6 +187,27 @@ func TestPrintPrepareActionsFailsJobOnDownloadError(t *testing.T) { assert.Zero(t, remaining.prepared) } +func TestNewStepsExecutorSkipsStepsAfterSetupFailure(t *testing.T) { + for name, tt := range map[string]struct { + preErr error + mainRan bool + }{ + "setup failure": {preErr: setupError{assert.AnError}}, + "pre step failure": {preErr: assert.AnError, mainRan: true}, // always() steps still run, as on GitHub + } { + t.Run(name, func(t *testing.T) { + ctx := common.WithJobErrorContainer(common.WithLogger(context.Background(), logging.New(logtest.New()))) + mainRan := false + err := newStepsExecutor(&RunContext{}, + []common.Executor{func(context.Context) error { return tt.preErr }}, + []common.Executor{func(context.Context) error { mainRan = true; return nil }}, + )(ctx) + require.ErrorIs(t, err, assert.AnError) + assert.Equal(t, tt.mainRan, mainRan) + }) + } +} + func TestPrintCompleteJobName(t *testing.T) { for name, tt := range map[string]struct { rc *RunContext @@ -249,8 +270,9 @@ func TestNewJobExecutorDownloadsAllActionsBeforeTheFirstStep(t *testing.T) { apm := &actionPreparerMock{reference: "actions/checkout@v" + stepModel.ID, ok: true} sfm.On("newStep", stepModel, rc).Return(&actionStepMock{stepMock: sm, actionPreparerMock: apm}, nil) - sm.On("pre").Return(func(context.Context) error { + sm.On("pre").Return(func(ctx context.Context) error { executorOrder = append(executorOrder, "pre"+stepModel.ID) + common.Logger(ctx).Infof("Run Pre %s", stepModel.ID) return nil }) sm.On("main").Return(func(context.Context) error { @@ -268,18 +290,25 @@ func TestNewJobExecutorDownloadsAllActionsBeforeTheFirstStep(t *testing.T) { assert.Equal(t, []string{"pre1", "pre2", "step1", "step2"}, executorOrder) + // the setup is logged ahead of the pre steps and the cleanup is tagged, so the reporter can show both as steps of their own setup := make([]string, 0) for _, entry := range hook.AllRecords() { if strings.HasPrefix(entry.Message, "Prepare all required actions") || strings.HasPrefix(entry.Message, "Download action") || - strings.HasPrefix(entry.Message, "Complete job name") { + strings.HasPrefix(entry.Message, "Complete job name") || strings.HasPrefix(entry.Message, "Run Pre") { setup = append(setup, entry.Message) } + if _, ok := entry.Attr("jobResult"); ok || strings.HasPrefix(entry.Message, "Cleaning up container") { + stage, _ := entry.Attr("stage") + assert.Equal(t, StageComplete, stage, entry.Message) + } } assert.Equal(t, []string{ "Prepare all required actions", "Download action repository 'actions/checkout@v1'", "Download action repository 'actions/checkout@v2'", "Complete job name: test", + "Run Pre 1", + "Run Pre 2", }, setup) } diff --git a/act/runner/logger.go b/act/runner/logger.go index 9195562a..16a67526 100644 --- a/act/runner/logger.go +++ b/act/runner/logger.go @@ -35,6 +35,16 @@ const ( gray = 37 ) +const ( + // The "stage" of the log entries of a step, and of the job level entries that finalize the job after its steps. + StagePre = "Pre" + StageMain = "Main" + StagePost = "Post" + StageComplete = "Complete" + // StepStartedField marks the log entry of a step that is about to run, unlike one that is skipped. + StepStartedField = "stepStarted" +) + const ( rawOutputField = "raw_output" scriptLineCyanField = "script_line_cyan" diff --git a/act/runner/step.go b/act/runner/step.go index adad9c83..9d84c779 100644 --- a/act/runner/step.go +++ b/act/runner/step.go @@ -47,11 +47,11 @@ const maxSymlinkDepth = 10 func (s stepStage) String() string { switch s { case stepStagePre: - return "Pre" + return StagePre case stepStageMain: - return "Main" + return StageMain case stepStagePost: - return "Post" + return StagePost } return "Unknown" } @@ -112,12 +112,13 @@ func runStepExecutor(step step, stage stepStage, executor common.Executor) commo if strings.Contains(stepString, "::add-mask::") { stepString = "add-mask command" } + startLogger := logger.With(StepStartedField, true) if stage == stepStageMain { // Main steps print their own raw "Run " header, so this line is redundant and // only leaks into the "Set up job" section for the first step; keep it as a debug trace. - logger.Debugf("Run %s %s", stage, stepString) + startLogger.Debugf("Run %s %s", stage, stepString) } else { - logger.Infof("Run %s %s", stage, stepString) + startLogger.Infof("Run %s %s", stage, stepString) } // Prepare and clean Runner File Commands diff --git a/act/runner/step_test.go b/act/runner/step_test.go index 1888ac49..9408efa8 100644 --- a/act/runner/step_test.go +++ b/act/runner/step_test.go @@ -522,6 +522,24 @@ func TestRunStepExecutorParity(t *testing.T) { }) } + t.Run("only a step that runs is marked started", func(t *testing.T) { + for _, test := range []struct { + condition string + started int + }{{"true", 1}, {"false", 0}} { + logger, hook := logtest.NewLogger() + step := newStep(t, &model.Step{ID: "pre", If: yaml.Node{Value: test.condition}}) + require.NoError(t, runStepExecutor(step, stepStagePre, noopExecutor)(common.WithLogger(context.Background(), logger))) + started := 0 + for _, entry := range hook.AllRecords() { + if v, _ := entry.Attr(StepStartedField); v == true { + started++ + } + } + assert.Equal(t, test.started, started, test.condition) + } + }) + t.Run("file command error honors continue-on-error", func(t *testing.T) { step := newStep(t, &model.Step{ID: "commands", RawContinueOnError: "true"}) logger, hook := logtest.NewLogger() diff --git a/go.mod b/go.mod index d0f69c69..5de3b2f0 100644 --- a/go.mod +++ b/go.mod @@ -7,7 +7,7 @@ toolchain go1.27.2 require ( connectrpc.com/connect v1.21.0 dario.cat/mergo v1.0.2 - gitea.dev/actionslib v1.3.0 + gitea.dev/actionslib v1.4.0 github.com/avast/retry-go/v5 v5.0.0 github.com/bmatcuk/doublestar/v4 v4.10.2 github.com/coder/websocket v1.8.15 diff --git a/go.sum b/go.sum index 4908e508..8e818ebb 100644 --- a/go.sum +++ b/go.sum @@ -4,8 +4,8 @@ cyphar.com/go-pathrs v0.2.5 h1:SnX9FBvnoyn3lUs1dkMgZ52bAETpirNu3FTRh5HlRik= cyphar.com/go-pathrs v0.2.5/go.mod h1:y8f1EMG7r+hCuFf/rXsKqMJrJAUoADZGNh5/vZPKcGc= dario.cat/mergo v1.0.2 h1:85+piFYR1tMbRrLcDwR18y4UKJ3aH1Tbzi24VRW1TK8= dario.cat/mergo v1.0.2/go.mod h1:E/hbnu0NxMFBjpMIE34DRGLWqDy0g5FuKDhCb31ngxA= -gitea.dev/actionslib v1.3.0 h1:xZRoTL1+sK/PaglT9uXkZJ9g0nk1WLMEH4FUXdXpKKw= -gitea.dev/actionslib v1.3.0/go.mod h1:svCefEsavx4jmn8vJ4wTG7Q0KBcGfrFmGUjrhxuNUAQ= +gitea.dev/actionslib v1.4.0 h1:k31smj22tnUZrrM5kkntXN8lKAe7dkoTEoiP1Fk5KJc= +gitea.dev/actionslib v1.4.0/go.mod h1:svCefEsavx4jmn8vJ4wTG7Q0KBcGfrFmGUjrhxuNUAQ= github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6 h1:He8afgbRMd7mFxO99hRNu+6tazq8nFF9lIwo9JFroBk= github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8= github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= diff --git a/internal/app/run/runner.go b/internal/app/run/runner.go index e2e7df22..85a0a697 100644 --- a/internal/app/run/runner.go +++ b/internal/app/run/runner.go @@ -471,7 +471,7 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report. return err } job := workflow.GetJob(jobID) - reporter.ResetSteps(len(job.Steps)) + reporter.ResetSteps(job.Steps) telemetry.SetJobName(ctx, cmp.Or(job.Name, jobID)) taskContext := task.Context.Fields diff --git a/internal/pkg/report/joblog_test.go b/internal/pkg/report/joblog_test.go index d43f867c..d65fe7ed 100644 --- a/internal/pkg/report/joblog_test.go +++ b/internal/pkg/report/joblog_test.go @@ -49,7 +49,7 @@ func TestJobLog_MirrorsUploadedRows(t *testing.T) { reporter := NewReporter(ctx, cancel, client, task, cfg) require.NotNil(t, reporter.jobLog) reporter.RunDaemon() - reporter.ResetSteps(1) + reporter.ResetSteps(newTestSteps(1)) fire := func(message string) { require.NoError(t, fire(reporter, logEntry{ diff --git a/internal/pkg/report/reporter.go b/internal/pkg/report/reporter.go index 17edd05b..f7546ab8 100644 --- a/internal/pkg/report/reporter.go +++ b/internal/pkg/report/reporter.go @@ -4,6 +4,7 @@ package report import ( + "cmp" "context" "errors" "fmt" @@ -21,6 +22,7 @@ import ( "gitea.com/gitea/runner/internal/pkg/metrics" "connectrpc.com/connect" + "gitea.dev/actionslib/pkg/model" runnerv1 "gitea.dev/actionslib/runner/v1" "github.com/avast/retry-go/v5" "google.golang.org/protobuf/proto" @@ -62,6 +64,11 @@ type Reporter struct { state *runnerv1.TaskState stateChanged bool + // The steps in state.ReportedSteps by stage and workflow step number, the step the next log row + // without a step of its own belongs to, and the workflow steps. Guarded by stateMu. + reportedSteps map[reportedStepKey]*runnerv1.StepState + currentReportedStep *runnerv1.StepState + stepModels []*model.Step // reportFailing keeps an outage to one log line at each end. reportFailing map[string]bool // serverResult is what the server decided, e.g. the zombie reaper failing @@ -157,14 +164,152 @@ func (r *Reporter) Result() runnerv1.Result { return r.state.Result } -func (r *Reporter) ResetSteps(l int) { +// ResetSteps sets the steps of the workflow job. They are reported both by their index in the +// workflow file, for Gitea versions deriving the steps from the file, and as the steps the runner +// runs, together with the setup, pre, post and cleanup steps, for Gitea versions that show those. +func (r *Reporter) ResetSteps(steps []*model.Step) { r.stateMu.Lock() defer r.stateMu.Unlock() - for i := range l { + for i := range steps { r.state.Steps = append(r.state.Steps, &runnerv1.StepState{ Id: int64(i), }) } + if len(steps) == 0 { + return // a reusable workflow caller, whose called jobs log their own steps through it + } + r.stepModels = steps + r.reportedSteps = map[reportedStepKey]*runnerv1.StepState{} + setup := r.addReportedStep(len(r.state.ReportedSteps), runnerv1.StepStage_STEP_STAGE_SETUP, 0, "Set up job") + setup.StartedAt = cmp.Or(r.state.StartedAt, timestamppb.Now()) + setup.LogLength = int64(r.logOffset + len(r.logRows)) // the runner's own lines logged before the job was parsed + r.currentReportedStep = setup + for i, step := range steps { + r.addReportedStep(len(r.state.ReportedSteps), runnerv1.StepStage_STEP_STAGE_MAIN, i, r.mask(stepDisplayName(step, ""))) + } + r.addReportedStep(len(r.state.ReportedSteps), runnerv1.StepStage_STEP_STAGE_CLEANUP, 0, "Complete job") +} + +type reportedStepKey struct { + stage runnerv1.StepStage + number int +} + +// stepDisplayName names a workflow step like Gitea and GitHub do, prefixing an unnamed one with "Run". +// evaluated is step.String() with its expressions evaluated, empty while the step has not run yet. +func stepDisplayName(step *model.Step, evaluated string) string { + if step.Name != "" { + return cmp.Or(evaluated, step.Name) + } + first, _, _ := strings.Cut(strings.TrimSpace(cmp.Or(evaluated, step.String())), "\n") + return "Run " + cmp.Or(strings.TrimSpace(first), step.String()) +} + +// addReportedStep inserts a step at pos of the reported steps. Caller holds stateMu. +func (r *Reporter) addReportedStep(pos int, stage runnerv1.StepStage, number int, name string) *runnerv1.StepState { + step := &runnerv1.StepState{Name: name, Stage: stage, Number: int64(number)} + r.state.ReportedSteps = slices.Insert(r.state.ReportedSteps, pos, step) + for i, v := range r.state.ReportedSteps[pos:] { + v.Id = int64(pos + i) + } + r.reportedSteps[reportedStepKey{stage, number}] = step + return step +} + +// reportedStepOf returns the reported step an entry belongs to, adding a pre or post step the +// first time it runs, and whether the entry names that step itself. Caller holds stateMu. +func (r *Reporter) reportedStepOf(entry logging.Record) (*runnerv1.StepState, bool) { + if r.reportedSteps == nil { + return nil, false + } + stage, _ := recordAttr[string](entry, "stage") + number, ok := recordAttr[int](entry, "stepNumber") + if stage == runner.StageComplete { + return r.reportedSteps[reportedStepKey{stage: runnerv1.StepStage_STEP_STAGE_CLEANUP}], false + } + if !ok || number < 0 || number >= len(r.stepModels) { + return r.currentReportedStep, false + } + var key reportedStepKey + switch stage { + case runner.StagePre: + key = reportedStepKey{runnerv1.StepStage_STEP_STAGE_PRE, number} + case runner.StageMain: + key = reportedStepKey{runnerv1.StepStage_STEP_STAGE_MAIN, number} + case runner.StagePost: + key = reportedStepKey{runnerv1.StepStage_STEP_STAGE_POST, number} + default: + return r.currentReportedStep, false + } + main, ok := r.reportedSteps[reportedStepKey{runnerv1.StepStage_STEP_STAGE_MAIN, number}] + if !ok { + return r.currentReportedStep, false + } + if evaluated, _ := recordAttr[string](entry, "step"); evaluated != "" { + // evaluated as GitHub shows it, and masked as an expression can name a secret + main.Name = r.mask(stepDisplayName(r.stepModels[number], evaluated)) + } + if step, ok := r.reportedSteps[key]; ok { + return step, true + } + if started, _ := recordAttr[bool](entry, runner.StepStartedField); !started { + return r.currentReportedStep, false // a pre or post step that does not run is not shown + } + mainName := main.Name + if stage == runner.StagePre { + // pre steps run in the order of the workflow steps, so each goes right before the main steps + pos := slices.IndexFunc(r.state.ReportedSteps, func(v *runnerv1.StepState) bool { return v.Stage == runnerv1.StepStage_STEP_STAGE_MAIN }) + return r.addReportedStep(pos, key.stage, number, "Pre "+mainName), true + } + return r.addReportedStep(len(r.state.ReportedSteps)-1, key.stage, number, "Post "+mainName), true +} + +// recordAttr returns the attribute key of rec if it is a T. +func recordAttr[T any](rec logging.Record, key string) (T, bool) { + v, _ := rec.Attr(key) + t, ok := v.(T) + return t, ok +} + +// enterReportedStep makes step the one log rows without a step of their own belong to, finishing +// the setup step once another step starts. It reports whether the step started. Caller holds stateMu. +func (r *Reporter) enterReportedStep(step *runnerv1.StepState, timestamp time.Time) bool { + if prev := r.currentReportedStep; step != prev && prev.Stage == runnerv1.StepStage_STEP_STAGE_SETUP && prev.StoppedAt == nil { + if prev.Result == runnerv1.Result_RESULT_UNSPECIFIED { + prev.Result = runnerv1.Result_RESULT_SUCCESS + } + prev.StoppedAt = timestamppb.New(timestamp) + } + r.currentReportedStep = step + if step.StartedAt != nil { + return false + } + step.StartedAt = timestamppb.New(timestamp) + return true +} + +// finishReportedSteps gives a result to the reported steps that have none when the job ends: +// the job's result to a step that was running, except the cleanup that ran in full. A running +// cleanup is left running unless closing, as it still logs after the job result. Caller holds stateMu. +func (r *Reporter) finishReportedSteps(jobResult runnerv1.Result, timestamp time.Time, closing bool) { + for _, v := range r.state.ReportedSteps { + if v.Result != runnerv1.Result_RESULT_UNSPECIFIED || !closing && v.StartedAt != nil && v.Stage == runnerv1.StepStage_STEP_STAGE_CLEANUP { + continue + } + switch { + case v.StartedAt == nil && jobResult == runnerv1.Result_RESULT_CANCELLED: + v.Result = runnerv1.Result_RESULT_CANCELLED + case v.StartedAt == nil: + v.Result = runnerv1.Result_RESULT_SKIPPED // never reached, such as the steps after a failed setup + case v.Stage == runnerv1.StepStage_STEP_STAGE_CLEANUP: + v.Result = runnerv1.Result_RESULT_SUCCESS + default: + v.Result = jobResult + } + if v.StartedAt != nil { + v.StoppedAt = timestamppb.New(timestamp) + } + } } // Enabled makes the Reporter see every record, the job logger's level filter runs before it. @@ -200,6 +345,12 @@ func (r *Reporter) appendLogRow(row *runnerv1.LogRow) { return } row.Content = r.mask(row.Content) + if step := r.currentReportedStep; step != nil { + if step.LogLength == 0 { + step.LogIndex = int64(r.logOffset + len(r.logRows)) + } + step.LogLength++ + } r.logRows = append(r.logRows, row) r.jobLog.write(row.Time.AsTime(), row.Content) } @@ -266,9 +417,26 @@ func (r *Reporter) Handle(_ context.Context, entry logging.Record) error { r.state.StartedAt = timestamppb.New(timestamp) } + if reported, own := r.reportedStepOf(entry); reported != nil { + if r.enterReportedStep(reported, timestamp) { + urgentState = true + } + if !own && entry.Level >= logging.LevelError && reported.Result == runnerv1.Result_RESULT_UNSPECIFIED { + reported.Result = runnerv1.Result_RESULT_FAILURE // a job level error, such as an action that cannot be downloaded + urgentState = true + } + if v, ok := entry.Attr("stepResult"); ok && own && isJobStepEntry(entry) { + if stepResult, ok := r.parseResult(v); ok { + reported.Result = stepResult + reported.StoppedAt = timestamppb.New(timestamp) + urgentState = true + } + } + } + stage, _ := entry.Attr("stage") - if stage != "Main" { + if stage != runner.StageMain { if v, ok := entry.Attr("jobResult"); ok { if jobResult, ok := r.parseResult(v); ok { // We need to ensure log upload before this upload @@ -282,6 +450,7 @@ func (r *Reporter) Handle(_ context.Context, entry logging.Record) error { } } } + r.finishReportedSteps(jobResult, timestamp, false) urgentState = true } } @@ -549,6 +718,7 @@ func (r *Reporter) Close(lastWords string) error { Content: lastWords, }) } + r.finishReportedSteps(r.state.Result, time.Now(), true) r.stateMu.Unlock() // Wake up the daemon loop so it detects closed promptly. diff --git a/internal/pkg/report/reporter_test.go b/internal/pkg/report/reporter_test.go index 98e38829..62bf0c81 100644 --- a/internal/pkg/report/reporter_test.go +++ b/internal/pkg/report/reporter_test.go @@ -255,7 +255,7 @@ func TestReporter_Handle(t *testing.T) { defer func() { require.NoError(t, reporter.Close("")) }() - reporter.ResetSteps(5) + reporter.ResetSteps(newTestSteps(5)) dataStep0 := map[string]any{ "stage": "Main", @@ -314,7 +314,7 @@ func TestReporter_LogLevelFiltering(t *testing.T) { defer func() { require.NoError(t, reporter.Close("")) }() - reporter.ResetSteps(2) + reporter.ResetSteps(newTestSteps(2)) dataStep0 := map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true} dataStep0Internal := map[string]any{"stage": "Main", "stepNumber": 0} @@ -380,7 +380,7 @@ func TestReporter_EphemeralRunnerDeletion(t *testing.T) { require.NoError(t, err) cfg, _ := config.LoadDefault("") reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg) - reporter.ResetSteps(1) + reporter.ResetSteps(newTestSteps(1)) // Fire a log entry to create pending data require.NoError(t, fire(reporter, logEntry{ @@ -446,7 +446,7 @@ func TestReporter_RunDaemonClose_Race(t *testing.T) { reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{ Context: taskCtx, }, cfg) - reporter.ResetSteps(1) + reporter.ResetSteps(newTestSteps(1)) // Start the daemon loop — RunDaemon spawns a goroutine internally. reporter.RunDaemon() @@ -493,7 +493,7 @@ func TestReporter_MaxLatencyTimer(t *testing.T) { cfg.Runner.LogReportBatchSize = 1000 // won't trigger batch flush reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg) - reporter.ResetSteps(1) + reporter.ResetSteps(newTestSteps(1)) reporter.RunDaemon() defer func() { _ = reporter.Close("") @@ -544,7 +544,7 @@ func TestReporter_BatchSizeFlush(t *testing.T) { cfg.Runner.LogReportBatchSize = 5 reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg) - reporter.ResetSteps(1) + reporter.ResetSteps(newTestSteps(1)) reporter.RunDaemon() defer func() { _ = reporter.Close("") @@ -591,7 +591,7 @@ func TestReporter_StateChangedNotLostDuringReport(t *testing.T) { require.NoError(t, err) cfg, _ := config.LoadDefault("") reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg) - reporter.ResetSteps(2) + reporter.ResetSteps(newTestSteps(2)) // Mark stateChanged=true so the first ReportState proceeds to UpdateTask. reporter.stateMu.Lock() @@ -652,7 +652,7 @@ func TestReporter_StateChangedRestoredOnError(t *testing.T) { require.NoError(t, err) cfg, _ := config.LoadDefault("") reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg) - reporter.ResetSteps(1) + reporter.ResetSteps(newTestSteps(1)) reporter.stateMu.Lock() reporter.stateChanged = true @@ -702,7 +702,7 @@ func TestReporter_StateNotifyFlush(t *testing.T) { cfg.Runner.LogReportInterval = 10 * time.Second reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg) - reporter.ResetSteps(1) + reporter.ResetSteps(newTestSteps(1)) reporter.RunDaemon() defer func() { _ = reporter.Close("") @@ -761,7 +761,7 @@ func TestReporter_CloseAlwaysSendsRowsWithNoMore(t *testing.T) { cfg.Runner.LogReportBatchSize = 1000 reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg) - reporter.ResetSteps(1) + reporter.ResetSteps(newTestSteps(1)) reporter.RunDaemon() // Simulate a successful job whose log buffer was already drained by the @@ -806,7 +806,7 @@ func TestReporter_StateHeartbeat(t *testing.T) { cfg, _ := config.LoadDefault("") cfg.Runner.StateReportInterval = 50 * time.Millisecond reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg) - reporter.ResetSteps(1) + reporter.ResetSteps(newTestSteps(1)) // First call has no prior report — sends to seed lastReportedAt. reporter.stateMu.Lock() @@ -877,7 +877,7 @@ func TestReporter_ServerCancelStillFlushesFinal(t *testing.T) { require.NoError(t, err) cfg, _ := config.LoadDefault("") reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg) - reporter.ResetSteps(1) + reporter.ResetSteps(newTestSteps(1)) // Force the first ReportState to actually call UpdateTask. reporter.stateMu.Lock() @@ -945,7 +945,7 @@ func TestReporter_CloseReportsCancelledOnCanceledCtx(t *testing.T) { require.NoError(t, err) cfg, _ := config.LoadDefault("") reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg) - reporter.ResetSteps(1) + reporter.ResetSteps(newTestSteps(1)) // Simulate the cancellation path: r.ctx is cancelled before Close() runs. cancel() @@ -1010,7 +1010,7 @@ func TestReporter_StopHeartbeats(t *testing.T) { cfg.Runner.LogReportInterval = time.Hour reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg) - reporter.ResetSteps(1) + reporter.ResetSteps(newTestSteps(1)) reporter.RunDaemon() reporter.stateMu.Lock() @@ -1216,7 +1216,7 @@ func TestReporter_AckIndexBeyondBuffer(t *testing.T) { require.NoError(t, err) cfg, _ := config.LoadDefault("") reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg) - reporter.ResetSteps(1) + reporter.ResetSteps(newTestSteps(1)) require.NoError(t, fire(reporter, logEntry{ Message: "hello", @@ -1256,7 +1256,7 @@ func TestReporter_CloseWithStuckDaemon(t *testing.T) { reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg) reporter.daemonWait = time.Millisecond reporter.RunDaemon() - reporter.ResetSteps(1) + reporter.ResetSteps(newTestSteps(1)) require.NoError(t, fire(reporter, logEntry{ Message: "hello", @@ -1434,7 +1434,7 @@ func TestReporter_StepRangeCoversAckedRows(t *testing.T) { cfg.Runner.LogReportBatchSize = 5 reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg) - reporter.ResetSteps(1) + reporter.ResetSteps(newTestSteps(1)) reporter.RunDaemon() defer func() { _ = reporter.Close("") @@ -1489,6 +1489,214 @@ func TestReporter_FlushWithoutAckedRowsSkipsState(t *testing.T) { reporter.reportLogWithState() } +func newTestSteps(n int) []*model.Step { + steps := make([]*model.Step, n) + for i := range steps { + steps[i] = &model.Step{Run: fmt.Sprintf("echo %d", i)} + } + return steps +} + +func TestReporter_ReportedSteps(t *testing.T) { + newReporter := func(t *testing.T, steps []*model.Step) (*Reporter, context.CancelFunc) { + client := mocks.NewClient(t) + client.On("UpdateLog", mock.Anything, mock.Anything).Maybe().Return(func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { + return connect_go.NewResponse(&runnerv1.UpdateLogResponse{AckIndex: req.Msg.Index + int64(len(req.Msg.Rows))}), nil + }) + client.On("UpdateTask", mock.Anything, mock.Anything).Maybe().Return(connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil) + taskCtx, err := structpb.NewStruct(map[string]any{}) + require.NoError(t, err) + cfg, _ := config.LoadDefault("") + ctx, cancel := context.WithCancel(t.Context()) + reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx, Secrets: map[string]string{"TOKEN": "s3cr3t"}}, cfg) + reporter.RunDaemon() + reporter.Logf("runner version") + reporter.ResetSteps(steps) + return reporter, cancel + } + send := func(t *testing.T, reporter *Reporter, level logging.Level, msg string, data map[string]any) { + t.Helper() + require.NoError(t, fire(reporter, logEntry{Message: msg, Data: data, Level: level})) + } + type reported struct { + stage runnerv1.StepStage + number int64 + name string + result runnerv1.Result + logIndex, logLen int64 + } + reportedOf := func(t *testing.T, reporter *Reporter) []reported { + reporter.stateMu.RLock() + defer reporter.stateMu.RUnlock() + var ret []reported + for i, v := range reporter.state.ReportedSteps { + assert.EqualValues(t, i, v.Id) + ret = append(ret, reported{v.Stage, v.Number, v.Name, v.Result, v.LogIndex, v.LogLength}) + } + return ret + } + const ( + setup = runnerv1.StepStage_STEP_STAGE_SETUP + pre = runnerv1.StepStage_STEP_STAGE_PRE + main = runnerv1.StepStage_STEP_STAGE_MAIN + post = runnerv1.StepStage_STEP_STAGE_POST + cleanup = runnerv1.StepStage_STEP_STAGE_CLEANUP + + unspecified = runnerv1.Result_RESULT_UNSPECIFIED + success = runnerv1.Result_RESULT_SUCCESS + failure = runnerv1.Result_RESULT_FAILURE + cancelled = runnerv1.Result_RESULT_CANCELLED + skipped = runnerv1.Result_RESULT_SKIPPED + ) + + t.Run("steps as run", func(t *testing.T) { + steps := newTestSteps(2) + steps[0].Name = "Echo ${{ matrix.n }}" + reporter, _ := newReporter(t, steps) + reporter.Logf("workflow prepared") + send(t, reporter, logging.LevelInfo, "Complete job name: test", map[string]any{"raw_output": true}) + send(t, reporter, logging.LevelDebug, "evaluating expression 'success()'", map[string]any{"stage": runner.StagePre, "stepNumber": 0}) + send(t, reporter, logging.LevelInfo, "Run Pre echo 0", map[string]any{"stage": runner.StagePre, "stepNumber": 0, "step": "Echo 1", runner.StepStartedField: true}) + send(t, reporter, logging.LevelInfo, "pre output", map[string]any{"stage": runner.StagePre, "stepNumber": 0, "raw_output": true}) + send(t, reporter, logging.LevelInfo, "Success - Pre echo 0", map[string]any{"stage": runner.StagePre, "stepNumber": 0, "stepResult": model.StepStatusSuccess}) + send(t, reporter, logging.LevelDebug, "evaluating expression 'false'", map[string]any{"stage": runner.StagePre, "stepNumber": 1}) + send(t, reporter, logging.LevelDebug, "Skipping step", map[string]any{"stage": runner.StagePre, "stepNumber": 1, "stepResult": model.StepStatusSkipped}) + send(t, reporter, logging.LevelInfo, "main output", map[string]any{"stage": runner.StageMain, "stepNumber": 0, "raw_output": true}) + send(t, reporter, logging.LevelInfo, "main output", map[string]any{"stage": runner.StageMain, "stepNumber": 0, "raw_output": true}) + send(t, reporter, logging.LevelInfo, "Success - Main echo 0", map[string]any{"stage": runner.StageMain, "stepNumber": 0, "stepResult": model.StepStatusSuccess}) + send(t, reporter, logging.LevelDebug, "Skipping step", map[string]any{"stage": runner.StageMain, "stepNumber": 1, "stepResult": model.StepStatusSkipped}) + send(t, reporter, logging.LevelInfo, "Run Post echo 0", map[string]any{"stage": runner.StagePost, "stepNumber": 0, runner.StepStartedField: true}) + send(t, reporter, logging.LevelInfo, "post output", map[string]any{"stage": runner.StagePost, "stepNumber": 0, "raw_output": true}) + send(t, reporter, logging.LevelInfo, "Success - Post echo 0", map[string]any{"stage": runner.StagePost, "stepNumber": 0, "stepResult": model.StepStatusSuccess}) + send(t, reporter, logging.LevelInfo, "Cleaning up container", map[string]any{"stage": runner.StageComplete}) + send(t, reporter, logging.LevelInfo, "Job succeeded", map[string]any{"stage": runner.StageComplete, "jobResult": model.StepStatusSuccess}) + + want := []reported{ + {setup, 0, "Set up job", success, 0, 3}, + {pre, 0, "Pre Echo 1", success, 3, 3}, + {main, 0, "Echo 1", success, 6, 2}, + {main, 1, "Run echo 1", skipped, 0, 0}, + {post, 0, "Post Echo 1", success, 8, 3}, + {cleanup, 0, "Complete job", unspecified, 11, 2}, // still logging after the job result + } + assert.Equal(t, want, reportedOf(t, reporter)) + + send(t, reporter, logging.LevelInfo, "Removing images", map[string]any{"stage": runner.StageComplete}) + require.NoError(t, reporter.Close("")) + want[5].result, want[5].logLen = success, 3 + assert.Equal(t, want, reportedOf(t, reporter)) + + // the workflow steps keep what Gitea versions deriving them from the workflow file expect + require.Len(t, reporter.state.Steps, 2) + assert.EqualValues(t, 6, reporter.state.Steps[0].LogIndex) + assert.EqualValues(t, 2, reporter.state.Steps[0].LogLength) + assert.Equal(t, success, reporter.state.Steps[0].Result) + assert.Equal(t, skipped, reporter.state.Steps[1].Result) + }) + + t.Run("failures and cancellation", func(t *testing.T) { + reporter, cancel := newReporter(t, newTestSteps(3)) + started := func(stage string, n int) map[string]any { + return map[string]any{"stage": stage, "stepNumber": n, runner.StepStartedField: true} + } + result := func(stage string, n int, result model.StepStatus, stepID ...string) map[string]any { + fields := map[string]any{"stage": stage, "stepNumber": n, "stepResult": result} + if len(stepID) > 0 { + fields["stepID"] = stepID + } + return fields + } + send(t, reporter, logging.LevelInfo, "Run Pre echo 0", started(runner.StagePre, 0)) + send(t, reporter, logging.LevelInfo, "Success - Pre echo 0", result(runner.StagePre, 0, model.StepStatusSuccess)) + send(t, reporter, logging.LevelDebug, "Skipping step", result(runner.StagePre, 1, model.StepStatusSkipped)) + send(t, reporter, logging.LevelInfo, "Run Pre echo 2", started(runner.StagePre, 2)) + send(t, reporter, logging.LevelInfo, "Success - Pre echo 2", result(runner.StagePre, 2, model.StepStatusSuccess)) + send(t, reporter, logging.LevelInfo, "composite step", result(runner.StageMain, 0, model.StepStatusSuccess, "0", "inner")) + send(t, reporter, logging.LevelInfo, "Failure - Main echo 0", result(runner.StageMain, 0, model.StepStatusFailure)) + send(t, reporter, logging.LevelInfo, "main output", map[string]any{"stage": runner.StageMain, "stepNumber": 1, "raw_output": true}) + send(t, reporter, logging.LevelInfo, "Run Post echo 0", started(runner.StagePost, 0)) + send(t, reporter, logging.LevelInfo, "Failure - Post echo 0", result(runner.StagePost, 0, model.StepStatusFailure)) + cancel() + require.NoError(t, reporter.Close("")) + + results := make(map[string]runnerv1.Result) + for _, v := range reportedOf(t, reporter) { + results[v.name] = v.result + } + assert.Equal(t, map[string]runnerv1.Result{ + "Set up job": success, + "Pre Run echo 0": success, + "Pre Run echo 2": success, + "Run echo 0": failure, // not the result of its composite step + "Run echo 1": cancelled, + "Run echo 2": cancelled, + "Post Run echo 0": failure, + "Complete job": cancelled, + }, results) + names := make([]string, 0, len(results)) + for _, v := range reportedOf(t, reporter) { + names = append(names, v.name) + } + assert.Equal(t, []string{"Set up job", "Pre Run echo 0", "Pre Run echo 2", "Run echo 0", "Run echo 1", "Run echo 2", "Post Run echo 0", "Complete job"}, names) + }) + + t.Run("job failing during setup", func(t *testing.T) { + reporter, _ := newReporter(t, newTestSteps(1)) + reporter.Logf("workflow prepared") + send(t, reporter, logging.LevelInfo, "Job failed", map[string]any{"jobResult": model.StepStatusFailure}) + require.NoError(t, reporter.Close("")) + + assert.Equal(t, []reported{ + {setup, 0, "Set up job", failure, 0, 3}, + {main, 0, "Run echo 0", skipped, 0, 0}, + {cleanup, 0, "Complete job", skipped, 0, 0}, + }, reportedOf(t, reporter)) + }) + + t.Run("action failing to download", func(t *testing.T) { + reporter, _ := newReporter(t, newTestSteps(1)) + send(t, reporter, logging.LevelInfo, "Prepare all required actions", map[string]any{"raw_output": true}) + send(t, reporter, logging.LevelError, "##[error]unable to clone 'invalid-owner/invalid-action'", nil) + send(t, reporter, logging.LevelInfo, "Cleaning up container", map[string]any{"stage": runner.StageComplete}) + send(t, reporter, logging.LevelInfo, "Job failed", map[string]any{"stage": runner.StageComplete, "jobResult": model.StepStatusFailure}) + require.NoError(t, reporter.Close("")) + + got := reportedOf(t, reporter) + assert.Equal(t, []runnerv1.Result{failure, skipped, success}, []runnerv1.Result{got[0].result, got[1].result, got[2].result}) + }) + + t.Run("step not reached in a succeeding job", func(t *testing.T) { + reporter, _ := newReporter(t, newTestSteps(1)) + send(t, reporter, logging.LevelInfo, "Job succeeded", map[string]any{"stage": runner.StageComplete, "jobResult": model.StepStatusSuccess}) + require.NoError(t, reporter.Close("")) + assert.Equal(t, skipped, reportedOf(t, reporter)[1].result) + }) + + t.Run("evaluated and masked names", func(t *testing.T) { + steps := newTestSteps(2) + steps[0].Run = "echo ${{ matrix.os }}\necho more" + steps[1].Name = "Deploy with ${{ secrets.TOKEN }}" + reporter, _ := newReporter(t, steps) + send(t, reporter, logging.LevelDebug, "Run Main", map[string]any{"stage": runner.StageMain, "stepNumber": 0, "step": "echo linux\necho more"}) + send(t, reporter, logging.LevelDebug, "Run Main", map[string]any{"stage": runner.StageMain, "stepNumber": 1, "step": "Deploy with s3cr3t"}) + require.NoError(t, reporter.Close("")) + + names := []string{} + for _, v := range reportedOf(t, reporter) { + names = append(names, v.name) + } + assert.Equal(t, []string{"Set up job", "Run echo linux", "Deploy with ***", "Complete job"}, names) + }) + + t.Run("reusable workflow caller", func(t *testing.T) { + reporter, _ := newReporter(t, nil) + send(t, reporter, logging.LevelInfo, "Cleaning up container", map[string]any{"stage": runner.StageComplete}) + send(t, reporter, logging.LevelInfo, "Job succeeded", map[string]any{"stage": runner.StageComplete, "jobResult": model.StepStatusSuccess}) + require.NoError(t, reporter.Close("")) + assert.Empty(t, reportedOf(t, reporter)) + }) +} + // logEntry is the record a job logger hands the Reporter, with the fields it accumulated as Data. type logEntry struct { Message string