hamlab/internal/metrics/zfswrite.go

package metrics

// The ZFS write path, for the ZFS tab: what is waiting in RAM to be written, how fast each
// pool is writing to its disks, and — where a pool has one — the landing tier in front of it.
//
// A landing tier is a mergerfs pool whose first branches are fast non-ZFS storage and whose
// last branch is a ZFS dataset: new files land on the fast branch and a mover drains them onto
// the pool. It is detected from mergerfs's own branch list (the user.mergerfs.branches xattr on
// the pool's .mergerfs control file), so it shows wherever one exists and nowhere else. A mover
// that writes a status file into /run/landing-mover/ (hamlet's scripts/landing-mover.py does)
// adds what it is doing: its queue, the file in flight, totals and the last error.
//
// Sources, all cheap enough for every sample:
//   - arcstats anon_size: dirty data in RAM, for all pools together (ZFS has no live per-pool
//     figure; txgs only records ndirty once a txg has synced)
//   - dmu_tx: dmu_tx_dirty_delay / dmu_tx_dirty_over_max — writers slowed by the write throttle,
//     or stopped at the hard limit
//   - <pool>/txgs: nwritten per synced txg — bytes the pool actually wrote to its vdevs
//   - <pool>/objset-*: nwritten per dataset — bytes programs wrote, before compression
// The landing branch's file walk is slower and runs every landingEvery.

import (
	"encoding/json"
	"io/fs"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"syscall"
	"time"
)

const (
	zfsKstatDir    = "/proc/spl/kstat/zfs/"
	zfsParamDir    = "/sys/module/zfs/parameters/"
	moverStateDir  = "/run/landing-mover"
	landingEvery   = 10 * time.Second
	landingWalkMax = 200000 // entries; past this the counts are marked partial
)

// ZFSWrites is the write path as one sample sees it. Nil without the ZFS module.
type ZFSWrites struct {
	Dirty         uint64       `json:"dirty"`            // bytes in RAM written but not yet on disk, all pools
	DirtyMax      uint64       `json:"dirty_max"`        // zfs_dirty_data_max
	DelayMinPct   int          `json:"delay_min_pct"`    // zfs_delay_min_dirty_percent: the throttle starts here
	Limit         uint64       `json:"limit"`            // what the buffer can really hold now: the smaller of dirty_max and half the ARC target
	DelayedPerSec float64      `json:"delayed_per_sec"`  // writes the throttle slowed, per second over the interval
	OverMaxPerSec float64      `json:"over_max_per_sec"` // writes stopped dead at the limit
	TxgTimeoutSec int          `json:"txg_timeout_sec"`  // zfs_txg_timeout
	Pools         []PoolWrites `json:"pools"`
}

// PoolWrites is one pool.
type PoolWrites struct {
	Pool       string   `json:"pool"`
	WriteBps   float64  `json:"write_bps"`   // to its vdevs, after compression, parity included
	LogicalBps float64  `json:"logical_bps"` // into its datasets, as programs wrote it
	Landing    *Landing `json:"landing,omitempty"`
}

// Landing is a landing tier in front of a pool.
type Landing struct {
	Mount          string  `json:"mount"`  // the mergerfs pool, e.g. /srv/tank
	Branch         string  `json:"branch"` // the landing branch, e.g. /srv/landing/tank
	PoolBranch     string  `json:"pool_branch"`
	Device         string  `json:"device,omitempty"` // block device under the branch, as in Sample.Disks
	FSType         string  `json:"fstype,omitempty"`
	Size           uint64  `json:"size"`
	Used           uint64  `json:"used"`
	PendingFiles   int     `json:"pending_files"`
	PendingBytes   uint64  `json:"pending_bytes"`
	PendingPartial bool    `json:"pending_partial,omitempty"` // the walk stopped at landingWalkMax
	OldestMs       int64   `json:"oldest_ms,omitempty"`       // mtime of the oldest waiting file
	CountedMs      int64   `json:"counted_ms"`                // when the walk ran
	InBps          float64 `json:"in_bps"`                    // the landing device's writes
	DrainBps       float64 `json:"drain_bps"`                 // the mover's progress onto the pool
	Mover          *Mover  `json:"mover,omitempty"`
}

