Z14 zfs-follower: the follower starts the apply on the events that end a wait and keeps its findings in its own state

Builds: design 42 zfs-owns-zfs §18, §27, A-10, A-16, A-32

modules/zfs/cmd/zfs-tools/bus.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 	"time"
 
+	"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
 	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/buscalls"
 	"git.novox.be/novox/mesh-sdk/go/process"
@@ -48,6 +49,7 @@ func (m *Machine) JoinBus() func() {
 	}
 	m.Emit = func(kind string, body any) error { return c.Emit(kind, body, stdio.EmitOptions{}) }
 	m.States = busStates{c}
+	m.Findings = meshconditionsclient.FindingWriter{State: c.State(StateFindings)}
 	m.Call = c.Call
 	return c.Close
 }

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

@@ -58,6 +58,8 @@ type Cursor struct {
 	EID  uint64    `json:"eid"`
 	Time time.Time `json:"time"`
 	path string
+	// OnEvent hears every event not read before, after the cursor passed it.
+	OnEvent func(Event) `json:"-"`
 }
 
 // LoadCursor is the cursor kept for this boot; an empty one for another boot or none kept.
@@ -149,6 +151,9 @@ func Follow(r io.Reader, out io.Writer, cur *Cursor) error {
 				cur.Time = at
 			}
 			_ = cur.Save(cur.path)
+			if cur.OnEvent != nil {
+				cur.OnEvent(e)
+			}
 		}
 		block = block[:0]
 	}
