From 0780e5cfa5b13a9a137fec95852cfd92f5fcd5cc Mon Sep 17 00:00:00 2001 From: Wesley Ellis Date: Fri, 10 Jul 2026 16:43:40 -0400 Subject: [PATCH] fix: ship buffered job logs on interval even when output is idle The append log processors only evaluated the max-time condition inside Process, which runs when a new line arrives. A job that printed output and then went quiet had those logs held until the next line, maxBytes, or job completion - breaking the contract of job-pod-log-max-interval. The log streamer's Run loop now calls an optional Tick() on processors every ticker interval, and both append processors ship their buffer from Tick once maxTime has elapsed since the last submit. This replaces the arrival-driven elapsed-time accounting (and removes the `elapsed = time.Since(time.Now())` typo in the faktory processor). Co-Authored-By: Claude Fable 5 --- src/pkg/faktoryRunnerAppendJobLogProcessor.go | 20 ++++++++-------- src/pkg/logs.go | 12 ++++++++++ src/pkg/logs_test.go | 24 +++++++++++++++++++ src/pkg/opslevelAppendLogProcessor.go | 19 ++++++++------- src/pkg/opslevelAppendLogProcessor_test.go | 24 +++++++++++++++++++ 5 files changed, 80 insertions(+), 19 deletions(-) create mode 100644 src/pkg/opslevelAppendLogProcessor_test.go diff --git a/src/pkg/faktoryRunnerAppendJobLogProcessor.go b/src/pkg/faktoryRunnerAppendJobLogProcessor.go index 33038f8..52d3aff 100644 --- a/src/pkg/faktoryRunnerAppendJobLogProcessor.go +++ b/src/pkg/faktoryRunnerAppendJobLogProcessor.go @@ -19,8 +19,7 @@ type FaktoryAppendJobLogProcessor struct { logLines []string logLinesBytesSize int firstLine bool - lastTime time.Time - elapsed time.Duration + lastSubmitTime time.Time } func NewFaktoryAppendJobLogProcessor(helper faktoryWorker.Helper, logger zerolog.Logger, jobId opslevel.ID, maxBytes int, maxTime time.Duration) *FaktoryAppendJobLogProcessor { @@ -33,7 +32,7 @@ func NewFaktoryAppendJobLogProcessor(helper faktoryWorker.Helper, logger zerolog logLines: []string{}, logLinesBytesSize: 0, firstLine: false, - lastTime: time.Now(), + lastSubmitTime: time.Now(), } } @@ -54,15 +53,16 @@ func (s *FaktoryAppendJobLogProcessor) Process(line string) string { s.submit() } - s.elapsed += time.Since(s.lastTime) - if s.elapsed > s.maxTime { + return line +} + +// Tick ships buffered logs once maxTime has elapsed since the last submit, +// so a job that goes quiet still has its logs delivered on schedule. +func (s *FaktoryAppendJobLogProcessor) Tick() { + if len(s.logLines) > 0 && time.Since(s.lastSubmitTime) > s.maxTime { s.logger.Trace().Msg("Shipping logs because of maxTime ...") - s.elapsed = time.Since(time.Now()) s.submit() } - s.lastTime = time.Now() - - return line } func (s *FaktoryAppendJobLogProcessor) ProcessStdout(line string) string { @@ -111,6 +111,6 @@ func (s *FaktoryAppendJobLogProcessor) submit() { } } s.logLinesBytesSize = 0 - s.logLines = nil s.logLines = []string{} + s.lastSubmitTime = time.Now() } diff --git a/src/pkg/logs.go b/src/pkg/logs.go index 98df7a8..3974aa3 100644 --- a/src/pkg/logs.go +++ b/src/pkg/logs.go @@ -15,6 +15,13 @@ type LogProcessor interface { Flush(outcome JobOutcome) } +// tickable is optionally implemented by processors that need to do periodic +// work even when no new lines arrive (e.g. time-based log shipping). Tick is +// called from the streamer's Run loop on every ticker interval. +type tickable interface { + Tick() +} + type LogStreamer struct { Stdout *SafeBuffer Stderr *SafeBuffer @@ -93,6 +100,11 @@ func (s *LogStreamer) Run(ctx context.Context) { s.processLine(stream) } } + for _, processor := range s.processors { + if t, ok := processor.(tickable); ok { + t.Tick() + } + } } } } diff --git a/src/pkg/logs_test.go b/src/pkg/logs_test.go index 8025235..97d23cc 100644 --- a/src/pkg/logs_test.go +++ b/src/pkg/logs_test.go @@ -2,6 +2,7 @@ package pkg import ( "context" + "sync/atomic" "testing" "time" @@ -25,6 +26,29 @@ func (c *captureProcessor) ProcessStderr(line string) string { func (c *captureProcessor) Flush(_ JobOutcome) {} +type tickCountProcessor struct { + captureProcessor + ticks atomic.Int32 +} + +func (c *tickCountProcessor) Tick() { + c.ticks.Add(1) +} + +func TestLogStreamerCallsTickWhileIdle(t *testing.T) { + proc := &tickCountProcessor{} + s := NewLogStreamer(zerolog.Nop(), proc) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go s.Run(ctx) + + // No lines are ever written; the streamer should still tick processors. + time.Sleep(200 * time.Millisecond) + + autopilot.Assert(t, proc.ticks.Load() > 0, "expected Tick to be called while idle") +} + func TestLogStreamerPartialLineStdout(t *testing.T) { cap := &captureProcessor{} s := NewLogStreamer(zerolog.Nop(), cap) diff --git a/src/pkg/opslevelAppendLogProcessor.go b/src/pkg/opslevelAppendLogProcessor.go index 33b5e6b..5738e7c 100644 --- a/src/pkg/opslevelAppendLogProcessor.go +++ b/src/pkg/opslevelAppendLogProcessor.go @@ -19,8 +19,7 @@ type OpsLevelAppendLogProcessor struct { logLines []string logLinesBytesSize int firstLine bool - lastTime time.Time - elapsed time.Duration + lastSubmitTime time.Time } func NewOpsLevelAppendLogProcessor(client *opslevel.Client, logger zerolog.Logger, runnerId opslevel.ID, jobId opslevel.ID, jobNumber string, maxBytes int, maxTime time.Duration) *OpsLevelAppendLogProcessor { @@ -35,7 +34,7 @@ func NewOpsLevelAppendLogProcessor(client *opslevel.Client, logger zerolog.Logge logLines: []string{}, logLinesBytesSize: 0, firstLine: false, - lastTime: time.Now(), + lastSubmitTime: time.Now(), } } @@ -56,15 +55,16 @@ func (s *OpsLevelAppendLogProcessor) Process(line string) string { s.submit() } - s.elapsed += time.Since(s.lastTime) - if s.elapsed > s.maxTime { + return line +} + +// Tick ships buffered logs once maxTime has elapsed since the last submit, +// so a job that goes quiet still has its logs delivered on schedule. +func (s *OpsLevelAppendLogProcessor) Tick() { + if len(s.logLines) > 0 && time.Since(s.lastSubmitTime) > s.maxTime { s.logger.Trace().Msg("Shipping logs because of maxTime ...") - s.elapsed = 0 s.submit() } - s.lastTime = time.Now() - - return line } func (s *OpsLevelAppendLogProcessor) ProcessStdout(line string) string { @@ -98,4 +98,5 @@ func (s *OpsLevelAppendLogProcessor) submit() { } s.logLinesBytesSize = 0 s.logLines = s.logLines[:0] + s.lastSubmitTime = time.Now() } diff --git a/src/pkg/opslevelAppendLogProcessor_test.go b/src/pkg/opslevelAppendLogProcessor_test.go new file mode 100644 index 0000000..9ae597c --- /dev/null +++ b/src/pkg/opslevelAppendLogProcessor_test.go @@ -0,0 +1,24 @@ +package pkg + +import ( + "testing" + "time" + + "github.com/rocktavious/autopilot/v2023" + "github.com/rs/zerolog" +) + +func TestOpsLevelAppendLogProcessorTickShipsIdleLogs(t *testing.T) { + // Arrange + p := NewOpsLevelAppendLogProcessor(nil, zerolog.Nop(), "1", "1", "1", 1024, 50*time.Millisecond) + p.Process("first") // first line ships immediately + p.Process("second") // buffered + autopilot.Equals(t, 1, len(p.logLines)) + // Act & Assert: maxTime has not elapsed yet, so nothing ships + p.Tick() + autopilot.Equals(t, 1, len(p.logLines)) + // Act & Assert: once maxTime elapses, Tick ships without any new lines + time.Sleep(100 * time.Millisecond) + p.Tick() + autopilot.Equals(t, 0, len(p.logLines)) +}