From dd1a4e083c52692c8f9adeff603614aa5e8d12c7 Mon Sep 17 00:00:00 2001 From: jochens Date: Fri, 13 Nov 2026 00:49:01 +0100 Subject: [PATCH] Z14 zfs-follower: the follower starts the apply on the events that end a wait and keeps its findings in its own state MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 42 zfs-owns-zfs §18, §27, A-10, A-16, A-32 --- modules/zfs/cmd/zfs-tools/bus.go | 2 + modules/zfs/cmd/zfs-tools/follow.go | 24 +++ modules/zfs/cmd/zfs-tools/follow_wake.go | 198 ++++++++++++++++++ modules/zfs/cmd/zfs-tools/follow_wake_test.go | 93 ++++++++ modules/zfs/cmd/zfs-tools/machine.go | 3 + modules/zfs/cmd/zfs-tools/manifest_test.go | 6 +- modules/zfs/go.mod | 5 + 7 files changed, 329 insertions(+), 2 deletions(-) create mode 100644 modules/zfs/cmd/zfs-tools/follow_wake.go create mode 100644 modules/zfs/cmd/zfs-tools/follow_wake_test.go diff --git a/modules/zfs/cmd/zfs-tools/bus.go b/modules/zfs/cmd/zfs-tools/bus.go index 41ec813b3..c846628ba 100644 --- a/modules/zfs/cmd/zfs-tools/bus.go +++ b/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 } diff --git a/modules/zfs/cmd/zfs-tools/follow.go b/modules/zfs/cmd/zfs-tools/follow.go index 495316ee4..0eea6d2e6 100644 --- a/modules/zfs/cmd/zfs-tools/follow.go +++ b/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(): diff --git a/modules/zfs/cmd/zfs-tools/follow_wake.go b/modules/zfs/cmd/zfs-tools/follow_wake.go new file mode 100644 index 000000000..51a8529ca --- /dev/null +++ b/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-service-manager.start`. It also keeps the findings: one key `pool..` per device of +// the 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..`, 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"} } diff --git a/modules/zfs/cmd/zfs-tools/follow_wake_test.go b/modules/zfs/cmd/zfs-tools/follow_wake_test.go new file mode 100644 index 000000000..ac65229fb --- /dev/null +++ b/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) + } +} diff --git a/modules/zfs/cmd/zfs-tools/machine.go b/modules/zfs/cmd/zfs-tools/machine.go index 669aa5039..ee817c47d 100644 --- a/modules/zfs/cmd/zfs-tools/machine.go +++ b/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 diff --git a/modules/zfs/cmd/zfs-tools/manifest_test.go b/modules/zfs/cmd/zfs-tools/manifest_test.go index 35e20bf7e..c7de1e430 100644 --- a/modules/zfs/cmd/zfs-tools/manifest_test.go +++ b/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 diff --git a/modules/zfs/go.mod b/modules/zfs/go.mod index 7b7046b1b..804a627cf 100644 --- a/modules/zfs/go.mod +++ b/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