From dbf8a6601b005b0600791f21bc9a7b1392b71564 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Tue, 28 Jul 2026 16:27:35 -0500 Subject: [PATCH 1/4] fix(log): route Writer through the structured encoder instead of raw passthrough log.Writer() is used as a subprocess's Stdout/Stderr at ~65 call sites across the codebase (docker build, buildkit, SSH, IDE installers, git clone, etc). Its levelWriter.Write wrote raw bytes directly to the stderr sink, bypassing zap's encoder entirely -- contradicting its own doc comment ("writes each line as a log entry") and producing unencoded lines in what's supposed to be a uniform JSON/text/logfmt stream in --log-output json mode. levelWriter now line-buffers and calls Info/Debug/Error per complete line, so subprocess output goes through the same encoder as every other log call. Close flushes any trailing line left without a newline (previously silently dropped). --- pkg/log/writer.go | 63 ++++++++++++++++++++------- pkg/log/writer_test.go | 99 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 147 insertions(+), 15 deletions(-) create mode 100644 pkg/log/writer_test.go diff --git a/pkg/log/writer.go b/pkg/log/writer.go index 4fa7d7b22..d3590549e 100644 --- a/pkg/log/writer.go +++ b/pkg/log/writer.go @@ -1,23 +1,20 @@ package log import ( + "bytes" "io" + "strings" + "sync" - "go.uber.org/zap" "go.uber.org/zap/zapcore" ) -// Writer returns an io.WriteCloser that writes each line as a log entry at the given level. -// Level uses the package constants: LevelInfo, LevelDebug, etc. +// Writer returns an io.WriteCloser that logs each complete line written to +// it as a structured entry at the given level, through the same encoder as +// every other log call. Primarily used as a subprocess's Stdout/Stderr. +// Close flushes any trailing line left without a newline. func Writer(level int) io.WriteCloser { - zapLevel := verbosityConstToZapLevel(level) - w, closer, _ := zap.Open("stderr") - _ = closer // stderr doesn't need closing - return &levelWriter{ - sink: w, - level: zapLevel, - core: sugar.Load().Desugar().Core(), - } + return &levelWriter{level: verbosityConstToZapLevel(level)} } func verbosityConstToZapLevel(level int) zapcore.Level { @@ -34,18 +31,54 @@ func verbosityConstToZapLevel(level int) zapcore.Level { } type levelWriter struct { - sink zapcore.WriteSyncer level zapcore.Level - core zapcore.Core + + mu sync.Mutex + buf bytes.Buffer } func (w *levelWriter) Write(p []byte) (int, error) { - if !w.core.Enabled(w.level) { + w.mu.Lock() + defer w.mu.Unlock() + + if !sugar.Load().Desugar().Core().Enabled(w.level) { return len(p), nil // discard if below current level } - return w.sink.Write(p) + + w.buf.Write(p) + for { + line, err := w.buf.ReadString('\n') + if err != nil { + // Incomplete line: put it back for the next Write/Close. + w.buf.Reset() + w.buf.WriteString(line) + break + } + w.logLine(strings.TrimSuffix(line, "\n")) + } + return len(p), nil +} + +func (w *levelWriter) logLine(line string) { + if line == "" { + return + } + switch w.level { + case zapcore.DebugLevel: + Debug(line) + case zapcore.ErrorLevel: + Error(line) + default: + Info(line) + } } func (w *levelWriter) Close() error { + w.mu.Lock() + defer w.mu.Unlock() + if w.buf.Len() > 0 { + w.logLine(w.buf.String()) + w.buf.Reset() + } return nil } diff --git a/pkg/log/writer_test.go b/pkg/log/writer_test.go new file mode 100644 index 000000000..aae50b3d2 --- /dev/null +++ b/pkg/log/writer_test.go @@ -0,0 +1,99 @@ +package log + +import ( + "bytes" + "strings" + "testing" +) + +func TestWriter_EmitsStructuredJSONLine(t *testing.T) { + Init(Config{Verbosity: 2, Format: "json"}) + + var sink bytes.Buffer + remove := AddSink(&sink) + defer remove() + + w := Writer(LevelInfo) + _, err := w.Write([]byte("Cloning into 'repo'...\n")) + if err != nil { + t.Fatalf("Write: %v", err) + } + _ = Sync() + + got := strings.TrimSpace(sink.String()) + if !strings.HasPrefix(got, "{") || !strings.HasSuffix(got, "}") { + t.Fatalf("expected a single JSON object, got %q", got) + } + if !strings.Contains(got, `"msg":"Cloning into 'repo'..."`) { + t.Errorf("missing expected msg field: %q", got) + } + if !strings.Contains(got, `"level":"info"`) { + t.Errorf("missing expected level field: %q", got) + } +} + +func TestWriter_SplitsMultipleLinesInOneWrite(t *testing.T) { + Init(Config{Verbosity: 2, Format: "json"}) + + var sink bytes.Buffer + remove := AddSink(&sink) + defer remove() + + w := Writer(LevelInfo) + _, err := w.Write([]byte("line one\nline two\n")) + if err != nil { + t.Fatalf("Write: %v", err) + } + _ = Sync() + + lines := strings.Split(strings.TrimSpace(sink.String()), "\n") + if len(lines) != 2 { + t.Fatalf("got %d lines, want 2: %q", len(lines), lines) + } + for _, l := range lines { + if !strings.HasPrefix(l, "{") || !strings.HasSuffix(l, "}") { + t.Errorf("line is not valid single-object JSON: %q", l) + } + } +} + +func TestWriter_FlushesTrailingPartialLineOnClose(t *testing.T) { + Init(Config{Verbosity: 2, Format: "json"}) + + var sink bytes.Buffer + remove := AddSink(&sink) + defer remove() + + w := Writer(LevelInfo) + _, _ = w.Write([]byte("no trailing newline")) + _ = Sync() + if sink.Len() != 0 { + t.Errorf("expected nothing logged before Close, got %q", sink.String()) + } + + if err := w.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + _ = Sync() + + if !strings.Contains(sink.String(), "no trailing newline") { + t.Errorf("expected trailing partial line flushed on Close, got %q", sink.String()) + } +} + +func TestWriter_DiscardsBelowConfiguredLevel(t *testing.T) { + Init(Config{Verbosity: 1, Format: "json"}) // info+ only, debug disabled + + var sink bytes.Buffer + remove := AddSink(&sink) + defer remove() + + w := Writer(LevelDebug) + _, _ = w.Write([]byte("should not appear\n")) + _ = w.Close() + _ = Sync() + + if sink.Len() != 0 { + t.Errorf("expected debug output discarded below configured level, got %q", sink.String()) + } +} From fe12612773f265b908c9291165136e7377c8c168 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Tue, 28 Jul 2026 16:43:53 -0500 Subject: [PATCH 2/4] fix(log): satisfy funcorder and goconst lint rules Reorder levelWriter's exported Close before the unexported logLine helper, and hoist the repeated "json" format literal in writer_test.go into a constant. --- pkg/log/writer.go | 20 ++++++++++---------- pkg/log/writer_test.go | 10 ++++++---- 2 files changed, 16 insertions(+), 14 deletions(-) diff --git a/pkg/log/writer.go b/pkg/log/writer.go index d3590549e..18fa95496 100644 --- a/pkg/log/writer.go +++ b/pkg/log/writer.go @@ -59,6 +59,16 @@ func (w *levelWriter) Write(p []byte) (int, error) { return len(p), nil } +func (w *levelWriter) Close() error { + w.mu.Lock() + defer w.mu.Unlock() + if w.buf.Len() > 0 { + w.logLine(w.buf.String()) + w.buf.Reset() + } + return nil +} + func (w *levelWriter) logLine(line string) { if line == "" { return @@ -72,13 +82,3 @@ func (w *levelWriter) logLine(line string) { Info(line) } } - -func (w *levelWriter) Close() error { - w.mu.Lock() - defer w.mu.Unlock() - if w.buf.Len() > 0 { - w.logLine(w.buf.String()) - w.buf.Reset() - } - return nil -} diff --git a/pkg/log/writer_test.go b/pkg/log/writer_test.go index aae50b3d2..e6ae3f3f4 100644 --- a/pkg/log/writer_test.go +++ b/pkg/log/writer_test.go @@ -6,8 +6,10 @@ import ( "testing" ) +const testFormatJSON = "json" + func TestWriter_EmitsStructuredJSONLine(t *testing.T) { - Init(Config{Verbosity: 2, Format: "json"}) + Init(Config{Verbosity: 2, Format: testFormatJSON}) var sink bytes.Buffer remove := AddSink(&sink) @@ -33,7 +35,7 @@ func TestWriter_EmitsStructuredJSONLine(t *testing.T) { } func TestWriter_SplitsMultipleLinesInOneWrite(t *testing.T) { - Init(Config{Verbosity: 2, Format: "json"}) + Init(Config{Verbosity: 2, Format: testFormatJSON}) var sink bytes.Buffer remove := AddSink(&sink) @@ -58,7 +60,7 @@ func TestWriter_SplitsMultipleLinesInOneWrite(t *testing.T) { } func TestWriter_FlushesTrailingPartialLineOnClose(t *testing.T) { - Init(Config{Verbosity: 2, Format: "json"}) + Init(Config{Verbosity: 2, Format: testFormatJSON}) var sink bytes.Buffer remove := AddSink(&sink) @@ -82,7 +84,7 @@ func TestWriter_FlushesTrailingPartialLineOnClose(t *testing.T) { } func TestWriter_DiscardsBelowConfiguredLevel(t *testing.T) { - Init(Config{Verbosity: 1, Format: "json"}) // info+ only, debug disabled + Init(Config{Verbosity: 1, Format: testFormatJSON}) // info+ only, debug disabled var sink bytes.Buffer remove := AddSink(&sink) From e2a84ec12e25e1aa3dcaf5ee29f7b2649a161425 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Tue, 28 Jul 2026 22:11:24 -0500 Subject: [PATCH 3/4] style: trim comment --- pkg/log/writer.go | 4 ---- 1 file changed, 4 deletions(-) diff --git a/pkg/log/writer.go b/pkg/log/writer.go index 18fa95496..b4d52d1e3 100644 --- a/pkg/log/writer.go +++ b/pkg/log/writer.go @@ -9,10 +9,6 @@ import ( "go.uber.org/zap/zapcore" ) -// Writer returns an io.WriteCloser that logs each complete line written to -// it as a structured entry at the given level, through the same encoder as -// every other log call. Primarily used as a subprocess's Stdout/Stderr. -// Close flushes any trailing line left without a newline. func Writer(level int) io.WriteCloser { return &levelWriter{level: verbosityConstToZapLevel(level)} } From 3b039178fd370fa3445f5457d5c7a128cf88e6eb Mon Sep 17 00:00:00 2001 From: Samuel K Date: Tue, 28 Jul 2026 23:07:43 -0500 Subject: [PATCH 4/4] fix(log): address CodeRabbit review on Writer - levelWriter.Write now scans only the bytes passed to that call for '\n' (via bytes.IndexByte) instead of resetting and re-copying the entire accumulated pending line on every partial write, which made writing one large unterminated line in many small chunks effectively O(n^2). Previously-seen bytes are never rescanned. - Bound the pending line to 64KiB: a subprocess that never emits a newline no longer grows levelWriter's buffer without limit. - logLine no longer discards empty lines. That guard was only ever needed to stop Close from emitting a phantom record when nothing was pending, which Close already handles itself via its buf.Len() check -- so it was silently dropping genuine blank lines from subprocess output. - Extended the multi-line test to cover a line split across two Write calls (as os/exec delivers chunks arbitrarily) and decode each emitted record to assert its exact msg, and added a test asserting blank lines are preserved as empty-message records. --- pkg/log/writer.go | 30 ++++++++++++-------- pkg/log/writer_test.go | 62 ++++++++++++++++++++++++++++++++++++++---- 2 files changed, 74 insertions(+), 18 deletions(-) diff --git a/pkg/log/writer.go b/pkg/log/writer.go index b4d52d1e3..c82f64074 100644 --- a/pkg/log/writer.go +++ b/pkg/log/writer.go @@ -3,12 +3,16 @@ package log import ( "bytes" "io" - "strings" "sync" "go.uber.org/zap/zapcore" ) +// maxPendingLine bounds how much unterminated output levelWriter buffers +// before logging it anyway, so a subprocess that never emits a newline +// can't grow the pending line without limit. +const maxPendingLine = 64 * 1024 + func Writer(level int) io.WriteCloser { return &levelWriter{level: verbosityConstToZapLevel(level)} } @@ -41,18 +45,23 @@ func (w *levelWriter) Write(p []byte) (int, error) { return len(p), nil // discard if below current level } - w.buf.Write(p) + total := len(p) for { - line, err := w.buf.ReadString('\n') - if err != nil { - // Incomplete line: put it back for the next Write/Close. - w.buf.Reset() - w.buf.WriteString(line) + i := bytes.IndexByte(p, '\n') + if i < 0 { break } - w.logLine(strings.TrimSuffix(line, "\n")) + w.buf.Write(p[:i]) + w.logLine(w.buf.String()) + w.buf.Reset() + p = p[i+1:] } - return len(p), nil + w.buf.Write(p) + if w.buf.Len() > maxPendingLine { + w.logLine(w.buf.String()) + w.buf.Reset() + } + return total, nil } func (w *levelWriter) Close() error { @@ -66,9 +75,6 @@ func (w *levelWriter) Close() error { } func (w *levelWriter) logLine(line string) { - if line == "" { - return - } switch w.level { case zapcore.DebugLevel: Debug(line) diff --git a/pkg/log/writer_test.go b/pkg/log/writer_test.go index e6ae3f3f4..3d90ce04e 100644 --- a/pkg/log/writer_test.go +++ b/pkg/log/writer_test.go @@ -2,6 +2,7 @@ package log import ( "bytes" + "encoding/json" "strings" "testing" ) @@ -46,16 +47,43 @@ func TestWriter_SplitsMultipleLinesInOneWrite(t *testing.T) { if err != nil { t.Fatalf("Write: %v", err) } + + // A line split across two Write calls (as os/exec delivers subprocess + // output in arbitrary chunks) must still be logged as one complete + // record, not two fragments. + if _, err := w.Write([]byte("line three\nline ")); err != nil { + t.Fatalf("Write: %v", err) + } + if _, err := w.Write([]byte("four\n")); err != nil { + t.Fatalf("Write: %v", err) + } _ = Sync() lines := strings.Split(strings.TrimSpace(sink.String()), "\n") - if len(lines) != 2 { - t.Fatalf("got %d lines, want 2: %q", len(lines), lines) + wantMsgs := []string{"line one", "line two", "line three", "line four"} + if len(lines) != len(wantMsgs) { + t.Fatalf("got %d lines, want %d: %q", len(lines), len(wantMsgs), lines) + } + for i, l := range lines { + assertJSONLineMsg(t, i, l, wantMsgs[i]) + } +} + +func assertJSONLineMsg(t *testing.T, i int, line, wantMsg string) { + t.Helper() + if !strings.HasPrefix(line, "{") || !strings.HasSuffix(line, "}") { + t.Errorf("line %d is not valid single-object JSON: %q", i, line) + return + } + var rec struct { + Msg string `json:"msg"` + } + if err := json.Unmarshal([]byte(line), &rec); err != nil { + t.Errorf("line %d: json.Unmarshal: %v", i, err) + return } - for _, l := range lines { - if !strings.HasPrefix(l, "{") || !strings.HasSuffix(l, "}") { - t.Errorf("line is not valid single-object JSON: %q", l) - } + if rec.Msg != wantMsg { + t.Errorf("line %d msg = %q, want %q", i, rec.Msg, wantMsg) } } @@ -83,6 +111,28 @@ func TestWriter_FlushesTrailingPartialLineOnClose(t *testing.T) { } } +func TestWriter_PreservesBlankLines(t *testing.T) { + Init(Config{Verbosity: 2, Format: testFormatJSON}) + + var sink bytes.Buffer + remove := AddSink(&sink) + defer remove() + + w := Writer(LevelInfo) + if _, err := w.Write([]byte("one\n\ntwo\n")); err != nil { + t.Fatalf("Write: %v", err) + } + _ = Sync() + + lines := strings.Split(strings.TrimSpace(sink.String()), "\n") + if len(lines) != 3 { + t.Fatalf("got %d lines, want 3 (blank line preserved): %q", len(lines), lines) + } + if !strings.Contains(lines[1], `"msg":""`) { + t.Errorf("line 1 = %q, want an empty msg field", lines[1]) + } +} + func TestWriter_DiscardsBelowConfiguredLevel(t *testing.T) { Init(Config{Verbosity: 1, Format: testFormatJSON}) // info+ only, debug disabled