fix: ship buffered job logs on interval even when output is idle

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

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

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)

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

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