mesh-catalog/modules/zfs/cmd/zfs-tools/follow.go

package main

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"os/exec"
	"path/filepath"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"syscall"
	"time"
)

// The module's second long-running process (`zfs-tools events`), run by the node-engine as root.
//
// ZFS keeps its events in a queue in the kernel, in memory, from the moment its module loads: a reboot
// loses them. `zpool events -f` gives the queue from its oldest event kept, then follows. The follower
// writes each error and change of state (resilver and scrub starts and ends among them) to its journal, so
// they outlive the reboot where the journal is kept on disk. It reads the queue from its start, so the
// pool's import and whatever came while the follower was down reach the journal too; a cursor of the boot
// and the last event read, in CursorFile, keeps it from writing an event twice when it restarts. A line
// ZFS gives when its queue was full and it dropped events is written as it comes.
//
// It writes no other file and changes nothing. It is judged by its unit (novox/hq ADR 0240): after
// MaxFollowFailures follows in a row that end within FollowSettle, it exits non-zero, so the service
// manager's restart counting and the unit's state say so.

// FollowArgv follows ZFS's events from the oldest kept, with their details.
func FollowArgv() []string { return []string{"events", "-f", "-v"} }

// CursorFile is where the follower keeps the boot and the last event it read. Under /run, so a reboot,
// which empties ZFS's queue, empties it too.
const CursorFile = "/run/zfs-events/cursor"

// BootID names this boot.
const BootID = "/proc/sys/kernel/random/boot_id"

const (
	// FollowRetry is how long the follower waits before following again after zpool ended.
	FollowRetry = 30 * time.Second
	// FollowSettle is how long a follow must last to count as having worked.
	FollowSettle = time.Minute
	// MaxFollowFailures in a row end the follower with an error.
	MaxFollowFailures = 5
)

// Cursor is the last event read in this boot: its id and its time. ZFS numbers events from its module's
// load, so after a reload a smaller id with a later time is a new event.
type Cursor struct {
	Boot string    `json:"boot"`
	EID  uint64    `json:"eid"`
	Time time.Time `json:"time"`
	path string
}

// LoadCursor is the cursor kept for this boot; an empty one for another boot or none kept.
func LoadCursor(path, boot string) *Cursor {
	c := &Cursor{Boot: boot, path: path}
	b, err := os.ReadFile(path)
	if err != nil {
		return c
	}
	var kept Cursor
	if json.Unmarshal(b, &kept) != nil || kept.Boot != boot {
		return c
	}
	kept.path = path
	return &kept
}

// Save keeps the cursor, atomically.
func (c *Cursor) Save(path string) error {
	if path == "" {
		return nil
	}
	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
		return err
	}
	b, _ := json.Marshal(c)
	tmp := path + ".new"
	if err := os.WriteFile(tmp, b, 0o600); err != nil {
		return err
	}
	return os.Rename(tmp, path)
}

// seen says whether an event was read already.
func (c *Cursor) seen(eid uint64, at time.Time, timed bool) bool {
	if c.EID == 0 && c.Time.IsZero() {
		return false
	}
	if eid > c.EID {
		return false
	}
	// A smaller or equal id: read, unless it is later than the cursor (ZFS's counter began again).
	return !timed || !at.After(c.Time)
}

// eventTime reads an event's time as `zpool events` gives it: "Oct 8 2026 21:01:55.889900112".
func eventTime(s string) (time.Time, bool) {
	t, err := time.ParseInLocation("Jan 2 2006 15:04:05.999999999", strings.Join(strings.Fields(s), " "), time.Local)
	return t, err == nil
}

