diff --git a/changes/20250530105438.feature b/changes/20250530105438.feature new file mode 100644 index 0000000000..299cd9d998 --- /dev/null +++ b/changes/20250530105438.feature @@ -0,0 +1 @@ +:sparkles: `logs` Add support for reading line by line in FIFO reader diff --git a/utils/logs/fifo_logger.go b/utils/logs/fifo_logger.go index ba8bbf16a9..de5297fc19 100644 --- a/utils/logs/fifo_logger.go +++ b/utils/logs/fifo_logger.go @@ -2,12 +2,19 @@ package logs import ( "bytes" + "context" "fmt" "io" + "iter" "log" "sync" + "time" + + "github.com/ARM-software/golang-utils/utils/parallelisation" ) +var _ Loggers = &FIFOLoggers{} + type FIFOWriter struct { io.WriteCloser mu sync.RWMutex @@ -39,6 +46,68 @@ func (w *FIFOWriter) Read() string { return string(bytes) } +func (w *FIFOWriter) ReadLines(ctx context.Context) iter.Seq[string] { + return func(yield func(string) bool) { + var partial []byte + for { + if err := parallelisation.DetermineContextError(ctx); err != nil { + return + } + + buf := func() []byte { + w.mu.Lock() + defer w.mu.Unlock() + defer w.Logs.Reset() + tmp := w.Logs.Bytes() + buf := make([]byte, len(tmp)) + copy(buf, tmp) + return buf + }() + + if len(buf) == 0 { + if err := parallelisation.DetermineContextError(ctx); err != nil { + if len(partial) > 0 { + yield(string(partial)) + } + return + } + + parallelisation.SleepWithContext(ctx, 50*time.Millisecond) + continue + } + + if len(partial) > 0 { + buf = append(partial, buf...) + partial = nil + } + + for { + idx := bytes.IndexByte(buf, '\n') + if idx < 0 { + break + } + line := buf[:idx] + + if len(line) > 0 && line[len(line)-1] == '\r' { + line = line[:len(line)-1] + } + buf = buf[idx+1:] + if len(line) == 0 { + continue + } + + if !yield(string(line)) { + return + } + } + + if len(buf) > 0 { + partial = buf + } + } + } +} + type FIFOLoggers struct { GenericLoggers LogWriter FIFOWriter @@ -52,6 +121,10 @@ func (l *FIFOLoggers) Read() string { return l.LogWriter.Read() } +func (l *FIFOLoggers) ReadLines(ctx context.Context) iter.Seq[string] { + return l.LogWriter.ReadLines(ctx) +} + // Close closes the logger func (l *FIFOLoggers) Close() (err error) { err = l.LogWriter.Close() diff --git a/utils/logs/fifo_logger_test.go b/utils/logs/fifo_logger_test.go index ce876e41ca..8aa757f98c 100644 --- a/utils/logs/fifo_logger_test.go +++ b/utils/logs/fifo_logger_test.go @@ -1,13 +1,17 @@ package logs import ( + "context" + "regexp" "strings" "testing" + "time" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) -func TestFIFOLogger(t *testing.T) { +func TestFIFOLoggerRead(t *testing.T) { loggers, err := NewFIFOLogger("Test") require.NoError(t, err) testLog(t, loggers) @@ -27,7 +31,7 @@ func TestFIFOLogger(t *testing.T) { require.Empty(t, contents) } -func TestPlainFIFOLogger(t *testing.T) { +func TestPlainFIFOLoggerRead(t *testing.T) { loggers, err := NewPlainFIFOLogger() require.NoError(t, err) testLog(t, loggers) @@ -46,3 +50,55 @@ func TestPlainFIFOLogger(t *testing.T) { contents = loggers.Read() require.Empty(t, contents) } + +func TestFIFOLoggerReadlines(t *testing.T) { + loggers, err := NewFIFOLogger("Test") + require.NoError(t, err) + testLog(t, loggers) + loggers.LogError("Test err\n") + loggers.Log("Test1") + count := 0 + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + var b strings.Builder + for line := range loggers.ReadLines(ctx) { + _, err := b.WriteString(line + "\n") + require.NoError(t, err) + count++ + } + + assert.Regexp(t, regexp.MustCompile(`\[Test\] Error: .* .* Test err\n\[Test\] Output: .* .* Test1\n`), b.String()) + assert.Equal(t, 2, count) +} + +func TestPlainFIFOLoggerReadlines(t *testing.T) { + loggers, err := NewPlainFIFOLogger() + require.NoError(t, err) + testLog(t, loggers) + + go func() { + time.Sleep(500 * time.Millisecond) + loggers.LogError("Test err") + loggers.Log("") + time.Sleep(100 * time.Millisecond) + loggers.Log("Test1") + loggers.Log("\n\n\n") + time.Sleep(200 * time.Millisecond) + loggers.Log("Test2") + }() + + count := 0 + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + var b strings.Builder + for line := range loggers.ReadLines(ctx) { + _, err := b.WriteString(line + "\n") + require.NoError(t, err) + count++ + } + + assert.Equal(t, "Test err\nTest1\nTest2\n", b.String()) + assert.Equal(t, 3, count) +}