refactor(logging): add a logging abstraction and move the job logger onto it (#1295)

Add `act/common/logging`: a printf-style `Logger` on a slog-shaped `Handler`, with a logrus-backed handler as the only backend, plus a `logtest` handler for tests. Move the job logger onto it: `common.Logger` returns `*logging.Logger`, the job log formatters and the `Reporter` hook become handlers, and composite step IDs travel in the context.

Logging is tightly coupled to logrus types (hooks, formatters, `Entry.Data`). Going through our own interface first lets the remaining call sites move over with unchanged behavior, and lets the backend be swapped for `log/slog` in one place afterwards. The global process logger and its masking formatter stay on logrus for now.

Assisted-by: Codet
Co-authored-by: bircni <[email protected]>
Reviewed-on: https://gitea.com/gitea/runner/pulls/1295
Reviewed-by: bircni <[email protected]>
This commit is contained in:
Lunny Xiaoandbircni committed 2026-10-11 23:26:00 +00:00
1 parent ad9eeb668a
commit de961b91eb
50 files changed
+1241 -438

No files matched your search

+2 -1
View File
@@ -8,6 +8,7 @@ import (
"fmt"
"gitea.com/gitea/runner/act/artifactcache"
"gitea.com/gitea/runner/act/common/logging"
"gitea.com/gitea/runner/internal/app/run"
"gitea.com/gitea/runner/internal/pkg/config"
@@ -60,7 +61,7 @@ func runCacheServer(configFile *string, cacheArgs *cacheServerArgs) func(cmd *co
Port: port,
InternalSecret: secret,
Policy: run.CachePolicy(cfg),
Logger: log.StandardLogger().WithField("module", "cache_request"),
Logger: logging.Default().With("module", "cache_request"),
})
if err != nil {
return err
+3 -2
View File
@@ -23,6 +23,7 @@ import (
"gitea.com/gitea/runner/act/artifactcache"
"gitea.com/gitea/runner/act/artifacts"
"gitea.com/gitea/runner/act/common"
"gitea.com/gitea/runner/act/common/logging"
"gitea.com/gitea/runner/act/runner"
"gitea.com/gitea/runner/internal/app/run"
"gitea.com/gitea/runner/internal/pkg/config"
@@ -482,7 +483,7 @@ func runExec(ctx context.Context, execArgs *executeArgs) func(cmd *cobra.Command
// init a cache server
handler, err := artifactcache.StartHandler(artifactcache.Options{
Policy: run.CachePolicy(&config.Config{Cache: config.DefaultCache()}),
Logger: log.StandardLogger().WithField("module", "cache_request"),
Logger: logging.Default().With("module", "cache_request"),
})
if err != nil {
return err
@@ -553,7 +554,7 @@ func runExec(ctx context.Context, execArgs *executeArgs) func(cmd *cobra.Command
}
if !execArgs.debug {
logLevel := log.InfoLevel
logLevel := logging.LevelInfo
config.JobLoggerLevel = &logLevel
}
+5 -11
View File
@@ -4,21 +4,15 @@
package run
import (
"io"
log "github.com/sirupsen/logrus"
"gitea.com/gitea/runner/act/common/logging"
)
// NullLogger is used to create a new JobLogger to discard logs. This
// will prevent these logs from being logged to the stdout, but
// forward them to the Reporter via its hook.
// forward them to the Reporter via its handler.
type NullLogger struct{}
// WithJobLogger creates a new logrus.Logger that will discard all logs.
func (n NullLogger) WithJobLogger() *log.Logger {
logger := log.New()
logger.SetOutput(io.Discard)
logger.SetLevel(log.TraceLevel)
return logger
// WithJobLogger returns a handler that discards all logs.
func (n NullLogger) WithJobLogger() logging.Handler {
return logging.DiscardHandler
}
+5 -5
View File
@@ -10,12 +10,12 @@ import (
"testing"
"time"
"gitea.com/gitea/runner/act/common/logging"
"gitea.com/gitea/runner/internal/pkg/config"
"gitea.com/gitea/runner/internal/pkg/metrics"
"gitea.com/gitea/runner/internal/pkg/report"
runnerv1 "gitea.dev/actionslib/runner/v1"
log "github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"google.golang.org/protobuf/types/known/structpb"
@@ -146,12 +146,12 @@ func TestRunPostTaskScriptIntegration(t *testing.T) {
func setReporterJobResult(t *testing.T, reporter *report.Reporter, result runnerv1.Result) {
t.Helper()
require.NoError(t, reporter.Fire(&log.Entry{
require.NoError(t, reporter.Handle(t.Context(), logging.Record{
Time: time.Now(),
Message: "job finished",
Data: log.Fields{
"stage": "Post",
"jobResult": metrics.ResultToStatusLabel(result),
Attrs: []logging.Attr{
{Key: "stage", Value: "Post"},
{Key: "jobResult", Value: metrics.ResultToStatusLabel(result)},
},
}))
}
+3 -2
View File
@@ -28,6 +28,7 @@ import (
"gitea.com/gitea/runner/act/artifactcache"
"gitea.com/gitea/runner/act/common"
"gitea.com/gitea/runner/act/common/logging"
"gitea.com/gitea/runner/act/container"
"gitea.com/gitea/runner/act/runner"
"gitea.com/gitea/runner/internal/pkg/client"
@@ -109,7 +110,7 @@ func NewRunner(cfg *config.Config, reg *config.Registration, cli client.Client)
Upstream: cfg.Cache.ExternalServer,
S3: cacheS3(cfg),
Policy: CachePolicy(cfg),
Logger: log.StandardLogger().WithField("module", "cache_request"),
Logger: logging.Default().With("module", "cache_request"),
})
if err != nil {
log.Errorf("cannot init cache server, it will be disabled: %v", err)
@@ -645,7 +646,7 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report.
reporter.Logf("workflow prepared")
// add logger recorders
ctx = common.WithLoggerHook(ctx, reporter)
ctx = common.WithLoggerHandler(ctx, reporter)
if !log.IsLevelEnabled(log.DebugLevel) {
ctx = runner.WithJobLoggerFactory(ctx, NullLogger{})
+4 -3
View File
@@ -18,6 +18,7 @@ import (
"time"
"gitea.com/gitea/runner/act/common"
"gitea.com/gitea/runner/act/common/logging"
"gitea.com/gitea/runner/act/runner"
clientmocks "gitea.com/gitea/runner/internal/pkg/client/mocks"
"gitea.com/gitea/runner/internal/pkg/config"
@@ -79,9 +80,9 @@ func TestRunnerRunningCountAndNullLogger(t *testing.T) {
r.runningCount.Add(2)
require.Equal(t, int64(2), r.RunningCount())
logger := NullLogger{}.WithJobLogger()
require.NotNil(t, logger)
require.NotNil(t, logger.Out)
handler := NullLogger{}.WithJobLogger()
require.NotNil(t, handler)
require.False(t, handler.Enabled(t.Context(), logging.LevelError))
}
func TestRunnerReclaimsVolumesAfterReporting(t *testing.T) {
+4 -4
View File
@@ -11,12 +11,12 @@ import (
"testing"
"time"
"gitea.com/gitea/runner/act/common/logging"
"gitea.com/gitea/runner/internal/pkg/client/mocks"
"gitea.com/gitea/runner/internal/pkg/config"
connect_go "connectrpc.com/connect"
runnerv1 "gitea.dev/actionslib/runner/v1"
log "github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
@@ -52,10 +52,10 @@ func TestJobLog_MirrorsUploadedRows(t *testing.T) {
reporter.ResetSteps(1)
fire := func(message string) {
require.NoError(t, reporter.Fire(&log.Entry{
require.NoError(t, fire(reporter, logEntry{
Message: message,
Level: log.InfoLevel,
Data: log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true},
Level: logging.LevelInfo,
Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true},
}))
}
fire("the token is s3cret-value")
+51 -18
View File
@@ -8,11 +8,13 @@ import (
"errors"
"fmt"
"regexp"
"slices"
"strings"
"sync"
"sync/atomic"
"time"
"gitea.com/gitea/runner/act/common/logging"
"gitea.com/gitea/runner/act/runner"
"gitea.com/gitea/runner/internal/pkg/client"
"gitea.com/gitea/runner/internal/pkg/config"
@@ -166,8 +168,30 @@ func (r *Reporter) ResetSteps(l int) {
}
}
func (r *Reporter) Levels() []log.Level {
return log.AllLevels
// Enabled makes the Reporter see every record, the job logger's level filter runs before it.
func (r *Reporter) Enabled(context.Context, logging.Level) bool { return true }
// WithAttrs returns a Handler feeding the same Reporter, with attrs added to every record.
func (r *Reporter) WithAttrs(attrs []logging.Attr) logging.Handler {
return &reporterHandler{r: r, attrs: attrs}
}
type reporterHandler struct {
r *Reporter
attrs []logging.Attr
}
func (h *reporterHandler) Enabled(ctx context.Context, level logging.Level) bool {
return h.r.Enabled(ctx, level)
}
func (h *reporterHandler) Handle(ctx context.Context, rec logging.Record) error {
rec.Attrs = slices.Concat(h.attrs, rec.Attrs)
return h.r.Handle(ctx, rec)
}
func (h *reporterHandler) WithAttrs(attrs []logging.Attr) logging.Handler {
return &reporterHandler{r: h.r, attrs: slices.Concat(h.attrs, attrs)}
}
// appendLogRow masks a row before buffering it for Gitea and the local job.log, the one point
@@ -181,11 +205,19 @@ func (r *Reporter) appendLogRow(row *runnerv1.LogRow) {
r.jobLog.write(row.Time.AsTime(), row.Content)
}
func attrArgs(attrs []logging.Attr) []any {
args := make([]any, len(attrs))
for i, a := range attrs {
args[i] = a
}
return args
}
// isJobStepEntry is used to not report composite step results incorrectly as step result
// returns true if the logentry is on job level
// returns false for composite action step messages
func isJobStepEntry(entry *log.Entry) bool {
if v, ok := entry.Data["stepID"]; ok {
func isJobStepEntry(rec logging.Record) bool {
if v, ok := rec.Attr("stepID"); ok {
if v, ok := v.([]string); ok && len(v) > 1 {
return false
}
@@ -219,15 +251,15 @@ func (r *Reporter) unlockAndNotify(urgentState bool) {
}
}
func (r *Reporter) Fire(entry *log.Entry) error {
func (r *Reporter) Handle(_ context.Context, entry logging.Record) error {
urgentState := false
r.stateMu.Lock()
r.stateChanged = true
if log.IsLevelEnabled(log.TraceLevel) {
log.WithFields(entry.Data).Trace(r.mask(entry.Message)) // the process masker has no ::add-mask:: value
if logging.Enabled(logging.LevelTrace) {
logging.With(attrArgs(entry.Attrs)...).Trace(r.mask(entry.Message)) // the process masker has no ::add-mask:: value
}
timestamp := entry.Time
@@ -235,10 +267,10 @@ func (r *Reporter) Fire(entry *log.Entry) error {
r.state.StartedAt = timestamppb.New(timestamp)
}
stage := entry.Data["stage"]
stage, _ := entry.Attr("stage")
if stage != "Main" {
if v, ok := entry.Data["jobResult"]; ok {
if v, ok := entry.Attr("jobResult"); ok {
if jobResult, ok := r.parseResult(v); ok {
// We need to ensure log upload before this upload
r.state.Result = jobResult
@@ -262,7 +294,7 @@ func (r *Reporter) Fire(entry *log.Entry) error {
}
var step *runnerv1.StepState
if v, ok := entry.Data["stepNumber"]; ok {
if v, ok := entry.Attr("stepNumber"); ok {
if v, ok := v.(int); ok && len(r.state.Steps) > v {
step = r.state.Steps[v]
}
@@ -284,11 +316,12 @@ func (r *Reporter) Fire(entry *log.Entry) error {
}
// Force reporting log errors as raw output to prevent silent failures
if entry.Level == log.ErrorLevel {
entry.Data["raw_output"] = true
v, ok := entry.Attr("raw_output")
if entry.Level >= logging.LevelError {
v, ok = true, true
}
if v, ok := entry.Data["raw_output"]; ok {
if ok {
if rawOutput, ok := v.(bool); ok && rawOutput {
if row := r.parseLogRow(entry); row != nil {
if step.LogLength == 0 {
@@ -301,7 +334,7 @@ func (r *Reporter) Fire(entry *log.Entry) error {
} else if r.shouldAppendLogRow(entry) {
r.appendLogRow(r.parseLogRow(entry))
}
if v, ok := entry.Data["stepResult"]; ok && isJobStepEntry(entry) {
if v, ok := entry.Attr("stepResult"); ok && isJobStepEntry(entry) {
if stepResult, ok := r.parseResult(v); ok {
if step.LogLength == 0 {
step.LogIndex = int64(r.logOffset + len(r.logRows))
@@ -671,7 +704,7 @@ func (r *Reporter) ReportState(reportResult bool) error {
r.outputsMu.Unlock()
// Consume stateChanged atomically with the snapshot; restored on error
// below so a concurrent Fire() during UpdateTask isn't silently lost.
// below so a log record handled during UpdateTask isn't silently lost.
// Heartbeat at stateReportInterval even when nothing changed, so the server
// doesn't time out long-running silent jobs as orphaned (#826).
last := r.lastReportedAtNanos.Load()
@@ -753,8 +786,8 @@ func (r *Reporter) duringSteps() bool {
// shouldAppendLogRow reports whether a non-raw_output entry should be written
// to the job log: only when we are between steps and the entry's level is
// within the globally configured log level.
func (r *Reporter) shouldAppendLogRow(entry *log.Entry) bool {
return !r.duringSteps() && entry.Level <= log.GetLevel()
func (r *Reporter) shouldAppendLogRow(entry logging.Record) bool {
return !r.duringSteps() && entry.Level >= logging.GetLevel()
}
var stringToResult = map[string]runnerv1.Result{
@@ -879,7 +912,7 @@ func cutPrefixFold(s, prefix string) (string, bool) {
return s[len(prefix):], true
}
func (r *Reporter) parseLogRow(entry *log.Entry) *runnerv1.LogRow {
func (r *Reporter) parseLogRow(entry logging.Record) *runnerv1.LogRow {
content := strings.TrimRight(entry.Message, "\r\n")
// cmdRegex only covers the ::cmd:: form, so the ##[add-mask] one would otherwise reach
+65 -50
View File
@@ -17,6 +17,7 @@ import (
"testing"
"time"
"gitea.com/gitea/runner/act/common/logging"
"gitea.com/gitea/runner/act/runner"
"gitea.com/gitea/runner/internal/pkg/client/mocks"
"gitea.com/gitea/runner/internal/pkg/config"
@@ -25,7 +26,6 @@ import (
connect_go "connectrpc.com/connect"
"gitea.dev/actionslib/pkg/model"
runnerv1 "gitea.dev/actionslib/runner/v1"
log "github.com/sirupsen/logrus"
logrustest "github.com/sirupsen/logrus/hooks/test"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
@@ -206,7 +206,7 @@ func TestReporter_parseLogRow(t *testing.T) {
debugOutputEnabled: tt.debugOutputEnabled,
}
for idx, arg := range tt.args {
rv := r.parseLogRow(&log.Entry{Message: arg})
rv := r.parseLogRow(logging.Record{Message: arg})
got := "<nil>"
if rv != nil {
@@ -225,13 +225,13 @@ func TestReporter_parseLogRowAddMask(t *testing.T) {
for _, line := range []string{"::add-mask::supersecret", "##[add-mask]supersecret"} {
r := &Reporter{logReplacer: strings.NewReplacer()}
assert.Nil(t, r.parseLogRow(&log.Entry{Message: line}), line)
assert.Nil(t, r.parseLogRow(logging.Record{Message: line}), line)
assert.Equal(t, "using *** now", r.mask("using supersecret now"), line)
}
}
func TestReporter_Fire(t *testing.T) {
func TestReporter_Handle(t *testing.T) {
t.Run("ignore command lines", func(t *testing.T) {
client := mocks.NewClient(t)
client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) {
@@ -263,13 +263,13 @@ func TestReporter_Fire(t *testing.T) {
"raw_output": true,
}
require.NoError(t, reporter.Fire(&log.Entry{Message: "regular log line", Data: dataStep0}))
require.NoError(t, reporter.Fire(&log.Entry{Message: "::debug::debug log line", Data: dataStep0}))
require.NoError(t, reporter.Fire(&log.Entry{Message: "regular log line", Data: dataStep0}))
require.NoError(t, reporter.Fire(&log.Entry{Message: "::debug::debug log line", Data: dataStep0}))
require.NoError(t, reporter.Fire(&log.Entry{Message: "::debug::debug log line", Data: dataStep0}))
require.NoError(t, reporter.Fire(&log.Entry{Message: "regular log line", Data: dataStep0}))
require.NoError(t, reporter.Fire(&log.Entry{Message: "composite step result", Data: map[string]any{
require.NoError(t, fire(reporter, logEntry{Message: "regular log line", Data: dataStep0}))
require.NoError(t, fire(reporter, logEntry{Message: "::debug::debug log line", Data: dataStep0}))
require.NoError(t, fire(reporter, logEntry{Message: "regular log line", Data: dataStep0}))
require.NoError(t, fire(reporter, logEntry{Message: "::debug::debug log line", Data: dataStep0}))
require.NoError(t, fire(reporter, logEntry{Message: "::debug::debug log line", Data: dataStep0}))
require.NoError(t, fire(reporter, logEntry{Message: "regular log line", Data: dataStep0}))
require.NoError(t, fire(reporter, logEntry{Message: "composite step result", Data: map[string]any{
"stage": "Main",
"stepID": []string{"0", "0"},
"stepNumber": 0,
@@ -277,7 +277,7 @@ func TestReporter_Fire(t *testing.T) {
"stepResult": model.StepStatusFailure,
}}))
assert.Equal(t, runnerv1.Result_RESULT_UNSPECIFIED, reporter.state.Steps[0].Result)
require.NoError(t, reporter.Fire(&log.Entry{Message: "step result", Data: map[string]any{
require.NoError(t, fire(reporter, logEntry{Message: "step result", Data: map[string]any{
"stage": "Main",
"stepNumber": 0,
"raw_output": true,
@@ -291,9 +291,9 @@ func TestReporter_Fire(t *testing.T) {
func TestReporter_LogLevelFiltering(t *testing.T) {
// Set global level to Info so Debug entries should be filtered.
origLevel := log.GetLevel()
log.SetLevel(log.InfoLevel)
defer log.SetLevel(origLevel)
origLevel := logging.GetLevel()
logging.SetLevel(logging.LevelInfo)
defer logging.SetLevel(origLevel)
client := mocks.NewClient(t)
client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) {
@@ -316,27 +316,27 @@ func TestReporter_LogLevelFiltering(t *testing.T) {
}()
reporter.ResetSteps(2)
dataStep0 := log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true}
dataStep0Internal := log.Fields{"stage": "Main", "stepNumber": 0}
dataStep0 := map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true}
dataStep0Internal := map[string]any{"stage": "Main", "stepNumber": 0}
// raw_output entries always appear in job log regardless of level.
require.NoError(t, reporter.Fire(&log.Entry{Message: "step output", Data: dataStep0, Level: log.InfoLevel}))
require.NoError(t, reporter.Fire(&log.Entry{Message: "step debug output", Data: dataStep0, Level: log.DebugLevel}))
require.NoError(t, fire(reporter, logEntry{Message: "step output", Data: dataStep0, Level: logging.LevelInfo}))
require.NoError(t, fire(reporter, logEntry{Message: "step debug output", Data: dataStep0, Level: logging.LevelDebug}))
assert.Equal(t, int64(2), reporter.state.Steps[0].LogLength, "raw_output entries must always be forwarded")
// Non-raw_output entries during steps are not added to logRows regardless of level.
require.NoError(t, reporter.Fire(&log.Entry{Message: "internal info", Data: dataStep0Internal, Level: log.InfoLevel}))
require.NoError(t, reporter.Fire(&log.Entry{Message: "internal debug", Data: dataStep0Internal, Level: log.DebugLevel}))
require.NoError(t, fire(reporter, logEntry{Message: "internal info", Data: dataStep0Internal, Level: logging.LevelInfo}))
require.NoError(t, fire(reporter, logEntry{Message: "internal debug", Data: dataStep0Internal, Level: logging.LevelDebug}))
// stepResult at DebugLevel (skipped step) must still update state even when filtered from log.
require.NoError(t, reporter.Fire(&log.Entry{
require.NoError(t, fire(reporter, logEntry{
Message: "Skipping step",
Data: log.Fields{
Data: map[string]any{
"stage": "Main",
"stepNumber": 1,
"stepResult": "skipped",
},
Level: log.DebugLevel,
Level: logging.LevelDebug,
}))
assert.Equal(t, runnerv1.Result_RESULT_SKIPPED, reporter.state.Steps[1].Result,
"stepResult at DebugLevel must update step state even when log entry is filtered from job log output")
@@ -383,9 +383,9 @@ func TestReporter_EphemeralRunnerDeletion(t *testing.T) {
reporter.ResetSteps(1)
// Fire a log entry to create pending data
require.NoError(t, reporter.Fire(&log.Entry{
require.NoError(t, fire(reporter, logEntry{
Message: "build output",
Data: log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true},
Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true},
}))
// Step 1: RunDaemon calls ReportLog(false) — runner is still alive
@@ -500,9 +500,9 @@ func TestReporter_MaxLatencyTimer(t *testing.T) {
}()
// Fire a single log line — not enough to trigger batch flush
require.NoError(t, reporter.Fire(&log.Entry{
require.NoError(t, fire(reporter, logEntry{
Message: "single log line",
Data: log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true},
Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true},
}))
// maxLatencyTimer should flush within ~100ms. Wait up to 500ms.
@@ -552,9 +552,9 @@ func TestReporter_BatchSizeFlush(t *testing.T) {
// Fire exactly batchSize log lines
for i := range 5 {
require.NoError(t, reporter.Fire(&log.Entry{
require.NoError(t, fire(reporter, logEntry{
Message: fmt.Sprintf("log line %d", i),
Data: log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true},
Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true},
}))
}
@@ -565,7 +565,7 @@ func TestReporter_BatchSizeFlush(t *testing.T) {
"batch size threshold should have triggered immediate flush")
}
// TestReporter_StateChangedNotLostDuringReport asserts that a Fire() arriving
// TestReporter_StateChangedNotLostDuringReport asserts that a log record arriving
// mid-UpdateTask re-dirties the flag so the change is picked up by the next report.
func TestReporter_StateChangedNotLostDuringReport(t *testing.T) {
var updateTaskCalls atomic.Int64
@@ -607,11 +607,11 @@ func TestReporter_StateChangedNotLostDuringReport(t *testing.T) {
// Wait until UpdateTask is in flight (snapshot taken, flag consumed).
<-inFlight
// Concurrent Fire() modifies state — must re-flip stateChanged so the
// A concurrent log record modifies state — must re-flip stateChanged so the
// change is not lost when the in-flight ReportState finishes.
require.NoError(t, reporter.Fire(&log.Entry{
require.NoError(t, fire(reporter, logEntry{
Message: "step starts",
Data: log.Fields{"stage": "Main", "stepNumber": 1, "raw_output": true},
Data: map[string]any{"stage": "Main", "stepNumber": 1, "raw_output": true},
}))
// Release the in-flight UpdateTask and wait for it to return.
@@ -619,15 +619,15 @@ func TestReporter_StateChangedNotLostDuringReport(t *testing.T) {
require.NoError(t, <-done)
// stateChanged must still be true so the next ReportState picks up the
// concurrent Fire()'s change instead of skipping via the early-return path.
// concurrent record's change instead of skipping via the early-return path.
reporter.stateMu.RLock()
changed := reporter.stateChanged
reporter.stateMu.RUnlock()
assert.True(t, changed, "stateChanged must remain true after a concurrent Fire() during in-flight ReportState")
assert.True(t, changed, "stateChanged must remain true after a concurrent log record during in-flight ReportState")
// And the next ReportState must actually send a second UpdateTask.
require.NoError(t, reporter.ReportState(false))
assert.Equal(t, int64(2), updateTaskCalls.Load(), "concurrent Fire() change must trigger a second UpdateTask, not be silently lost")
assert.Equal(t, int64(2), updateTaskCalls.Load(), "concurrent log record change must trigger a second UpdateTask, not be silently lost")
}
// TestReporter_StateChangedRestoredOnError verifies that when UpdateTask fails,
@@ -709,9 +709,9 @@ func TestReporter_StateNotifyFlush(t *testing.T) {
}()
// Fire a log entry that starts a step — this triggers stateNotify
require.NoError(t, reporter.Fire(&log.Entry{
require.NoError(t, fire(reporter, logEntry{
Message: "step starting",
Data: log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true},
Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true},
}))
// stateNotify should trigger immediate UpdateTask call
@@ -1054,8 +1054,8 @@ func TestAppendLogRow(t *testing.T) {
assert.Equal(t, "***", r.logRows[0].Content)
}
func TestReporter_Levels(t *testing.T) {
assert.Equal(t, log.AllLevels, (&Reporter{}).Levels())
func TestReporter_Enabled(t *testing.T) {
assert.True(t, (&Reporter{}).Enabled(t.Context(), logging.LevelTrace))
}
func TestReporter_Result(t *testing.T) {
@@ -1218,10 +1218,10 @@ func TestReporter_AckIndexBeyondBuffer(t *testing.T) {
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
reporter.ResetSteps(1)
require.NoError(t, reporter.Fire(&log.Entry{
require.NoError(t, fire(reporter, logEntry{
Message: "hello",
Data: log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true},
Level: log.InfoLevel,
Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true},
Level: logging.LevelInfo,
}))
require.NoError(t, reporter.ReportLog(false))
@@ -1258,10 +1258,10 @@ func TestReporter_CloseWithStuckDaemon(t *testing.T) {
reporter.RunDaemon()
reporter.ResetSteps(1)
require.NoError(t, reporter.Fire(&log.Entry{
require.NoError(t, fire(reporter, logEntry{
Message: "hello",
Data: log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true},
Level: log.InfoLevel,
Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true},
Level: logging.LevelInfo,
}))
select {
@@ -1441,17 +1441,17 @@ func TestReporter_StepRangeCoversAckedRows(t *testing.T) {
_ = reporter.Close("")
}()
stepData := log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true}
stepData := map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true}
// The step's first row is a transition, so it flushes state as well as logs.
require.NoError(t, reporter.Fire(&log.Entry{Message: "step starting", Data: stepData}))
require.NoError(t, fire(reporter, logEntry{Message: "step starting", Data: stepData}))
require.Eventually(t, func() bool {
rows, claimed := server.snapshot()
return rows == 1 && claimed == rows
}, time.Second, 10*time.Millisecond, "the step transition should have flushed both log and state")
for i := range cfg.Runner.LogReportBatchSize {
require.NoError(t, reporter.Fire(&log.Entry{
require.NoError(t, fire(reporter, logEntry{
Message: fmt.Sprintf("step output %d", i),
Data: stepData,
}))
@@ -1489,3 +1489,18 @@ func TestReporter_FlushWithoutAckedRowsSkipsState(t *testing.T) {
reporter.Logf("row")
reporter.reportLogWithState()
}
// logEntry is the record a job logger hands the Reporter, with the fields it accumulated as Data.
type logEntry struct {
Message string
Data map[string]any
Level logging.Level
}
func fire(r *Reporter, e logEntry) error {
attrs := make([]logging.Attr, 0, len(e.Data))
for k, v := range e.Data {
attrs = append(attrs, logging.Attr{Key: k, Value: v})
}
return r.WithAttrs(attrs).Handle(context.Background(), logging.Record{Message: e.Message, Level: e.Level})
}