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/catfile.go | 14 +- internal/io/fs/filereader.go | 8 +- internal/io/fs/permissions/permission.go | 1 - internal/io/fs/permissions/permission_linuxacl.go | 1 - internal/io/fs/permissions/permission_test.go | 1 - internal/io/fs/readfile.go | 238 ++---- internal/io/fs/readfile_nozstd.go | 16 + internal/io/fs/readfile_processor.go | 359 +++++++++ internal/io/fs/readfile_processor_optimized.go | 430 +++++++++++ internal/io/fs/readfile_processor_test.go | 869 ++++++++++++++++++++++ internal/io/fs/readfile_zstd.go | 20 + internal/io/fs/readfilelcontext.go | 209 ------ internal/io/fs/rootedpath.go | 96 +++ internal/io/fs/rootedpath_test.go | 57 ++ internal/io/fs/tailfile.go | 14 +- internal/io/fs/validatedreadtarget.go | 148 ++++ internal/io/fs/validatedreadtarget_test.go | 218 ++++++ 17 files changed, 2297 insertions(+), 402 deletions(-) create mode 100644 internal/io/fs/readfile_nozstd.go create mode 100644 internal/io/fs/readfile_processor.go create mode 100644 internal/io/fs/readfile_processor_optimized.go create mode 100644 internal/io/fs/readfile_processor_test.go create mode 100644 internal/io/fs/readfile_zstd.go delete mode 100644 internal/io/fs/readfilelcontext.go create mode 100644 internal/io/fs/rootedpath.go create mode 100644 internal/io/fs/rootedpath_test.go create mode 100644 internal/io/fs/validatedreadtarget.go create mode 100644 internal/io/fs/validatedreadtarget_test.go (limited to 'internal/io/fs') diff --git a/internal/io/fs/catfile.go b/internal/io/fs/catfile.go index e4676f3..ac42fc0 100644 --- a/internal/io/fs/catfile.go +++ b/internal/io/fs/catfile.go @@ -6,7 +6,9 @@ type CatFile struct { } // NewCatFile returns a new file catter. -func NewCatFile(filePath string, globID string, serverMessages chan<- string) CatFile { +func NewCatFile(filePath string, globID string, serverMessages chan<- string, + maxLineLength int) CatFile { + return CatFile{ readFile: readFile{ filePath: filePath, @@ -15,6 +17,16 @@ func NewCatFile(filePath string, globID string, serverMessages chan<- string) Ca retry: false, canSkipLines: false, seekEOF: false, + maxLineLength: maxLineLength, }, } } + +// NewValidatedCatFile returns a new file catter backed by a rooted open target. +func NewValidatedCatFile(filePath string, target ValidatedReadTarget, globID string, + serverMessages chan<- string, maxLineLength int) CatFile { + + cat := NewCatFile(filePath, globID, serverMessages, maxLineLength) + cat.readFile.validatedTarget = &target + return cat +} diff --git a/internal/io/fs/filereader.go b/internal/io/fs/filereader.go index e27d2a7..7f36a29 100644 --- a/internal/io/fs/filereader.go +++ b/internal/io/fs/filereader.go @@ -9,9 +9,13 @@ import ( ) // FileReader is the interface used on the dtail server to read/cat/grep/mapr... -// a file. +// a file. Line delivery is processor-based (line.Processor); the historic +// channel-based Start(chan<- *line.Line) method was removed once every read path +// migrated to the processor pipeline (task iv0). type FileReader interface { - Start(ctx context.Context, ltx lcontext.LContext, lines chan<- *line.Line, + StartWithProcessor(ctx context.Context, ltx lcontext.LContext, processor line.Processor, + re regex.Regex) error + StartWithProcessorOptimized(ctx context.Context, ltx lcontext.LContext, processor line.Processor, re regex.Regex) error FilePath() string Retry() bool diff --git a/internal/io/fs/permissions/permission.go b/internal/io/fs/permissions/permission.go index aaab9e7..889fe9e 100644 --- a/internal/io/fs/permissions/permission.go +++ b/internal/io/fs/permissions/permission.go @@ -1,5 +1,4 @@ //go:build !linuxacl -// +build !linuxacl package permissions diff --git a/internal/io/fs/permissions/permission_linuxacl.go b/internal/io/fs/permissions/permission_linuxacl.go index bfac7e2..0a334e6 100644 --- a/internal/io/fs/permissions/permission_linuxacl.go +++ b/internal/io/fs/permissions/permission_linuxacl.go @@ -1,5 +1,4 @@ //go:build linuxacl -// +build linuxacl package permissions diff --git a/internal/io/fs/permissions/permission_test.go b/internal/io/fs/permissions/permission_test.go index e28c67b..3d9f8e3 100644 --- a/internal/io/fs/permissions/permission_test.go +++ b/internal/io/fs/permissions/permission_test.go @@ -1,5 +1,4 @@ //go:build linuxacl -// +build linuxacl package permissions diff --git a/internal/io/fs/readfile.go b/internal/io/fs/readfile.go index dc1d8ea..7969cf0 100644 --- a/internal/io/fs/readfile.go +++ b/internal/io/fs/readfile.go @@ -2,7 +2,6 @@ package fs import ( "bufio" - "bytes" "compress/gzip" "context" "errors" @@ -10,25 +9,18 @@ import ( "io" "os" "strings" - "sync" "time" - "github.com/mimecast/dtail/internal/config" "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" - - "github.com/DataDog/zstd" ) type readStatus int const ( - nothing readStatus = iota - abortReading readStatus = iota - continueReading readStatus = iota + nothing readStatus = iota + abortReading readStatus = iota + continueReading readStatus = iota + defaultMaxLineLength = 1024 * 1024 ) // Used to tail and filter a local log file. @@ -37,6 +29,8 @@ type readFile struct { stats // Path of log file to tail. filePath string + // Rooted target used for validated server-side re-opens. + validatedTarget *ValidatedReadTarget // The glob identifier of the file. globID string // Channel to send a server message to the dtail client @@ -49,6 +43,8 @@ type readFile struct { seekEOF bool // Warned already about a long line. warnedAboutLongLine bool + // Maximum line length before a line is split. + maxLineLength int } // String returns the string representation of the readFile @@ -72,52 +68,42 @@ func (f readFile) Retry() bool { return f.retry } -// Start tailing a log file. -func (f readFile) Start(ctx context.Context, ltx lcontext.LContext, - lines chan<- *line.Line, re regex.Regex) error { - - reader, fd, err := f.makeReader() - if fd != nil { - defer fd.Close() +func (f *readFile) lineLimit() int { + if f.maxLineLength <= 0 { + return defaultMaxLineLength } - if err != nil { - return err - } - - rawLines := make(chan *bytes.Buffer, 100) - truncate := make(chan struct{}) - - readCtx, readCancel := context.WithCancel(ctx) - var filterWg sync.WaitGroup - filterWg.Add(1) + return f.maxLineLength +} - go f.periodicTruncateCheck(ctx, truncate) - go func() { - f.filter(ctx, ltx, rawLines, lines, re) - filterWg.Done() - // If the filter stopped, make the reader stop too, no need to read - // more data if there is nothing more the filter wants to filter for! - // E.g. it could be that we only want to filter N matches but not more. - readCancel() - }() +func (f *readFile) warnAboutLongLine(ctx context.Context) bool { + if f.warnedAboutLongLine { + return true + } - err = f.read(readCtx, fd, reader, rawLines, truncate) - close(rawLines) - // Filter may sends some data still. So wait until it is done here. - filterWg.Wait() + if f.serverMessages == nil { + f.warnedAboutLongLine = true + return true + } - return err + select { + case f.serverMessages <- dlog.Common.Warn(f.filePath, + "Long log line, splitting into multiple lines") + "\n": + f.warnedAboutLongLine = true + return true + case <-ctx.Done(): + return false + } } -func (f *readFile) makeReader() (*bufio.Reader, *os.File, error) { +func (f *readFile) makeReader() (*bufio.Reader, *os.File, io.Closer, error) { if f.filePath == "" && f.globID == "-" { return f.makePipeReader() } return f.makeFileReader() } -func (f *readFile) makeFileReader() (reader *bufio.Reader, fd *os.File, err error) { - if fd, err = os.Open(f.filePath); err != nil { +func (f *readFile) makeFileReader() (reader *bufio.Reader, fd *os.File, decompressor io.Closer, err error) { + if fd, err = f.openFile(); err != nil { return } @@ -127,21 +113,32 @@ func (f *readFile) makeFileReader() (reader *bufio.Reader, fd *os.File, err erro } } - reader, err = f.makeCompressedFileReader(fd) + reader, decompressor, err = f.makeCompressedFileReader(fd) return } -func (f *readFile) makePipeReader() (*bufio.Reader, *os.File, error) { - return bufio.NewReader(os.Stdin), nil, nil +func (f *readFile) openFile() (*os.File, error) { + if f.validatedTarget != nil { + return f.validatedTarget.Open() + } + return os.Open(f.filePath) } -func (f *readFile) periodicTruncateCheck(ctx context.Context, truncate chan struct{}) { +func (f *readFile) makePipeReader() (*bufio.Reader, *os.File, io.Closer, error) { + return bufio.NewReader(os.Stdin), nil, nil, nil +} + +func (f *readFile) periodicTruncateCheck(ctx context.Context, truncate chan<- struct{}) { + ticker := time.NewTicker(time.Second * 3) + defer ticker.Stop() + for { select { - case <-time.After(time.Second * 3): + case <-ticker.C: select { case truncate <- struct{}{}: case <-ctx.Done(): + return } case <-ctx.Done(): return @@ -149,7 +146,7 @@ func (f *readFile) periodicTruncateCheck(ctx context.Context, truncate chan stru } } -func (f *readFile) makeCompressedFileReader(fd *os.File) (reader *bufio.Reader, err error) { +func (f *readFile) makeCompressedFileReader(fd *os.File) (reader *bufio.Reader, decompressor io.Closer, err error) { switch { case strings.HasSuffix(f.FilePath(), ".gz"): fallthrough @@ -160,82 +157,16 @@ func (f *readFile) makeCompressedFileReader(fd *os.File) (reader *bufio.Reader, if err != nil { return } + decompressor = gzipReader reader = bufio.NewReader(gzipReader) case strings.HasSuffix(f.FilePath(), ".zst"): - dlog.Common.Info(f.FilePath(), "Detected zstd compression format") - reader = bufio.NewReader(zstd.NewReader(fd)) + return f.makeZstdReader(fd) default: reader = bufio.NewReader(fd) } return } -func (f *readFile) read(ctx context.Context, fd *os.File, reader *bufio.Reader, - rawLines chan *bytes.Buffer, truncate <-chan struct{}) error { - - var offset uint64 - message := pool.BytesBuffer.Get().(*bytes.Buffer) - - for { - b, err := reader.ReadByte() - if err != nil { - status, err := f.handleReadError(ctx, err, fd, rawLines, truncate, message) - if abortReading == status { - return err - } - time.Sleep(time.Millisecond * 100) - continue - } - - offset++ - message.WriteByte(b) - - status, newMessage := f.handleReadByte(ctx, b, rawLines, message) - if status == abortReading { - return nil - } - message = newMessage - } -} - -// Filter log lines matching a given regular expression. -func (f *readFile) filter(ctx context.Context, ltx lcontext.LContext, - rawLines <-chan *bytes.Buffer, lines chan<- *line.Line, re regex.Regex) { - - // Do we have any kind of local context settings? If so then run the more complex - // filterWithLContext method. - if ltx.Has() { - // We can not skip transmitting any lines to the client with a local - // grep context specified. - f.canSkipLines = false - f.filterWithLContext(ctx, ltx, rawLines, lines, re) - return - } - - f.filterWithoutLContext(ctx, rawLines, lines, re) -} - -func (f *readFile) transmittable(rawLine *bytes.Buffer, length, capacity int, - re regex.Regex) (*line.Line, bool) { - - newLine := line.Null() - if !re.Match(rawLine.Bytes()) { - f.updateLineNotMatched() - f.updateLineNotTransmitted() - return newLine, false - } - f.updateLineMatched() - - // Can we actually send more messages, channel capacity reached? - if f.canSkipLines && length >= capacity { - f.updateLineNotTransmitted() - return newLine, false - } - f.updateLineTransmitted() - - return line.New(rawLine, f.totalLineCount(), f.transmittedPerc(), f.globID), true -} - // Check wether log file is truncated. Returns nil if not. func (f *readFile) truncated(fd *os.File) (bool, error) { if fd == nil { @@ -250,7 +181,7 @@ func (f *readFile) truncated(fd *os.File) (bool, error) { return true, err } // Can not open file at original path. - pathFd, err := os.Open(f.filePath) + pathFd, err := f.openFile() if err != nil { return true, err } @@ -267,68 +198,3 @@ func (f *readFile) truncated(fd *os.File) (bool, error) { return false, nil } -// Deal with the scenario that nothing could be read from the fd. -func (f *readFile) handleReadError(ctx context.Context, err error, fd *os.File, - rawLines chan *bytes.Buffer, truncate <-chan struct{}, - message *bytes.Buffer) (readStatus, error) { - - if err != io.EOF { - return abortReading, err - } - - select { - case <-truncate: - if isTruncated, err := f.truncated(fd); isTruncated { - return abortReading, err - } - case <-ctx.Done(): - return abortReading, nil - default: - } - - if !f.seekEOF { - dlog.Common.Info(f.FilePath(), "End of file reached") - if len(message.Bytes()) > 0 { - select { - case rawLines <- message: - case <-ctx.Done(): - } - } - return abortReading, nil - } - - return nothing, nil -} - -// Now process the byte we just read from the fd. -func (f *readFile) handleReadByte(ctx context.Context, b byte, - rawLines chan *bytes.Buffer, message *bytes.Buffer) (readStatus, *bytes.Buffer) { - - switch b { - case '\n': - select { - case rawLines <- message: - message = pool.BytesBuffer.Get().(*bytes.Buffer) - f.warnedAboutLongLine = false - case <-ctx.Done(): - return abortReading, message - } - default: - if message.Len() >= config.Server.MaxLineLength { - if !f.warnedAboutLongLine { - f.serverMessages <- dlog.Common.Warn(f.filePath, - "Long log line, splitting into multiple lines") + "\n" - f.warnedAboutLongLine = true - } - message.WriteByte('\n') - select { - case rawLines <- message: - message = pool.BytesBuffer.Get().(*bytes.Buffer) - case <-ctx.Done(): - return abortReading, message - } - } - } - - return nothing, message -} diff --git a/internal/io/fs/readfile_nozstd.go b/internal/io/fs/readfile_nozstd.go new file mode 100644 index 0000000..afd4523 --- /dev/null +++ b/internal/io/fs/readfile_nozstd.go @@ -0,0 +1,16 @@ +//go:build nozstd + +package fs + +import ( + "bufio" + "fmt" + "io" + "os" +) + +func (f *readFile) makeZstdReader(fd *os.File) (reader *bufio.Reader, decompressor io.Closer, err error) { + _ = fd + err = fmt.Errorf("%s: zstd is not supported in this build (built with -tags nozstd)", f.FilePath()) + return +} diff --git a/internal/io/fs/readfile_processor.go b/internal/io/fs/readfile_processor.go new file mode 100644 index 0000000..37809e0 --- /dev/null +++ b/internal/io/fs/readfile_processor.go @@ -0,0 +1,359 @@ +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" +) + +// StartWithProcessor starts reading a log file using a LineProcessor for handling lines. +// This is a channel-less implementation for better performance. +func (f *readFile) StartWithProcessor(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 + } + + truncateCtx, cancelTruncate := context.WithCancel(ctx) + defer cancelTruncate() + + truncate := make(chan struct{}) + + go f.periodicTruncateCheck(truncateCtx, truncate) + + // Process file with direct callbacks instead of channels + err = f.readWithProcessor(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 +} + +// readWithProcessor reads from the file and processes lines directly without channels +func (f *readFile) readWithProcessor(ctx context.Context, fd *os.File, reader *bufio.Reader, + truncate <-chan struct{}, ltx lcontext.LContext, processor line.Processor, re regex.Regex) error { + + var offset uint64 + message := pool.BytesBuffer.Get().(*bytes.Buffer) + // Use a closure so that the CURRENT value of `message` is recycled on + // return, not the pointer captured at defer-registration time. Downstream + // code paths that take ownership of the buffer set `message = nil` before + // reassigning or returning, preventing a double-recycle. + defer func() { + if message != nil { + pool.RecycleBytesBuffer(message) + } + }() + + // Create a line filter processor that wraps the given processor + filterProcessor := &filteringProcessor{ + processor: processor, + re: re, + ltx: ltx, + stats: &f.stats, + globID: f.globID, + } + + for { + b, err := reader.ReadByte() + if err != nil { + // handleReadErrorProcessor may hand `message` to ProcessFilteredLine + // (which takes ownership); in that case it sets *messagePtr = nil so + // the caller's defer does not recycle an already-recycled buffer. + status, err := f.handleReadErrorProcessor(ctx, err, fd, truncate, &message, filterProcessor) + if abortReading == status { + return err + } + if !ctxutil.Sleep(ctx, 100*time.Millisecond) { + return nil + } + continue + } + + offset++ + message.WriteByte(b) + + status := f.handleReadByteProcessor(ctx, b, message, filterProcessor) + if status == abortReading { + // ProcessFilteredLine took ownership; avoid defer double-recycle. + message = nil + return nil + } + if status == continueReading { + // Previous buffer was consumed by ProcessFilteredLine; acquire a fresh one. + message = pool.BytesBuffer.Get().(*bytes.Buffer) + } + } +} + +// handleReadByteProcessor processes a byte read from the file +func (f *readFile) handleReadByteProcessor(ctx context.Context, b byte, + message *bytes.Buffer, processor *filteringProcessor) readStatus { + + switch b { + case '\n': + // Process the complete line + f.updatePosition() + if err := processor.ProcessFilteredLine(message); err != nil { + return abortReading + } + + f.warnedAboutLongLine = false + return continueReading + + default: + if message.Len() >= f.lineLimit() { + if !f.warnAboutLongLine(ctx) { + return abortReading + } + // Force a line break + message.WriteByte('\n') + + // Process the line + f.updatePosition() + if err := processor.ProcessFilteredLine(message); err != nil { + return abortReading + } + return continueReading + } + } + + return nothing +} + +// handleReadErrorProcessor handles read errors in processor mode. When it hands +// the buffer to ProcessFilteredLine it nils out *messagePtr, signalling to the +// caller that ownership has been transferred downstream. +func (f *readFile) handleReadErrorProcessor(ctx context.Context, err error, fd *os.File, + truncate <-chan struct{}, messagePtr **bytes.Buffer, processor *filteringProcessor) (readStatus, error) { + + if err != io.EOF { + return abortReading, err + } + + select { + case <-truncate: + if isTruncated, err := f.truncated(fd); isTruncated { + return abortReading, err + } + case <-ctx.Done(): + return abortReading, nil + default: + } + + if !f.seekEOF { + dlog.Common.Info(f.FilePath(), "End of file reached") + message := *messagePtr + if len(message.Bytes()) > 0 { + // Process the last line if it doesn't end with newline. + f.updatePosition() + *messagePtr = nil + if processErr := processor.ProcessFilteredLine(message); processErr != nil { + return abortReading, processErr + } + } + return abortReading, nil + } + + return nothing, nil +} + +// filteringProcessor wraps a LineProcessor to add regex filtering +type filteringProcessor struct { + processor line.Processor + re regex.Regex + ltx lcontext.LContext + stats *stats + globID string + + // For local context handling + beforeBuf []*bytes.Buffer + afterCount int + maxCount int + maxReached bool +} + +// ProcessFilteredLine applies regex filtering before passing to the underlying processor +func (fp *filteringProcessor) ProcessFilteredLine(rawLine *bytes.Buffer) error { + // Update stats + lineNum := fp.stats.totalLineCount() + + // Simple case: no local context + if !fp.ltx.Has() { + if !fp.re.Match(rawLine.Bytes()) { + fp.stats.updateLineNotMatched() + fp.stats.updateLineNotTransmitted() + pool.RecycleBytesBuffer(rawLine) + return nil + } + + fp.stats.updateLineMatched() + fp.stats.updateLineTransmitted() + + // Process the line. Per the line.Processor contract (processor.go), + // ownership of rawLine transfers to the processor, which recycles it on + // every return path. The only processors on the fs read path - + // DirectLineProcessor and AggregateProcessor - recycle unconditionally, + // even when ProcessLine returns a write error (e.g. a client disconnect / + // broken pipe). Recycling here on error would Put the same buffer into the + // shared pool.BytesBuffer a second time; the pool would then hand one object + // to two Get callers whose concurrent writes race and corrupt data. So do + // not recycle rawLine here. + return fp.processor.ProcessLine(rawLine, lineNum, fp.globID) + } + + // Complex case: handle local context (before/after/max) + return fp.processWithContext(rawLine, lineNum) +} + +// ProcessFilteredRaw is the zero-copy fast path for the no-local-context case. +// It runs the regex match directly on the scanner-owned byte slice and only +// acquires+fills a pooled buffer when the line actually matches. At low hit +// rates this avoids a pool.Get + copy + pool.Put for the (vast majority of) +// non-matching lines, which profiling showed as ~10-15% of serverless +// dgrep CPU (sync.Pool Get/Put + bytes.Buffer.Write). +// +// Semantics are identical to the !ltx.Has() branch of ProcessFilteredLine: the +// same regex, the same stats bookkeeping, and the same lineNum are used, so +// output is byte-identical. It MUST only be called when fp.ltx.Has() is false; +// the local-context path deliberately buffers non-matching lines (before/after +// context) and cannot skip the copy. +// +// The caller passes raw = scanner.Bytes(), which is only valid until the next +// Scan(). On a match we copy it into a pooled buffer before returning, so the +// buffer handed to the underlying processor is a stable copy and never aliases +// the scanner's transient slice. +func (fp *filteringProcessor) ProcessFilteredRaw(raw []byte) error { + lineNum := fp.stats.totalLineCount() + + if !fp.re.Match(raw) { + fp.stats.updateLineNotMatched() + fp.stats.updateLineNotTransmitted() + // No buffer was acquired, so there is nothing to recycle. + return nil + } + + fp.stats.updateLineMatched() + fp.stats.updateLineTransmitted() + + // Only now, on a confirmed match, pay for the buffer and the copy. + lineBuf := pool.BytesBuffer.Get().(*bytes.Buffer) + lineBuf.Write(raw) + + // Ownership of lineBuf transfers to the processor, which recycles it on every + // return path (see ProcessFilteredLine for the full rationale). Recycling here + // on error would return the same buffer to the shared pool a second time and + // race, so leave it to the processor. + return fp.processor.ProcessLine(lineBuf, lineNum, fp.globID) +} + +// processWithContext handles lines when local context is enabled +func (fp *filteringProcessor) processWithContext(rawLine *bytes.Buffer, lineNum uint64) error { + matched := fp.re.Match(rawLine.Bytes()) + + if !matched { + fp.stats.updateLineNotMatched() + + // Handle after context + if fp.ltx.AfterContext > 0 && fp.afterCount > 0 { + fp.afterCount-- + fp.stats.updateLineTransmitted() + // Ownership transfers to the processor, which recycles rawLine on every + // return path; recycling here on error would double Put into the shared + // pool and race (see ProcessFilteredLine). + return fp.processor.ProcessLine(rawLine, lineNum, fp.globID) + } + + // Handle before context buffer + if fp.ltx.BeforeContext > 0 { + // Add to before buffer + if len(fp.beforeBuf) >= fp.ltx.BeforeContext { + // Recycle oldest buffer + pool.RecycleBytesBuffer(fp.beforeBuf[0]) + fp.beforeBuf = fp.beforeBuf[1:] + } + fp.beforeBuf = append(fp.beforeBuf, rawLine) + } else { + pool.RecycleBytesBuffer(rawLine) + } + + fp.stats.updateLineNotTransmitted() + return nil + } + + // Line matched + fp.stats.updateLineMatched() + + // Check if we've reached max count + if fp.maxReached { + pool.RecycleBytesBuffer(rawLine) + return io.EOF // Stop processing + } + + // Process before context + if fp.ltx.BeforeContext > 0 && len(fp.beforeBuf) > 0 { + for i, buf := range fp.beforeBuf { + fp.stats.updateLineTransmitted() + if err := fp.processor.ProcessLine(buf, lineNum-uint64(len(fp.beforeBuf)-i), fp.globID); err != nil { + // Clean up remaining buffers + for j := i + 1; j < len(fp.beforeBuf); j++ { + pool.RecycleBytesBuffer(fp.beforeBuf[j]) + } + pool.RecycleBytesBuffer(rawLine) + return err + } + } + fp.beforeBuf = fp.beforeBuf[:0] // Clear the buffer + } + + // Process the matched line. Ownership transfers to the processor, which + // recycles rawLine on every return path; recycling here on error would double + // Put into the shared pool and race (see ProcessFilteredLine). + fp.stats.updateLineTransmitted() + if err := fp.processor.ProcessLine(rawLine, lineNum, fp.globID); err != nil { + return err + } + + // Update max count + if fp.ltx.MaxCount > 0 { + fp.maxCount++ + if fp.maxCount >= fp.ltx.MaxCount { + if fp.ltx.AfterContext == 0 { + return io.EOF // Stop processing + } + fp.maxReached = true + } + } + + // Reset after context + if fp.ltx.AfterContext > 0 { + fp.afterCount = fp.ltx.AfterContext + } + + return nil +} 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: + } + } +} diff --git a/internal/io/fs/readfile_processor_test.go b/internal/io/fs/readfile_processor_test.go new file mode 100644 index 0000000..cd5d180 --- /dev/null +++ b/internal/io/fs/readfile_processor_test.go @@ -0,0 +1,869 @@ +package fs + +import ( + "bufio" + "bytes" + "context" + "errors" + "os" + "path/filepath" + "reflect" + "strings" + "testing" + "time" + + "github.com/mimecast/dtail/internal/io/pool" + "github.com/mimecast/dtail/internal/lcontext" + "github.com/mimecast/dtail/internal/regex" +) + +type captureProcessor struct { + lines []string + lineNums []uint64 + errAtLine int + processErr error + flushErr error +} + +func (p *captureProcessor) ProcessLine(lineContent *bytes.Buffer, lineNum uint64, _ string) error { + p.lines = append(p.lines, lineContent.String()) + p.lineNums = append(p.lineNums, lineNum) + pool.RecycleBytesBuffer(lineContent) + + if p.errAtLine > 0 && len(p.lines) == p.errAtLine { + return p.processErr + } + return nil +} + +func (p *captureProcessor) Flush() error { + return p.flushErr +} + +func (p *captureProcessor) Close() error { + return nil +} + +func TestStartWithProcessorOptimizedReadsAllLines(t *testing.T) { + filePath := writeProcessorTestFile(t, "alpha\nbeta\n") + re := regex.NewNoop() + + cat := NewCatFile(filePath, "glob-id", make(chan string, 1), defaultMaxLineLength) + processor := &captureProcessor{} + + if err := cat.readFile.StartWithProcessorOptimized( + context.Background(), + lcontext.LContext{}, + processor, + re, + ); err != nil { + t.Fatalf("optimized reader start failed: %v", err) + } + + want := []string{"alpha\n", "beta\n"} + if !reflect.DeepEqual(processor.lines, want) { + t.Fatalf("unexpected processed lines: got=%v want=%v", processor.lines, want) + } +} + +// TestReadWithProcessorOptimizedDetectsTruncation proves that after the +// per-line time.Since truncate gate was removed (task 2t0), the non-follow +// read loop still detects truncation: when the periodicTruncateCheck goroutine +// signals on the truncate channel, the loop re-stats the file and returns the +// truncation error. The reader (line source) is decoupled from the fd (stat +// source) so the scenario is deterministic without relying on the 3s cadence: +// the file on disk is shorter than the fd's current read position, exactly the +// state truncated() flags. A signal is pre-loaded on the truncate channel so +// the very first loop iteration performs the check. +func TestReadWithProcessorOptimizedDetectsTruncation(t *testing.T) { + resetCommonLogger(t) + + // The on-disk file is intentionally tiny; the fd is then seeked well past + // its end to emulate having read a file that shrank underneath us. + filePath := writeProcessorTestFile(t, "short") + + fd, err := os.Open(filePath) + if err != nil { + t.Fatalf("open file: %v", err) + } + defer fd.Close() + if _, err := fd.Seek(4096, 0); err != nil { + t.Fatalf("seek fd past end: %v", err) + } + + // The scanner reads its lines from an independent in-memory reader so the + // loop actually iterates and reaches the truncate check. + reader := bufio.NewReader(strings.NewReader("l1\nl2\nl3\nl4\nl5\n")) + + // Pre-load one truncate signal (buffered) so the first iteration checks. + truncate := make(chan struct{}, 1) + truncate <- struct{}{} + + rf := readFile{ + filePath: filePath, + globID: "glob-id", + maxLineLength: defaultMaxLineLength, + } + + err = rf.readWithProcessorOptimized( + context.Background(), + fd, + reader, + truncate, + lcontext.LContext{}, + &captureProcessor{}, + regex.NewNoop(), + ) + if err == nil { + t.Fatal("expected truncation to be detected, got nil error") + } + if !strings.Contains(err.Error(), "truncated") { + t.Fatalf("expected truncation error, got: %v", err) + } +} + +func TestProcessorVariantsReturnOpenError(t *testing.T) { + re := regex.NewNoop() + missingFile := filepath.Join(t.TempDir(), "missing.log") + + tests := []struct { + name string + start func(*readFile, context.Context, lcontext.LContext, *captureProcessor, regex.Regex) error + }{ + { + name: "standard", + start: func(rf *readFile, ctx context.Context, ltx lcontext.LContext, p *captureProcessor, re regex.Regex) error { + return rf.StartWithProcessor(ctx, ltx, p, re) + }, + }, + { + name: "optimized", + start: func(rf *readFile, ctx context.Context, ltx lcontext.LContext, p *captureProcessor, re regex.Regex) error { + return rf.StartWithProcessorOptimized(ctx, ltx, p, re) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cat := NewCatFile(missingFile, "glob-id", make(chan string, 1), defaultMaxLineLength) + err := tt.start(&cat.readFile, context.Background(), lcontext.LContext{}, &captureProcessor{}, re) + if err == nil { + t.Fatalf("expected error for missing file") + } + }) + } +} + +func TestStartWithProcessorOptimizedPropagatesProcessError(t *testing.T) { + filePath := writeProcessorTestFile(t, "alpha\nbeta\n") + re := regex.NewNoop() + expectedErr := errors.New("processor failure") + + cat := NewCatFile(filePath, "glob-id", make(chan string, 1), defaultMaxLineLength) + processor := &captureProcessor{ + errAtLine: 1, + processErr: expectedErr, + } + + err := cat.readFile.StartWithProcessorOptimized( + context.Background(), + lcontext.LContext{}, + processor, + re, + ) + if !errors.Is(err, expectedErr) { + t.Fatalf("expected process error %v, got %v", expectedErr, err) + } +} + +func TestStartWithProcessorOptimizedUsesInjectedMaxLineLength(t *testing.T) { + resetCommonLogger(t) + + filePath := writeProcessorTestFile(t, "abcdef\n") + re := regex.NewNoop() + + cat := NewCatFile(filePath, "glob-id", make(chan string, 1), 3) + processor := &captureProcessor{} + + if err := cat.readFile.StartWithProcessorOptimized( + context.Background(), + lcontext.LContext{}, + processor, + re, + ); err != nil { + t.Fatalf("optimized reader start failed: %v", err) + } + + want := []string{"abc", "def\n"} + if !reflect.DeepEqual(processor.lines, want) { + t.Fatalf("unexpected processed lines: got=%v want=%v", processor.lines, want) + } +} + +func TestStartWithProcessorOptimizedWaitsOnLiveLongLineWarningUntilCanceled(t *testing.T) { + resetCommonLogger(t) + + filePath := writeProcessorTestFile(t, strings.Repeat("a", 8)) + re := regex.NewNoop() + + cat := NewCatFile(filePath, "glob-id", make(chan string), 1) + processor := &captureProcessor{} + ctx, cancel := context.WithCancel(context.Background()) + + done := make(chan error, 1) + go func() { + done <- cat.readFile.StartWithProcessorOptimized( + ctx, + lcontext.LContext{}, + processor, + re, + ) + }() + + select { + case err := <-done: + t.Fatalf("optimized reader returned before cancellation: %v", err) + case <-time.After(100 * time.Millisecond): + } + + cancel() + + select { + case err := <-done: + if err != nil && !errors.Is(err, context.Canceled) { + t.Fatalf("expected canceled optimized reader to stop with nil or context.Canceled, got %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("optimized reader did not return after cancellation") + } +} + +// TestStartWithProcessorExitsWhenContextCanceledDuringLongLineWarning proves the +// byte-by-byte processor reader (StartWithProcessor) returns cleanly when the +// context is canceled while a long-line warning would otherwise block. The +// optimized reader has equivalent coverage in +// TestStartWithProcessorOptimizedWaitsOnLiveLongLineWarningUntilCanceled. The +// historic channel-based Start reader was removed in task iv0, so only the +// processor variant remains here. +func TestStartWithProcessorExitsWhenContextCanceledDuringLongLineWarning(t *testing.T) { + resetCommonLogger(t) + + filePath := writeProcessorTestFile(t, strings.Repeat("a", 8)) + re := regex.NewNoop() + + cat := NewCatFile(filePath, "glob-id", make(chan string), 1) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + errCh := make(chan error, 1) + go func() { + errCh <- cat.readFile.StartWithProcessor(ctx, lcontext.LContext{}, &captureProcessor{}, re) + }() + + select { + case err := <-errCh: + if err != nil { + t.Fatalf("expected canceled start to exit cleanly, got %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("start did not return after context cancellation") + } +} + +func TestTailWithProcessorOptimizedExitsWhenContextCanceledDuringLongLineWarning(t *testing.T) { + resetCommonLogger(t) + + filePath := writeProcessorTestFile(t, strings.Repeat("a", 8)) + re := regex.NewNoop() + + rf := readFile{ + filePath: filePath, + globID: "glob-id", + serverMessages: make(chan string), + retry: true, + canSkipLines: true, + seekEOF: false, + maxLineLength: 1, + } + + reader, fd, decompressor, err := rf.makeReader() + if fd != nil { + defer fd.Close() + } + if decompressor != nil { + defer func() { + if closeErr := decompressor.Close(); closeErr != nil { + t.Fatalf("unable to close decompressor: %v", closeErr) + } + }() + } + if err != nil { + t.Fatalf("make reader: %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + done := make(chan error, 1) + go func() { + done <- rf.tailWithProcessorOptimized( + ctx, + fd, + reader, + make(chan struct{}), + lcontext.LContext{}, + &captureProcessor{}, + re, + ) + }() + + select { + case err := <-done: + if err != nil { + t.Fatalf("expected canceled optimized tail to exit cleanly, got %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("optimized tail did not return after context cancellation") + } +} + +// TestReadWithProcessorNoDoubleRecycle verifies that readWithProcessor does not +// Put the same *bytes.Buffer back into the pool twice. The bug: a stale +// `defer pool.RecycleBytesBuffer(message)` captured the initial buffer pointer +// at defer-registration time; after that buffer was handed off downstream (and +// recycled there) and `message` was reassigned on continueReading, the deferred +// call recycled the already-recycled original buffer. A trailing partial line +// (no final newline) makes the bug deterministic because handleReadErrorProcessor +// also hands the current buffer to ProcessFilteredLine (which recycles it). +func TestReadWithProcessorNoDoubleRecycle(t *testing.T) { + resetCommonLogger(t) + drainBytesBufferPool() + + filePath := writeProcessorTestFile(t, "alpha\nbeta") + re := regex.NewNoop() + + cat := NewCatFile(filePath, "glob-id", make(chan string, 1), defaultMaxLineLength) + processor := &captureProcessor{} + + if err := cat.readFile.StartWithProcessor( + context.Background(), + lcontext.LContext{}, + processor, + re, + ); err != nil { + t.Fatalf("reader start failed: %v", err) + } + + want := []string{"alpha\n", "beta"} + if !reflect.DeepEqual(processor.lines, want) { + t.Fatalf("unexpected processed lines: got=%v want=%v", processor.lines, want) + } + + seen := make(map[*bytes.Buffer]int) + for i := 0; i < 512; i++ { + b := pool.BytesBuffer.Get().(*bytes.Buffer) + seen[b]++ + if seen[b] > 1 { + t.Fatalf("buffer %p observed in pool more than once: "+ + "double-recycle detected (Put twice into sync.Pool)", b) + } + } +} + +// drainBytesBufferPool empties the global buffer pool of any previously-Put +// entries so that pool inspection in a subsequent test is not polluted by +// artifacts from earlier test runs. +func drainBytesBufferPool() { + for i := 0; i < 1024; i++ { + _ = pool.BytesBuffer.Get() + } +} + +// TestReadWithProcessorOptimizedFastPathByteIdentical proves that the no-context +// zero-copy fast path (match on scanner.Bytes() before copying) yields exactly +// the same emitted lines as the previous copy-every-line behavior, across grep +// hit rates, inverted matching, zero matches, and the cat noop (match-all) case. +func TestReadWithProcessorOptimizedFastPathByteIdentical(t *testing.T) { + const content = "apple\nbanana\napricot\ncherry\navocado\n" + + mustRegex := func(pattern string, flag regex.Flag) regex.Regex { + re, err := regex.New(pattern, flag) + if err != nil { + t.Fatalf("build regex %q: %v", pattern, err) + } + return re + } + + // wantNums, when non-nil, pins the exact lineNum argument passed to + // ProcessLine for each emitted line. Because f.updatePosition() runs for + // every scanned line (matching or not) before the filter, non-matching lines + // still advance the counter, so a match after N non-matches must report + // lineNum N+1 (1-based) - never restarting at 1. This locks in that the + // zero-copy fast path counts lines identically to the old copy-every-line + // path. (apple=1, banana=2, apricot=3, cherry=4, avocado=5.) + tests := []struct { + name string + re regex.Regex + want []string + wantNums []uint64 + }{ + { + name: "low hit default", + re: mustRegex("ap", regex.Default), + want: []string{"apple\n", "apricot\n"}, + wantNums: []uint64{1, 3}, + }, + { + name: "high hit default", + re: mustRegex("a", regex.Default), + want: []string{"apple\n", "banana\n", "apricot\n", "avocado\n"}, + }, + { + name: "zero match", + re: mustRegex("zzz", regex.Default), + want: nil, + }, + { + name: "invert", + re: mustRegex("ap", regex.Invert), + want: []string{"banana\n", "cherry\n", "avocado\n"}, + wantNums: []uint64{2, 4, 5}, + }, + { + name: "noop matches all (cat)", + re: regex.NewNoop(), + want: []string{"apple\n", "banana\n", "apricot\n", "cherry\n", "avocado\n"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + filePath := writeProcessorTestFile(t, content) + cat := NewCatFile(filePath, "glob-id", make(chan string, 1), defaultMaxLineLength) + processor := &captureProcessor{} + + if err := cat.readFile.StartWithProcessorOptimized( + context.Background(), + lcontext.LContext{}, + processor, + tt.re, + ); err != nil { + t.Fatalf("optimized reader start failed: %v", err) + } + + if !reflect.DeepEqual(processor.lines, tt.want) { + t.Fatalf("unexpected processed lines: got=%v want=%v", processor.lines, tt.want) + } + if tt.wantNums != nil && !reflect.DeepEqual(processor.lineNums, tt.wantNums) { + t.Fatalf("unexpected line numbers: got=%v want=%v", processor.lineNums, tt.wantNums) + } + }) + } +} + +// TestReadWithProcessorOptimizedContextPathUnchanged exercises the local-context +// path (ltx.Has() == true), which must keep buffering every line so before/after +// context lines are still emitted. The fast path must NOT be taken here. +func TestReadWithProcessorOptimizedContextPathUnchanged(t *testing.T) { + const content = "a\nb\nHIT\nd\ne\n" + re, err := regex.New("HIT", regex.Default) + if err != nil { + t.Fatalf("build regex: %v", err) + } + + filePath := writeProcessorTestFile(t, content) + cat := NewCatFile(filePath, "glob-id", make(chan string, 1), defaultMaxLineLength) + processor := &captureProcessor{} + + // One line of before context and one line of after context around the match. + ltx := lcontext.LContext{BeforeContext: 1, AfterContext: 1} + if err := cat.readFile.StartWithProcessorOptimized( + context.Background(), + ltx, + processor, + re, + ); err != nil { + t.Fatalf("optimized reader start failed: %v", err) + } + + want := []string{"b\n", "HIT\n", "d\n"} + if !reflect.DeepEqual(processor.lines, want) { + t.Fatalf("unexpected context lines: got=%v want=%v", processor.lines, want) + } +} + +// TestProcessFilteredRawZeroAllocOnNonMatch locks in the win: a non-matching line +// on the fast path must not acquire a pooled buffer or copy anything, so it +// allocates nothing. A matching line does allocate (buffer copy + emit). +func TestProcessFilteredRawZeroAllocOnNonMatch(t *testing.T) { + re, err := regex.New("MATCHME", regex.Default) + if err != nil { + t.Fatalf("build regex: %v", err) + } + + var st stats + fp := &filteringProcessor{ + processor: &captureProcessor{}, + re: re, + ltx: lcontext.LContext{}, + stats: &st, + globID: "glob-id", + } + + nonMatch := []byte("this line does not contain the needle\n") + allocs := testing.AllocsPerRun(100, func() { + if err := fp.ProcessFilteredRaw(nonMatch); err != nil { + t.Fatalf("ProcessFilteredRaw returned error: %v", err) + } + }) + if allocs != 0 { + t.Fatalf("expected zero allocations on non-matching fast-path line, got %v", allocs) + } +} + +// TestProcessorMaxCountEarlyStopNoErrorLeak is a regression test for the +// optimized read path leaking the io.EOF early-stop sentinel that +// filteringProcessor.processWithContext returns once a -m/-max (MaxCount) limit +// is reached. The byte-by-byte path (StartWithProcessor) already swallowed that +// sentinel and returned nil; the optimized path (StartWithProcessorOptimized) +// used to surface it as an error, which the server then logged as a spurious +// SERVER|...|ERROR|...|EOF line. Both paths must now return nil AND emit +// byte-identical lines for the same MaxCount, proving the sentinel is handled as +// a clean early stop, not a genuine I/O error. +func TestProcessorMaxCountEarlyStopNoErrorLeak(t *testing.T) { + const content = "match 1\nother\nmatch 2\nother\nmatch 3\nother\nmatch 4\n" + re, err := regex.New("match", regex.Default) + if err != nil { + t.Fatalf("build regex: %v", err) + } + // MaxCount without after-context: processWithContext returns io.EOF as soon + // as the second match is emitted (the -max 2 early stop). + ltx := lcontext.LContext{MaxCount: 2} + + run := func(start func(*readFile, context.Context, lcontext.LContext, *captureProcessor, regex.Regex) error) *captureProcessor { + filePath := writeProcessorTestFile(t, content) + cat := NewCatFile(filePath, "glob-id", make(chan string, 1), defaultMaxLineLength) + processor := &captureProcessor{} + if err := start(&cat.readFile, context.Background(), ltx, processor, re); err != nil { + // A non-nil return here is exactly the leaked sentinel the server + // would log as ERROR|...|EOF. + t.Fatalf("reader returned error; max-count early-stop sentinel must be swallowed: %v", err) + } + return processor + } + + byteByByte := run(func(rf *readFile, ctx context.Context, l lcontext.LContext, p *captureProcessor, r regex.Regex) error { + return rf.StartWithProcessor(ctx, l, p, r) + }) + optimized := run(func(rf *readFile, ctx context.Context, l lcontext.LContext, p *captureProcessor, r regex.Regex) error { + return rf.StartWithProcessorOptimized(ctx, l, p, r) + }) + + want := []string{"match 1\n", "match 2\n"} + if !reflect.DeepEqual(optimized.lines, want) { + t.Fatalf("optimized -max lines: got=%v want=%v", optimized.lines, want) + } + // Byte-identical -max output between the byte-by-byte and optimized paths. + if !reflect.DeepEqual(byteByByte.lines, optimized.lines) { + t.Fatalf("-max output differs between byte-by-byte and optimized: byteByByte=%v optimized=%v", + byteByByte.lines, optimized.lines) + } +} + +// TestReadWithProcessorOptimizedMaxCountWithContextEarlyStop covers -m combined +// with after-context (-A). Here processWithContext returns the io.EOF sentinel +// from its maxReached branch (a distinct return site from plain -m: it fires on +// the NEXT match after the after-context window drains, not on the match that +// reaches the count). The optimized path must still swallow the sentinel +// (return nil), emit the after-context line, and stay byte-identical to the +// byte-by-byte path. Pre-fix the optimized run returns io.EOF and goes red. +func TestReadWithProcessorOptimizedMaxCountWithContextEarlyStop(t *testing.T) { + const content = "x\nHIT one\ny\nHIT two\nz\nHIT three\n" + // MaxCount 1 with AfterContext 1: emit the first match plus its single + // trailing context line, then stop at the next match via the maxReached + // sentinel. + ltx := lcontext.LContext{MaxCount: 1, AfterContext: 1} + + run := func(start func(*readFile, context.Context, lcontext.LContext, *captureProcessor, regex.Regex) error) *captureProcessor { + re, err := regex.New("HIT", regex.Default) + if err != nil { + t.Fatalf("build regex: %v", err) + } + filePath := writeProcessorTestFile(t, content) + cat := NewCatFile(filePath, "glob-id", make(chan string, 1), defaultMaxLineLength) + processor := &captureProcessor{} + if err := start(&cat.readFile, context.Background(), ltx, processor, re); err != nil { + t.Fatalf("reader returned error; max-count+context sentinel must be swallowed: %v", err) + } + return processor + } + + byteByByte := run(func(rf *readFile, ctx context.Context, l lcontext.LContext, p *captureProcessor, r regex.Regex) error { + return rf.StartWithProcessor(ctx, l, p, r) + }) + optimized := run(func(rf *readFile, ctx context.Context, l lcontext.LContext, p *captureProcessor, r regex.Regex) error { + return rf.StartWithProcessorOptimized(ctx, l, p, r) + }) + + want := []string{"HIT one\n", "y\n"} + if !reflect.DeepEqual(optimized.lines, want) { + t.Fatalf("optimized -m+context lines: got=%v want=%v", optimized.lines, want) + } + if !reflect.DeepEqual(byteByByte.lines, optimized.lines) { + t.Fatalf("-m+context output differs between byte-by-byte and optimized: byteByByte=%v optimized=%v", + byteByByte.lines, optimized.lines) + } +} + +// TestTailWithProcessorOptimizedMaxCountEarlyStop proves the follow/tail +// optimized path (tailWithProcessorOptimized) treats the io.EOF max-count early-stop +// sentinel as a clean stop (return nil) at ALL THREE of its processPartialLine +// call sites: the newline-terminated line site, the long-line split site, and +// the context-cancel trailing-partial cleanup site. Pre-fix each site returned +// io.EOF straight to the caller (logged as SERVER|...|ERROR|...|EOF), so every +// subtest goes red on the unfixed code. The reader is driven in-memory so the +// follow loop is deterministic; serverMessages is nil so warnAboutLongLine never +// blocks (it returns true immediately). +func TestTailWithProcessorOptimizedMaxCountEarlyStop(t *testing.T) { + resetCommonLogger(t) + + newReadFile := func(maxLineLength int) readFile { + return readFile{ + filePath: "test.log", + globID: "glob-id", + maxLineLength: maxLineLength, + } + } + + mustRegex := func(pattern string) regex.Regex { + re, err := regex.New(pattern, regex.Default) + if err != nil { + t.Fatalf("build regex %q: %v", pattern, err) + } + return re + } + + // runTail drives tailWithProcessorOptimized directly. fd is nil because the + // truncate channel is never signaled, so f.truncated(fd) is never reached. + runTail := func(t *testing.T, ctx context.Context, rf *readFile, input string, + ltx lcontext.LContext, re regex.Regex) *captureProcessor { + + processor := &captureProcessor{} + reader := bufio.NewReader(strings.NewReader(input)) + if err := rf.tailWithProcessorOptimized(ctx, nil, reader, + make(chan struct{}), ltx, processor, re); err != nil { + t.Fatalf("tail returned error; max-count sentinel must be swallowed: %v", err) + } + return processor + } + + t.Run("newline terminated line site", func(t *testing.T) { + rf := newReadFile(defaultMaxLineLength) + // The second complete (newline-terminated) match hits the count and stops + // via the newline branch's processPartialLine call. + p := runTail(t, context.Background(), &rf, "match1\nmatch2\nmatch3\n"