summaryrefslogtreecommitdiff
path: root/internal/io/fs
diff options
context:
space:
mode:
Diffstat (limited to 'internal/io/fs')
-rw-r--r--internal/io/fs/catfile.go3
-rw-r--r--internal/io/fs/filereader.go6
-rw-r--r--internal/io/fs/filter.go167
-rw-r--r--internal/io/fs/permissions/permission.go4
-rw-r--r--internal/io/fs/permissions/permission_linuxacl.go2
-rw-r--r--internal/io/fs/readfile.go334
-rw-r--r--internal/io/fs/readfilelcontext.go214
-rw-r--r--internal/io/fs/tailfile.go3
-rw-r--r--internal/io/fs/truncate.go61
9 files changed, 449 insertions, 345 deletions
diff --git a/internal/io/fs/catfile.go b/internal/io/fs/catfile.go
index 7f387bc..e4676f3 100644
--- a/internal/io/fs/catfile.go
+++ b/internal/io/fs/catfile.go
@@ -6,7 +6,7 @@ type CatFile struct {
}
// NewCatFile returns a new file catter.
-func NewCatFile(filePath string, globID string, serverMessages chan<- string, limiter chan struct{}) CatFile {
+func NewCatFile(filePath string, globID string, serverMessages chan<- string) CatFile {
return CatFile{
readFile: readFile{
filePath: filePath,
@@ -15,7 +15,6 @@ func NewCatFile(filePath string, globID string, serverMessages chan<- string, li
retry: false,
canSkipLines: false,
seekEOF: false,
- limiter: limiter,
},
}
}
diff --git a/internal/io/fs/filereader.go b/internal/io/fs/filereader.go
index efd410e..e27d2a7 100644
--- a/internal/io/fs/filereader.go
+++ b/internal/io/fs/filereader.go
@@ -8,9 +8,11 @@ import (
"github.com/mimecast/dtail/internal/regex"
)
-// FileReader is the interface used on the dtail server to read/cat/grep/mapr... a file.
+// FileReader is the interface used on the dtail server to read/cat/grep/mapr...
+// a file.
type FileReader interface {
- Start(ctx context.Context, lContext lcontext.LContext, lines chan<- line.Line, re regex.Regex) error
+ Start(ctx context.Context, ltx lcontext.LContext, lines chan<- *line.Line,
+ re regex.Regex) error
FilePath() string
Retry() bool
}
diff --git a/internal/io/fs/filter.go b/internal/io/fs/filter.go
deleted file mode 100644
index c4f605e..0000000
--- a/internal/io/fs/filter.go
+++ /dev/null
@@ -1,167 +0,0 @@
-package fs
-
-import (
- "context"
-
- "github.com/mimecast/dtail/internal/io/line"
- "github.com/mimecast/dtail/internal/lcontext"
- "github.com/mimecast/dtail/internal/regex"
-)
-
-func (f readFile) filter(ctx context.Context, rawLines <-chan []byte, lines chan<- line.Line, re regex.Regex, lContext lcontext.LContext) {
- // Do we have any kind of local context settings? If so then run the more complex
- // filterWithLContext method.
- if lContext.Has() {
- // We can not skip transmitting any lines to the client with a local
- // grep context specified.
- f.canSkipLines = false
- f.filterWithLContext(ctx, rawLines, lines, re, lContext)
- return
- }
-
- f.filterWithoutLContext(ctx, rawLines, lines, re)
-}
-
-// Filter log lines matching a given regular expression, however with local grep context.
-func (f readFile) filterWithLContext(ctx context.Context, rawLines <-chan []byte, lines chan<- line.Line, re regex.Regex, lContext lcontext.LContext) {
- // Scenario 1: Finish once maxCount hits found
- maxCount := lContext.MaxCount
- processMaxCount := maxCount > 0
- maxReached := false
-
- // Scenario 2: Print prev. N lines when current line matches.
- before := lContext.BeforeContext
- processBefore := before > 0
- var beforeBuf chan []byte
- if processBefore {
- beforeBuf = make(chan []byte, before)
- }
-
- // Screnario 3: Print next N lines when current line matches.
- after := 0
- processAfter := lContext.AfterContext > 0
-
- for rawLine := range rawLines {
- // logger.Debug("rawLine", string(rawLine))
- f.updatePosition()
-
- if !re.Match(rawLine) {
- f.updateLineNotMatched()
-
- if processAfter && after > 0 {
- after--
- myLine := line.Line{Content: rawLine, SourceID: f.globID, Count: f.totalLineCount(), TransmittedPerc: 100}
- select {
- case lines <- myLine:
- case <-ctx.Done():
- return
- }
-
- } else if processBefore {
- // Keep last num BeforeContext raw messages.
- select {
- case beforeBuf <- rawLine:
- default:
- <-beforeBuf
- beforeBuf <- rawLine
- }
- }
- continue
- }
-
- f.updateLineMatched()
-
- if processAfter {
- if maxReached {
- return
- }
- after = lContext.AfterContext
- }
-
- if processBefore {
- i := uint64(len(beforeBuf))
- for {
- select {
- case myRawLine := <-beforeBuf:
- myLine := line.Line{Content: myRawLine, SourceID: f.globID, Count: f.totalLineCount() - i, TransmittedPerc: 100}
- i--
- select {
- case lines <- myLine:
- case <-ctx.Done():
- return
- }
- default:
- // beforeBuf is now empty.
- }
- if len(beforeBuf) == 0 {
- break
- }
- }
- }
-
- line := line.Line{Content: rawLine, SourceID: f.globID, Count: f.totalLineCount(), TransmittedPerc: 100}
-
- select {
- case lines <- line:
- if processMaxCount {
- maxCount--
- if maxCount == 0 {
- if !processAfter || after == 0 {
- return
- }
- // Unfortunatley we have to continue filter, as there might be more lines to print
- maxReached = true
- }
- }
- case <-ctx.Done():
- return
- }
- }
-}
-
-// Filter log lines matching a given regular expression, there is no local grep context specified.
-func (f readFile) filterWithoutLContext(ctx context.Context, rawLines <-chan []byte, lines chan<- line.Line, re regex.Regex) {
- for {
- select {
- case rawLine, ok := <-rawLines:
- f.updatePosition()
- if !ok {
- return
- }
-
- if f.lineUntransmittable(rawLine, len(lines), cap(lines), re) {
- continue
- }
-
- line := line.Line{Content: rawLine, SourceID: f.globID, Count: f.totalLineCount(), TransmittedPerc: f.transmittedPerc()}
-
- select {
- case lines <- line:
- continue
- case <-ctx.Done():
- return
- }
- }
- }
-}
-
-func (f readFile) lineUntransmittable(rawLine []byte, length, capacity int, re regex.Regex) bool {
- if !re.Match(rawLine) {
- f.updateLineNotMatched()
- f.updateLineNotTransmitted()
- // Regex dosn't match, so not interested in it.
- return true
- }
- f.updateLineMatched()
-
- // Can we actually send more messages, channel capacity reached?
- if f.canSkipLines && length >= capacity {
- f.updateLineNotTransmitted()
- // Matching, not transmittable
- return true
- }
- f.updateLineTransmitted()
-
- // Matching, transmittable
- return false
-}
diff --git a/internal/io/fs/permissions/permission.go b/internal/io/fs/permissions/permission.go
index cc5dd9b..d621c09 100644
--- a/internal/io/fs/permissions/permission.go
+++ b/internal/io/fs/permissions/permission.go
@@ -3,12 +3,12 @@
package permissions
import (
- "github.com/mimecast/dtail/internal/io/logger"
+ "github.com/mimecast/dtail/internal/io/dlog"
)
// ToRead is to check whether user has read permissions to a given file.
func ToRead(user, filePath string) (bool, error) {
// Only implemented for Linux, always expect true
- logger.Warn(user, filePath, "Not performing ACL check, not supported on this platform")
+ dlog.Common.Debug(user, filePath, "Not performing ACL check as not compiled in")
return true, nil
}
diff --git a/internal/io/fs/permissions/permission_linuxacl.go b/internal/io/fs/permissions/permission_linuxacl.go
index 7d2d7ca..904b90f 100644
--- a/internal/io/fs/permissions/permission_linuxacl.go
+++ b/internal/io/fs/permissions/permission_linuxacl.go
@@ -13,7 +13,7 @@ import (
"unsafe"
)
-// ToRead checks whether user has Linux file system permissions to read a given file.
+// ToRead checks whether user has Linux file system permissions to read a file.
func ToRead(user, filePath string) (bool, error) {
cUser := C.CString(user)
cFilePath := C.CString(filePath)
diff --git a/internal/io/fs/readfile.go b/internal/io/fs/readfile.go
index 161e3f0..90c5966 100644
--- a/internal/io/fs/readfile.go
+++ b/internal/io/fs/readfile.go
@@ -2,22 +2,35 @@ package fs
import (
"bufio"
+ "bytes"
"compress/gzip"
"context"
+ "errors"
"fmt"
"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/logger"
+ "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
+)
+
// Used to tail and filter a local log file.
type readFile struct {
// Various statistics (e.g. regex hit percentage, transfer percentage).
@@ -34,34 +47,14 @@ type readFile struct {
canSkipLines bool
// Seek to the EOF before processing file?
seekEOF bool
- limiter chan struct{}
-}
-
-func (f readFile) makeReader(fd *os.File) (reader *bufio.Reader, err error) {
- switch {
- case strings.HasSuffix(f.FilePath(), ".gz"):
- fallthrough
- case strings.HasSuffix(f.FilePath(), ".gzip"):
- logger.Info(f.FilePath(), "Detected gzip compression format")
- var gzipReader *gzip.Reader
- gzipReader, err = gzip.NewReader(fd)
- if err != nil {
- return
- }
- reader = bufio.NewReader(gzipReader)
- case strings.HasSuffix(f.FilePath(), ".zst"):
- logger.Info(f.FilePath(), "Detected zstd compression format")
- reader = bufio.NewReader(zstd.NewReader(fd))
- default:
- reader = bufio.NewReader(fd)
- }
-
- return
+ // Warned already about a long line.
+ warnedAboutLongLine bool
}
// String returns the string representation of the readFile
func (f readFile) String() string {
- return fmt.Sprintf("readFile(filePath:%s,globID:%s,retry:%v,canSkipLines:%v,seekEOF:%v)",
+ return fmt.Sprintf(
+ "readFile(filePath:%s,globID:%s,retry:%v,canSkipLines:%v,seekEOF:%v)",
f.filePath,
f.globID,
f.retry,
@@ -80,141 +73,266 @@ func (f readFile) Retry() bool {
}
// Start tailing a log file.
-func (f readFile) Start(ctx context.Context, lContext lcontext.LContext, lines chan<- line.Line, re regex.Regex) error {
- logger.Debug("readFile", f)
- defer func() {
- select {
- case <-f.limiter:
- default:
- }
- }()
+func (f readFile) Start(ctx context.Context, ltx lcontext.LContext,
+ lines chan<- *line.Line, re regex.Regex) error {
- select {
- case f.limiter <- struct{}{}:
- default:
- select {
- case f.serverMessages <- logger.Warn(f.filePath, f.globID, "Server limit reached. Queuing file..."):
- case <-ctx.Done():
- return nil
- }
- f.limiter <- struct{}{}
+ reader, fd, err := f.makeReader()
+ if fd != nil {
+ defer fd.Close()
}
-
- fd, err := os.Open(f.filePath)
if err != nil {
return err
}
- defer fd.Close()
- if f.seekEOF {
- fd.Seek(0, io.SeekEnd)
- }
+ rawLines := make(chan *bytes.Buffer, 100)
+ truncate := make(chan struct{})
- rawLines := make(chan []byte, 100)
readCtx, readCancel := context.WithCancel(ctx)
+ var filterWg sync.WaitGroup
+ filterWg.Add(1)
- filterDone := make(chan struct{})
+ go f.periodicTruncateCheck(ctx, truncate)
go func() {
- f.filter(ctx, rawLines, lines, re, lContext)
- close(filterDone)
+ 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()
}()
- err = f.read(readCtx, fd, rawLines)
+ err = f.read(readCtx, fd, reader, rawLines, truncate)
close(rawLines)
-
- // Filter may flushes some data still. So wait until it is done here.
- <-filterDone
+ // Filter may sends some data still. So wait until it is done here.
+ filterWg.Wait()
return err
}
-func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan []byte) error {
- var offset uint64
+func (f *readFile) makeReader() (*bufio.Reader, *os.File, error) {
+ if f.filePath == "" && f.globID == "-" {
+ return f.makePipeReader()
+ }
+ return f.makeFileReader()
+}
+
+func (f *readFile) makeFileReader() (*bufio.Reader, *os.File, error) {
+ var reader *bufio.Reader
+ fd, err := os.Open(f.filePath)
+ if err != nil {
+ return reader, fd, err
+ }
+
+ if f.seekEOF {
+ fd.Seek(0, io.SeekEnd)
+ }
- reader, err := f.makeReader(fd)
+ reader, err = f.makeCompressedFileReader(fd)
if err != nil {
- return err
+ return reader, fd, err
}
- rawLine := make([]byte, 0, 512)
- lineLengthThreshold := 1024 * 1024 // 1mb
- longLineWarning := false
+ return reader, fd, nil
+}
- checkTruncate := f.truncateTimer(ctx)
+func (f *readFile) makePipeReader() (*bufio.Reader, *os.File, error) {
+ return bufio.NewReader(os.Stdin), nil, nil
+}
+func (f *readFile) periodicTruncateCheck(ctx context.Context, truncate chan struct{}) {
for {
select {
+ case <-time.After(time.Second * 3):
+ select {
+ case truncate <- struct{}{}:
+ case <-ctx.Done():
+ }
case <-ctx.Done():
- return nil
- default:
+ return
}
+ }
+}
- select {
- case <-checkTruncate:
- if isTruncated, err := f.truncated(fd); isTruncated {
- return err
- }
- logger.Info(f.filePath, "Current offset", offset)
- default:
+func (f *readFile) makeCompressedFileReader(fd *os.File) (reader *bufio.Reader, err error) {
+ switch {
+ case strings.HasSuffix(f.FilePath(), ".gz"):
+ fallthrough
+ case strings.HasSuffix(f.FilePath(), ".gzip"):
+ dlog.Common.Info(f.FilePath(), "Detected gzip compression format")
+ var gzipReader *gzip.Reader
+ gzipReader, err = gzip.NewReader(fd)
+ if err != nil {
+ return
}
+ 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))
+ 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 {
- // Read some bytes (max 4k at once as of go 1.12). isPrefix will
- // be set if line does not fit into 4k buffer.
- bytes, isPrefix, err := reader.ReadLine()
+ var offset uint64
+ message := pool.BytesBuffer.Get().(*bytes.Buffer)
+ for {
+ b, err := reader.ReadByte()
if err != nil {
- // If EOF, sleep a couple of ms and return with nil error.
- // If other error, return with non-nil error.
- if err != io.EOF {
+ status, err := f.handleReadError(ctx, err, fd, rawLines, truncate, message)
+ if abortReading == status {
return err
}
- if !f.seekEOF {
- logger.Debug(f.FilePath(), "End of file reached")
- return nil
- }
time.Sleep(time.Millisecond * 100)
continue
}
- rawLine = append(rawLine, bytes...)
- offset += uint64(len(bytes))
+ 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
+}
- if !isPrefix {
- // last LineRead call returned contend until end of line.
- rawLine = append(rawLine, '\n')
+// Check wether log file is truncated. Returns nil if not.
+func (f *readFile) truncated(fd *os.File) (bool, error) {
+ if fd == nil {
+ return false, nil
+ }
+
+ dlog.Common.Debug(f.filePath, "File truncation check")
+
+ // Can not seek currently open FD.
+ currentPosition, err := fd.Seek(0, os.SEEK_CUR)
+ if err != nil {
+ return true, err
+ }
+ // Can not open file at original path.
+ pathFd, err := os.Open(f.filePath)
+ if err != nil {
+ return true, err
+ }
+ defer pathFd.Close()
+
+ // Can not seek file at original path.
+ pathPosition, err := pathFd.Seek(0, io.SeekEnd)
+ if err != nil {
+ return true, err
+ }
+ if currentPosition > pathPosition {
+ return true, errors.New("File got truncated")
+ }
+ 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 <- rawLine:
+ case rawLines <- message:
case <-ctx.Done():
- return nil
- }
- rawLine = make([]byte, 0, 512)
- if longLineWarning {
- longLineWarning = false
}
- continue
}
+ return abortReading, nil
+ }
- // Last LineRead call could not read content until end of line, buffer
- // was too small. Determine whether we exceed the max line length we
- // want dtail to send to the client at once. Possibly split up log line
- // into multiple log lines.
- if len(rawLine) >= lineLengthThreshold {
- if !longLineWarning {
- f.serverMessages <- logger.Warn(f.filePath, "Long log line, splitting into multiple lines")
- // Only print out one warning per long log line.
- longLineWarning = true
+ 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")
+ f.warnedAboutLongLine = true
}
- rawLine = append(rawLine, '\n')
+ message.WriteByte('\n')
select {
- case rawLines <- rawLine:
+ case rawLines <- message:
+ message = pool.BytesBuffer.Get().(*bytes.Buffer)
case <-ctx.Done():
- return nil
+ return abortReading, message
}
- rawLine = make([]byte, 0, 512)
}
}
+
+ return nothing, message
}
diff --git a/internal/io/fs/readfilelcontext.go b/internal/io/fs/readfilelcontext.go
new file mode 100644
index 0000000..44ce17d
--- /dev/null
+++ b/internal/io/fs/readfilelcontext.go
@@ -0,0 +1,214 @@
+package fs
+
+import (
+ "bytes"
+ "context"
+
+ "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"
+)
+
+// The local context state.
+type ltxState struct {
+ // Max state
+ maxCount int
+ processMaxCount bool
+ maxReached bool
+
+ // Before state
+ before int
+ processBefore bool
+ beforeBuf chan *bytes.Buffer
+
+ // After state
+ after int
+ processAfter bool
+}
+
+// We don't have any local grep context, which makes life much simpler and more efficient.
+func (f *readFile) filterWithoutLContext(ctx context.Context, rawLines <-chan *bytes.Buffer,
+ lines chan<- *line.Line, re regex.Regex) {
+
+ for {
+ select {
+ case rawLine, ok := <-rawLines:
+ f.updatePosition()
+ if !ok {
+ return
+ }
+ if newLine, ok := f.transmittable(rawLine, len(lines), cap(lines), re); ok {
+ select {
+ case lines <- newLine:
+ case <-ctx.Done():
+ return
+ }
+ }
+ }
+ }
+}
+
+// Filter log lines matching a given regular expression, however with local grep context.
+func (f *readFile) filterWithLContext(ctx context.Context, ltx lcontext.LContext,
+ rawLines <-chan *bytes.Buffer, lines chan<- *line.Line, re regex.Regex) {
+
+ var ls ltxState
+
+ // The following 3 scenarios may also be used at once/any combination together.
+
+ // Scenario 1: Finish once maxCount hits found
+ ls.maxCount = ltx.MaxCount
+ ls.processMaxCount = ls.maxCount > 0
+ ls.maxReached = false
+
+ // Scenario 2: Print prev. N lines when current line matches.
+ ls.before = ltx.BeforeContext
+ ls.processBefore = ls.before > 0
+ if ls.processBefore {
+ ls.beforeBuf = make(chan *bytes.Buffer, ls.before)
+ }
+
+ // Screnario 3: Print next N lines when current line matches.
+ ls.after = 0
+ ls.processAfter = ltx.AfterContext > 0
+
+ // No go through all raw lines read to determine with they satisfy the local
+ // context or not. "Matching" lines will be sent to the lines channel.
+ for rawLine := range rawLines {
+ status := f.filterLineWithLContext(ctx, &ltx, &ls, rawLines, lines, &re, rawLine)
+ switch status {
+ case abortReading:
+ return
+ default:
+ }
+ }
+}
+
+// Filter log lines matching a given regular expression, however with local grep context.
+func (f *readFile) filterLineWithLContext(ctx context.Context, ltx *lcontext.LContext,
+ ls *ltxState, rawLines <-chan *bytes.Buffer, lines chan<- *line.Line, re *regex.Regex,
+ rawLine *bytes.Buffer) readStatus {
+
+ f.updatePosition()
+
+ if !re.Match(rawLine.Bytes()) {
+ f.updateLineNotMatched()
+ status := f.lContextNotMatched(ctx, ls, lines, rawLine)
+ switch status {
+ case nothing:
+ default:
+ return status
+ }
+ }
+
+ f.updateLineMatched()
+
+ // If we have an "after" context to worry about...
+ if ls.processAfter {
+ if ls.maxReached {
+ // We have reached the "max" hits. Stop/abort reading.
+ return abortReading
+ }
+ // Reset the "after" context.
+ ls.after = ltx.AfterContext
+ }
+
+ // If we have a "before" context to worry about...
+ if ls.processBefore {
+ status := f.lContextProcessBefore(ctx, ls, lines, rawLine)
+ switch status {
+ case nothing:
+ default:
+ return status
+ }
+ }
+
+ line := line.New(rawLine, f.totalLineCount(), 100, f.globID)
+
+ select {
+ case lines <- line:
+ // If we have a "max" context to worry about...
+ if ls.processMaxCount {
+ status := f.lContextProcessMaxCount(ctx, ls)
+ switch status {
+ case nothing:
+ default:
+ return status
+ }
+ }
+ case <-ctx.Done():
+ return abortReading
+ }
+
+ return nothing
+}
+
+// Do some post-processing for the "after" and the "before" contexts in case the
+// line didn't match the regex.
+func (f *readFile) lContextNotMatched(ctx context.Context, ls *ltxState,
+ lines chan<- *line.Line, rawLine *bytes.Buffer) readStatus {
+
+ if ls.processAfter && ls.after > 0 {
+ ls.after--
+ myLine := line.New(rawLine, f.totalLineCount(), 100, f.globID)
+
+ select {
+ case lines <- myLine:
+ case <-ctx.Done():
+ return abortReading
+ }
+
+ } else if ls.processBefore {
+ // Keep last num BeforeContext raw messages.
+ select {
+ case ls.beforeBuf <- rawLine:
+ default:
+ pool.RecycleBytesBuffer(<-ls.beforeBuf)
+ ls.beforeBuf <- rawLine
+ }
+ }
+
+ return continueReading
+}
+
+// Do some processing for the "before" context.
+func (f *readFile) lContextProcessBefore(ctx context.Context,
+ ls *ltxState, lines chan<- *line.Line, rawLine *bytes.Buffer) readStatus {
+
+ i := uint64(len(ls.beforeBuf))
+ for {
+ select {
+ case rawLine := <-ls.beforeBuf:
+ myLine := line.New(rawLine, f.totalLineCount()-i, 100, f.globID)
+ i--
+
+ select {
+ case lines <- myLine:
+ case <-ctx.Done():
+ return abortReading
+ }
+ default:
+ // beforeBuf is now empty.
+ }
+ if len(ls.beforeBuf) == 0 {
+ break
+ }
+ }
+
+ return nothing
+}
+
+// Do some processing for the "max" context.
+func (f *readFile) lContextProcessMaxCount(ctx context.Context, ls *ltxState) readStatus {
+ ls.maxCount--
+ if ls.maxCount == 0 {
+ if !ls.processAfter || ls.after == 0 {
+ return abortReading
+ }
+ // Unfortunatley we have to continue filter, as there might be more lines to print
+ ls.maxReached = true
+ }
+
+ return nothing
+}
diff --git a/internal/io/fs/tailfile.go b/internal/io/fs/tailfile.go
index 14994e5..7a40ac4 100644
--- a/internal/io/fs/tailfile.go
+++ b/internal/io/fs/tailfile.go
@@ -6,7 +6,7 @@ type TailFile struct {
}
// NewTailFile returns a new file tailer.
-func NewTailFile(filePath string, globID string, serverMessages chan<- string, limiter chan struct{}) TailFile {
+func NewTailFile(filePath string, globID string, serverMessages chan<- string) TailFile {
return TailFile{
readFile: readFile{
filePath: filePath,
@@ -15,7 +15,6 @@ func NewTailFile(filePath string, globID string, serverMessages chan<- string, l
retry: true,
canSkipLines: true,
seekEOF: true,
- limiter: limiter,
},
}
}
diff --git a/internal/io/fs/truncate.go b/internal/io/fs/truncate.go
deleted file mode 100644
index a8d59ac..0000000
--- a/internal/io/fs/truncate.go
+++ /dev/null
@@ -1,61 +0,0 @@
-package fs
-
-import (
- "context"
- "errors"
- "io"
- "os"
- "time"
-
- "github.com/mimecast/dtail/internal/io/logger"
-)
-
-func (f readFile) truncateTimer(ctx context.Context) (checkTruncate chan struct{}) {
- checkTruncate = make(chan struct{})
-
- go func() {
- for {
- select {
- case <-time.After(time.Second * 3):
- select {
- case checkTruncate <- struct{}{}:
- case <-ctx.Done():
- }
- case <-ctx.Done():
- return
- }
- }
- }()
-
- return
-}
-
-// Check wether log file is truncated. Returns nil if not.
-func (f readFile) truncated(fd *os.File) (bool, error) {
- logger.Debug(f.filePath, "File truncation check")
-
- // Can not seek currently open FD.
- curPos, err := fd.Seek(0, os.SEEK_CUR)
- if err != nil {
- return true, err
- }
-
- // Can not open file at original path.
- pathFd, err := os.Open(f.filePath)
- if err != nil {
- return true, err
- }
- defer pathFd.Close()
-
- // Can not seek file at original path.
- pathPos, err := pathFd.Seek(0, io.SeekEnd)
- if err != nil {
- return true, err
- }
-
- if curPos > pathPos {
- return true, errors.New("File got truncated")
- }
-
- return false, nil
-}
generated by cgit v1.2.3 (git 2.25.1) at 2026-08-11 10:38:50 +0000