From 1d90070ad1d6d1c88d34275f3bba57fcaa544b27aedcbd7df7fc718e2477db17 Mon Sep 17 00:00:00 2001 From: Lixen Wraith Date: Thu, 15 Oct 2026 03:05:51 -0400 Subject: [PATCH] v0.37.1 match bytes to follow a file tail, preventing reset on edit in most cases --- doc/sources.md | 42 ++- internal/source/file/file.go | 32 +- internal/source/file/file_test.go | 263 +++++++++++++++++ internal/source/file/file_watcher.go | 422 ++++++++++++++++----------- 4 files changed, 567 insertions(+), 192 deletions(-) diff --git a/doc/sources.md b/doc/sources.md index 1637077..2bb0c79 100644 --- a/doc/sources.md +++ b/doc/sources.md @@ -59,18 +59,36 @@ Options, each as type and default: written meanwhile is not read. `from = "start"` instead reads each file whole when its watcher is created, which a process writing beside LogWisp needs, at the cost of replaying files already on disk at every restart and reload. -- Rotation is detected from size decrease, a modification time that went back - without growth, a position more than 1 KiB beyond end-of-file, or an inode - change. An inode change where the new file is non-empty and at least as large - as the recorded position is treated as an atomic save, not a rotation, and - the position is preserved. Each rotation also publishes an INFO entry - `Log rotation detected (#N): REASON` with the file's name as `Source`; it - reaches every sink, so exclude it with a filter (`Log rotation detected`) - where unwanted. -- A rotation that renames in place — what a size-capped writer does — puts the - same inode back under a name `pattern` also matches. Its watcher resumes at - the position the original reached, so `from = "start"` reads the tail an - unfinished read left behind rather than the whole archive a second time. +- A file is followed by its content, not its inode or size. lw keeps the last + 8 lines read (1 KiB at most), matches them only whole (never a line still + being written), and assumes lines are mostly distinct, as timestamped logs + are; runs of identical lines can make it resume after a later copy. + - Appended to, it reads on. + - Rewritten, in place or by an editor's rename: reading resumes after the + newest of those lines still in the file's last 64 KiB, wherever lines were + deleted or inserted before it; inserted lines are not read. Nothing read + is replayed. + - None of those lines left (logrotate's `create` or `copytruncate`, a file + emptied and written again, an edit deleting them all), and no other + watcher having read the file (below), is a rotation: read from its start, + after an INFO entry + `Log rotation detected (#N): the file starts anew` with the file's name as + `Source`, which reaches every sink (filter `Log rotation detected` out + where unwanted). The NULs `copytruncate` leaves when a writer that does + not append goes on at its old offset are skipped, unless the file stayed + empty for a whole poll. + - While the line read last is missing and the file still changes between + polls, as when a slow save is rewriting it in place, lw holds that file, + for 2 s at most, and decides once it settles. + - Limits: a save appending more than 64 KiB after those lines loses them, + so it is read from its start. A rewrite still being written after 2 s is + judged as it stands: read from its start if it has not reached those + lines yet, else what it writes after that is replayed. +- A rotation that renames in place (a size-capped writer, a numbered logrotate + chain shifting `app.log.1` to `app.log.2`) puts the same inode back under a + name `pattern` also matches. The watcher there, new or already following + that name, resumes where the file was read to: it reads only the tail an + unfinished read left behind, never the whole archive a second time. - A line is parsed as JSON only when it is a non-empty object whose top-level keys are all drawn from `time`, `level`, `msg` and `fields` (the four an entry can carry), with any `level` or `msg` a string. `time` is read as diff --git a/internal/source/file/file.go b/internal/source/file/file.go index d309436..f060b15 100644 --- a/internal/source/file/file.go +++ b/internal/source/file/file.go @@ -10,7 +10,6 @@ import ( "strings" "sync" "sync/atomic" - "syscall" "time" "github.com/lixenwraith/logwisp/internal/config" @@ -276,6 +275,7 @@ func (fs *FileSource) ensureWatcher(path string) { } w := newFileWatcher(path, fs.config.Raw, fs.config.From == "start", fs.publish, fs.logger) + w.others = fs.othersOf(w) // A rotation renames the file out from under its watcher, so the same inode // reappears here under the archive name. Resume where it was left: from the // start would re-emit every record the file has already delivered. @@ -304,22 +304,44 @@ func (fs *FileSource) ensureWatcher(path string) { // readPosition reports how far a running watcher has read the file now at path. // Callers hold fs.mu. func (fs *FileSource) readPosition(path string) (int64, bool) { - info, err := os.Stat(path) + file, err := os.Open(path) if err != nil { return 0, false } - stat, ok := info.Sys().(*syscall.Stat_t) - if !ok { + defer file.Close() + info, err := file.Stat() + if err != nil { return 0, false } + return fs.readFile(file, info, nil) +} + +// readFile reports how far a running watcher other than but read the file, by +// its inode, if it still holds what that one read last before there: an inode +// can be reused, and more than one watcher can claim it. Callers hold fs.mu. +func (fs *FileSource) readFile(file *os.File, info os.FileInfo, but *fileWatcher) (int64, bool) { for _, w := range fs.watchers { - if position, ok := w.readTo(stat.Ino); ok { + if w == but { + continue + } + if position, tail, ok := w.readTo(inodeOf(info)); ok && leads(file, position, info.Size(), tail) { return position, true } } return 0, false } +// othersOf asks the source's other watchers, from w's goroutine, how far they +// read a file: w holds no lock of its own then, so the order is fs.mu, then a +// watcher's mu, as everywhere. +func (fs *FileSource) othersOf(w *fileWatcher) func(*os.File, os.FileInfo) (int64, bool) { + return func(file *os.File, info os.FileInfo) (int64, bool) { + fs.mu.RLock() + defer fs.mu.RUnlock() + return fs.readFile(file, info, w) + } +} + // removeWatcher removes only the watcher that finished. A deleted file can be // recreated before its old watcher observes stop; in that case ensureWatcher // has already installed a replacement under the same path, which must survive. diff --git a/internal/source/file/file_test.go b/internal/source/file/file_test.go index 7b58a3d..57c7e8c 100644 --- a/internal/source/file/file_test.go +++ b/internal/source/file/file_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "errors" + "fmt" "os" "path/filepath" "slices" @@ -210,3 +211,265 @@ func TestRotatedFileResumesInsteadOfReplaying(t *testing.T) { t.Error("a file no watcher has read was treated as rotated") } } + +// lines are log lines stamped with day, from line first up to but not +// including last +func lines(day, first, last int) string { + var b strings.Builder + for i := first; i < last; i++ { + fmt.Fprintf(&b, "Oct %d 00:00:%02d host app[1]: line %d\n", day, i%60, i) + } + return b.String() +} + +// appendFile adds text to the file's end, as a writer would +func appendFile(t *testing.T, path, text string) { + t.Helper() + f, err := os.OpenFile(path, os.O_WRONLY|os.O_APPEND, 0) + if err != nil { + t.Fatal(err) + } + if _, err := f.WriteString(text); err != nil { + t.Fatal(err) + } + if err := f.Close(); err != nil { + t.Fatal(err) + } +} + +// tailFrom starts a watcher at the file's end; each call of the check it +// returns polls once and gives the messages read +func tailFrom(t *testing.T, path string) func() []string { + t.Helper() + var got []string + w := newFileWatcher(path, true, false, func(e core.LogEntry) { got = append(got, e.Message) }, log.NewLogger()) + if err := w.initPosition(); err != nil { + t.Fatal(err) + } + return func() []string { + t.Helper() + got = nil + if err := w.checkFile(); err != nil { + t.Fatal(err) + } + return got + } +} + +// An edited file is still the file being read, however it is saved and +// wherever lines are deleted or inserted: reading resumes after the newest +// line read that it still holds, and replays nothing +func TestEditedFileResumesAfterTheLastLineRead(t *testing.T) { + saves := map[string]func(t *testing.T, check func() []string, path, text string){ + "by rename": func(t *testing.T, _ func() []string, path, text string) { + testutil.WriteFile(t, path+".tmp", text) + if err := os.Rename(path+".tmp", path); err != nil { + t.Fatal(err) + } + }, + "in place": func(t *testing.T, _ func() []string, path, text string) { testutil.WriteFile(t, path, text) }, + "in place, polled while empty": func(t *testing.T, check func() []string, path, text string) { + testutil.WriteFile(t, path, "") + if got := check(); len(got) != 0 { + t.Fatalf("read %q from the emptied file", got) + } + testutil.WriteFile(t, path, text) + }, + "in place, polled partway": func(t *testing.T, check func() []string, path, text string) { + testutil.WriteFile(t, path, text[:len(text)/2]) + if got := check(); len(got) != 0 { + t.Fatalf("read %q from the file half written", got) + } + testutil.WriteFile(t, path, text) + }, + } + for name, save := range saves { + t.Run(name, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "messages") + base := lines(14, 0, 20) + testutil.WriteFile(t, path, base) + check := tailFrom(t, path) + steps := []struct { + text string + want []string + }{ + {base + "ssh in\n", []string{"ssh in"}}, + {base, nil}, // the line read last, deleted + {base + "after\n", []string{"after"}}, + {lines(14, 0, 15) + lines(14, 16, 20) + "after\nmore\n", []string{"more"}}, // a line before it deleted + {lines(14, 0, 5) + "inserted\n" + lines(14, 5, 15) + lines(14, 16, 20) + "after\nmore\nagain\n", []string{"again"}}, + {lines(14, 10, 15) + lines(14, 16, 20) + "after\nmore\nagain\nlast\n", []string{"last"}}, // the top trimmed + {lines(14, 10, 15) + lines(14, 16, 20) + "after\nmore\nagain\nfinal\n", []string{"final"}}, // the line read last replaced + {"final\nend\n", []string{"end"}}, // trimmed down to the line read last + } + for i, step := range steps { + save(t, check, path, step.text) + // a missing line read last holds the file for one check, until it settles + if got := append(check(), check()...); !slices.Equal(got, step.want) { + t.Fatalf("save %d read %q, want %q", i+1, got, step.want) + } + } + }) + } +} + +// A file that starts anew is a rotation: it is read from its start, after the +// entry saying so and once it settles, however it was replaced, whatever its +// size, and wherever in a line the last read ended +func TestRotatedFileIsReadFromItsStart(t *testing.T) { + fresh := lines(15, 0, 3) + rotations := map[string]func(t *testing.T, check func() []string, path string){ + "truncated and written again": func(t *testing.T, _ func() []string, path string) { testutil.WriteFile(t, path, fresh) }, + "renamed away and made anew": func(t *testing.T, _ func() []string, path string) { + if err := os.Rename(path, path+".1"); err != nil { + t.Fatal(err) + } + testutil.WriteFile(t, path, fresh) + }, + "made anew past the position read": func(t *testing.T, _ func() []string, path string) { + testutil.WriteFile(t, path, fresh+lines(15, 3, 60)) + }, + "emptied, polled, then written": func(t *testing.T, check func() []string, path string) { + testutil.WriteFile(t, path, "") + if got := check(); len(got) != 0 { + t.Fatalf("read %q from the emptied file", got) + } + testutil.WriteFile(t, path, fresh) + }, + // a block-buffered writer leaves a line half written: no mark, though + // the new lines start as it does + "after a line read half written": func(t *testing.T, check func() []string, path string) { + appendFile(t, path, "Oct 1") + check() + testutil.WriteFile(t, path, fresh) + }, + // copytruncate under a writer that does not append: it goes on at its + // offset, past a hole of NULs, which is no entry + "truncated under a writer at its offset": func(t *testing.T, _ func() []string, path string) { + testutil.WriteFile(t, path, strings.Repeat("\x00", len(lines(14, 0, 20)))+fresh) + }, + } + for name, rotate := range rotations { + t.Run(name, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "messages") + testutil.WriteFile(t, path, lines(14, 0, 20)) + check := tailFrom(t, path) + rotate(t, check, path) + got := append(check(), check()...) + want := slices.Collect(strings.Lines(fresh)) + if len(got) < len(want)+1 || !strings.HasPrefix(got[0], "Log rotation detected (#1)") { + t.Fatalf("read %q, want the rotation entry, then the new file from its start", got) + } + for i, line := range want { + if got[i+1] != strings.TrimSuffix(line, "\n") { + t.Fatalf("read %q, want the new file from its start", got) + } + } + }) + } +} + +// readTo answers only for files the watcher has read: never the inode an +// editor's save freed, and while a check holds a replaced file, the one renamed +// away, not its replacement +func TestReadToAnswersOnlyForFilesRead(t *testing.T) { + path := filepath.Join(t.TempDir(), "messages") + inode := func(name string) uint64 { + info, err := os.Stat(name) + if err != nil { + t.Fatal(err) + } + return inodeOf(info) + } + text := lines(14, 0, 20) + testutil.WriteFile(t, path, text) + w := newFileWatcher(path, true, false, func(core.LogEntry) {}, log.NewLogger()) + if err := w.initPosition(); err != nil { + t.Fatal(err) + } + + freed := inode(path) + text += "edited\n" + testutil.WriteFile(t, path+".tmp", text) + if err := errors.Join(os.Rename(path+".tmp", path), w.checkFile()); err != nil { + t.Fatal(err) + } + if _, _, ok := w.readTo(freed); ok { + t.Error("the inode an editor's save freed is taken for an archive") + } + + archive := path + ".1" + if err := os.Rename(path, archive); err != nil { + t.Fatal(err) + } + testutil.WriteFile(t, path, "") + if err := w.checkFile(); err != nil { + t.Fatal(err) + } + if position, _, ok := w.readTo(inode(archive)); !ok || position != int64(len(text)) { + t.Errorf("held: the renamed file read to %d, %v; want %d, true", position, ok, len(text)) + } + if _, _, ok := w.readTo(inode(path)); ok { + t.Error("held: the replacement is answered for before it is judged") + } + if err := w.checkFile(); err != nil { + t.Fatal(err) + } + if position, _, ok := w.readTo(inode(archive)); !ok || position != int64(len(text)) { + t.Errorf("judged: the renamed file read to %d, %v; want %d, true", position, ok, len(text)) + } +} + +// A rotation shifts archives along their names (app.log to app.log.1, that to +// app.log.2): each watcher finds the file the one before it read and goes on +// from where that one got, so nothing is replayed and what it left is read +// once. An inode reused, claimed by a watcher of a deleted file, is not it. +func TestShiftedArchiveResumesWhereItWasRead(t *testing.T) { + for name, order := range map[string][]int{"the archive's watcher first": {1, 0}, "the file's watcher first": {0, 1}} { + t.Run(name, func(t *testing.T) { + dir := t.TempDir() + paths := []string{filepath.Join(dir, "app.log"), filepath.Join(dir, "app.log.1"), filepath.Join(dir, "app.log.2")} + testutil.WriteFile(t, paths[0], lines(15, 0, 20)) + testutil.WriteFile(t, paths[1], lines(14, 0, 20)) + var got []string + source := &FileSource{watchers: map[string]*fileWatcher{}} + watchers := make([]*fileWatcher, 2) + for i := range watchers { + w := newFileWatcher(paths[i], true, false, func(e core.LogEntry) { got = append(got, e.Message) }, log.NewLogger()) + w.others = source.othersOf(w) + if err := w.initPosition(); err != nil { + t.Fatal(err) + } + source.watchers[paths[i]], watchers[i] = w, w + } + + appendFile(t, paths[0], "unread\n") + for i := 1; i >= 0; i-- { + if err := os.Rename(paths[i], paths[i+1]); err != nil { + t.Fatal(err) + } + } + testutil.WriteFile(t, paths[0], "fresh\n") + info, err := os.Stat(paths[0]) + if err != nil { + t.Fatal(err) + } + source.watchers["deleted"] = &fileWatcher{inode: inodeOf(info), position: 3, tail: []byte("ab\n")} + source.watchers["deleted too"] = &fileWatcher{prevInode: watchers[0].inode, prevPosition: 3, prevTail: []byte("ab\n")} + for range 2 { + for _, i := range order { + if err := watchers[i].checkFile(); err != nil { + t.Fatal(err) + } + } + } + want := []string{"Log rotation detected (#1): the file starts anew", "fresh", "unread"} + if slices.Sort(got); !slices.Equal(got, want) { + t.Errorf("read %q, want %q", got, want) + } + if position, ok := source.readPosition(paths[2]); !ok || position != int64(len(lines(14, 0, 20))) { + t.Errorf("app.log.2 read to %d, %v; want %d, true", position, ok, len(lines(14, 0, 20))) + } + }) + } +} diff --git a/internal/source/file/file_watcher.go b/internal/source/file/file_watcher.go index 29ebe71..cca9578 100644 --- a/internal/source/file/file_watcher.go +++ b/internal/source/file/file_watcher.go @@ -2,6 +2,7 @@ package file import ( "bufio" + "bytes" "context" "encoding/json" "fmt" @@ -30,6 +31,23 @@ type WatcherInfo struct { Rotations int64 } +// The lines read last that are kept, to find the place again in a file an edit +// rewrote, newest first, are tailLines and tailBytes at most; findWithin is how +// much of its end is searched, what a save may append after them +const ( + tailLines = 8 + tailBytes = 1 << 10 + findWithin = 64 << 10 +) + +// settleChecks is how many checks a file whose line read last is missing may go +// on changing, a slow rewrite still being written, before lw decides +const settleChecks = 20 + +// holeProbe is how much at each end of a rotated file's bytes before the +// position must be NUL to take them for a hole (see pastHole) +const holeProbe = 4 << 10 + // fileWatcher tails a single file, handles rotations, and sends new lines to a callback type fileWatcher struct { directory string @@ -37,17 +55,23 @@ type fileWatcher struct { raw bool position int64 size int64 - inode uint64 + inode uint64 // the file position is in, set once a check has judged it modTime time.Time mu sync.Mutex stopped bool rotationSeq int64 prevInode uint64 prevPosition int64 + prevTail []byte entriesRead atomic.Uint64 lastReadTime atomic.Value // time.Time logger *log.Logger - failed string // the last check's error, "" for none; watch's + failed string // the last check's error, "" for none; watch's + tail []byte // the lines before position (lastLines), set with it under mu + seen os.FileInfo // the file as the last check found it; watch's + holds int // checks waited for the file to settle; watch's + // how far another watcher of the source read a file (FileSource.othersOf) + others func(*os.File, os.FileInfo) (int64, bool) } // newFileWatcher creates a new watcher for a specific file path. @@ -63,6 +87,7 @@ func newFileWatcher(directory string, raw, fromStart bool, callback func(core.Lo raw: raw, position: position, logger: logger, + others: func(*os.File, os.FileInfo) (int64, bool) { return 0, false }, } w.lastReadTime.Store(time.Time{}) return w @@ -100,14 +125,12 @@ func (w *fileWatcher) note(err error) { } } -// stop signals the watcher to terminate its loop func (w *fileWatcher) stop() { w.mu.Lock() w.stopped = true w.mu.Unlock() } -// getInfo returns a snapshot of the watcher's current statistics func (w *fileWatcher) getInfo() WatcherInfo { w.mu.Lock() info := WatcherInfo{ @@ -127,12 +150,16 @@ func (w *fileWatcher) getInfo() WatcherInfo { return info } -// checkFile examines the file for changes, rotations, or new content +// checkFile reads what the file has gained, whatever its inode or size. Where +// an edit rewrote it, reading resumes after the newest line read that it still +// holds; with none the place is lost: a file another watcher read goes on from +// where that one got, any other is a rotation, read from the start. While the +// line read last is missing and the file still changes, it is held, up to +// settleChecks, so that a slow rewrite is judged once written. func (w *fileWatcher) checkFile() error { file, err := os.Open(w.directory) if err != nil { if os.IsNotExist(err) { - // File doesn't exist yet, keep watching return nil } return err @@ -143,166 +170,217 @@ func (w *fileWatcher) checkFile() error { if err != nil { return err } + size, same := info.Size(), w.seen != nil && os.SameFile(info, w.seen) && info.Size() == w.seen.Size() && info.ModTime().Equal(w.seen.ModTime()) + w.seen = info w.mu.Lock() - oldPos := w.position - oldSize := w.size - oldInode := w.inode - oldModTime := w.modTime + pos := w.position + w.size, w.modTime = size, info.ModTime() w.mu.Unlock() - - currentSize := info.Size() - currentModTime := info.ModTime() - var currentInode uint64 - - if stat, ok := info.Sys().(*syscall.Stat_t); ok { - currentInode = stat.Ino + if same && pos >= size && w.holds == 0 { + return nil } - // Handle first time seeing a file that didn't exist before - if oldInode == 0 && currentInode != 0 { - // File just appeared, don't treat as rotation - w.mu.Lock() - w.inode = currentInode - w.size = currentSize - w.modTime = currentModTime - // Position stays at 0 for new files - w.mu.Unlock() - // Don't return here - continue to read content - } - - // Check for rotation - rotated := false - rotationReason := "" - startPos := oldPos - - // Rotation detection - if currentSize < oldSize { - // File was truncated - rotated = true - rotationReason = "size decrease" - } else if currentModTime.Before(oldModTime) && currentSize <= oldSize { - // Modification time went backwards (logrotate behavior) - rotated = true - rotationReason = "modification time reset" - } else if oldPos > currentSize+1024 { - // Our position is way beyond file size - rotated = true - rotationReason = "position beyond file size" - } else if oldInode != 0 && currentInode != 0 && currentInode != oldInode { - // Inode changed - distinguish between rotation and atomic save - if currentSize == 0 { - // Empty file with new inode = likely rotation - rotated = true - rotationReason = "inode change with empty file" - } else if currentSize < oldPos { - // New file is smaller than our position = rotation - rotated = true - rotationReason = "inode change with size less than position" - } else { - // Inode changed but file has content and size >= position - // This is likely an atomic save by an editor - // Update inode but keep position - w.mu.Lock() - w.inode = currentInode - w.mu.Unlock() - - w.logger.Debug("msg", "Atomic file update detected", - "component", "file_watcher", - "directory", w.directory, - "old_inode", oldInode, - "new_inode", currentInode, - "position", oldPos, - "size", currentSize) - } - } - - if rotated { - startPos = 0 - w.mu.Lock() - w.rotationSeq++ - seq := w.rotationSeq - // Retained for the source: the renamed file is about to be discovered - // under its archive name, and only this says how much of it was read. - w.prevInode, w.prevPosition = oldInode, oldPos - w.inode = currentInode - w.position = 0 // Reset position on rotation - w.mu.Unlock() - - w.callback(core.LogEntry{ - Time: time.Now(), - Source: filepath.Base(w.directory), - Level: "INFO", - Message: fmt.Sprintf("Log rotation detected (#%d): %s", seq, rotationReason), - }) - - w.logger.Info("msg", "Log rotation detected", - "component", "file_watcher", - "directory", w.directory, - "sequence", seq, - "reason", rotationReason) - } - - // Read if there's new content OR if we need to continue from position - if currentSize > startPos { - if _, err := file.Seek(startPos, io.SeekStart); err != nil { - return err - } - - scanner := bufio.NewScanner(file) - scanner.Buffer(make([]byte, 0, 64*1024), core.MaxLogEntryBytes) - scanner.Split(scanLinesCapped) - - for scanner.Scan() { - line := scanner.Text() - if line == "" { - continue - } - - rawSize := int64(len(line)) - entry := w.parseLine(line) - entry.RawSize = rawSize - - w.callback(entry) - w.entriesRead.Add(1) - w.lastReadTime.Store(time.Now()) - } - - if err := scanner.Err(); err != nil { - return err - } - - // Update position after successful read - currentPos, err := file.Seek(0, io.SeekCurrent) + rotated, lost, tail := false, false, w.tail + if !leads(file, pos, size, w.tail) { + at, which, err := w.find(file, pos, size) if err != nil { - // Log error but don't fail - best effort position tracking - w.logger.Warn("msg", "Failed to get file position", "error", err) - // Use size as fallback position - currentPos = currentSize + return err } - - w.mu.Lock() - w.position = currentPos - w.size = currentSize - w.modTime = currentModTime - if !rotated && currentInode != 0 { - w.inode = currentInode + switch { + case which != 0 && !same && w.holds < settleChecks: + w.holds++ + return nil + case which < 0: + // as a rotation shifts archives along their names, each watcher + // finds the file the one before it read + if p, read := w.others(file, info); read { + at = p + w.logger.Info("msg", "File replaced by one another watcher read; resuming where it got", + "component", "file_watcher", "path", w.directory, "resumed_at", at, "size", size) + } else { + rotated, at = true, pastHole(file, pos, size) + } + lost = true + default: + w.logger.Info("msg", "File rewritten; resuming after the newest line read it still holds", + "component", "file_watcher", "path", w.directory, "position", pos, "resumed_at", at, "size", size) + } + pos = at + if tail, err = lastLines(file, pos); err != nil { + return err } - w.mu.Unlock() - - return scanner.Err() } - - // Update metadata even if no new content w.mu.Lock() - w.size = currentSize - w.modTime = currentModTime - if currentInode != 0 { - w.inode = currentInode + if lost { + // Retained for the source: the renamed file is about to be found under + // its archive name, and only this says how much of it was read. + w.prevInode, w.prevPosition, w.prevTail = w.inode, w.position, w.tail } + w.inode, w.position, w.tail = inodeOf(info), pos, tail + w.mu.Unlock() + w.holds = 0 + if rotated { + w.rotated() + } + + if _, err := file.Seek(pos, io.SeekStart); err != nil { + return err + } + scanner := bufio.NewScanner(file) + scanner.Buffer(make([]byte, 0, 64*1024), core.MaxLogEntryBytes) + scanner.Split(scanLinesCapped) + for scanner.Scan() { + line := scanner.Text() + if line == "" { + continue + } + + entry := w.parseLine(line) + entry.RawSize = int64(len(line)) + + w.callback(entry) + w.entriesRead.Add(1) + w.lastReadTime.Store(time.Now()) + } + if err := scanner.Err(); err != nil { + return err + } + + end, err := file.Seek(0, io.SeekCurrent) + if err != nil { + w.logger.Warn("msg", "Failed to get file position", "error", err) + end = max(pos, size) + } + if tail, err = lastLines(file, end); err != nil { + return err + } + w.mu.Lock() + w.position, w.tail = end, tail + w.mu.Unlock() + return nil +} + +// rotated counts a rotation and says so, in lw's log and to the pipeline +func (w *fileWatcher) rotated() { + w.mu.Lock() + w.rotationSeq++ + seq := w.rotationSeq w.mu.Unlock() - return nil + w.callback(core.LogEntry{ + Time: time.Now(), + Source: filepath.Base(w.directory), + Level: "INFO", + Message: fmt.Sprintf("Log rotation detected (#%d): the file starts anew", seq), + }) + w.logger.Info("msg", "Log rotation detected", + "component", "file_watcher", + "directory", w.directory, + "sequence", seq) +} + +// leads reports whether the file holds tail just before pos +func leads(file *os.File, pos, size int64, tail []byte) bool { + if pos > size { + return false + } + got, err := readAt(file, pos-int64(len(tail)), int64(len(tail))) + return err == nil && bytes.Equal(got, tail) +} + +// find is where reading resumes in a file an edit rewrote: after the newest of +// the lines read last that its last findWithin bytes still hold, each matched +// whole, the file's start counting as a newline. which counts back from the +// newest, -1 where none is. +func (w *fileWatcher) find(file *os.File, pos, size int64) (at int64, which int, err error) { + lo := max(0, size-findWithin) + buf, err := readAt(file, lo, size-lo) + if err != nil { + return 0, -1, err + } + if lo == 0 { + buf, lo = append([]byte{'\n'}, buf...), -1 + } + end := len(w.tail) + if end > 0 && w.tail[end-1] != '\n' { // a line still being written is no mark + end, which = bytes.LastIndexByte(w.tail, '\n')+1, 1 + } + for ; end > 0; which++ { + start := bytes.LastIndexByte(w.tail[:end-1], '\n') + line := w.tail[max(0, start):end] + if start < 0 && pos == int64(len(w.tail)) { + line = append([]byte{'\n'}, line...) + } + if len(bytes.TrimSpace(line)) > 0 { + if i := bytes.LastIndex(buf, line); i >= 0 { + return lo + int64(i+len(line)), which, nil + } + } + end = start + 1 + } + return 0, -1, nil +} + +// pastHole is where a rotated file is read from: its start, or its first data +// past the position when the bytes before it are NUL, the hole copytruncate +// leaves under a writer that does not append and goes on at its offset +func pastHole(file *os.File, pos, size int64) int64 { + at := min(pos, size) + for _, from := range []int64{0, max(0, at-holeProbe)} { + if b, err := readAt(file, from, min(at-from, holeProbe)); err != nil || len(b) == 0 || len(bytes.Trim(b, "\x00")) > 0 { + return 0 + } + } + buf := make([]byte, holeProbe) + for at < size { + n, err := file.ReadAt(buf[:min(int64(len(buf)), size-at)], at) + if i := bytes.IndexFunc(buf[:n], func(r rune) bool { return r != 0 }); i >= 0 { + return at + int64(i) + } + if at += int64(n); err != nil { + break + } + } + return at +} + +// lastLines are the lines that end at end, tailLines and tailBytes at most, +// each with the newline before it so that it is found only whole +func lastLines(file *os.File, end int64) ([]byte, error) { + n := min(end, tailBytes) + buf, err := readAt(file, end-n, n) + cut := 0 + if end > n { // the first line is cut: from the newline after it + cut = max(0, bytes.IndexByte(buf[:max(0, len(buf)-1)], '\n')) + } + for i, lines := len(buf)-2, 0; i > cut; i-- { + if buf[i] == '\n' { + if lines++; lines == tailLines { + cut = i + break + } + } + } + return buf[cut:], err +} + +// readAt reads n bytes at off, fewer where the file ends first +func readAt(file *os.File, off, n int64) ([]byte, error) { + buf := make([]byte, n) + got, err := file.ReadAt(buf, off) + if err == io.EOF { + err = nil + } + return buf[:got], err +} + +// inodeOf is the file's inode, 0 where the platform gives none +func inodeOf(info os.FileInfo) uint64 { + if stat, ok := info.Sys().(*syscall.Stat_t); ok { + return stat.Ino + } + return 0 } // scanLinesCapped is bufio.ScanLines, except that a line longer than @@ -316,8 +394,8 @@ func scanLinesCapped(data []byte, atEOF bool) (int, []byte, error) { return advance, token, err } -// initPosition records the file's metadata and, unless the watcher was created -// to read from the start, sets the initial read position to the end. +// initPosition records the file's metadata and its marks and, unless the +// watcher was created to read from the start, sets the read position to the end. func (w *fileWatcher) initPosition() error { file, err := os.Open(w.directory) if err != nil { @@ -342,41 +420,35 @@ func (w *fileWatcher) initPosition() error { w.mu.Lock() defer w.mu.Unlock() - if w.position == -1 { - pos, err := file.Seek(0, io.SeekEnd) - if err != nil { - return err - } - w.position = pos - } - w.size = info.Size() w.modTime = info.ModTime() - if stat, ok := info.Sys().(*syscall.Stat_t); ok { - w.inode = stat.Ino + w.inode = inodeOf(info) + w.seen = info + if w.position == -1 || w.position > w.size { // a resumed position past the end is not this file's + w.position = w.size } - - return nil + w.tail, err = lastLines(file, w.position) + return err } -// readTo reports how far this watcher read the given inode: the file it tails -// now, or the one a rotation renamed out from under it. -func (w *fileWatcher) readTo(inode uint64) (int64, bool) { +// readTo reports how far this watcher read the given inode, and the lines it +// read last before there: the file it tails now, as last judged (not one a held +// check has yet to), or the one a rotation renamed out from under it. +func (w *fileWatcher) readTo(inode uint64) (int64, []byte, bool) { if inode == 0 { - return 0, false + return 0, nil, false } w.mu.Lock() defer w.mu.Unlock() switch inode { case w.inode: - return w.position, true + return w.position, w.tail, true case w.prevInode: - return w.prevPosition, true + return w.prevPosition, w.prevTail, true } - return 0, false + return 0, nil, false } -// isStopped checks if the watcher has been instructed to stop func (w *fileWatcher) isStopped() bool { w.mu.Lock() defer w.mu.Unlock()