package stats import ( "bytes" "errors" "io" "io/fs" "os" ) // Tailer follows a log file that named rotates (renaming it to FILE.0, // FILE.1, ... and starting a new FILE). It remembers its position as // (inode, offset) so reading resumes after a restart. type Tailer struct { Path string f *os.File ino uint64 offset int64 partial []byte // draining is set while finishing a rotated file before switching // to the current one. draining bool } // Position is where the tailer is, for persisting. type Position struct { Inode uint64 `json:"inode"` Offset int64 `json:"offset"` } // Position returns the current position (of complete lines). func (t *Tailer) Position() Position { return Position{Inode: t.ino, Offset: t.offset - int64(len(t.partial))} } // Resume continues from a saved position: in the current file, or in the // newest rotated file if named rotated it in the meantime. Without a // usable position, reading starts at the beginning of the current file. func (t *Tailer) Resume(p Position) { if p.Inode == 0 { return } for _, path := range []string{t.Path, t.Path + ".0"} { fi, err := os.Stat(path) if err != nil || inode(fi) != p.Inode || fi.Size() < p.Offset { continue } f, err := os.Open(path) if err != nil { return } if _, err := f.Seek(p.Offset, io.SeekStart); err != nil { f.Close() return } t.f, t.ino, t.offset, t.draining = f, p.Inode, p.Offset, path != t.Path return } } // open opens the current file from the start. A missing file is not an // error: named creates it when it logs the first response. func (t *Tailer) open() (bool, error) { f, err := os.Open(t.Path) if errors.Is(err, fs.ErrNotExist) { return false, nil } if err != nil { return false, err } fi, err := f.Stat() if err != nil { f.Close() return false, err } t.f, t.ino, t.offset, t.partial, t.draining = f, inode(fi), 0, nil, false return true, nil } // Read calls fn for every complete new line and returns how many lines it // read. Call it periodically. func (t *Tailer) Read(fn func(line string)) (int, error) { if t.f == nil { ok, err := t.open() if !ok || err != nil { return 0, err } } n := 0 buf := make([]byte, 64<<10) for { k, err := t.f.Read(buf) if k > 0 { t.offset += int64(k) data := append(t.partial, buf[:k]...) for { i := bytes.IndexByte(data, '\n') if i < 0 { break } fn(string(data[:i])) n++ data = data[i+1:] } t.partial = append([]byte(nil), data...) if len(t.partial) > 1<<20 { // not a log line; drop it t.partial = nil } } if err == io.EOF || k == 0 { break } if err != nil { return n, err } } // At the end of the file: has named rotated or truncated it? fi, err := os.Stat(t.Path) switch { case errors.Is(err, fs.ErrNotExist): return n, nil // rotated and not recreated yet case err != nil: return n, err case t.draining || inode(fi) != t.ino: t.f.Close() t.f = nil if _, err := t.open(); err != nil { return n, err } if t.f != nil { m, err := t.Read(fn) return n + m, err } case fi.Size() < t.offset: // Truncated (e.g. by hand): start again from the top. if _, err := t.f.Seek(0, io.SeekStart); err != nil { return n, err } t.offset, t.partial = 0, nil } return n, nil } // Close closes the file. func (t *Tailer) Close() { if t.f != nil { t.f.Close() t.f = nil } }