summaryrefslogtreecommitdiff
path: root/internal/io/dlog/loggers
diff options
context:
space:
mode:
Diffstat (limited to 'internal/io/dlog/loggers')
-rw-r--r--internal/io/dlog/loggers/file.go122
-rw-r--r--internal/io/dlog/loggers/file_test.go183
-rw-r--r--internal/io/dlog/loggers/fout.go68
-rw-r--r--internal/io/dlog/loggers/fout_test.go199
-rw-r--r--internal/io/dlog/loggers/stdout.go86
-rw-r--r--internal/io/dlog/loggers/stdout_test.go190
6 files changed, 815 insertions, 33 deletions
diff --git a/internal/io/dlog/loggers/file.go b/internal/io/dlog/loggers/file.go
index 8e567bc..b3b6a10 100644
--- a/internal/io/dlog/loggers/file.go
+++ b/internal/io/dlog/loggers/file.go
@@ -12,6 +12,26 @@ import (
"github.com/mimecast/dtail/internal/config"
)
+const (
+ // fileWriterBufSize is the size of the bufio buffer wrapping the log file
+ // descriptor. A real buffer (instead of the old 1-byte writer that forced a
+ // write syscall per line) lets bulk payload — e.g. dcat/dgrep tee — batch
+ // into ~one syscall per bufferful, cutting the client receive-path syscall
+ // count and CPU by ~5x. bufio auto-flushes when full, so high-throughput
+ // output never stalls in the buffer.
+ fileWriterBufSize = 64 * 1024
+ // fileIdleFlushInterval bounds how long buffered data may sit unwritten when
+ // output goes idle (follow/interactive mode trickling a few lines). Without
+ // it, low-volume output would be stuck behind the buffer until it fills or
+ // the logger shuts down, so follow/tail would appear frozen on disk.
+ fileIdleFlushInterval = 100 * time.Millisecond
+ // fileFlushTimeout bounds how long a synchronous Flush() waits for the
+ // logger goroutine to acknowledge. It exists purely as a deadlock guard for
+ // the rare case where the goroutine is paused or already gone (e.g. Flush
+ // racing shutdown); under normal operation the ack is near-instant.
+ fileFlushTimeout = 2 * time.Second
+)
+
type fileMessageBuf struct {
now time.Time
message string
@@ -23,7 +43,13 @@ type file struct {
pauseCh chan struct{}
resumeCh chan struct{}
rotateCh chan struct{}
- flushCh chan struct{}
+ // flushCh carries a per-call reply channel so Flush() can block until the
+ // logger goroutine has actually drained the buffer channel and flushed the
+ // bufio writer to disk. This makes Flush() synchronous, which the crash path
+ // (dlog.FatalPanic -> Flush -> panic) relies on: an async signal could let
+ // the process unwind before the goroutine drains, dropping up to one buffer
+ // (64KB) of Fatal diagnostics.
+ flushCh chan chan struct{}
fd *os.File
writer *bufio.Writer
mutex sync.Mutex
@@ -32,13 +58,20 @@ type file struct {
strategy Strategy
}
+var _ Logger = (*file)(nil)
+
func newFile(strategy Strategy) *file {
+ // Pause/Resume/Rotate use capacity-1, non-blocking coalescing sends so
+ // callers never block on the logger goroutine (repeated signals collapse
+ // into one pending notification). flushCh is unbuffered and carries a reply
+ // channel because Flush() is synchronous: it must wait for the goroutine to
+ // drain and write before returning.
return &file{
bufferCh: make(chan *fileMessageBuf, runtime.NumCPU()*100),
- pauseCh: make(chan struct{}),
- resumeCh: make(chan struct{}),
- rotateCh: make(chan struct{}),
- flushCh: make(chan struct{}),
+ pauseCh: make(chan struct{}, 1),
+ resumeCh: make(chan struct{}, 1),
+ rotateCh: make(chan struct{}, 1),
+ flushCh: make(chan chan struct{}),
strategy: strategy,
}
}
@@ -67,17 +100,43 @@ func (f *file) Start(ctx context.Context, wg *sync.WaitGroup) {
go func() {
defer wg.Done()
+ // Idle-flush ticker: with a real (64KB) buffer, low-volume output
+ // (follow/interactive) would otherwise sit in the buffer until it
+ // fills. The ticker flushes any partial buffer promptly so follow/tail
+ // output reaches disk within fileIdleFlushInterval. flush() is cheap
+ // when nothing is buffered.
+ ticker := time.NewTicker(fileIdleFlushInterval)
+ defer ticker.Stop()
for {
select {
case m := <-f.bufferCh:
f.write(m)
+ case <-ticker.C:
+ f.flush()
case <-f.pauseCh:
+ // Flush before pausing so all output produced so far is on
+ // disk before the caller (e.g. an interactive prompt) writes
+ // directly to the terminal/file; preserves ordering.
+ f.flush()
pause(ctx)
- case <-f.flushCh:
+ case done := <-f.flushCh:
+ // Synchronous flush: drain + write, then acknowledge so the
+ // blocked Flush() caller can proceed (used by FatalPanic).
f.flush()
+ close(done)
+ case <-f.rotateCh:
+ // Force re-opening the outfile on the next write.
+ // Drained here (not only from write()) so that Rotate()
+ // makes progress even when no log messages arrive.
+ f.lastFileName = ""
case <-ctx.Done():
f.flush()
- f.fd.Close()
+ // f.fd is only populated after the first getWriter() call;
+ // guard against a nil pointer when the logger is shut down
+ // before anything has been written.
+ if f.fd != nil {
+ f.fd.Close()
+ }
return
}
}
@@ -100,21 +159,43 @@ func (f *file) RawWithColors(now time.Time, message, coloredMessage string) {
panic("Colors not supported in file logger")
}
-func (f *file) Pause() { f.pauseCh <- struct{}{} }
-func (f *file) Resume() { f.resumeCh <- struct{}{} }
-func (f *file) Flush() { f.flushCh <- struct{}{} }
+// signal performs a non-blocking, coalescing send on a capacity-1 control
+// channel. If a signal is already pending the new one is dropped, which is
+// the desired behaviour for idempotent operations such as Pause/Rotate/Flush.
+func signal(ch chan struct{}) {
+ select {
+ case ch <- struct{}{}:
+ default:
+ }
+}
-func (f *file) Rotate() { f.rotateCh <- struct{}{} }
-func (*file) SupportsColors() bool { return false }
+func (f *file) Pause() { signal(f.pauseCh) }
+func (f *file) Resume() { signal(f.resumeCh) }
+func (f *file) Rotate() { signal(f.rotateCh) }
-func (f *file) write(m *fileMessageBuf) {
+// Flush synchronously drains any queued messages and writes the bufio buffer to
+// disk, blocking until the logger goroutine acknowledges. The crash path
+// (dlog.FatalPanic) depends on this: with an async signal the process could
+// panic and unwind before the goroutine drained, losing buffered diagnostics.
+// A bounded timeout guards against a deadlock when the goroutine is paused or
+// has already exited (Flush racing shutdown), in which case the ctx.Done path
+// has already flushed or will flush.
+func (f *file) Flush() {
+ done := make(chan struct{})
select {
- case <-f.rotateCh:
- // Force re-opening the outfile next time in getWriter.
- f.lastFileName = ""
- default:
+ case f.flushCh <- done:
+ case <-time.After(fileFlushTimeout):
+ return
+ }
+ select {
+ case <-done:
+ case <-time.After(fileFlushTimeout):
}
+}
+
+func (*file) SupportsColors() bool { return false }
+func (f *file) write(m *fileMessageBuf) {
var writer *bufio.Writer
if f.strategy.Rotation == DailyRotation {
writer = f.getWriter(m.now.Format("20060102"))
@@ -150,9 +231,12 @@ func (f *file) getWriter(name string) *bufio.Writer {
f.writer.Flush()
f.fd.Close()
}
- // Set new writer.
+ // Set new writer. Use a real buffer (fileWriterBufSize) so bulk payload
+ // batches into few write syscalls instead of one-or-two per line. The
+ // logger goroutine's idle ticker and the ctx.Done/flush/pause paths keep
+ // low-volume and shutdown output from being stuck in the buffer.
f.fd = newFd
- f.writer = bufio.NewWriterSize(f.fd, 1)
+ f.writer = bufio.NewWriterSize(f.fd, fileWriterBufSize)
f.lastFileName = name
return f.writer
diff --git a/internal/io/dlog/loggers/file_test.go b/internal/io/dlog/loggers/file_test.go
new file mode 100644
index 0000000..0436327
--- /dev/null
+++ b/internal/io/dlog/loggers/file_test.go
@@ -0,0 +1,183 @@
+package loggers
+
+import (
+ "context"
+ "os"
+ "path/filepath"
+ "strconv"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/mimecast/dtail/internal/config"
+)
+
+// withTempLogDir points config.Common.LogDir at a fresh temp dir for the
+// duration of a test and restores the previous config afterwards. The file
+// logger resolves its output path from config.Common.LogDir at write time.
+func withTempLogDir(t *testing.T) string {
+ t.Helper()
+ dir := t.TempDir()
+ prev := config.Common
+ config.Common = &config.CommonConfig{LogDir: dir}
+ t.Cleanup(func() { config.Common = prev })
+ return dir
+}
+
+// startFileLogger starts f and returns a stop func that cancels the context and
+// JOINS the logger goroutine (wg.Wait). Tests must defer stop() so the goroutine
+// has fully exited before returning: withTempLogDir's t.Cleanup restores the
+// global config.Common, and a still-running goroutine reading config.Common.LogDir
+// would otherwise race that restore. For the same reason none of these tests may
+// call t.Parallel — they mutate the process-global config.Common.
+func startFileLogger(t *testing.T, f *file) func() {
+ t.Helper()
+ ctx, cancel := context.WithCancel(context.Background())
+ var wg sync.WaitGroup
+ wg.Add(1)
+ f.Start(ctx, &wg)
+ return func() {
+ cancel()
+ wg.Wait()
+ }
+}
+
+func readLogFile(t *testing.T, dir, base string) string {
+ t.Helper()
+ data, err := os.ReadFile(filepath.Join(dir, base+".log"))
+ if err != nil {
+ if os.IsNotExist(err) {
+ return ""
+ }
+ t.Fatalf("reading log file: %v", err)
+ }
+ return string(data)
+}
+
+// TestFileLoggerNothingLostOnClose verifies every logged line reaches disk when
+// the context is cancelled (clean shutdown): the goroutine drains the buffer
+// channel and flushes the 64KB writer before closing the fd.
+func TestFileLoggerNothingLostOnClose(t *testing.T) {
+ dir := withTempLogDir(t)
+ base := "close-test"
+ f := newFile(Strategy{Rotation: SignalRotation, FileBase: base})
+ stop := startFileLogger(t, f)
+
+ const n = 500
+ var want strings.Builder
+ for i := 0; i < n; i++ {
+ line := "line-" + strconv.Itoa(i)
+ f.Log(time.Now(), line)
+ want.WriteString(line + "\n")
+ }
+
+ // stop() cancels the context and joins the goroutine, which flushes and
+ // closes the fd on the way out — so all output must be on disk afterwards.
+ stop()
+
+ if got := readLogFile(t, dir, base); got != want.String() {
+ t.Fatalf("lost output on close: got %d bytes, want %d bytes",
+ len(got), want.Len())
+ }
+}
+
+// TestFileLoggerIdleFlush verifies a single low-volume line (follow/tail style)
+// is not stuck behind the 64KB buffer: the idle ticker flushes it to disk
+// promptly without any explicit Flush or shutdown. The logger goroutine is
+// joined via stop() before returning so it cannot outlive config.Common.
+func TestFileLoggerIdleFlush(t *testing.T) {
+ dir := withTempLogDir(t)
+ base := "idle-test"
+ f := newFile(Strategy{Rotation: SignalRotation, FileBase: base})
+ stop := startFileLogger(t, f)
+ defer stop()
+
+ f.Log(time.Now(), "follow-line")
+
+ deadline := time.Now().Add(2 * time.Second)
+ for time.Now().Before(deadline) {
+ if strings.Contains(readLogFile(t, dir, base), "follow-line") {
+ return
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+ t.Fatal("follow-style line stuck behind buffer; idle flush did not emit it")
+}
+
+// TestFileLoggerExplicitFlush verifies Flush() is SYNCHRONOUS: once it returns,
+// the buffered data is already on disk (no polling needed). This is the property
+// dlog.FatalPanic relies on to not drop Fatal diagnostics before panicking.
+func TestFileLoggerExplicitFlush(t *testing.T) {
+ dir := withTempLogDir(t)
+ base := "flush-test"
+ f := newFile(Strategy{Rotation: SignalRotation, FileBase: base})
+ stop := startFileLogger(t, f)
+ defer stop()
+
+ f.Log(time.Now(), "flush-me")
+ f.Flush()
+
+ if got := readLogFile(t, dir, base); !strings.Contains(got, "flush-me") {
+ t.Fatalf("synchronous Flush() did not persist data before returning; got %q", got)
+ }
+}
+
+// TestFileLoggerRotateDoesNotBlockWithoutWrites verifies that Rotate() does
+// not deadlock when no log messages have been produced. Previously rotateCh
+// was unbuffered and only drained opportunistically from write(), so a SIGHUP
+// before any Log() call would block the caller forever.
+func TestFileLoggerRotateDoesNotBlockWithoutWrites(t *testing.T) {
+ f := newFile(Strategy{Rotation: SignalRotation, FileBase: "unit-test"})
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ var wg sync.WaitGroup
+ wg.Add(1)
+ f.Start(ctx, &wg)
+
+ done := make(chan struct{})
+ go func() {
+ f.Rotate()
+ close(done)
+ }()
+
+ select {
+ case <-done:
+ case <-time.After(500 * time.Millisecond):
+ t.Fatal("Rotate() blocked without any writes; expected prompt return")
+ }
+
+ cancel()
+ wg.Wait()
+}
+
+// TestFileLoggerCancelBeforeFirstWriteDoesNotPanic verifies that cancelling
+// the context before any write has happened does not panic. Previously the
+// goroutine called f.fd.Close() unconditionally, but f.fd is only populated
+// by the first getWriter() call, so a ctx cancel with no prior writes
+// panicked on a nil pointer.
+func TestFileLoggerCancelBeforeFirstWriteDoesNotPanic(t *testing.T) {
+ f := newFile(Strategy{Rotation: SignalRotation, FileBase: "unit-test"})
+
+ ctx, cancel := context.WithCancel(context.Background())
+
+ var wg sync.WaitGroup
+ wg.Add(1)
+ f.Start(ctx, &wg)
+
+ cancel()
+
+ doneCh := make(chan struct{})
+ go func() {
+ wg.Wait()
+ close(doneCh)
+ }()
+
+ select {
+ case <-doneCh:
+ case <-time.After(1 * time.Second):
+ t.Fatal("file logger goroutine did not exit after ctx cancel")
+ }
+}
diff --git a/internal/io/dlog/loggers/fout.go b/internal/io/dlog/loggers/fout.go
index 6888d40..0d3fde5 100644
--- a/internal/io/dlog/loggers/fout.go
+++ b/internal/io/dlog/loggers/fout.go
@@ -4,16 +4,47 @@ import (
"context"
"sync"
"time"
+
+ "github.com/mimecast/dtail/internal/config"
)
+// fout logs to both a file and stdout. It is the default client logger.
+//
+// The two things a client emits are deliberately split at this seam:
+// - Diagnostics (connection INFO/WARN/ERROR/etc.) arrive via Log/LogWithColors
+// and are ALWAYS written to both stdout and the file — they are the small,
+// useful audit trail the daily log file is meant to keep.
+// - Retrieved payload (the bulk dcat/dgrep/dtail output) arrives via
+// Raw/RawWithColors. It always reaches stdout/terminal, but it is teed to
+// the file only when logPayload is set (opt-in via --log-payload /
+// Client.LogPayload). By default the file receives no payload, so a bulk
+// dcat no longer silently grows the daily log file by the full payload size.
type fout struct {
- file *file
- stdout *stdout
+ file Logger
+ stdout Logger
+ logPayload bool
}
-// Logs to both, a file and stdout
+// newFout builds the default client logger. Whether retrieved payload is teed
+// to the file is decided once at construction from the client config.
func newFout(strategy Strategy) *fout {
- return &fout{file: newFile(strategy), stdout: newStdout()}
+ return newFoutWithSinks(newFile(strategy), newStdout(), clientLogPayloadEnabled())
+}
+
+// newFoutWithSinks builds a fout over injectable sinks and an explicit payload
+// switch. Production uses newFout (concrete file+stdout, config-driven switch);
+// tests inject fakes to assert that diagnostics always reach the file while
+// payload reaches it only when opted in.
+func newFoutWithSinks(file, stdout Logger, logPayload bool) *fout {
+ return &fout{file: file, stdout: stdout, logPayload: logPayload}
+}
+
+// clientLogPayloadEnabled reports whether the client has opted in to teeing the
+// full retrieved payload into the daily log file. Default (false) keeps only
+// diagnostics in the file. config.Client is nil-guarded because a logger can be
+// constructed in early/unit contexts before config.Setup has populated it.
+func clientLogPayloadEnabled() bool {
+ return config.Client != nil && config.Client.LogPayload
}
func (f *fout) Start(ctx context.Context, wg *sync.WaitGroup) {
@@ -35,17 +66,42 @@ func (f *fout) Log(now time.Time, message string) {
func (f *fout) LogWithColors(now time.Time, message, coloredMessage string) {
f.stdout.LogWithColors(now, "", coloredMessage)
+ // The file logger does not support colors, so write the plain message via
+ // Log (its LogWithColors would route to RawWithColors, which panics).
f.file.Log(now, message)
}
+// Raw writes retrieved payload. It always reaches stdout/terminal; it is teed
+// to the file sink only when the client opted in via --log-payload /
+// Client.LogPayload. By default the file is left payload-free.
func (f *fout) Raw(now time.Time, message string) {
f.stdout.Raw(now, message)
- f.file.Raw(now, message)
+ if f.logPayload {
+ f.file.Raw(now, message)
+ }
}
func (f *fout) RawWithColors(now time.Time, message, coloredMessage string) {
f.stdout.RawWithColors(now, "", coloredMessage)
- f.file.Raw(now, message)
+ // Same opt-in gate as Raw; the file gets the plain (uncolored) payload.
+ if f.logPayload {
+ f.file.Raw(now, message)
+ }
+}
+
+// RawFileOnly tees retrieved payload into the daily log FILE sink only, never to
+// stdout, honoring the same --log-payload / Client.LogPayload opt-in as Raw.
+//
+// It exists for the serverless direct-output path: that path writes
+// payload straight to its own stdout sink and bypasses Raw entirely, so without
+// this hook --log-payload would silently no longer tee payload to the file in
+// serverless mode. The caller (the serverless output writer) already emits the
+// payload bytes to stdout itself, so this method deliberately writes ONLY to the
+// file to keep stdout byte-identical whether or not --log-payload is set.
+func (f *fout) RawFileOnly(now time.Time, message string) {
+ if f.logPayload {
+ f.file.Raw(now, message)
+ }
}
func (f *fout) Flush() { f.stdout.Flush(); f.file.Flush() }
diff --git a/internal/io/dlog/loggers/fout_test.go b/internal/io/dlog/loggers/fout_test.go
new file mode 100644
index 0000000..0788f01
--- /dev/null
+++ b/internal/io/dlog/loggers/fout_test.go
@@ -0,0 +1,199 @@
+package loggers
+
+import (
+ "context"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/mimecast/dtail/internal/config"
+)
+
+// recordingSink is an injectable Logger that records which messages reached it
+// via the diagnostic path (Log) versus the payload path (Raw). It lets the fout
+// routing tests assert exactly which sink gets diagnostics and which gets
+// payload without touching real files or stdout.
+type recordingSink struct {
+ mutex sync.Mutex
+ logs []string
+ raws []string
+}
+
+func (r *recordingSink) Log(now time.Time, message string) {
+ r.mutex.Lock()
+ defer r.mutex.Unlock()
+ r.logs = append(r.logs, message)
+}
+
+func (r *recordingSink) LogWithColors(now time.Time, message, colored string) {
+ r.Log(now, message)
+}
+
+func (r *recordingSink) Raw(now time.Time, message string) {
+ r.mutex.Lock()
+ defer r.mutex.Unlock()
+ r.raws = append(r.raws, message)
+}
+
+func (r *recordingSink) RawWithColors(now time.Time, message, colored string) {
+ r.Raw(now, message)
+}
+
+func (r *recordingSink) Start(ctx context.Context, wg *sync.WaitGroup) { wg.Done() }
+func (r *recordingSink) Flush() {}
+func (r *recordingSink) Pause() {}
+func (r *recordingSink) Resume() {}
+func (r *recordingSink) Rotate() {}
+func (r *recordingSink) SupportsColors() bool { return false }
+
+func (r *recordingSink) logCount() int {
+ r.mutex.Lock()
+ defer r.mutex.Unlock()
+ return len(r.logs)
+}
+
+func (r *recordingSink) rawCount() int {
+ r.mutex.Lock()
+ defer r.mutex.Unlock()
+ return len(r.raws)
+}
+
+var _ Logger = (*recordingSink)(nil)
+
+// TestFoutDefaultKeepsPayloadOutOfFile proves the footgun fix: with LogPayload
+// disabled (the default), diagnostics reach the file sink but retrieved payload
+// does not — while stdout still receives BOTH, so the terminal output is
+// unchanged.
+func TestFoutDefaultKeepsPayloadOutOfFile(t *testing.T) {
+ file := &recordingSink{}
+ stdout := &recordingSink{}
+ f := newFoutWithSinks(file, stdout, false)
+
+ now := time.Now()
+ f.Log(now, "diagnostic-line") // connection INFO/WARN/ERROR audit line
+ f.Raw(now, "payload-line-1\n") // bulk dcat/dgrep/dtail output
+ f.Raw(now, "payload-line-2\n")
+
+ // File: exactly the diagnostic, no payload.
+ if got := file.logCount(); got != 1 {
+ t.Fatalf("file diagnostics: got %d, want 1", got)
+ }
+ if got := file.rawCount(); got != 0 {
+ t.Fatalf("file payload leaked: got %d raws, want 0 (default must be diagnostics-only)", got)
+ }
+ // Stdout: diagnostic + both payload lines (terminal output unchanged).
+ if got := stdout.logCount(); got != 1 {
+ t.Fatalf("stdout diagnostics: got %d, want 1", got)
+ }
+ if got := stdout.rawCount(); got != 2 {
+ t.Fatalf("stdout payload: got %d, want 2", got)
+ }
+}
+
+// TestFoutOptInTeesPayloadToFile proves --log-payload / Client.LogPayload
+// restores the legacy behaviour: payload is teed to the file too, and
+// diagnostics still land in the file.
+func TestFoutOptInTeesPayloadToFile(t *testing.T) {
+ file := &recordingSink{}
+ stdout := &recordingSink{}
+ f := newFoutWithSinks(file, stdout, true)
+
+ now := time.Now()
+ f.Log(now, "diagnostic-line")
+ f.Raw(now, "payload-line-1\n")
+ f.Raw(now, "payload-line-2\n")
+
+ if got := file.logCount(); got != 1 {
+ t.Fatalf("file diagnostics: got %d, want 1", got)
+ }
+ if got := file.rawCount(); got != 2 {
+ t.Fatalf("file payload with opt-in: got %d, want 2", got)
+ }
+ if got := stdout.rawCount(); got != 2 {
+ t.Fatalf("stdout payload: got %d, want 2", got)
+ }
+}
+
+// TestFoutServerErrorDiagnosticReachesFileByDefault is the end-to-end guard for
+// the ReportServerError footgun: a server-error audit line, sent via the
+// diagnostic (Log) path over a REAL on-disk file sink with LogPayload=false,
+// must actually land in the daily log file — while bulk payload (Raw) must not.
+// This exercises the real file-write path the regression would have skipped.
+func TestFoutServerErrorDiagnosticReachesFileByDefault(t *testing.T) {
+ tmp := t.TempDir()
+ prevCommon := config.Common
+ config.Common = &config.CommonConfig{LogDir: tmp}
+ t.Cleanup(func() { config.Common = prevCommon })
+
+ // SignalRotation with a fixed FileBase gives a deterministic file name.
+ fileSink := newFile(Strategy{Rotation: SignalRotation, FileBase: "servererr"})
+ stdout := &recordingSink{}
+ f := newFoutWithSinks(fileSink, stdout, false)
+
+ ctx, cancel := context.WithCancel(context.Background())
+ var wg sync.WaitGroup
+ wg.Add(1)
+ f.Start(ctx, &wg)
+
+ now := time.Now()
+ serverError := "SERVER|srv1|ERROR|journal file targets require server capability journal-v1"
+ f.Log(now, serverError) // diagnostic / audit line (ReportServerError path)
+ f.Raw(now, "payload\n") // bulk payload, must stay out of the file
+
+ f.Flush()
+ cancel()
+ wg.Wait()
+
+ content, err := os.ReadFile(filepath.Join(tmp, "servererr.log"))
+ if err != nil {
+ t.Fatalf("reading log file: %v", err)
+ }
+ if !strings.Contains(string(content), serverError) {
+ t.Fatalf("server-error diagnostic missing from default log file:\n%s", content)
+ }
+ if strings.Contains(string(content), "payload") {
+ t.Fatalf("payload leaked into the default log file:\n%s", content)
+ }
+ // Stdout still receives both (terminal output unchanged).
+ if stdout.logCount() != 1 || stdout.rawCount() != 1 {
+ t.Fatalf("stdout must receive both diagnostic and payload; logs=%d raws=%d",
+ stdout.logCount(), stdout.rawCount())
+ }
+}
+
+// TestFoutWithColorsRouting mirrors the two tests above for the colored paths
+// (LogWithColors always to file, RawWithColors gated by the opt-in), because the
+// client uses the *WithColors variants when terminal colors are enabled.
+func TestFoutWithColorsRouting(t *testing.T) {
+ now := time.Now()
+
+ t.Run("default", func(t *testing.T) {
+ file := &recordingSink{}
+ stdout := &recordingSink{}
+ f := newFoutWithSinks(file, stdout, false)
+ f.LogWithColors(now, "diag", "\x1b[1mdiag\x1b[0m")
+ f.RawWithColors(now, "payload\n", "\x1b[1mpayload\x1b[0m\n")
+ if got := file.logCount(); got != 1 {
+ t.Fatalf("file diagnostics: got %d, want 1", got)
+ }
+ if got := file.rawCount(); got != 0 {
+ t.Fatalf("file payload leaked: got %d, want 0", got)
+ }
+ if got := stdout.rawCount(); got != 1 {
+ t.Fatalf("stdout payload: got %d, want 1", got)
+ }
+ })
+
+ t.Run("optin", func(t *testing.T) {
+ file := &recordingSink{}
+ stdout := &recordingSink{}
+ f := newFoutWithSinks(file, stdout, true)
+ f.RawWithColors(now, "payload\n", "\x1b[1mpayload\x1b[0m\n")
+ if got := file.rawCount(); got != 1 {
+ t.Fatalf("file payload with opt-in: got %d, want 1", got)
+ }
+ })
+}
diff --git a/internal/io/dlog/loggers/stdout.go b/internal/io/dlog/loggers/stdout.go
index b024243..7067305 100644
--- a/internal/io/dlog/loggers/stdout.go
+++ b/internal/io/dlog/loggers/stdout.go
@@ -1,27 +1,76 @@
package loggers
import (
+ "bufio"
"context"
"fmt"
+ "io"
+ "os"
"sync"
"time"
)
+const (
+ // stdoutWriterBufSize is the size of the bufio buffer wrapping os.Stdout.
+ // The old path did one fmt.Println (one write syscall) per received line;
+ // buffering lets bulk payload batch into ~one syscall per bufferful. bufio
+ // auto-flushes when full so high-throughput output never stalls.
+ stdoutWriterBufSize = 64 * 1024
+ // stdoutIdleFlushInterval bounds how long buffered output may sit unwritten
+ // when output goes idle (follow/interactive trickling a few lines). Without
+ // it, low-volume output would be stuck behind the buffer, so follow/tail
+ // would appear frozen on the terminal.
+ stdoutIdleFlushInterval = 100 * time.Millisecond
+)
+
type stdout struct {
pauseCh chan struct{}
resumeCh chan struct{}
+ writer *bufio.Writer
mutex sync.Mutex
}
+var _ Logger = (*stdout)(nil)
+
func newStdout() *stdout {
+ return newStdoutWriter(os.Stdout)
+}
+
+// newStdoutWriter builds a stdout logger over an arbitrary sink. Production
+// uses os.Stdout; tests inject a counting writer to assert that buffering
+// batches many lines into few underlying writes. The bufio writer is created
+// eagerly so the logger is usable even when Start() is never called (e.g. in
+// isolated unit tests); idle/shutdown flushing is only driven once Start()
+// spawns the flush goroutine.
+func newStdoutWriter(w io.Writer) *stdout {
return &stdout{
pauseCh: make(chan struct{}),
resumeCh: make(chan struct{}),
+ writer: bufio.NewWriterSize(w, stdoutWriterBufSize),
}
}
func (s *stdout) Start(ctx context.Context, wg *sync.WaitGroup) {
- wg.Done()
+ // Background flusher: with a real buffer, low-volume (follow/interactive)
+ // output would otherwise sit unwritten until the buffer fills. The ticker
+ // flushes any partial buffer promptly, and ctx.Done triggers a final flush
+ // so no buffered output is lost on clean shutdown. wg.Done is deferred to
+ // the goroutine so callers (ClientRuntime.Stop -> wg.Wait) block until the
+ // final flush has happened.
+ go func() {
+ defer wg.Done()
+ ticker := time.NewTicker(stdoutIdleFlushInterval)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-ticker.C:
+ s.Flush()
+ case <-ctx.Done():
+ s.Flush()
+ return
+ }
+ }
+ }()
}
func (s *stdout) Log(now time.Time, message string) {
@@ -42,27 +91,48 @@ func (s *stdout) RawWithColors(now time.Time, message, coloredMessage string) {
func (s *stdout) log(message string, nl bool) {
s.mutex.Lock()
- defer s.mutex.Unlock()
-
select {
case <-s.pauseCh:
- // Pause until resumed.
+ // Wait for Resume without holding the mutex: the prompt path calls
+ // dlog after the user answers while Pause is still active; holding the
+ // mutex here would deadlock (Info blocks on Lock, Resume never runs).
+ s.mutex.Unlock()
<-s.resumeCh
+ s.mutex.Lock()
default:
}
+ defer s.mutex.Unlock()
+ // Buffered writes: fmt.Fprint(ln) into the bufio.Writer batches many lines
+ // into one write syscall. Errors are intentionally ignored — a logger that
+ // cannot write to stdout has nowhere to report the failure.
if nl {
- fmt.Println(message)
+ _, _ = fmt.Fprintln(s.writer, message)
return
}
- fmt.Print(message)
+ _, _ = fmt.Fprint(s.writer, message)
+}
+
+func (s *stdout) Pause() {
+ // Flush before pausing so all output produced so far is visible before the
+ // caller (interactive prompt / stats interrupt) writes directly to stdout,
+ // preserving the ordering the unbuffered path used to give for free. The
+ // pauseCh handshake below is unchanged so the pause semantics (and the
+ // deadlock guarantees exercised by the unit tests) are preserved.
+ s.mutex.Lock()
+ _ = s.writer.Flush()
+ s.mutex.Unlock()
+ s.pauseCh <- struct{}{}
}
-func (s *stdout) Pause() { s.pauseCh <- struct{}{} }
func (s *stdout) Resume() { s.resumeCh <- struct{}{} }
func (s *stdout) Flush() {
- // This is empty because it isn't doing anything but has to satisfy the interface.
+ s.mutex.Lock()
+ defer s.mutex.Unlock()
+ // bufio.Flush is a no-op when nothing is buffered, so calling this on every
+ // idle tick is cheap.
+ _ = s.writer.Flush()
}
func (s *stdout) Rotate() {
diff --git a/internal/io/dlog/loggers/stdout_test.go b/internal/io/dlog/loggers/stdout_test.go
new file mode 100644
index 0000000..af8c9e5
--- /dev/null
+++ b/internal/io/dlog/loggers/stdout_test.go
@@ -0,0 +1,190 @@
+package loggers
+
+import (
+ "bytes"
+ "context"
+ "strconv"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+)
+
+// countingWriter records how many times Write is called and accumulates all
+// bytes, so tests can assert that buffering batches many logical lines into a
+// small number of underlying writes while preserving content and order.
+type countingWriter struct {
+ mutex sync.Mutex
+ buf bytes.Buffer
+ writes int
+}
+
+func (c *countingWriter) Write(p []byte) (int, error) {
+ c.mutex.Lock()
+ defer c.mutex.Unlock()
+ c.writes++
+ return c.buf.Write(p)
+}
+
+func (c *countingWriter) String() string {
+ c.mutex.Lock()
+ defer c.mutex.Unlock()
+ return c.buf.String()
+}
+
+func (c *countingWriter) Writes() int {
+ c.mutex.Lock()
+ defer c.mutex.Unlock()
+ return c.writes
+}
+
+// TestStdoutBuffersAndPreservesOrder proves the buffered stdout path batches
+// many lines into far fewer underlying writes than one-per-line, and that a
+// Flush emits the exact content in order (nothing dropped or reordered).
+func TestStdoutBuffersAndPreservesOrder(t *testing.T) {
+ cw := &countingWriter{}
+ s := newStdoutWriter(cw)
+
+ const n = 1000
+ var want strings.Builder
+ for i := 0; i < n; i++ {
+ line := "line-" + strconv.Itoa(i)
+ s.Raw(time.Now(), line+"\n")
+ want.WriteString(line + "\n")
+ }
+
+ // Before flush the small lines must still be batched in the bufio buffer:
+ // with per-line writes this would already be n writes.
+ if got := cw.Writes(); got >= n {
+ t.Fatalf("expected buffering to batch writes, got %d writes for %d lines", got, n)
+ }
+
+ s.Flush()
+
+ if got := cw.String(); got != want.String() {
+ t.Fatalf("content mismatch after flush:\n got %q\nwant %q", got, want.String())
+ }
+ // 1000 short lines fit in a handful of 64KB flushes, definitely far below n.
+ if got := cw.Writes(); got > n/10 {
+ t.Fatalf("expected far fewer than %d writes, got %d", n/10, got)
+ }
+}
+
+// TestStdoutFlushOnPause proves output produced before Pause() is flushed to
+// the sink before Pause returns, so an interactive prompt writing directly to
+// the terminal never appears ahead of already-logged output.
+func TestStdoutFlushOnPause(t *testing.T) {
+ cw := &countingWriter{}
+ s := newStdoutWriter(cw)
+
+ s.Raw(time.Now(), "before-pause\n")
+
+ // Pause blocks on the pauseCh handshake until a concurrent log() consumes
+ // the token (the existing pause semantics). Drive a stream of log() calls
+ // so one is guaranteed to pick up the token and let Pause() return,
+ // mirroring the real prompt flow where logging goroutines are active.
+ paused := make(chan struct{})
+ go func() {
+ s.Pause()
+ close(paused)
+ }()
+ go func() {
+ for {
+ select {
+ case <-paused:
+ return
+ default:
+ s.Log(time.Now(), "consumes-pause")
+ time.Sleep(time.Millisecond)
+ }
+ }
+ }()
+ <-paused
+
+ // Pause() flushes before the handshake, so the pre-pause line must already
+ // be in the sink now regardless of the buffer.
+ if got := cw.String(); !strings.Contains(got, "before-pause") {
+ t.Fatalf("expected buffered output flushed on Pause, got %q", got)
+ }
+ s.Resume()
+}
+
+// TestStdoutIdleFlush proves that a single low-volume line (follow/tail style)
+// is not stuck behind the buffer: the Start() idle ticker flushes it promptly
+// without any explicit Flush call.
+func TestStdoutIdleFlush(t *testing.T) {
+ cw := &countingWriter{}
+ s := newStdoutWriter(cw)
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ var wg sync.WaitGroup
+ wg.Add(1)
+ s.Start(ctx, &wg)
+
+ s.Raw(time.Now(), "follow-line\n")
+
+ deadline := time.Now().Add(2 * time.Second)
+ for time.Now().Before(deadline) {
+ if strings.Contains(cw.String(), "follow-line") {
+ cancel()
+ wg.Wait()
+ return
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+ cancel()
+ wg.Wait()
+ t.Fatal("follow-style line stuck behind buffer; idle flush did not emit it")
+}
+
+// TestStdoutFinalFlushOnClose proves no buffered output is lost on clean
+// shutdown: data logged just before ctx cancel is flushed before the Start
+// goroutine (and thus wg.Wait) returns.
+func TestStdoutFinalFlushOnClose(t *testing.T) {
+ cw := &countingWriter{}
+ s := newStdoutWriter(cw)
+
+ ctx, cancel := context.WithCancel(context.Background())
+ var wg sync.WaitGroup
+ wg.Add(1)
+ s.Start(ctx, &wg)
+
+ s.Raw(time.Now(), "last-line-before-exit\n")
+ cancel()
+ wg.Wait()
+
+ if got := cw.String(); !strings.Contains(got, "last-line-before-exit") {
+ t.Fatalf("buffered output lost on shutdown, got %q", got)
+ }
+}
+
+// Regression: during an interactive prompt, dlog.Common.Pause() unblocks when some
+// goroutine hits stdout.log(); that goroutine must not hold the stdout mutex while
+// waiting on resume, or dlog.Client.Info from the prompt callback deadlocks forever.
+func TestStdoutSecondLogDuringPauseWaitDoesNotDeadlock(t *testing.T) {
+ s := newStdout()
+
+ go s.Pause()
+ time.Sleep(50 * time.Millisecond)
+
+ go func() {
+ s.Log(time.Now(), "first log consumes pause and waits on resume")
+ }()
+ time.Sleep(50 * time.Millisecond)
+
+ secondDone := make(chan struct{})
+ go func() {
+ s.Log(time.Now(), "second log must acquire mutex while first waits for Resume")
+ close(secondDone)
+ }()
+
+ select {
+ case <-secondDone:
+ case <-time.After(2 * time.Second):
+ t.Fatal("deadlock: second Log blocked on mutex while first waits for Resume")
+ }
+
+ s.Resume()
+ time.Sleep(50 * time.Millisecond)
+}
generated by cgit v1.2.3 (git 2.25.1) at 2026-08-07 03:23:20 +0000