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/fs/readfile_processor_optimized.go | 430 +++++++++++++++++++++++++ 1 file changed, 430 insertions(+) create mode 100644 internal/io/fs/readfile_processor_optimized.go (limited to 'internal/io/fs/readfile_processor_optimized.go') diff --git a/internal/io/fs/readfile_processor_optimized.go b/internal/io/fs/readfile_processor_optimized.go new file mode 100644 index 0000000..6426962 --- /dev/null +++ b/internal/io/fs/readfile_processor_optimized.go @@ -0,0 +1,430 @@ +package fs + +import ( + "bufio" + "bytes" + "context" + "io" + "os" + "time" + + "github.com/mimecast/dtail/internal/ctxutil" + "github.com/mimecast/dtail/internal/io/dlog" + "github.com/mimecast/dtail/internal/io/line" + "github.com/mimecast/dtail/internal/io/pool" + "github.com/mimecast/dtail/internal/lcontext" + "github.com/mimecast/dtail/internal/regex" +) + +// readWithProcessorOptimized reads from the file using buffered line reading +// instead of byte-by-byte reading for better performance +func (f *readFile) readWithProcessorOptimized(ctx context.Context, fd *os.File, reader *bufio.Reader, + truncate <-chan struct{}, ltx lcontext.LContext, processor line.Processor, re regex.Regex) error { + + // Create a line filter processor that wraps the given processor + filterProcessor := &filteringProcessor{ + processor: processor, + re: re, + ltx: ltx, + stats: &f.stats, + globID: f.globID, + } + + // Compute the local-context predicate once. When no context is requested we + // can take the zero-copy fast path (match before copy); when it is, every + // line must be buffered so surrounding before/after lines remain available. + hasContext := ltx.Has() + + // Use a scanner for efficient line reading + scanner := bufio.NewScanner(reader) + + // Get a buffer from the pool instead of allocating a new one + bufPtr := pool.GetScannerBuffer() + buf := *bufPtr + maxTokenSize := 1024 * 1024 // 1MB max token size + scanner.Buffer(buf, maxTokenSize) + + // Ensure we return the buffer to the pool when done + defer pool.PutScannerBuffer(bufPtr) + + // Use the cancellation-aware split function so long-line warnings can be + // abandoned if the caller cancels while the reader is blocked. + scanner.Split(func(data []byte, atEOF bool) (advance int, token []byte, err error) { + return f.scanLinesWithMaxLength(ctx, data, atEOF) + }) + + for scanner.Scan() { + // Check context cancellation + select { + case <-ctx.Done(): + return nil + default: + } + + // Check for file truncation. The periodicTruncateCheck goroutine + // (started in StartWithProcessorOptimized) already ticks every 3s and + // signals on the unbuffered truncate channel, so this non-blocking + // receive only re-stats the file on that cadence. Keeping the timing in + // the goroutine lets the per-line cost be a single atomic load on an + // empty channel (Go's non-blocking chanrecv fast path) instead of a + // per-line time.Since/runtime.nanotime call, which profiling showed as + // 10-18% of serverless dcat/dgrep CPU. This path is non-follow + // (cat/grep) only; follow mode uses tailWithProcessorOptimized, which + // has its own truncate handling. + select { + case <-truncate: + if isTruncated, err := f.truncated(fd); isTruncated { + return err + } + default: + } + + // Get the line data. scanner.Bytes() is only valid until the next + // Scan(); we must not retain it across iterations. + lineData := scanner.Bytes() + f.updatePosition() + + if !hasContext { + // Fast path: run the regex on the scanner's slice directly and only + // copy into a pooled buffer on a match. At low hit rates this skips + // the pool.Get + copy for the discarded (non-matching) lines. + if err := filterProcessor.ProcessFilteredRaw(lineData); err != nil { + if isEarlyStop(err) { + return nil + } + return err + } + continue + } + + // Local-context path: buffer every line (before/after context needs the + // surrounding non-matching lines), so copy into a pooled buffer first. + lineBuf := pool.BytesBuffer.Get().(*bytes.Buffer) + lineBuf.Write(lineData) + if err := filterProcessor.ProcessFilteredLine(lineBuf); err != nil { + if isEarlyStop(err) { + return nil + } + return err + } + } + + // Check for scanner errors + if err := scanner.Err(); err != nil { + // Handle EOF specially for tailing + if err == io.EOF && f.seekEOF { + // For tail mode, we want to keep reading + return nil + } + return err + } + + return nil +} + +// isEarlyStop reports whether err is the io.EOF sentinel that filteringProcessor +// returns from processWithContext once a max-count (-m/-max) limit is reached. +// This is a NORMAL early stop, not a genuine I/O error: bufio.Scanner signals +// real end-of-input via Scan()==false and never returns io.EOF from +// ProcessFiltered*, so any io.EOF bubbling up from the filter can only be the +// max-count sentinel. The byte-by-byte path (readWithProcessor) swallows it and +// returns nil; the optimized path must do the same, otherwise the sentinel leaks +// out to the caller and is logged as a spurious SERVER|...|ERROR|...|EOF line. +// Real (non-EOF) processor errors are left untouched so they still surface. +// +// Bare equality (== io.EOF) is intentional and must NOT become errors.Is: the +// sentinel is returned bare by processWithContext, so exact identity matches it +// precisely. A WRAPPED io.EOF, by contrast, can only originate from a genuine +// downstream failure (e.g. an ssh channel Write after the peer closed), which we +// deliberately do NOT want to mistake for a clean early stop. +func isEarlyStop(err error) bool { + return err == io.EOF +} + +// scanLinesPreserveEndings is a custom split function that preserves original line endings +// and respects MaxLineLength +func (f *readFile) scanLinesPreserveEndings(data []byte, atEOF bool) (advance int, token []byte, err error) { + if atEOF && len(data) == 0 { + return 0, nil, nil + } + + maxLineLen := f.lineLimit() + + // Look for a newline + if i := bytes.IndexByte(data, '\n'); i >= 0 { + // Check if the line before the newline exceeds max length + if i > maxLineLen { + // Line is too long, split it silently at maxLineLen + return maxLineLen, data[0:maxLineLen], nil + } + + // Line is within limit, include the line ending in the token + // Check if there's a \r before the \n + if i > 0 && data[i-1] == '\r' { + // Windows line ending (\r\n) - include both in token + return i + 1, data[0 : i+1], nil + } + // Unix line ending (\n) - include it in token + return i + 1, data[0 : i+1], nil + } + + // If we're at EOF, we have a final, non-terminated line + if atEOF { + if len(data) > maxLineLen { + // Even at EOF, respect max line length (split silently) + return maxLineLen, data[0:maxLineLen], nil + } + return len(data), data, nil + } + + // If the line is too long, split it + if len(data) >= maxLineLen { + // Return a chunk up to MaxLineLength (split silently) + return maxLineLen, data[0:maxLineLen], nil + } + + // Request more data + return 0, nil, nil +} + +// scanLinesWithMaxLength is a custom split function for bufio.Scanner that respects MaxLineLength. +// It is kept context-aware so long-line warnings can still be dropped when the reader is canceled. +func (f *readFile) scanLinesWithMaxLength(ctx context.Context, data []byte, atEOF bool) (advance int, token []byte, err error) { + if atEOF && len(data) == 0 { + return 0, nil, nil + } + + maxLineLen := f.lineLimit() + + // Look for a newline + if i := bytes.IndexByte(data, '\n'); i >= 0 { + // Check if the line before the newline exceeds max length + if i > maxLineLen { + // Line is too long, split it at maxLineLen + if !f.warnAboutLongLine(ctx) { + return 0, nil, ctx.Err() + } + return maxLineLen, data[0:maxLineLen], nil + } + // We have a full line within the limit + f.warnedAboutLongLine = false // Reset warning for next long line sequence + return i + 1, data[0 : i+1], nil + } + + // If we're at EOF, we have a final, non-terminated line + if atEOF { + if len(data) > maxLineLen { + // Even at EOF, respect max line length + if !f.warnAboutLongLine(ctx) { + return 0, nil, ctx.Err() + } + return maxLineLen, data[0:maxLineLen], nil + } + return len(data), data, nil + } + + // If the line is too long, split it + if len(data) >= maxLineLen { + // Warn about long line (only once) + if !f.warnAboutLongLine(ctx) { + return 0, nil, ctx.Err() + } + + // Return a chunk up to MaxLineLength + return maxLineLen, data[0:maxLineLen], nil + } + + // Request more data + return 0, nil, nil +} + +// StartWithProcessorOptimized starts reading a log file using an optimized LineProcessor implementation. +// This version uses buffered line reading instead of byte-by-byte reading. +func (f *readFile) StartWithProcessorOptimized(ctx context.Context, ltx lcontext.LContext, + processor line.Processor, re regex.Regex) error { + + reader, fd, decompressor, err := f.makeReader() + if fd != nil { + defer fd.Close() + } + if decompressor != nil { + defer func() { + if closeErr := decompressor.Close(); closeErr != nil { + dlog.Common.Warn(f.filePath, "Unable to close compressed reader", closeErr) + } + }() + } + if err != nil { + return err + } + + // Create a cancelable context for the truncate check goroutine + truncateCtx, cancelTruncate := context.WithCancel(ctx) + defer cancelTruncate() + + truncate := make(chan struct{}) + + go f.periodicTruncateCheck(truncateCtx, truncate) + + // For tail mode, we need to handle continuous reading + if f.seekEOF { + return f.tailWithProcessorOptimized(ctx, fd, reader, truncate, ltx, processor, re) + } + + // For cat/grep mode, just read once + err = f.readWithProcessorOptimized(ctx, fd, reader, truncate, ltx, processor, re) + + // Ensure any buffered data is flushed + if flushErr := processor.Flush(); flushErr != nil && err == nil { + err = flushErr + } + + return err +} + +// tailWithProcessorOptimized handles continuous reading for tail mode +func (f *readFile) tailWithProcessorOptimized(ctx context.Context, fd *os.File, reader *bufio.Reader, + truncate <-chan struct{}, ltx lcontext.LContext, processor line.Processor, re regex.Regex) error { + + // Create a line filter processor + filterProcessor := &filteringProcessor{ + processor: processor, + re: re, + ltx: ltx, + stats: &f.stats, + globID: f.globID, + } + + // Compute the local-context predicate once (see readWithProcessorOptimized): + // without context we take the zero-copy match-before-copy fast path. + hasContext := ltx.Has() + + // Buffer for partial lines + partialLine := pool.BytesBuffer.Get().(*bytes.Buffer) + defer pool.RecycleBytesBuffer(partialLine) + + // Get a buffer from the pool for reading + bufPtr := pool.GetMediumBuffer() + defer pool.PutMediumBuffer(bufPtr) + + // processPartialLine advances the line position and hands the currently + // accumulated partialLine to the filter. Without local context it takes the + // zero-copy fast path (match on partialLine.Bytes(), copy only on a match); + // with context it copies into a pooled buffer so surrounding lines stay + // buffered. partialLine is owned by this loop (reset after each call), so the + // fast path never retains its slice past the copy-on-match. + processPartialLine := func() error { + f.updatePosition() + if !hasContext { + return filterProcessor.ProcessFilteredRaw(partialLine.Bytes()) + } + lineBuf := pool.BytesBuffer.Get().(*bytes.Buffer) + lineBuf.Write(partialLine.Bytes()) + return filterProcessor.ProcessFilteredLine(lineBuf) + } + + for { + // Read available data using pooled buffer + buf := (*bufPtr)[:cap(*bufPtr)] // Reset to full capacity + n, err := reader.Read(buf) + + if n > 0 { + // Process the data we read + data := buf[:n] + + // Process complete lines + for len(data) > 0 { + // Find newline + idx := bytes.IndexByte(data, '\n') + + if idx >= 0 { + // Complete line found + partialLine.Write(data[:idx]) + + // Process the line if it's not empty + if partialLine.Len() > 0 { + if err := processPartialLine(); err != nil { + // Max-count early stop is a clean stop, not an error + // (see isEarlyStop); mirror the byte-by-byte path. + if isEarlyStop(err) { + return nil + } + return err + } + } + + partialLine.Reset() + data = data[idx+1:] + + // Reset long line warning + f.warnedAboutLongLine = false + } else { + // No newline, add to partial line + partialLine.Write(data) + + // Check if line is too long + if partialLine.Len() >= f.lineLimit() { + if !f.warnAboutLongLine(ctx) { + return nil + } + + // Process the partial line + if err := processPartialLine(); err != nil { + if isEarlyStop(err) { + return nil + } + return err + } + + partialLine.Reset() + } + + break + } + } + + // Flush processor periodically + if err := processor.Flush(); err != nil { + return err + } + } + + // Handle read errors + if err != nil { + if err != io.EOF { + return err + } + + waitForMoreData := true + + // EOF handling + select { + case <-ctx.Done(): + return nil + case <-truncate: + if isTruncated, err := f.truncated(fd); isTruncated { + return err + } + waitForMoreData = false + default: + } + + if waitForMoreData && !ctxutil.Sleep(ctx, 100*time.Millisecond) { + return nil + } + } + + // Check for cancellation + select { + case <-ctx.Done(): + // Process any remaining partial line + if partialLine.Len() > 0 { + if err := processPartialLine(); err != nil && !isEarlyStop(err) { + return err + } + } + return nil + default: + } + } +} -- cgit v1.2.3