// Package tailer watches a log file and delivers new bytes to a consumer. // // It uses fsnotify for event-driven notifications with a polling fallback // ticker, and detects log rotation by comparing the file's inode against // the inode stored in the state file. On rotation the offset resets to 0 // (subject to MaxRotationRecoverySize) and the state is updated. package tailer import ( "encoding/json" "fmt" "io" "os" "path/filepath" "sync" "syscall" "time" "github.com/fsnotify/fsnotify" "workshop.perforce.com/p4lf/internal/config" ) // State is the persistent checkpoint written to StateFile. type State struct { Inode uint64 `json:"inode"` Offset int64 `json:"offset"` } // Chunk is a slice of bytes read from the log file since the last flush. type Chunk struct { Data []byte StartOffset int64 EndOffset int64 } // Tailer watches a P4LOG file and delivers accumulated chunks on demand. type Tailer struct { cfg config.Config log Logger mu sync.Mutex file *os.File state State buf []byte startOff int64 // offset at the start of current accumulation window done chan struct{} watcher *fsnotify.Watcher } // Logger is the minimal interface Tailer needs for logging. type Logger interface { Infof(format string, args ...interface{}) Warnf(format string, args ...interface{}) Debugf(format string, args ...interface{}) Errorf(format string, args ...interface{}) } // New creates and starts a Tailer. func New(cfg config.Config, log Logger) (*Tailer, error) { t := &Tailer{ cfg: cfg, log: log, done: make(chan struct{}), } // Load or initialise state. if err := t.loadState(); err != nil { return nil, fmt.Errorf("loading state: %w", err) } // Open the log file. if err := t.openFile(); err != nil { return nil, err } // Start fsnotify watcher on the log file's directory (more reliable than // watching the file directly, since rotation replaces the file). w, err := fsnotify.NewWatcher() if err != nil { t.file.Close() return nil, fmt.Errorf("creating fsnotify watcher: %w", err) } t.watcher = w if err := w.Add(filepath.Dir(cfg.P4LogFile)); err != nil { t.file.Close() w.Close() return nil, fmt.Errorf("watching directory %q: %w", filepath.Dir(cfg.P4LogFile), err) } // Background goroutine: drain fsnotify events so the channel doesn't fill. // The actual reads happen in Flush(), called on the LogTailDelay ticker. go t.drainEvents() return t, nil } // loadState reads the state file, or initialises fresh state. func (t *Tailer) loadState() error { data, err := os.ReadFile(t.cfg.StateFile) if os.IsNotExist(err) { // No state file — fresh start. t.state = State{} return nil } if err != nil { return fmt.Errorf("reading state file %q: %w", t.cfg.StateFile, err) } if err := json.Unmarshal(data, &t.state); err != nil { t.log.Warnf("state file %q is corrupt (%v); starting fresh", t.cfg.StateFile, err) t.state = State{} } return nil } // openFile opens the P4LOG and positions the read cursor, performing rotation // detection and handling the ReadFromStart / MaxRotationRecoverySize logic. func (t *Tailer) openFile() error { f, err := os.Open(t.cfg.P4LogFile) if err != nil { return fmt.Errorf("opening P4LogFile %q: %w", t.cfg.P4LogFile, err) } inode, err := fileInode(f) if err != nil { f.Close() return fmt.Errorf("stat P4LogFile: %w", err) } info, err := f.Stat() if err != nil { f.Close() return fmt.Errorf("stat P4LogFile: %w", err) } switch { case t.state.Inode == 0: // First ever run (no state file). t.handleFirstRun(f, inode, info.Size()) case inode == t.state.Inode: // Same file — resume from saved offset. if t.state.Offset > info.Size() { // Truncation without inode change (copytruncate style). t.log.Warnf("P4LOG appears truncated (saved offset %d > file size %d); treating as rotation", t.state.Offset, info.Size()) t.handleRotation(f, inode, info.Size()) } else { if _, err := f.Seek(t.state.Offset, io.SeekStart); err != nil { f.Close() return fmt.Errorf("seeking to saved offset %d: %w", t.state.Offset, err) } t.log.Infof("Resuming P4LOG from offset %d (inode %d)", t.state.Offset, inode) } default: // Inode changed — log rotation detected. t.log.Infof("Log rotation detected (inode changed %d → %d)", t.state.Inode, inode) t.handleRotation(f, inode, info.Size()) } t.file = f t.startOff = t.state.Offset return nil } func (t *Tailer) handleFirstRun(f *os.File, inode uint64, size int64) { t.state.Inode = inode if t.cfg.ReadFromStart { t.log.Infof("First run: reading P4LOG from beginning (inode %d, size %d)", inode, size) t.state.Offset = 0 } else { t.log.Infof("First run: starting from end of P4LOG (inode %d, size %d)", inode, size) f.Seek(0, io.SeekEnd) t.state.Offset = size } } func (t *Tailer) handleRotation(f *os.File, inode uint64, size int64) { t.state.Inode = inode if t.cfg.MaxRotationRecoverySize == 0 || size <= t.cfg.MaxRotationRecoverySize { t.log.Infof("Reading new P4LOG from beginning (size %d ≤ MaxRotationRecoverySize %d)", size, t.cfg.MaxRotationRecoverySize) t.state.Offset = 0 } else { t.log.Warnf("New P4LOG size %d exceeds MaxRotationRecoverySize %d; starting from EOF to avoid overload", size, t.cfg.MaxRotationRecoverySize) f.Seek(0, io.SeekEnd) t.state.Offset = size } } // drainEvents keeps the fsnotify channel clear so it never blocks. // Actual data reads happen in Flush(), not here. func (t *Tailer) drainEvents() { for { select { case <-t.done: return case _, ok := <-t.watcher.Events: if !ok { return } // We don't act on individual events; Flush() reads everything // available each tick. The watcher just prevents us from missing // a rotation between ticks. case err, ok := <-t.watcher.Errors: if !ok { return } t.log.Errorf("fsnotify error: %v", err) } } } // Flush reads all bytes available since the last call and returns a Chunk. // It also checks for log rotation (inode change or truncation). // Returns a nil Chunk if no new data was available. func (t *Tailer) Flush() (*Chunk, error) { t.mu.Lock() defer t.mu.Unlock() if err := t.checkRotation(); err != nil { return nil, err } buf := make([]byte, 32*1024) for { n, err := t.file.Read(buf) if n > 0 { t.buf = append(t.buf, buf[:n]...) t.state.Offset += int64(n) } if err == io.EOF { break } if err != nil { return nil, fmt.Errorf("reading P4LOG: %w", err) } } if len(t.buf) == 0 { return nil, nil } chunk := &Chunk{ Data: t.buf, StartOffset: t.startOff, EndOffset: t.state.Offset, } t.buf = nil t.startOff = t.state.Offset return chunk, nil } // checkRotation detects inode changes and truncation, reopening the file // as needed. Must be called with t.mu held. func (t *Tailer) checkRotation() error { // Stat the path (not the open fd) to see if the file was replaced. pathInfo, err := os.Stat(t.cfg.P4LogFile) if os.IsNotExist(err) { // File temporarily absent (mid-rotation); wait for next tick. t.log.Warnf("P4LOG %q not found; waiting for rotation to complete", t.cfg.P4LogFile) return nil } if err != nil { return fmt.Errorf("stat %q: %w", t.cfg.P4LogFile, err) } newInode := inodeFromInfo(pathInfo) truncated := t.state.Offset > pathInfo.Size() if newInode == t.state.Inode && !truncated { return nil // no rotation } // Rotation or truncation detected. if truncated && newInode == t.state.Inode { t.log.Warnf("P4LOG truncated (copytruncate rotation?); resetting to offset 0") } else { // Flush whatever remains in the current file before switching. remaining, readErr := io.ReadAll(t.file) if readErr == nil && len(remaining) > 0 { t.buf = append(t.buf, remaining...) t.state.Offset += int64(len(remaining)) t.log.Infof("Captured %d trailing bytes from rotated P4LOG before switching", len(remaining)) } t.log.Infof("Log rotation detected during flush (inode %d → %d)", t.state.Inode, newInode) } t.file.Close() newFile, err := os.Open(t.cfg.P4LogFile) if err != nil { return fmt.Errorf("reopening P4LOG after rotation: %w", err) } t.file = newFile t.handleRotation(newFile, newInode, pathInfo.Size()) t.startOff = t.state.Offset return nil } // SaveState atomically writes the current state to StateFile. func (t *Tailer) SaveState() error { t.mu.Lock() state := t.state t.mu.Unlock() data, err := json.MarshalIndent(state, "", " ") if err != nil { return err } tmp := t.cfg.StateFile + ".tmp" if err := os.WriteFile(tmp, data, 0600); err != nil { return fmt.Errorf("writing state file: %w", err) } return os.Rename(tmp, t.cfg.StateFile) } // Close shuts down the tailer. func (t *Tailer) Close() { close(t.done) t.watcher.Close() t.mu.Lock() defer t.mu.Unlock() if t.file != nil { t.file.Close() } } // WaitForFile blocks until the P4LOG file appears, retrying every retryInterval. // Useful at startup if the file doesn't exist yet. func WaitForFile(path string, retryInterval time.Duration, log Logger, done <-chan struct{}) error { for { if _, err := os.Stat(path); err == nil { return nil } log.Warnf("P4LOG %q not found; retrying in %s", path, retryInterval) select { case <-done: return fmt.Errorf("shutdown before P4LOG appeared") case <-time.After(retryInterval): } } } // fileInode returns the inode number of an open file. func fileInode(f *os.File) (uint64, error) { info, err := f.Stat() if err != nil { return 0, err } return inodeFromInfo(info), nil } // inodeFromInfo extracts the inode from an os.FileInfo. // On Linux/macOS this is sys.Ino; falls back to 0 on unsupported platforms. func inodeFromInfo(info os.FileInfo) uint64 { if sys, ok := info.Sys().(*syscall.Stat_t); ok { return sys.Ino } return 0 }