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