// Mover is what a landing mover says about itself in its status file.
type Mover struct {
	State        string  `json:"state"` // idle | moving | holding (a mount it needs is missing)
	Src          string  `json:"src"`
	Dst          string  `json:"dst"`
	PID          int     `json:"pid"`
	StartedMs    int64   `json:"started_ms"`
	UpdatedMs    int64   `json:"updated_ms"`
	Stale        bool    `json:"stale"` // not updated for a while: stopped or hung
	Queue        int     `json:"queue"`
	Settle       float64 `json:"settle"`
	BWLimit      string  `json:"bwlimit"`
	BatchFiles   int     `json:"batch_files"`
	BatchBytes   uint64  `json:"batch_bytes"`
	BatchStartMs int64   `json:"batch_start_ms"`
	BatchFirst   string  `json:"batch_first"`
	Current      string  `json:"current"`      // the file being copied, when it can tell
	CurrentSize  uint64  `json:"current_size"` // its full size
	Partial      uint64  `json:"partial"`      // bytes of it already on the pool
	MovedFiles   uint64  `json:"moved_files"`  // since the mover started
	MovedBytes   uint64  `json:"moved_bytes"`
	LastMove     string  `json:"last_move"`
	LastMoveMs   int64   `json:"last_move_ms"`
	LastError    string  `json:"last_error"`
	LastErrorMs  int64   `json:"last_error_ms"`
}

type zfsWritePrev struct {
	at        time.Time
	delay     uint64
	overMax   uint64
	txg       map[string]uint64  // pool -> highest synced txg counted
	logical   map[string]uint64  // pool -> sum of objset nwritten
	drained   map[string]uint64  // landing branch -> moved + partial bytes
	landing   map[string]Landing // branch -> last walk
	landingAt time.Time
}

func readParamU(name string) uint64 {
	v, _ := strconv.ParseUint(strings.TrimSpace(readStr(zfsParamDir+name)), 10, 64)
	return v
}

// poolsWithKstats lists the imported pools: the directories under the kstat tree.
func poolsWithKstats() []string {
	ents, err := os.ReadDir(zfsKstatDir)
	if err != nil {
		return nil
	}
	var out []string
	for _, e := range ents {
		if e.IsDir() {
			if _, err := os.Stat(zfsKstatDir + e.Name() + "/txgs"); err == nil {
				out = append(out, e.Name())
			}
		}
	}
	sort.Strings(out)
	return out
}

// parseTxgs reads a pool's txgs kstat and returns, for every committed txg above `after`, the
// bytes it wrote, plus the highest committed txg seen.
func parseTxgs(text string, after uint64) (written, top uint64) {
	top = after
	for _, line := range strings.Split(text, "\n") {
		f := strings.Fields(line)
		// txg birth state ndirty nread nwritten reads writes otime qtime wtime stime
		if len(f) < 12 || f[2] != "C" {
			continue
		}
		txg, err := strconv.ParseUint(f[0], 10, 64)
		if err != nil {
			continue
		}
		if txg > top {
			top = txg
		}
		if txg > after {
			n, _ := strconv.ParseUint(f[5], 10, 64)
			written += n
		}
	}
	return written, top
}

// objsetWritten sums nwritten over a pool's datasets.
func objsetWritten(pool string) uint64 {
	m, _ := filepath.Glob(zfsKstatDir + pool + "/objset-*")
	var t uint64
	for _, p := range m {
		for _, line := range strings.Split(readStr(p), "\n") {
			f := strings.Fields(line)
			if len(f) == 3 && f[0] == "nwritten" {
				v, _ := strconv.ParseUint(f[2], 10, 64)
				t += v
			}
		}
	}
	return t
}

// mountEntry is one line of /proc/mounts.
type mountEntry struct{ src, point, fstype string }

func readMounts() []mountEntry {
	var out []mountEntry
	for _, line := range strings.Split(readStr("/proc/mounts"), "\n") {
		f := strings.Fields(line)
		if len(f) >= 3 {
			out = append(out, mountEntry{src: unescapeMount(f[0]), point: unescapeMount(f[1]), fstype: f[2]})
		}
	}
	return out
}

