jarl/internal/stats/tail.go

package stats

import (
	"bytes"
	"errors"
	"io"
	"io/fs"
	"os"
	"syscall"
)

// 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"`
}

func inode(fi fs.FileInfo) uint64 {
	if st, ok := fi.Sys().(*syscall.Stat_t); ok {
		return st.Ino
	}
	return 0
}

// 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
	}
}