From 849951be1d1a7ee9f9302006ccb187bf5b4e36f3 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Wed, 22 Jul 2026 23:51:18 +0300 Subject: =?UTF-8?q?feat:=20DTail=20fork=20=E2=80=94=20server/client=20feat?= =?UTF-8?q?ure=20development?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Squashed development of the snonux/dtail fork's product code (internal/, cmd/) since diverging from mimecast/dtail. Major areas: - Read/output path: the former "turbo" channel-less path is now the single, default server-side read/output path for cat/grep/tail and MapReduce; the old channel-based path and its config/env toggles were removed. - MapReduce: single aggregate implementation (server + serverless) fed directly by a processor pipeline, with input-exhausted finalization via the shutdown coordinator; high-concurrency and data-race fixes. - Journal source reads (journal:unit.service) via journalctl, Linux-gated behind a journal-v1 capability. - Auth-key fast reconnect: in-memory per-user public-key cache with TTL/max-keys, registered over an authenticated session (AUTHKEY), checked before authorized_keys. - Interactive query reload (--interactive-query) with SESSION START/UPDATE generation boundaries and capability negotiation. - Client-side deadlines: --timeout / --shutdownAfter as context deadlines; follow shutdown handling. - Client logging: diagnostics-only daily log by default, opt-in payload tee via --log-payload. - Numerous correctness fixes (buffer-pool double-recycle races, EOF-sentinel leaks, glob-expansion cap, TOCTOU in CSV parsing) with accompanying unit tests. Co-Authored-By: Claude Opus 4.8 --- internal/io/dlog/loggers/stdout.go | 86 ++++++++++++++++++++++++++++++++++---- 1 file changed, 78 insertions(+), 8 deletions(-) (limited to 'internal/io/dlog/loggers/stdout.go') 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() { -- cgit v1.2.3