// unescapeMount undoes /proc/mounts' octal escapes (\040 for a space, and so on).
func unescapeMount(s string) string {
	if !strings.Contains(s, `\`) {
		return s
	}
	var b strings.Builder
	for i := 0; i < len(s); i++ {
		if s[i] == '\\' && i+3 < len(s) {
			if v, err := strconv.ParseUint(s[i+1:i+4], 8, 8); err == nil {
				b.WriteByte(byte(v))
				i += 3
				continue
			}
		}
		b.WriteByte(s[i])
	}
	return b.String()
}

// mountFor is the mount a path lives on: the longest mount point that is a prefix of it.
func mountFor(ms []mountEntry, path string) (mountEntry, bool) {
	var best mountEntry
	found := false
	for _, m := range ms {
		if m.point == path || m.point == "/" || strings.HasPrefix(path, strings.TrimSuffix(m.point, "/")+"/") {
			if !found || len(m.point) > len(best.point) {
				best, found = m, true
			}
		}
	}
	return best, found
}

// parseBranches reads mergerfs's branch list: "/srv/landing/tank=RW:/tank=RW,50G".
func parseBranches(v string) (paths []string, modes []string) {
	for _, b := range strings.Split(strings.TrimSpace(v), ":") {
		if b == "" {
			continue
		}
		p, mode, _ := strings.Cut(b, "=")
		mode, _, _ = strings.Cut(mode, ",")
		if mode == "" {
			mode = "RW"
		}
		paths, modes = append(paths, p), append(modes, mode)
	}
	return paths, modes
}

func getxattr(path, name string) string {
	buf := make([]byte, 4096)
	n, err := syscall.Getxattr(path, name, buf)
	if err != nil || n <= 0 {
		return ""
	}
	return string(buf[:n])
}

// findLandings maps pool -> the landing tiers in front of it.
func findLandings(ms []mountEntry) map[string][]Landing {
	out := map[string][]Landing{}
	for _, m := range ms {
		if m.fstype != "fuse.mergerfs" {
			continue
		}
		paths, modes := parseBranches(getxattr(filepath.Join(m.point, ".mergerfs"), "user.mergerfs.branches"))
		// the pool branch is the first ZFS one; landing branches are writable non-ZFS ones before it
		for i, p := range paths {
			bm, ok := mountFor(ms, p)
			if !ok || bm.fstype != "zfs" {
				continue
			}
			pool, _, _ := strings.Cut(bm.src, "/")
			for j := 0; j < i; j++ {
				lm, ok := mountFor(ms, paths[j])
				if !ok || lm.fstype == "zfs" || modes[j] == "RO" {
					continue
				}
				out[pool] = append(out[pool], Landing{Mount: m.point, Branch: paths[j], PoolBranch: p, FSType: lm.fstype, Device: blockDisk(lm.src)})
			}
			break
		}
	}
	return out
}

// blockDisk turns a mount source into the name Sample.Disks uses: /dev/nvme0n1p1 -> nvme0,
// /dev/sdb2 -> sdb. Empty for anything that is not a plain block device.
func blockDisk(src string) string {
	dev, err := filepath.EvalSymlinks(src)
	if err != nil || !strings.HasPrefix(dev, "/dev/") {
		return ""
	}
	name := filepath.Base(dev)
	// a partition's sysfs entry sits inside its disk's
	if p, err := filepath.EvalSymlinks("/sys/class/block/" + name); err == nil {
		if _, err := os.Stat(p + "/partition"); err == nil {
			name = filepath.Base(filepath.Dir(p))
		}
	}
	if m := nsRe.FindStringSubmatch(name); m != nil {
		return m[1] // nvme0n1 -> nvme0, as foldNVMe names it
	}
	return name
}

// walkLanding counts what is waiting on a landing branch.
func walkLanding(l *Landing, now time.Time) {
	var st syscall.Statfs_t
	if syscall.Statfs(l.Branch, &st) == nil {
		l.Size = st.Blocks * uint64(st.Bsize)
		l.Used = (st.Blocks - st.Bfree) * uint64(st.Bsize)
	}
	seen := 0
	l.PendingFiles, l.PendingBytes, l.OldestMs, l.PendingPartial = 0, 0, 0, false
	_ = filepath.WalkDir(l.Branch, func(p string, d fs.DirEntry, err error) error {
		if err != nil {
			return nil
		}
		if seen++; seen > landingWalkMax {
			l.PendingPartial = true
			return filepath.SkipAll
		}
		if !d.Type().IsRegular() {
			return nil
		}
		if fi, err := d.Info(); err == nil {
			l.PendingFiles++
			l.PendingBytes += uint64(fi.Size())
			if ms := fi.ModTime().UnixMilli(); l.OldestMs == 0 || ms < l.OldestMs {
				l.OldestMs = ms
			}
		}
		return nil
	})
	l.CountedMs = now.UnixMilli()
}

// readMovers loads every mover status file, keyed by the branch it drains.
func readMovers(now time.Time) map[string]*Mover {
	out := map[string]*Mover{}
	m, _ := filepath.Glob(moverStateDir + "/*.json")
	for _, p := range m {
		b, err := os.ReadFile(p)
		if err != nil {
			continue
		}
		var mv Mover
		if json.Unmarshal(b, &mv) != nil || mv.Src == "" {
			continue
		}
		mv.Stale = now.UnixMilli()-mv.UpdatedMs > 30_000
		if mv.Stale {
			mv.State = "stopped"
		}
		out[filepath.Clean(mv.Src)] = &mv
	}
	return out
}

// zfsWrites samples the write path; nil when this machine has no ZFS module loaded.
func (c *Collector) zfsWrites(now time.Time, arc *ARC, disks []Disk) *ZFSWrites {
	if arc == nil {
		return nil
	}
	p := &c.prevZW
	dt := 0.0
	if !p.at.IsZero() {
		dt = now.Sub(p.at).Seconds()
	}
	w := &ZFSWrites{Dirty: arc.Anon, DirtyMax: readParamU("zfs_dirty_data_max"),
		DelayMinPct: int(readParamU("zfs_delay_min_dirty_percent")), TxgTimeoutSec: int(readParamU("zfs_txg_timeout")), Pools: []PoolWrites{}}
	w.Limit = w.DirtyMax
	if half := arc.Target / 2; half > 0 && half < w.Limit {
		w.Limit = half
	}
	tx := readKstat(zfsKstatDir + "dmu_tx")
	if dt > 0 && tx["dmu_tx_dirty_delay"] >= p.delay && tx["dmu_tx_dirty_over_max"] >= p.overMax {
		w.DelayedPerSec = round1(float64(tx["dmu_tx_dirty_delay"]-p.delay) / dt)
		w.OverMaxPerSec = round1(float64(tx["dmu_tx_dirty_over_max"]-p.overMax) / dt)
	}
	p.delay, p.overMax = tx["dmu_tx_dirty_delay"], tx["dmu_tx_dirty_over_max"]
	if p.txg == nil {
		p.txg, p.logical, p.drained, p.landing = map[string]uint64{}, map[string]uint64{}, map[string]uint64{}, map[string]Landing{}
	}

	ms := readMounts()
	landings := findLandings(ms)
	walk := now.Sub(p.landingAt) >= landingEvery
	if walk {
		p.landingAt = now
	}
	movers := readMovers(now)
	diskW := map[string]float64{}
	for _, d := range disks {
		diskW[d.Name] = d.WriteBps
	}

	for _, pool := range poolsWithKstats() {
		pw := PoolWrites{Pool: pool}
		prevTop, had := p.txg[pool]
		written, top := parseTxgs(readStr(zfsKstatDir+pool+"/txgs"), prevTop)
		if had && dt > 0 {
			pw.WriteBps = round1(float64(written) / dt)
		}
		p.txg[pool] = top
		lw := objsetWritten(pool)
		if prev, ok := p.logical[pool]; ok && dt > 0 && lw >= prev {
			pw.LogicalBps = round1(float64(lw-prev) / dt)
		}
		p.logical[pool] = lw

		if ls := landings[pool]; len(ls) > 0 {
			l := ls[0] // one landing tier per pool is what the view shows
			if last, ok := p.landing[l.Branch]; ok && !walk {
				l.Size, l.Used, l.PendingFiles, l.PendingBytes = last.Size, last.Used, last.PendingFiles, last.PendingBytes
				l.OldestMs, l.PendingPartial, l.CountedMs = last.OldestMs, last.PendingPartial, last.CountedMs
			} else {
				walkLanding(&l, now)
				p.landing[l.Branch] = l
			}
			l.InBps = diskW[l.Device]
			if mv := movers[filepath.Clean(l.Branch)]; mv != nil {
				l.Mover = mv
				done := mv.MovedBytes + mv.Partial
				if prev, ok := p.drained[l.Branch]; ok && dt > 0 && done >= prev {
					l.DrainBps = round1(float64(done-prev) / dt)
				}
				p.drained[l.Branch] = done
			}
			pw.Landing = &l
		}
		w.Pools = append(w.Pools, pw)
	}
	p.at = now
	return w
}