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 <bircni@icloud.com>
This commit is contained in:
bircni authored and Lunny Xiao committed 2026-10-12 20:35:02 +00:00
1 parent 20c8b4abc4
commit 160e18e705
11 files changed
+497 -47

No files matched your search

+29 -15
View File
@@ -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())
+31 -2
View File
@@ -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)
}
+10
View File
@@ -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"
+6 -5
View File
@@ -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 <title>" 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
+18
View File
@@ -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()
+1 -1
View File
@@ -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
+2 -2
View File
@@ -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=
+1 -1
View File
@@ -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
+1 -1
View File
@@ -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{
+173 -3
View File
@@ -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.
+225 -17
View File
@@ -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