@@ -260,6 +265,25 @@ func StartZpoolFollow(ctx context.Context) (FollowProcess, error) {
 func Follower(ctx context.Context, out io.Writer) error {
 	boot, _ := os.ReadFile(BootID)
 	cur := LoadCursor(CursorFile, strings.TrimSpace(string(boot)))
+	m := Here()
+	defer m.JoinBus()()
+	warn := func(what string, err error) {
+		if err != nil {
+			fmt.Fprintf(out, "%s: %v\n", what, err)
+		}
+	}
+	// Once at its own start (A-16), then on the events that end a wait.
+	warn("starting the apply", m.StartApply())
+	warn("reading the findings", m.SyncFindings(ctx))
+	cur.OnEvent = func(e Event) {
+		if EndsAWait(e.Class) {
+			warn("starting the apply", m.StartApply())
+		}
+		if ChangesAFinding(e.Class) {
+			warn("reading the findings", m.SyncFindings(ctx))
+		}
+	}
+	go func() { warn("following udev", m.StartUdev(ctx)) }()
 	return RunFollower(ctx, out, StartZpoolFollow, cur, CursorFile, func(d time.Duration) {
 		select {
 		case <-ctx.Done():

modules/zfs/cmd/zfs-tools/follow_wake.go

@@ -0,0 +1,198 @@
+package main
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"io"
+	"os/exec"
+	"path/filepath"
+	"sort"
+	"strings"
+	"syscall"
+	"time"
+
+	"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
+	"git.novox.be/novox/mesh-sdk/go/views"
+)
+
+// What the event follower does besides the journal (design 42 zfs-owns-zfs §18, §27; A-10, A-16).
+//
+// **A wait is ended by an event.** The follower, which runs from boot, starts the apply unit once at its own
+// start and when it hears a scrub or resilver finish (`scrub_finish`, `resilver_finish`) or a declared device
+// appear (a udev `add` of a block device whose link under /dev/disk/by-id is declared), through
+// `<node>/node-service-manager.start`. It also keeps the findings: one key `pool.<pool>.<device>` per device of
+// the pool, `pool.<pool>.pool` for the pool, not ONLINE, its value the state, deleted when the device is ONLINE
+// again or leaves the pool.
+
+// ApplyUnit is the unit of the apply process (`zfs-config-apply`).
+const ApplyUnit = "zfs-config-apply.service"
+
+// UdevArgv follows the kernel's block device events.
+func UdevArgv() []string { return []string{"monitor", "--udev", "--subsystem-match=block"} }
+
+// StartApply starts the apply unit through the service manager.
+func (m *Machine) StartApply() error {
+	if m.Call == nil {
+		return fmt.Errorf("this process reaches no seat, so the apply was not started")
+	}
+	_, err := m.Call("seat:node-service-manager.start@"+m.Node, map[string]any{"scope": "system", "unit": ApplyUnit})
+	return err
+}
+
+// EndsAWait says whether an event of ZFS's ends a wait of the apply.
+func EndsAWait(class string) bool {
+	return strings.HasSuffix(class, ".scrub_finish") || strings.HasSuffix(class, ".resilver_finish")
+}
+
+// ChangesAFinding says whether an event of ZFS's may change a device's state.
+func ChangesAFinding(class string) bool {
+	return strings.HasSuffix(class, ".statechange") || strings.HasPrefix(class, "resource.fs.zfs.") ||
+		strings.HasSuffix(class, ".config_sync") || EndsAWait(class)
+}
+
+// DeclaredAdded says whether a line of `udevadm monitor` is the add of a block device a declared name links to.
+// The declared names are read from the config file at each add, the links resolved now.
+func (m *Machine) DeclaredAdded(line string) bool {
+	f := strings.Fields(line)
+	if len(f) < 4 || f[0] != "UDEV" || f[2] != "add" {
+		return false
+	}
+	added := filepath.Base(f[3])
+	text, err := m.ReadFile(ConfigFile)
+	if err != nil {
+		return false
+	}
+	c, refused := ParseConfig(text, m.memMiB())
+	if len(refused) > 0 {
+		return false
+	}
+	for _, d := range append(m.resolveDeclared(c.CacheDevices, "cache"), m.resolveDeclared(c.LogDevices, "log")...) {
+		if d.Present && partOf(added, filepath.Base(d.Real)) {
+			return true
+		}
+	}
+	return false
+}
+
+// FollowUdev reads `udevadm monitor` and starts the apply on the add of a declared device.
+func (m *Machine) FollowUdev(r io.Reader) error {
+	sc := bufio.NewScanner(r)
+	for sc.Scan() {
+		if m.DeclaredAdded(sc.Text()) {
+			if err := m.StartApply(); err != nil {
+				return err
+			}
+		}
+	}
+	return sc.Err()
+}
+
+// StartUdev starts `udevadm monitor` and follows it until the context ends.
+func (m *Machine) StartUdev(ctx context.Context) error {
+	cmd := exec.Command("udevadm", UdevArgv()...)
+	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+	out, err := cmd.StdoutPipe()
+	if err != nil {
+		return err
+	}
+	if err := cmd.Start(); err != nil {
+		return err
+	}
+	go func() {
+		<-ctx.Done()
+		_ = syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
+	}()
+	ferr := m.FollowUdev(out)
+	_ = cmd.Wait()
+	return ferr
+}
+
+// FindingsOf are the findings `zpool status -P` shows now: key to state, for a pool or a device not ONLINE.
+func FindingsOf(out string) (map[string]string, error) {
+	ps, err := ParseStatus(out)
+	if err != nil {
+		return nil, err
+	}
+	found := map[string]string{}
+	var walk func(pool string, d *Device)
+	walk = func(pool string, d *Device) {
+		for _, c := range d.Children {
+			if c.Kind != "section" && c.State != "" && c.State != "ONLINE" && c.State != "AVAIL" && c.State != "INUSE" {
+				found["pool."+pool+"."+c.Name] = c.State
+			}
+			walk(pool, c)
+		}
+	}
+	for _, p := range ps {
+		if p.State != "ONLINE" {
+			found["pool."+p.Name+".pool"] = p.State
+		}
+		if p.Root != nil {
+			walk(p.Name, p.Root)
+		}
+	}
+	return found, nil
+}
+
+// FindingsObserver is what the findings are written as: the module is their observer, as every observer on the mesh
+// writes its own state `findings` with mesh-sdk's views.Finding and the seat mesh-conditions' writer (A-32).
+const FindingsObserver = "zfs"
+
+// findingOf is the finding of one device or pool not ONLINE: the module is its subject, the pool and device its
+// kind, `pool.<pool>.<device>`, so each is a key of its own and goes when the device is ONLINE again or gone.
+func findingOf(kind, state string, at time.Time) views.Finding {
+	name := strings.TrimPrefix(kind, "pool.")
+	return views.Finding{Observer: FindingsObserver, Kind: kind, ID: FindingsObserver,
+		Summary:     fmt.Sprintf("%s is %s", name, state),
+		Headline:    fmt.Sprintf("ZFS %s is %s", name, state),
+		Explanation: "ZFS reports this pool or device not ONLINE; zfs_pool_layout and zfs_pools say more. The finding clears when it is ONLINE again or leaves the pool.",
+		At:          at}
+}
+
+// SyncFindings reads `zpool status -P` against the keys of the state `findings`, writes what is new or changed,
+// and clears what is gone, each through the writer.
+func (m *Machine) SyncFindings(ctx context.Context) error {
+	if m.States == nil || m.Findings == nil {
+		return fmt.Errorf("this process keeps no state, so the findings were not read")
+	}
+	out, err := m.ask(ctx, "zpool", StatusPathsArgv()...)
+	if err != nil {
+		return err
+	}
+	now, err := FindingsOf(out)
+	if err != nil {
+		return err
+	}
+	keys, err := m.States.Keys(StateFindings)
+	if err != nil {
+		return err
+	}
+	prefix := strings.TrimSuffix(meshconditionsclient.Key(meshconditionsclient.ScopeModule, FindingsObserver, "x"), "x")
+	wanted := map[string]bool{}
+	for _, kind := range sortedKeys(now) {
+		f := findingOf(kind, now[kind], m.Now())
+		key := meshconditionsclient.Key(meshconditionsclient.ScopeModule, FindingsObserver, kind)
+		wanted[key] = true
+		if raw, ok, _ := m.States.Get(StateFindings, key); ok && strings.Contains(string(raw), `"summary":"`+f.Summary+`"`) {
+			continue
+		}
+		if err := m.Findings.Put(f); err != nil {
+			return err
+		}
+	}
+	sort.Strings(keys)
+	for _, k := range keys {
+		kind, mine := strings.CutPrefix(k, prefix)
+		if !mine || wanted[k] {
+			continue
+		}
+		if err := m.Findings.Delete(views.Finding{ID: FindingsObserver, Kind: kind}); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// StatusPathsArgv reads every pool's status with the paths its devices are held by.
+func StatusPathsArgv() []string { return []string{"status", "-P"} }

modules/zfs/cmd/zfs-tools/follow_wake_test.go

@@ -0,0 +1,93 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+
+	"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
+	"git.novox.be/novox/mesh-sdk/go/views"
+)
+
+type findingWriter struct {
+	st   memStates
+	puts []string
+	dels []string
+}
+
+func (w *findingWriter) Put(f views.Finding) error {
+	key := meshconditionsclient.Key(meshconditionsclient.ScopeModule, f.ID, f.Kind)
+	w.puts = append(w.puts, key)
+	return w.st.Put(StateFindings, key, map[string]string{"summary": f.Summary})
+}
+func (w *findingWriter) Delete(f views.Finding) error {
+	key := meshconditionsclient.Key(meshconditionsclient.ScopeModule, f.ID, f.Kind)
+	w.dels = append(w.dels, key)
+	return w.st.Delete(StateFindings, key)
+}
+
+// §18, §27, A-16: the apply is started once at the follower's start and on scrub_finish, resilver_finish and a
+// declared device's udev add, through the service manager; a device not ONLINE is a finding, written once,
+// deleted when ONLINE again or gone.
+func TestTheFollowerStartsTheApplyOnTheEventsThatEndAWaitAndKeepsTheFindings(t *testing.T) {
+	for class, want := range map[string]bool{"sysevent.fs.zfs.scrub_finish": true, "sysevent.fs.zfs.resilver_finish": true,
+		"sysevent.fs.zfs.scrub_start": false, "ereport.fs.zfs.io": false} {
+		if EndsAWait(class) != want {
+			t.Errorf("%s: %v", class, !want)
+		}
+	}
+	f := newFake(t)
+	f.uid = 0
+	f.files[ConfigFile] = `{"pool":"tank","cache-devices":["c1"]}`
+	f.files[MemInfoFile] = "MemTotal:       16384000 kB\n"
+	f.links["/dev/disk/by-id/c1"] = "/dev/nvme1n1"
+	m := f.machine()
+	var key string
+	var body any
+	m.Node = "ace"
+	m.Call = func(k string, b any) (json.RawMessage, error) { key, body = k, b; return nil, nil }
+	udev := map[string]bool{
+		"UDEV  [12.3] add      /devices/pci0000:00/nvme/nvme1/nvme1n1 (block)":           true,
+		"UDEV  [12.3] add      /devices/pci0000:00/nvme/nvme1/nvme1n1/nvme1n1p1 (block)": true,
+		"UDEV  [12.3] add      /devices/pci0000:00/nvme/nvme2/nvme2n1 (block)":           false,
+		"UDEV  [12.3] remove   /devices/pci0000:00/nvme/nvme1/nvme1n1 (block)":           false,
+		"KERNEL[12.3] add      /devices/pci0000:00/nvme/nvme1/nvme1n1 (block)":           false,
+	}
+	for line, want := range udev {
+		if m.DeclaredAdded(line) != want {
+			t.Errorf("%q: %v", line, !want)
+		}
+	}
+	if err := m.FollowUdev(strings.NewReader("UDEV  [1] add /devices/x/nvme2n1 (block)\nUDEV  [2] add /devices/x/nvme1n1 (block)\n")); err != nil {
+		t.Fatal(err)
+	}
+	args, _ := body.(map[string]any)
+	if key != "seat:node-service-manager.start@ace" || args["unit"] != ApplyUnit || args["scope"] != "system" {
+		t.Errorf("called %s with %v", key, body)
+	}
+
+	// The findings: a degraded pool, then healthy again.
+	st := memStates{}
+	w := &findingWriter{st: st}
+	m.States, m.Findings = st, w
+	f.on(line("zpool", StatusPathsArgv()), fixture(t, "zpool-status-degraded.txt"))
+	if err := m.SyncFindings(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	for _, k := range []string{"pool_tank_pool", "pool_tank_raidz2-0", "pool_tank_ata-WDC_WD120EFBX-68B0EN0_D7JA0002"} {
+		if _, ok := st["findings/module.zfs."+k]; !ok {
+			t.Errorf("no finding %s in %v", k, st)
+		}
+	}
+	if len(w.puts) != 4 {
+		t.Errorf("puts %v", w.puts)
+	}
+	if err := m.SyncFindings(context.Background()); err != nil || len(w.puts) != 4 {
+		t.Errorf("an unchanged finding was written again: %v %v", err, w.puts)
+	}
+	f.on(line("zpool", StatusPathsArgv()), fixture(t, "zpool-status.txt"))
+	if err := m.SyncFindings(context.Background()); err != nil || len(w.dels) != 4 || len(st) != 0 {
+		t.Errorf("deleted %v, left %v, %v", w.dels, st, err)
+	}
+}

modules/zfs/cmd/zfs-tools/machine.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"git.novox.be/novox/mesh-sdk/go/views"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -127,6 +128,8 @@ type Machine struct {
 	// Await waits at most the bound for an event of the module whose kind is one of kinds, and answers it; false
 	// when none came.
 	Await func(kinds []string, bound time.Duration) (map[string]any, bool)
+	// Findings writes the module's own state `findings` as the seat mesh-conditions' layout says (A-32).
+	Findings views.FindingWriter
 	// Warnings are what the apply saw that is no failure (a change the bus did not take, a parameter awaiting
 	// a load).
 	Warnings []string

modules/zfs/cmd/zfs-tools/manifest_test.go

@@ -228,7 +228,7 @@ func TestTheBundleRunsOnlyReadingVerbsSnapshotAndScrub(t *testing.T) {
 		"DatasetSetArgv": {"zfs", "set"}, "DatasetInheritArgv": {"zfs", "inherit"},
 		"DeviceStatusArgv": {"zpool", "status"}, "AllRealStatusArgv": {"zpool", "status"}, "RemoveArgv": {"zpool", "remove"},
 		"AddArgv": {"zpool", "add"}, "WipefsArgv": {"wipefs", "--no-act"}, "ZdbLabelArgv": {"zdb", "-l"},
-		"SnapshotSetArgv": {"zfs", "set"}, "HoldsArgv": {"zfs", "holds"}, "SnapshotPropertyArgv": {"zfs", "get"}, "DestroyArgv": {"zfs", "destroy"}, "ZdbConfigArgv": {"zdb", "-C"}, "ZgenhostidArgv": {"zgenhostid", "-f"},
+		"UdevArgv": {"udevadm", "monitor"}, "SnapshotSetArgv": {"zfs", "set"}, "StatusPathsArgv": {"zpool", "status"}, "HoldsArgv": {"zfs", "holds"}, "SnapshotPropertyArgv": {"zfs", "get"}, "DestroyArgv": {"zfs", "destroy"}, "ZdbConfigArgv": {"zdb", "-C"}, "ZgenhostidArgv": {"zgenhostid", "-f"},
 	}
 	verbs := map[string][]string{"DatasetsArgv": DatasetsArgv(), "SnapshotsArgv": SnapshotsArgv(), "PoolsArgv": PoolsArgv(),
 		"StatusArgv": StatusArgv(), "IOArgv": IOArgv(3), "EventsArgv": EventsArgv(), "HistoryArgv": HistoryArgv("tank"),
@@ -241,6 +241,7 @@ func TestTheBundleRunsOnlyReadingVerbsSnapshotAndScrub(t *testing.T) {
 		"RemoveArgv": RemoveArgv("tank", "x"), "AddArgv": AddArgv("tank", "cache", false, "/dev/x"),
 		"WipefsArgv": WipefsArgv("/dev/x"), "ZdbLabelArgv": ZdbLabelArgv("/dev/x"),
 		"HoldsArgv": HoldsArgv("tank@x"), "SnapshotPropertyArgv": SnapshotPropertyArgv("tank@x", "clones"), "DestroyArgv": DestroyArgv("tank", "x"),
+		"StatusPathsArgv": StatusPathsArgv(), "UdevArgv": UdevArgv(),
 		"SnapshotSetArgv": SnapshotSetArgv("tank@mesh-hourly-x", "hourly"), "ZdbConfigArgv": ZdbConfigArgv("tank"), "ZgenhostidArgv": ZgenhostidArgv("00bab10c")}
 	for fn, argv := range verbs {
 		for _, a := range argv {
@@ -262,7 +263,7 @@ func TestTheBundleRunsOnlyReadingVerbsSnapshotAndScrub(t *testing.T) {
 		}
 	}
 	call := regexp.MustCompile(`(\.(ask|AsRoot|Escalated|Run|Start|Output|CombinedOutput)\(|exec\.Command(Context)?\(|os\.StartProcess\(|syscall\.(Exec|ForkExec)\()`)
-	site := regexp.MustCompile(`(?:\.(?:ask|AsRoot|Escalated)|exec\.Command)\((?:\w+, )?"(zfs|zpool|systemctl|wipefs|zdb|zgenhostid)", (\w+)\(`)
+	site := regexp.MustCompile(`(?:\.(?:ask|AsRoot|Escalated)|exec\.Command)\((?:\w+, )?"(zfs|zpool|systemctl|wipefs|zdb|zgenhostid|udevadm)", (\w+)\(`)
 	// The bodies that run what they are given, each once.
 	bodies := map[string]bool{
 		"machine.go:	r := m.Run(ctx, name, args...)":              true,
@@ -271,6 +272,7 @@ func TestTheBundleRunsOnlyReadingVerbsSnapshotAndScrub(t *testing.T) {
 		"machine.go:	cmd := exec.Command(name, args...)":          true,
 		"machine.go:	if err := cmd.Start(); err != nil {":         true,
 		"follow.go:	if err := cmd.Start(); err != nil {":          true,
+		"follow_wake.go:	if err := cmd.Start(); err != nil {":     true,
 	}
 	files, _ := filepath.Glob("*.go")
 	sites := 0

modules/zfs/go.mod

@@ -4,6 +4,11 @@ go 1.26.0
 
 require git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000
 
+require (
+	git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000
+	git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 // indirect
+)
+
 require (
 	github.com/klauspost/compress v1.20.1 // indirect
 	github.com/nats-io/nats.go v1.54.0 // indirect