// EventLine is an event in one journal line.
func EventLine(e Event) string {
	line := e.Time + " " + e.Class
	keys := make([]string, 0, len(e.Details))
	for k := range e.Details {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	for _, k := range keys {
		line += fmt.Sprintf(" %s=%s", k, e.Details[k])
	}
	return line
}

var droppedLine = regexp.MustCompile(`^dropped (\d+) events$`)

// Follow reads `zpool events -f -v` and writes each notable event the cursor has not seen in a line, and
// each line saying ZFS dropped events; the cursor moves past every event read and is saved as it goes.
func Follow(r io.Reader, out io.Writer, cur *Cursor) error {
	sc := bufio.NewScanner(r)
	sc.Buffer(make([]byte, 64*1024), 1024*1024)
	var block []string
	flush := func() {
		if len(block) == 0 {
			return
		}
		for _, e := range ParseEvents(strings.Join(block, "\n")) {
			at, timed := eventTime(e.Time)
			eid, _ := strconv.ParseUint(strings.TrimPrefix(e.Details["eid"], "0x"), 16, 64)
			if cur.seen(eid, at, timed) {
				continue
			}
			if Notable(e.Class) {
				fmt.Fprintln(out, EventLine(e))
			}
			cur.EID = eid
			if timed {
				cur.Time = at
			}
			_ = cur.Save(cur.path)
		}
		block = block[:0]
	}
	for sc.Scan() {
		l := sc.Text()
		if m := droppedLine.FindStringSubmatch(strings.TrimSpace(l)); m != nil {
			flush()
			fmt.Fprintf(out, "ZFS dropped %s events before they were read: its queue was full\n", m[1])
			continue
		}
		if strings.TrimSpace(l) == "" {
			flush()
			continue
		}
		block = append(block, l)
	}
	flush()
	return sc.Err()
}

// FollowProcess is a running `zpool events -f`.
type FollowProcess interface {
	Stdout() io.Reader
	Kill() error
	Wait() error
}

// FollowOnce follows until zpool ends or its output cannot be read. On a failure to read, zpool is
// killed before it is waited for: it would otherwise follow on, and the wait never end.
func FollowOnce(p FollowProcess, out io.Writer, cur *Cursor) error {
	ferr := Follow(p.Stdout(), out, cur)
	if ferr != nil {
		_ = p.Kill()
	}
	werr := p.Wait()
	if ferr != nil {
		return fmt.Errorf("reading zpool events: %w", ferr)
	}
	if werr != nil {
		return fmt.Errorf("zpool events -f ended: %w", werr)
	}
	return errors.New("zpool events -f ended")
}

// RunFollower follows again after each end, and gives up with an error after MaxFollowFailures follows in
// a row that did not last FollowSettle.
func RunFollower(ctx context.Context, out io.Writer, start func(context.Context) (FollowProcess, error), cur *Cursor,
	cursorPath string, wait func(time.Duration)) error {
	if cursorPath != "" {
		cur.path = cursorPath
	}
	failures := 0
	for ctx.Err() == nil {
		began := time.Now()
		var err error
		p, serr := start(ctx)
		if serr != nil {
			err = serr
		} else {
			err = FollowOnce(p, out, cur)
		}
		if ctx.Err() != nil {
			return nil
		}
		if time.Since(began) >= FollowSettle {
			failures = 0
		}
		failures++
		fmt.Fprintf(out, "following ZFS's events stopped (%v); %d of %d in a row\n", err, failures, MaxFollowFailures)
		if failures >= MaxFollowFailures {
			return fmt.Errorf("following ZFS's events failed %d times in a row, the last: %v", failures, err)
		}
		wait(FollowRetry)
	}
	return nil
}

// zpoolFollow is the real `zpool events -f -v`, in a process group of its own.
type zpoolFollow struct {
	cmd *exec.Cmd
	out io.ReadCloser
}

func (z *zpoolFollow) Stdout() io.Reader { return z.out }
func (z *zpoolFollow) Kill() error       { return syscall.Kill(-z.cmd.Process.Pid, syscall.SIGKILL) }
func (z *zpoolFollow) Wait() error       { return z.cmd.Wait() }

// StartZpoolFollow starts `zpool events -f -v`; the context's end kills it.
func StartZpoolFollow(ctx context.Context) (FollowProcess, error) {
	cmd := exec.Command("zpool", FollowArgv()...)
	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
	cmd.WaitDelay = KillGrace
	out, err := cmd.StdoutPipe()
	if err != nil {
		return nil, err
	}
	if err := cmd.Start(); err != nil {
		return nil, err
	}
	z := &zpoolFollow{cmd: cmd, out: out}
	go func() {
		<-ctx.Done()
		_ = z.Kill()
	}()
	return z, nil
}

// Follower is the events process: the cursor for this boot, then RunFollower.
func Follower(ctx context.Context, out io.Writer) error {
	boot, _ := os.ReadFile(BootID)
	cur := LoadCursor(CursorFile, strings.TrimSpace(string(boot)))
	return RunFollower(ctx, out, StartZpoolFollow, cur, CursorFile, func(d time.Duration) {
		select {
		case <-ctx.Done():
		case <-time.After(d):
		}
	})
}