From 5771cc3bad0c24235018d9dbaf48082b31b3da8a Mon Sep 17 00:00:00 2001 From: joshjennings98 Date: Fri, 30 May 2025 11:10:43 +0100 Subject: [PATCH 1/5] :sparkles: `logs` Add support for reading line by line in FIFO reader --- changes/20250530105438.feature | 1 + utils/logs/fifo_logger.go | 68 ++++++++++++++++++++++++++++++++++ utils/logs/fifo_logger_test.go | 60 +++++++++++++++++++++++++++++- 3 files changed, 127 insertions(+), 2 deletions(-) create mode 100644 changes/20250530105438.feature 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..cfc52333c0 100644 --- a/utils/logs/fifo_logger.go +++ b/utils/logs/fifo_logger.go @@ -2,10 +2,13 @@ package logs import ( "bytes" + "context" "fmt" "io" "log" "sync" + + "github.com/ARM-software/golang-utils/utils/parallelisation" ) type FIFOWriter struct { @@ -39,6 +42,67 @@ func (w *FIFOWriter) Read() string { return string(bytes) } +func (w *FIFOWriter) ReadLines(ctx context.Context, delim byte) func(yield func(string) bool) { + 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 + } + + continue + } + + if len(partial) > 0 { + buf = append(partial, buf...) + partial = nil + } + + for { + idx := bytes.IndexByte(buf, delim) + 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 +116,10 @@ func (l *FIFOLoggers) Read() string { return l.LogWriter.Read() } +func (l *FIFOLoggers) ReadLines(ctx context.Context, delim byte) func(yield func(string) bool) { + return l.LogWriter.ReadLines(ctx, delim) +} + // 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..41f5a8abe9 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, '\n') { + _, 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, '\n') { + _, 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) +} From 8567d021a51251e8b75b04b65cbefdf751f38ab4 Mon Sep 17 00:00:00 2001 From: joshjennings98 Date: Fri, 30 May 2025 11:14:01 +0100 Subject: [PATCH 2/5] remove delimiter option since it is for reading lines --- utils/logs/fifo_logger.go | 8 ++++---- utils/logs/fifo_logger_test.go | 4 ++-- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/utils/logs/fifo_logger.go b/utils/logs/fifo_logger.go index cfc52333c0..fdfccd0ff8 100644 --- a/utils/logs/fifo_logger.go +++ b/utils/logs/fifo_logger.go @@ -42,7 +42,7 @@ func (w *FIFOWriter) Read() string { return string(bytes) } -func (w *FIFOWriter) ReadLines(ctx context.Context, delim byte) func(yield func(string) bool) { +func (w *FIFOWriter) ReadLines(ctx context.Context) func(yield func(string) bool) { return func(yield func(string) bool) { var partial []byte for { @@ -77,7 +77,7 @@ func (w *FIFOWriter) ReadLines(ctx context.Context, delim byte) func(yield func( } for { - idx := bytes.IndexByte(buf, delim) + idx := bytes.IndexByte(buf, '\n') if idx < 0 { break } @@ -116,8 +116,8 @@ func (l *FIFOLoggers) Read() string { return l.LogWriter.Read() } -func (l *FIFOLoggers) ReadLines(ctx context.Context, delim byte) func(yield func(string) bool) { - return l.LogWriter.ReadLines(ctx, delim) +func (l *FIFOLoggers) ReadLines(ctx context.Context) func(yield func(string) bool) { + return l.LogWriter.ReadLines(ctx) } // Close closes the logger diff --git a/utils/logs/fifo_logger_test.go b/utils/logs/fifo_logger_test.go index 41f5a8abe9..8aa757f98c 100644 --- a/utils/logs/fifo_logger_test.go +++ b/utils/logs/fifo_logger_test.go @@ -62,7 +62,7 @@ func TestFIFOLoggerReadlines(t *testing.T) { defer cancel() var b strings.Builder - for line := range loggers.ReadLines(ctx, '\n') { + for line := range loggers.ReadLines(ctx) { _, err := b.WriteString(line + "\n") require.NoError(t, err) count++ @@ -93,7 +93,7 @@ func TestPlainFIFOLoggerReadlines(t *testing.T) { defer cancel() var b strings.Builder - for line := range loggers.ReadLines(ctx, '\n') { + for line := range loggers.ReadLines(ctx) { _, err := b.WriteString(line + "\n") require.NoError(t, err) count++ From 32848dfda5b2255a59070747a3ca08a02044fa5a Mon Sep 17 00:00:00 2001 From: joshjennings98 Date: Fri, 30 May 2025 11:18:56 +0100 Subject: [PATCH 3/5] use iter.Seq[string] for iterator return type --- utils/logs/fifo_logger.go | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/utils/logs/fifo_logger.go b/utils/logs/fifo_logger.go index fdfccd0ff8..ab033ab7bc 100644 --- a/utils/logs/fifo_logger.go +++ b/utils/logs/fifo_logger.go @@ -5,6 +5,7 @@ import ( "context" "fmt" "io" + "iter" "log" "sync" @@ -42,7 +43,7 @@ func (w *FIFOWriter) Read() string { return string(bytes) } -func (w *FIFOWriter) ReadLines(ctx context.Context) func(yield func(string) bool) { +func (w *FIFOWriter) ReadLines(ctx context.Context) iter.Seq[string] { return func(yield func(string) bool) { var partial []byte for { @@ -116,7 +117,7 @@ func (l *FIFOLoggers) Read() string { return l.LogWriter.Read() } -func (l *FIFOLoggers) ReadLines(ctx context.Context) func(yield func(string) bool) { +func (l *FIFOLoggers) ReadLines(ctx context.Context) iter.Seq[string] { return l.LogWriter.ReadLines(ctx) } From a0889233e2354cc535cfc849a3932f35ecf322db Mon Sep 17 00:00:00 2001 From: joshjennings98 Date: Fri, 30 May 2025 11:26:13 +0100 Subject: [PATCH 4/5] add sleep before continue --- utils/logs/fifo_logger.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/utils/logs/fifo_logger.go b/utils/logs/fifo_logger.go index ab033ab7bc..90ff186ae2 100644 --- a/utils/logs/fifo_logger.go +++ b/utils/logs/fifo_logger.go @@ -8,6 +8,7 @@ import ( "iter" "log" "sync" + "time" "github.com/ARM-software/golang-utils/utils/parallelisation" ) @@ -69,6 +70,7 @@ func (w *FIFOWriter) ReadLines(ctx context.Context) iter.Seq[string] { return } + parallelisation.SleepWithContext(ctx, 50*time.Millisecond) continue } From 9d3d21f88df55e726d447472c591b1a185da97e7 Mon Sep 17 00:00:00 2001 From: Josh Date: Fri, 30 May 2025 17:24:53 +0100 Subject: [PATCH 5/5] review --- utils/logs/fifo_logger.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/utils/logs/fifo_logger.go b/utils/logs/fifo_logger.go index 90ff186ae2..de5297fc19 100644 --- a/utils/logs/fifo_logger.go +++ b/utils/logs/fifo_logger.go @@ -13,6 +13,8 @@ import ( "github.com/ARM-software/golang-utils/utils/parallelisation" ) +var _ Loggers = &FIFOLoggers{} + type FIFOWriter struct { io.WriteCloser mu sync.RWMutex