From 86cfcd2bad4e688d322579fc80b9fc09546a8077 Mon Sep 17 00:00:00 2001 From: Mitchell-Johnson Date: Sun, 25 Oct 2026 22:28:40 -0700 Subject: [PATCH] =?UTF-8?q?ZFS=20tab:=20Writes=20=E2=80=94=20the=20RAM=20w?= =?UTF-8?q?rite=20buffer,=20each=20pool's=20write=20rate,=20and=20a=20land?= =?UTF-8?q?ing=20tier=20with=20its=20mover?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Live from arcstats (dirty), dmu_tx (throttled writes), each pool's txgs (bytes synced to disk) and objset kstats (bytes from programs). A landing tier — a mergerfs pool whose first branch is fast storage and whose last is a ZFS dataset — is detected from mergerfs's branch xattr; a mover status file in /run/landing-mover/ adds its queue, the file in flight and totals. Sensors zfs.write.*, zfs..write_bps|logical_bps and zfs..landing.* carry the history. Co-Authored-By: Claude Opus 5.5 --- internal/api/zfs.go | 6 +- internal/metrics/fold.go | 45 ++++ internal/metrics/metrics.go | 5 +- internal/metrics/sensors.go | 26 ++ internal/metrics/zfswrite.go | 433 ++++++++++++++++++++++++++++++ internal/metrics/zfswrite_test.go | 86 ++++++ web/src/ZFS.tsx | 142 +++++++++- web/src/api.ts | 18 +- web/src/styles.css | 2 + 9 files changed, 754 insertions(+), 9 deletions(-) create mode 100644 internal/metrics/zfswrite.go create mode 100644 internal/metrics/zfswrite_test.go diff --git a/internal/api/zfs.go b/internal/api/zfs.go index 2095254..5b9e5ae 100644 --- a/internal/api/zfs.go +++ b/internal/api/zfs.go @@ -138,11 +138,13 @@ func (s *Server) zfsOverview(w http.ResponseWriter, r *http.Request) { // zfsARC is the cache: the sampler's live ARC (size, the hit rate over the last interval) // beside the storage package's fuller picture (since-boot counters, tunables, and each -// pool's cache/log/special devices or the lack of them). +// pool's cache/log/special devices or the lack of them), and the write path — the dirty +// buffer in RAM, each pool's write rate, and any landing tier in front of a pool. func (s *Server) zfsARC(w http.ResponseWriter, r *http.Request) { - out := map[string]any{"detail": s.storage.ARC(r.Context()), "live": nil} + out := map[string]any{"detail": s.storage.ARC(r.Context()), "live": nil, "writes": nil} if smp, ok := s.metrics.Latest(); ok && smp.ARC != nil { out["live"] = smp.ARC + out["writes"] = smp.ZFSWrite } jsonOK(w, 200, out) } diff --git a/internal/metrics/fold.go b/internal/metrics/fold.go index 2aa52b2..2ba7aa8 100644 --- a/internal/metrics/fold.go +++ b/internal/metrics/fold.go @@ -122,9 +122,54 @@ func fold(b []Sample) Sample { out.Disks[i].ReadBps, out.Disks[i].WriteBps, out.Disks[i].UtilPct = v[0], v[1], v[2] } } + out.ZFSWrite = foldZFSWrites(b, out.ZFSWrite) return out } +// foldZFSWrites averages the write path's rates and the buffer's fill over a bucket, per pool; +// everything else (limits, the mover's state, the last walk) is the newest sample's. +func foldZFSWrites(b []Sample, last *ZFSWrites) *ZFSWrites { + if last == nil { + return nil + } + w := *last + w.Pools = append([]PoolWrites(nil), last.Pools...) + var dirty, delayed, over, n float64 + for _, s := range b { + if s.ZFSWrite != nil { + dirty, delayed, over, n = dirty+float64(s.ZFSWrite.Dirty), delayed+s.ZFSWrite.DelayedPerSec, over+s.ZFSWrite.OverMaxPerSec, n+1 + } + } + if n > 0 { + w.Dirty, w.DelayedPerSec, w.OverMaxPerSec = uint64(dirty/n), round1(delayed/n), round1(over/n) + } + per := avgBy(b, func(s Sample) map[string][]float64 { + m := map[string][]float64{} + if s.ZFSWrite == nil { + return m + } + for _, p := range s.ZFSWrite.Pools { + v := []float64{p.WriteBps, p.LogicalBps, 0, 0} + if p.Landing != nil { + v[2], v[3] = p.Landing.InBps, p.Landing.DrainBps + } + m[p.Pool] = v + } + return m + }) + for i, p := range w.Pools { + if v, ok := per[p.Pool]; ok { + w.Pools[i].WriteBps, w.Pools[i].LogicalBps = round1(v[0]), round1(v[1]) + if p.Landing != nil { + l := *p.Landing + l.InBps, l.DrainBps = round1(v[2]), round1(v[3]) + w.Pools[i].Landing = &l + } + } + } + return &w +} + // avgBy averages each sample's named numbers across the bucket, per name. func avgBy(b []Sample, of func(Sample) map[string][]float64) map[string][]float64 { sums := map[string][]float64{} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index c95e0bf..872d2df 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -150,7 +150,8 @@ type Sample struct { FS []FS `json:"fs"` UPS *UPS `json:"ups"` GPUs []GPU `json:"gpus"` - ARC *ARC `json:"arc"` // the ZFS read cache; nil without the module + ARC *ARC `json:"arc"` // the ZFS read cache; nil without the module + ZFSWrite *ZFSWrites `json:"zfs_writes"` // the ZFS write path; nil without the module Uptime float64 `json:"uptime"` fsT, upsT int64 // when the slow sections were really read; sensors report these, not T @@ -182,6 +183,7 @@ type Collector struct { gpu gpuState power cpuPowerState prevARC arcPrev + prevZW zfsWritePrev subMu sync.Mutex subs map[chan struct{}]struct{} // told after every sample; see Subscribe @@ -368,6 +370,7 @@ func (c *Collector) sample() { s.UPS, s.upsT = c.lastUPS, c.upsAt.UnixMilli() s.GPUs = c.gpus(now, cpuTemp(s.Temps)) s.ARC = c.arc(now) + s.ZFSWrite = c.zfsWrites(now, s.ARC, s.Disks) if first { return // deltas are meaningless on the priming pass diff --git a/internal/metrics/sensors.go b/internal/metrics/sensors.go index aa23da5..2d338bd 100644 --- a/internal/metrics/sensors.go +++ b/internal/metrics/sensors.go @@ -203,6 +203,32 @@ func flatten(s Sample, want func(id string) bool) []sensors.Sensor { } } + if w := s.ZFSWrite; w != nil { + f.add("zfs.write.dirty", "ZFS write buffer", "bytes", "B", float64(w.Dirty), t, upTo(float64(w.Limit))) + f.add("zfs.write.limit", "ZFS write buffer limit", "bytes", "B", float64(w.Limit), t, none) + f.add("zfs.write.delayed", "ZFS writes throttled", "rate", "/s", w.DelayedPerSec, t, none) + for _, p := range w.Pools { + pp := "zfs." + sensors.Seg(p.Pool) + "." + f.add(pp+"write_bps", p.Pool+" writes to disk", "rate", "B/s", p.WriteBps, t, none, "pool", p.Pool) + f.add(pp+"logical_bps", p.Pool+" writes by programs", "rate", "B/s", p.LogicalBps, t, none, "pool", p.Pool) + if l := p.Landing; l != nil { + lp := pp + "landing." + lbl := []string{"pool", p.Pool, "branch", l.Branch} + f.add(lp+"pending_bytes", p.Pool+" landing: waiting", "bytes", "B", float64(l.PendingBytes), t, upTo(float64(l.Size)), lbl...) + f.add(lp+"pending_files", p.Pool+" landing: files waiting", "count", "", float64(l.PendingFiles), t, none, lbl...) + if l.Size > 0 { + f.add(lp+"used_pct", p.Pool+" landing: used", "usage", "%", round1(float64(l.Used)/float64(l.Size)*100), t, pct, lbl...) + } + f.add(lp+"in_bps", p.Pool+" landing: arriving", "rate", "B/s", l.InBps, t, none, lbl...) + f.add(lp+"drain_bps", p.Pool+" landing: moving to the pool", "rate", "B/s", l.DrainBps, t, none, lbl...) + if m := l.Mover; m != nil { + f.add(lp+"mover", p.Pool+" landing: mover", "state", "", m.State, t, none, lbl...) + f.add(lp+"queue", p.Pool+" landing: mover queue", "count", "", float64(m.Queue), t, none, lbl...) + } + } + } + } + if u := s.UPS; u != nil { ut := s.upsT f.add("ups.battery", "UPS battery", "usage", "%", u.BatteryPct, ut, pct, "model", u.Model) diff --git a/internal/metrics/zfswrite.go b/internal/metrics/zfswrite.go new file mode 100644 index 0000000..1bb8846 --- /dev/null +++ b/internal/metrics/zfswrite.go @@ -0,0 +1,433 @@ +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 +// - /txgs: nwritten per synced txg — bytes the pool actually wrote to its vdevs +// - /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 +} diff --git a/internal/metrics/zfswrite_test.go b/internal/metrics/zfswrite_test.go new file mode 100644 index 0000000..1c1652d --- /dev/null +++ b/internal/metrics/zfswrite_test.go @@ -0,0 +1,86 @@ +package metrics + +import "testing" + +// real rows from grid's /proc/spl/kstat/zfs/tank/txgs, 2026-10-25 +const txgsSample = `81 1 0x01 26 7072 20029594359 13953814329135 +txg birth state ndirty nread nwritten reads writes otime qtime wtime stime +15812833 13936408791659 C 0 0 0 0 0 5119980242 2040 58385 309580 +15812834 13941528771901 C 0 0 0 0 0 5119948815 8471 65826 112612 +15812835 13946648720716 C 885760 0 2293760 0 367 5120100556 1704 56762 271264374 +15812836 13951768821272 S 4194304 0 0 0 0 5120000000 1000 50000 0 +15812837 13956888821272 O 0 0 0 0 0 0 0 0 0 +` + +func TestParseTxgs(t *testing.T) { + w, top := parseTxgs(txgsSample, 0) + if w != 2293760 || top != 15812835 { + t.Fatalf("from 0: written %d top %d", w, top) + } + // only txgs synced since the last look count, and an open or syncing txg never does + w, top = parseTxgs(txgsSample, 15812834) + if w != 2293760 || top != 15812835 { + t.Fatalf("after 15812834: written %d top %d", w, top) + } + w, top = parseTxgs(txgsSample, 15812835) + if w != 0 || top != 15812835 { + t.Fatalf("nothing new: written %d top %d", w, top) + } +} + +func TestParseBranches(t *testing.T) { + p, m := parseBranches("/srv/landing/tank=RW:/tank=NC,50G:/mnt/old=RO\n") + if len(p) != 3 || p[0] != "/srv/landing/tank" || p[1] != "/tank" || p[2] != "/mnt/old" { + t.Fatalf("paths %q", p) + } + if m[0] != "RW" || m[1] != "NC" || m[2] != "RO" { + t.Fatalf("modes %q", m) + } + if p, _ := parseBranches("/a:/b"); len(p) != 2 { + t.Fatalf("bare paths %q", p) + } +} + +func TestMountFor(t *testing.T) { + ms := []mountEntry{ + {"/dev/sdj2", "/", "ext4"}, + {"/dev/nvme0n1p1", "/srv/landing", "ext4"}, + {"tank", "/tank", "zfs"}, + {"tank/media", "/tank/media", "zfs"}, + {"tank-landing", "/srv/tank", "fuse.mergerfs"}, + } + cases := map[string]string{ + "/srv/landing/tank": "/srv/landing", "/tank": "/tank", "/tank/media/x": "/tank/media", + "/tankfoo": "/", "/srv/landingx": "/", + } + for path, want := range cases { + if m, ok := mountFor(ms, path); !ok || m.point != want { + t.Errorf("%s: got %q want %q", path, m.point, want) + } + } +} + +func TestUnescapeMount(t *testing.T) { + if got := unescapeMount(`/mnt/my\040disk`); got != "/mnt/my disk" { + t.Fatalf("got %q", got) + } +} + +func TestFoldZFSWrites(t *testing.T) { + mk := func(dirty uint64, bps, in, drain float64) Sample { + return Sample{ZFSWrite: &ZFSWrites{Dirty: dirty, Limit: 100, Pools: []PoolWrites{{Pool: "tank", WriteBps: bps, + Landing: &Landing{Branch: "/srv/landing/tank", InBps: in, DrainBps: drain, Mover: &Mover{State: "moving"}}}}}} + } + b := []Sample{mk(10, 100, 400, 0), mk(30, 300, 0, 60)} + got := fold(b).ZFSWrite + if got.Dirty != 20 || got.Pools[0].WriteBps != 200 || got.Pools[0].Landing.InBps != 200 || got.Pools[0].Landing.DrainBps != 30 { + t.Fatalf("folded %+v / %+v", got, got.Pools[0].Landing) + } + // the fold must not write through to the newest sample it copied from + if b[1].ZFSWrite.Pools[0].Landing.InBps != 0 || b[1].ZFSWrite.Dirty != 30 { + t.Fatalf("fold changed its input") + } + if got.Pools[0].Landing.Mover.State != "moving" { + t.Fatalf("mover state lost") + } +} diff --git a/web/src/ZFS.tsx b/web/src/ZFS.tsx index 23dac84..b022b1a 100644 --- a/web/src/ZFS.tsx +++ b/web/src/ZFS.tsx @@ -9,8 +9,8 @@ // disagree over. An existing cron job is detected and shown, so the page tells the truth // about the machine whether Ham Lab or something else is doing the work. import { FormEvent, Fragment, useCallback, useEffect, useState } from 'react' -import { api, ARCView, CacheDev, PoolCaching, ZFSExternal, ZFSInfo, ZFSOverview, ZFSPolicy, ZFSRun, ZFSSnapshot } from './api' -import { fmtBytes } from './fmt' +import { api, ARCView, CacheDev, Landing, PoolCaching, PoolWrites, ZFSExternal, ZFSInfo, ZFSOverview, ZFSPolicy, ZFSRun, ZFSSnapshot, ZFSWrites } from './api' +import { fmtBytes, fmtRate } from './fmt' import { useRefreshMs } from './prefs' import { fetchHistories } from './sensors' import Sparkline from './Sparkline' @@ -367,18 +367,142 @@ function PoolTiers({ p, v }: { p: PoolCaching; v: ARCView }) { ) } +// ---- the write path: the dirty buffer in RAM, what each pool writes, and a landing tier ---- +// +// A write reaches a pool in up to three steps. If a landing tier fronts the pool (a mergerfs +// pool whose first branch is fast storage and whose last is the pool), new files land there +// first and a mover drains them. Either way the data then waits in RAM as dirty buffers, and +// each txg sync writes it to the disks. Each step's live number is here, with its history. + +const dur = (s: number) => (!Number.isFinite(s) || s <= 0 ? '—' : s < 90 ? `${Math.round(s)} s` : s < 5400 ? `${Math.round(s / 60)} min` : `${(s / 3600).toFixed(1)} h`) +type WHist = Record +const wIds = (w: ZFSWrites) => ['zfs.write.dirty', ...w.pools.flatMap((p) => { + const b = `zfs.${p.pool}.` + return [b + 'write_bps', b + 'logical_bps', ...(p.landing ? [b + 'landing.pending_bytes', b + 'landing.in_bps', b + 'landing.drain_bps'] : [])] +})] + +function MoverLine({ l }: { l: Landing }) { + const m = l.mover + if (!m) return
No mover is reporting for this landing tier (nothing in /run/landing-mover/). Files that land here stay here until something moves them.
+ const frac = m.current_size > 0 ? m.partial / m.current_size : 0 + return ( + <> +
+
Mover + {m.state} + {' '}{m.state === 'moving' && m.batch_files > 1 ? `batch of ${m.batch_files} files, ${fmtBytes(m.batch_bytes)}` : m.state === 'holding' ? 'a mount it needs is missing — waiting' : m.state === 'stopped' ? `no word since ${ago(m.updated_ms)}` : `queue ${m.queue}`} + {m.state === 'moving' && m.queue > 0 ? ` · ${m.queue} more queued` : ''} +
+ {m.state === 'moving' && m.current &&
Now{m.current}{m.current_size ? ` · ${fmtBytes(m.partial)} of ${fmtBytes(m.current_size)}` : ''}
} +
Since start{fmtN(m.moved_files)} files · {fmtBytes(m.moved_bytes)} moved · running {ago(m.started_ms).replace(' ago', '')}
+ {m.last_move &&
Last moved{m.last_move} · {ago(m.last_move_ms)}
} +
Rulesa file moves once it has been closed and left alone for {m.settle} s · limit {m.bwlimit && m.bwlimit !== '0' ? `${m.bwlimit}B/s` : 'none'} · oldest first
+ {m.last_error &&
Last error{m.last_error} · {ago(m.last_error_ms)}
} +
+ {m.state === 'moving' && m.current_size > 0 && } + + ) +} + +function LandingBlockInner({ l, pending, inn, drain }: { l: Landing; pending: number[]; inn: number[]; drain: number[] }) { + const eta = l.drain_bps > 0 ? l.pending_bytes / l.drain_bps : NaN + const fill = l.size > 0 ? l.used / l.size : 0 + return ( +
+
Landing + {l.branch}{l.device ? ` · ${l.device}` : ''}{l.fstype ? ` · ${l.fstype}` : ''} · in front of {l.pool_branch} as {l.mount} +
+
+ {fmtBytes(l.used)} of {fmtBytes(l.size)} used + {l.pending_files ? `${fmtN(l.pending_files)}${l.pending_partial ? '+' : ''} waiting · ${fmtBytes(l.pending_bytes)}` : 'nothing waiting'} +
+ +
+
Waiting{l.pending_files ? `${fmtN(l.pending_files)}${l.pending_partial ? '+' : ''} files, ${fmtBytes(l.pending_bytes)} · oldest landed ${ago(l.oldest_ms)}` : 'nothing — everything that landed is on the pool'} · counted {ago(l.counted_ms)}
+
Arriving{fmtRate(l.in_bps)}{l.device ? · writes to {l.device} : ''}
+
To the pool{fmtRate(l.drain_bps)}{l.pending_bytes > 0 ? ` · ${Number.isFinite(eta) ? `clear in about ${dur(eta)}` : 'not draining right now'}` : ''}
+
+ +
+
+
Waiting on the landing tier · {fmtBytes(l.pending_bytes)}
+ +
+
+
Arriving {fmtRate(l.in_bps)} · to the pool {fmtRate(l.drain_bps)}
+ +
+
+
+ ) +} + +function WritePath({ w, p, h }: { w: ZFSWrites; p: PoolWrites; h: WHist }) { + const b = `zfs.${p.pool}.` + const throttling = w.delayed_per_sec > 0 || w.over_max_per_sec > 0 + const delayAt = (w.limit * w.delay_min_pct) / 100 + const ratio = p.write_bps > 0 && p.logical_bps > 0 ? p.logical_bps / p.write_bps : 0 + return ( +
+
+ 0 ? 'bad' : throttling ? 'warn' : 'ok')} /> + {p.pool} + write path + + {fmtRate(p.write_bps)} to disk +
+ {p.landing + ? + :
Landingnone — writes go straight into the pool's RAM buffer
} +
+
+
RAM write buffer · {fmtBytes(w.dirty)} of {fmtBytes(w.limit)}
+ +
+
+
Writes · {fmtRate(p.write_bps)} to disk · {fmtRate(p.logical_bps)} from programs
+ +
+
+
+
In RAM{fmtBytes(w.dirty)} waiting · holds up to {fmtBytes(w.limit)} {w.limit < w.dirty_max ? `(half the ARC target; zfs_dirty_data_max is ${fmtBytes(w.dirty_max)})` : '(zfs_dirty_data_max)'}
+
Throttle + {w.over_max_per_sec > 0 ? {fmtN(w.over_max_per_sec)}/s writes stopped at the limit : w.delayed_per_sec > 0 ? slowing {fmtN(w.delayed_per_sec)} writes/s — the disks are the limit now : 'not slowing anything'} + · starts at {w.delay_min_pct}% ({fmtBytes(delayAt)})
+
To disk{fmtRate(p.write_bps)} written by syncs{ratio > 0 ? ` · ${fmtRate(p.logical_bps)} from programs (${ratio >= 1 ? `${ratio.toFixed(2)}x compressed` : 'parity and metadata included'})` : ''} · a sync at least every {w.txg_timeout_sec} s, sooner as the buffer fills
+
+
+ ) +} + +function LandingBlockWithHist({ l, h, b }: { l: Landing; h: WHist; b: string }) { + return +} + function ARCSection() { const [v, setV] = useState(null) const [hist, setHist] = useState<{ size: number[]; hit: number[] }>({ size: [], hit: [] }) + const [whist, setWhist] = useState({}) const [err, setErr] = useState('') const rate = useRefreshMs() useEffect(() => { let stop = false const poll = async () => { - try { const r = await api.zfsARC(); if (!stop) { setV(r); setErr('') } } catch (e) { if (!stop) setErr(e instanceof Error ? e.message : 'failed') } + let r: ARCView | null = null + try { r = await api.zfsARC(); if (!stop) { setV(r); setErr('') } } catch (e) { if (!stop) setErr(e instanceof Error ? e.message : 'failed') } try { - const hs = await fetchHistories('zfs.arc.size,zfs.arc.hit_pct') - if (!stop) setHist({ size: hs.find((h) => h.id === 'zfs.arc.size')?.v ?? [], hit: hs.find((h) => h.id === 'zfs.arc.hit_pct')?.v ?? [] }) + const extra = r?.writes ? wIds(r.writes) : [] + const hs = await fetchHistories(['zfs.arc.size', 'zfs.arc.hit_pct', ...extra].join(',')) + if (!stop) { + setHist({ size: hs.find((h) => h.id === 'zfs.arc.size')?.v ?? [], hit: hs.find((h) => h.id === 'zfs.arc.hit_pct')?.v ?? [] }) + setWhist(Object.fromEntries(hs.filter((h) => extra.includes(h.id)).map((h) => [h.id, h.v]))) + } } catch { /* the sparklines stay empty; the numbers still show */ } } poll() @@ -445,6 +569,14 @@ function ARCSection() { {d.pools.length > 0 &&
{d.pools.map((p) => )}
} + {v.writes && v.writes.pools.length > 0 && ( + <> +

Writes

+ what is waiting to reach each pool, and how fast it is getting there +
+
{v.writes.pools.map((p) => )}
+ + )} ) } diff --git a/web/src/api.ts b/web/src/api.ts index b618a6e..b819c0f 100644 --- a/web/src/api.ts +++ b/web/src/api.ts @@ -252,7 +252,23 @@ export interface ARCDetail { dbuf: { Size: number; Target: number; Hits: number; Misses: number } pools: PoolCaching[] } -export interface ARCView { detail: ARCDetail; live: ARCLive | null } +export interface LandingMover { + state: 'idle' | 'moving' | 'holding' | 'stopped' | string; src: string; dst: string; pid: number; started_ms: number; updated_ms: number; stale: boolean + queue: number; settle: number; bwlimit: string; batch_files: number; batch_bytes: number; batch_start_ms: number; batch_first: string + current: string; current_size: number; partial: number; moved_files: number; moved_bytes: number + last_move: string; last_move_ms: number; last_error: string; last_error_ms: number +} +export interface Landing { + mount: string; branch: string; pool_branch: string; device?: string; fstype?: string; size: number; used: number + pending_files: number; pending_bytes: number; pending_partial?: boolean; oldest_ms?: number; counted_ms: number + in_bps: number; drain_bps: number; mover?: LandingMover +} +export interface PoolWrites { pool: string; write_bps: number; logical_bps: number; landing?: Landing } +export interface ZFSWrites { + dirty: number; dirty_max: number; delay_min_pct: number; limit: number; delayed_per_sec: number; over_max_per_sec: number; txg_timeout_sec: number + pools: PoolWrites[] +} +export interface ARCView { detail: ARCDetail; live: ARCLive | null; writes: ZFSWrites | null } export interface FileEntry { name: string; type: 'dir' | 'file' | 'link' | 'other'; size: number; mode: string; perm: string; mtime: number; user: string; group: string; target?: string; broken?: boolean; preview?: 'image' | 'video' | 'audio' } export interface PlayTrack { codec: string; profile?: string; pix_fmt?: string; width?: number; height?: number; fps?: number; channels?: number; spherical?: boolean } diff --git a/web/src/styles.css b/web/src/styles.css index 680e586..c9351d2 100644 --- a/web/src/styles.css +++ b/web/src/styles.css @@ -884,6 +884,8 @@ pre.diff .add { color: var(--ok); background: color-mix(in srgb, var(--ok) 12%, .arc-tier { display: flex; gap: 8px; min-width: 0; } .arc-tier .k { color: var(--muted); width: 80px; flex-shrink: 0; cursor: help; } .arc-tier-v { display: flex; flex-direction: column; gap: 1px; min-width: 0; } .arc-tier-v > span { overflow: hidden; text-overflow: ellipsis; white-space: nowrap; } .arc-tier-v .none { color: var(--muted); white-space: normal; } +.landing-tier { border: 1px solid var(--border); border-radius: 8px; padding: 8px 10px; margin: 4px 0 6px; font-size: 13px; } +.landing-tier .props .badge { margin-right: 2px; } /* ---- Space: what is using a disk ---- */ .space { position: absolute; inset: 0; display: flex; flex-direction: column; background: var(--bg); min-height: 0; }