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)) +}