Watchman: raise machine-network, network-rewritten, network-unreachable, machine-units, relogin-needed and directory-used-as-found from each node's own health report, under the controller's keys, so these alerts outlive the controller's lease (issue 774)

Pins mesh-sdk 94d0de5.
This commit is contained in:
2026-10-29 23:01:03 +01:00
parent 64a5074d58
commit 8a91871de6
10 changed files with 457 additions and 7 deletions
@@ -321,6 +321,25 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}()
})
}
// Each node's own health report (the node-engine's state health): read before the findings are reconciled when the
// bucket is there, bound once it appears when it is not; each change is heard on the watch, never polled.
healthEach := func(e jetstream.KeyValueEntry) { w.NodeHealth(e.Key(), e.Value(), opOf(e)) }
var healthWatch jetstream.KeyWatcher
if kv, err := js.KeyValue(ctx, delivery.NodeHealthBucket); err == nil {
if healthWatch, err = settle(ctx, kv, healthEach); err != nil {
process.NotWatched(logger.Printf, delivery.NodeHealthBucket, err, "no node's own health is said until the watchman starts again")
}
} else {
process.AwaitBucket(ctx, js, conn, delivery.NodeHealthBucket, "no node's own health is said",
logger.Printf).Ready(func(kv jetstream.KeyValue) {
hw, err := settle(ctx, kv, healthEach)
if err != nil {
process.NotWatched(logger.Printf, delivery.NodeHealthBucket, err, "no node's own health is said until the watchman starts again")
return
}
go func() { _ = follow(ctx, hw, healthEach) }()
})
}
dead := &deadKeeper{ctx: ctx, js: js, w: w, logf: logger.Printf}
w.SetRecount(dead.count)
dead.count()
@@ -332,7 +351,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return err
}
errs := make(chan error, 8)
errs := make(chan error, 16)
if healthWatch != nil {
go func() { errs <- follow(ctx, healthWatch, healthEach) }()
}
if provWatch != nil {
go func() {
errs <- follow(ctx, provWatch, func(e jetstream.KeyValueEntry) { w.Provision(e.Key(), e.Value(), opOf(e)) })
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-watchman
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029213215-4a967933b368
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029215805-94d0de5c19b6
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
)
+2 -2
View File
@@ -1,5 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029213215-4a967933b368 h1:LaBxsyU3HeTIPKYmvpJuw+Wm2plpf6O7qHO91O+mAi0=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029213215-4a967933b368/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029215805-94d0de5c19b6 h1:eQIsSB3Z72MNjirW3TbKbvFgkhwjCAgOAqWd0U4ldCE=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029215805-94d0de5c19b6/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op h1:1BOWQJweNyvZMlpAHXGLiZQn9S+QXGcz3xh94lC0w6E=
github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
@@ -173,6 +173,59 @@ func TestTheWatchmansFindings(t *testing.T) {
w.Provision("ace.web.oidc", enc(provisions.Waits, "x"), Put)
w.Provision("ace.web.oidc", nil, Delete)
}, map[string]string{}},
{"774: a node's own report: its network, the resolver file rewritten, failed units, a relogin, a directory used as found", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
report := func(h delivery.NodeHealth) []byte {
b, err := delivery.EncodeNodeHealth(h)
if err != nil {
t.Fatal(err)
}
return b
}
w.Have("novox.mesh-controller", haveOf(t, delivery.PhaseRunning, delivery.HealthOK, ""), Put) // the control node
w.NodeHealth("novox", report(delivery.NodeHealth{Node: "novox", Network: &delivery.NetworkHealth{State: "unhealthy", Parts: []delivery.NetworkPart{
{Part: "resolv-conf", State: "unhealthy", Writer: "NetworkManager", Owner: "systemd-resolved", Said: "rewritten"},
{Part: "names", State: "unhealthy", Reason: "names do not resolve.", Said: "x"},
{Part: "route", State: "unhealthy", Reason: "no default route", Said: "y"}}},
Units: &delivery.UnitsHealth{State: "degraded", Failed: []delivery.FailedUnit{{Unit: "cups.service", Scope: "system", Result: "exit-code"}}},
Relogin: []delivery.NodeAccount{{Account: "jochen", Module: "docker", Reason: "a group"}},
Found: []delivery.NodeDirectory{{Module: "photos", Resource: "data", Path: "/srv/photos", Reason: "owner differs"}}}), Put)
for key, sev := range map[string]string{"machine.novox.network": Urgent, "machine.novox.systemd-resolved.rewritten": Urgent,
"machine.novox.units": Warning, "module.docker.novox.relogin-needed": Warning, "module.photos.novox.directory-used-as-found": Warning} {
if o.findings[key].Severity != sev {
t.Errorf("%s = %+v", key, o.findings[key])
}
}
if f := o.findings["machine.novox.network"]; strings.Contains(f.Summary, "names") {
t.Errorf("the names failing through the rewritten file are said twice: %s", f.Summary)
}
w.NodeHealth("novox", report(delivery.NodeHealth{Node: "novox"}), Put) // all well again
}, map[string]string{}},
{"774: a node two nodes' tunnels no longer see is unreachable, said once there; not while it is silent", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
for node, key := range map[string]string{"ace": "kA", "g14": "kG", "lap": "kL"} {
b, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: node, TunnelKey: key}})
w.Have(node+".node", b, Put)
}
peers := func(from string, seen bool) []byte {
b, _ := delivery.EncodeNodeHealth(delivery.NodeHealth{Node: from, Peers: []delivery.PeerSeen{{Key: "kL", Seen: seen}}})
return b
}
w.NodeHealth("ace", peers("ace", false), Put)
if len(o.findings) != 0 {
t.Fatalf("one node alone said it: %v", o.findings)
}
w.NodeHealth("g14", peers("g14", false), Put)
if f := o.findings["machine.lap.unreachable"]; len(f.Also) != 2 {
t.Fatalf("findings = %v", o.findings)
}
w.Connect("node.lap")
w.Disconnect("node.lap")
expire(w, c, pulseKeyOf(t, presenceNS, "lap"), 120*time.Second)
if _, ok := o.findings["machine.lap.unreachable"]; ok {
t.Error("a silent node is also said unreachable")
}
w.Connect("node.lap")
w.NodeHealth("g14", peers("g14", true), Put)
}, map[string]string{}},
{"WM3: a have removed ends its findings", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthFailed, "x"), Put)
w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthFailed, "x"), Put)
@@ -106,11 +106,22 @@ func (w *Watchman) Want(key string, value []byte, op Op, at time.Time) {
// fingerprint of the node's want for the module ends the want's pulse. A node's own key and a key deleted are left.
func (w *Watchman) Have(key string, value []byte, op Op) {
node, module, isNode, err := delivery.ParseHaveKey(key)
if err != nil || isNode {
if err != nil {
return
}
w.mu.Lock()
defer w.mu.Unlock()
if isNode {
// The node's own record: its tunnel key, whose peers' sight of it is network-unreachable.
var record *delivery.NodeRecord
if op == Put {
if hn, err := delivery.DecodeHaveNode(value); err == nil {
record = hn.Record
}
}
w.tunnelKey(node, record)
return
}
switch op {
case Expired:
return
@@ -0,0 +1,358 @@
package watch
import (
"fmt"
"maps"
"slices"
"sort"
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// A node's own health report (novox/hq ADR 0241, issue 315, the operator's alert table of 2026-10-29): each node's engine
// writes `mesh-host_health.<node>` when what it says changes, and the watchman raises from it what the controller raised
// from the passing health message, under the same keys and words (mesh-control cmd/mesh-controller machine_network.go,
// machine_units.go, module_health.go at e04846e9):
// - machine-network `machine.<node>.network`: the machine's own unhealthy parts; urgent on the control node or for the bus;
// - network-rewritten `machine.<node>[.<owner>].rewritten`: the resolver file rewritten by another program;
// - network-unreachable `machine.<node>.unreachable`: two or more nodes' tunnels no longer see it (their peers' handshake),
// said once there, never while its own network or its silence says it;
// - machine-units `machine.<node>.units`: failed units no module places;
// - relogin-needed `module.<module>.<node>.relogin-needed` and directory-used-as-found
// `module.<module>.<node>.directory-used-as-found`.
//
// What the controller held at another machine by the addresses its parts point at (Toward) needs each node's private
// address and the hub, which no report carries: a part pointing elsewhere is the machine's own here, and the machine all
// point at is said by its peers instead.
// The kinds a node's health report raises.
const (
KindMachineNetwork = "machine-network"
KindNetworkRewritten = "network-rewritten"
KindNetworkUnreachable = "network-unreachable"
KindMachineUnits = "machine-units"
KindReloginNeeded = "relogin-needed"
KindUsedAsFound = "directory-used-as-found"
)
// The parts and states a report names (mesh-control internal/link).
const (
partResolvConf = "resolv-conf"
partNames = "names"
partTunnel = "tunnel"
partBus = "bus"
stateUnhealthy = "unhealthy"
)
func stampSec(t time.Time) string { return t.UTC().Format("2006-01-02 15:04:05 MST") }
// NodeHealth is a change of a node's report: written, or deleted (value nil).
func (w *Watchman) NodeHealth(node string, value []byte, op Op) {
w.mu.Lock()
defer w.mu.Unlock()
if op == Put {
h, err := delivery.DecodeNodeHealth(value)
if err != nil {
w.say("the health report of %s cannot be read, and is left: %v", node, err)
return
}
w.health[node] = h
} else {
delete(w.health, node)
}
w.nodeFindings(node)
w.unreachable()
}
// TunnelKey is a node's own tunnel key, from its record in its have: whose peers see it is read by it.
func (w *Watchman) tunnelKey(node string, record *delivery.NodeRecord) {
key := ""
if record != nil {
key = record.TunnelKey
}
if w.tunnels[node] == key {
return
}
if key == "" {
delete(w.tunnels, node)
} else {
w.tunnels[node] = key
}
w.unreachable()
}
// nodeFindings writes and deletes one node's own findings from its report.
func (w *Watchman) nodeFindings(node string) {
h, ok := w.health[node]
want := map[string]bool{}
if ok {
for _, f := range w.networkOf(node, h) {
want[f.Key()] = true
w.find(f)
}
if f, raise := unitsOf(node, h.Units); raise {
want[f.Key()] = true
w.find(f)
}
for _, f := range reloginOf(node, h.Relogin) {
want[f.Key()] = true
w.find(f)
}
for _, f := range foundOf(node, h.Found) {
want[f.Key()] = true
w.find(f)
}
}
for key := range w.findings {
if !want[key] && w.ownsHealthKey(key, node) {
w.unfind(key)
}
}
}
// ownsHealthKey says a finding key is one of node's own report findings.
func (w *Watchman) ownsHealthKey(key, node string) bool {
f, err := DecodeFinding(w.findings[key])
if err != nil || f.Machine != node || f.Source != "health-report" {
return false
}
return f.Kind != KindNetworkUnreachable
}
// networkOf is a node's own network findings: the resolver file rewritten, and the rest of its unhealthy parts.
func (w *Watchman) networkOf(m string, h delivery.NodeHealth) []Finding {
if h.Network == nil {
return nil
}
var parts []delivery.NetworkPart
for _, p := range h.Network.Parts {
if p.State == stateUnhealthy {
parts = append(parts, p)
}
}
var out []Finding
var own []delivery.NetworkPart
var rewritten *delivery.NetworkPart
for i, p := range parts {
if p.Part == partResolvConf {
rewritten = &parts[i]
continue
}
own = append(own, p)
}
if rewritten != nil {
out = append(out, w.rewrittenOf(m, *rewritten, parts))
kept := own[:0]
for _, p := range own {
if p.Part != partNames {
kept = append(kept, p)
}
}
own = kept
}
if len(own) > 0 {
var words, said []string
severity := Warning
for _, p := range own {
words = append(words, p.Reason)
said = append(said, fmt.Sprintf("%s since %s: %s", p.Part, stampSec(p.Since), p.Said))
if p.Part == partBus {
severity = Urgent
}
}
if w.control[m] {
severity = Urgent
}
out = append(out, Finding{Scope: ScopeMachine, ID: m, Token: "network", Kind: KindMachineNetwork, Machine: m,
Severity: severity, Source: "health-report", Since: h.Network.Since,
Summary: fmt.Sprintf("%s's network is not healthy: %s", m, strings.Join(words, "; ")), Said: strings.Join(said, " | ")})
}
return out
}
func (w *Watchman) rewrittenOf(m string, p delivery.NetworkPart, all []delivery.NetworkPart) Finding {
writer := ""
if p.Writer != "" {
writer = " (" + p.Writer + ")"
}
cost := "the names still resolve through what it wrote"
said := []string{p.Part + ": " + p.Said}
for _, q := range all {
if q.Part == partNames {
cost = strings.TrimSuffix(q.Reason, ".")
said = append(said, q.Part+": "+q.Said)
}
}
id := m
summary := fmt.Sprintf("the resolver file on %s was rewritten by another program%s — %s until the node-engine writes it "+
"back at its next reconcile, or that program gives it back", m, writer, cost)
if p.Owner != "" {
id = m + "." + p.Owner
summary = fmt.Sprintf("the resolver file %s writes on %s was rewritten by another program%s — %s until the node-engine "+
"writes it back at its next reconcile, or that program gives it back", p.Owner, m, writer, cost)
}
severity := Warning
if w.control[m] {
severity = Urgent
}
return Finding{Scope: ScopeMachine, ID: id, Token: "rewritten", Kind: KindNetworkRewritten, Machine: m, Severity: severity,
Source: "health-report", Since: p.Since, Summary: summary,
Said: fmt.Sprintf("since %s: %s", stampSec(p.Since), strings.Join(said, " | "))}
}
// unitsOf is a machine's failed units no module places, one finding.
func unitsOf(node string, u *delivery.UnitsHealth) (Finding, bool) {
if u == nil || len(u.Failed) == 0 {
return Finding{}, false
}
var names, said []string
since := u.Failed[0].Since
for _, f := range u.Failed {
where := ""
if f.Scope == "user" {
where = " (the account's own manager)"
}
names = append(names, f.Unit+where)
scope, result := f.Scope, f.Result
if scope == "" {
scope = "system"
}
if result == "" {
result = "no reason said"
}
line := fmt.Sprintf("%s in the %s manager: %s", f.Unit, scope, result)
if f.Load != "" && f.Load != "loaded" {
line += ", its unit " + f.Load
}
if f.Resource != "" {
line += ", named by the mesh's own " + f.Resource
}
said = append(said, line+", since "+stampSec(f.Since))
if f.Since.Before(since) {
since = f.Since
}
}
failed := "1 failed service"
if len(u.Failed) > 1 {
failed = fmt.Sprintf("%d failed services", len(u.Failed))
}
return Finding{Scope: ScopeMachine, ID: node, Token: "units", Kind: KindMachineUnits, Machine: node, Severity: Warning,
Source: "health-report", Since: since,
Summary: fmt.Sprintf("%s's service manager is degraded: %d failed unit(s) no module places — %s. Each is the machine's own: "+
"mend or remove it there, or have a module place it", node, len(u.Failed), strings.Join(names, ", ")),
Said: strings.Join(said, "; "),
Headline: strings.ToUpper(failed[:1]) + failed[1:] + " on " + node,
Needs: fmt.Sprintf("mend or remove them on %s, or silence this if they do not matter.", node),
Resolved: "No failed services on " + node + " any more"}, true
}
// reloginOf is one finding per module whose accounts wait for a new login.
func reloginOf(node string, accounts []delivery.NodeAccount) []Finding {
by := map[string][]delivery.NodeAccount{}
for _, a := range accounts {
by[a.Module] = append(by[a.Module], a)
}
var out []Finding
for _, module := range slices.Sorted(maps.Keys(by)) {
var names []string
since := by[module][0].Since
for _, a := range by[module] {
names = append(names, a.Account)
if a.Since.Before(since) {
since = a.Since
}
}
sort.Strings(names)
said := fmt.Sprintf("relogin needed on %s: %s waits for a new login of %s, which is in its group and whose login began "+
"before it was; log out of %[1]s completely and log in again, or restart it", node, orModule(module), strings.Join(names, ", "))
out = append(out, Finding{Scope: ScopeModule, ID: orModule(module) + "." + node, Kind: KindReloginNeeded, Machine: node,
Severity: Warning, Source: "health-report", Since: since, Summary: said, Said: said})
}
return out
}
// foundOf is one finding per module with a directory used as found.
func foundOf(node string, dirs []delivery.NodeDirectory) []Finding {
by := map[string][]delivery.NodeDirectory{}
for _, d := range dirs {
by[d.Module] = append(by[d.Module], d)
}
var out []Finding
for _, module := range slices.Sorted(maps.Keys(by)) {
var ids, why []string
since := by[module][0].Since
for _, d := range by[module] {
ids = append(ids, d.Resource)
why = append(why, d.Reason)
if d.Since.Before(since) {
since = d.Since
}
}
m := orModule(module)
said := fmt.Sprintf("%s on %s uses %s as found: %s", m, node, strings.Join(ids, ", "), strings.Join(why, "; "))
out = append(out, Finding{Scope: ScopeModule, ID: m + "." + node, Kind: KindUsedAsFound, Machine: node, Severity: Warning,
Source: "health-report", Since: since, Summary: said, Said: said,
Headline: fmt.Sprintf("%s waits for a directory on %s", m, node),
Explanation: fmt.Sprintf("Needs you: hand the directory over from the control-node, as the operator; the details name it "+
"and the line to type. A directory of %s was already on %s, with another owner or mode than %s declares.", m, node, m),
Needs: "hand the directory over from the control-node, as the operator; the details name it and the line to type.",
Resolved: fmt.Sprintf("%s's directory on %s is the mesh's", m, node)})
}
return out
}
func orModule(m string) string {
if m == "" {
return "_"
}
return m
}
// unreachable is network-unreachable across every report: a node whose tunnel key two or more other nodes' peers list
// as not seen, said once there, unless its own network finding or its silence says it.
func (w *Watchman) unreachable() {
byKey := map[string]string{}
for node, key := range w.tunnels {
byKey[key] = node
}
notSeen := map[string][]string{}
for from, h := range w.health {
for _, p := range h.Peers {
x, ok := byKey[p.Key]
if !ok || x == from || p.Seen {
continue
}
notSeen[x] = append(notSeen[x], from)
}
}
want := map[string]bool{}
for x, from := range notSeen {
if len(from) < 2 {
continue
}
if _, own := w.findings[FindingKey(ScopeMachine, x, "network")]; own {
continue
}
if _, silent := w.findings[FindingKey(ScopeMachine, x, KindSilent)]; silent {
continue
}
sort.Strings(from)
severity := Warning
if w.control[x] {
severity = Urgent
}
f := Finding{Scope: ScopeMachine, ID: x, Token: "unreachable", Kind: KindNetworkUnreachable, Machine: x, Also: from,
Severity: severity, Source: "health-report",
Summary: fmt.Sprintf("%s cannot be reached from %s: the tunnel to it", x, strings.Join(from, ", ")),
Said: fmt.Sprintf("%s: tunnel: no handshake with %s within five minutes", strings.Join(from, ", "), x)}
want[f.Key()] = true
w.find(f)
}
for key := range w.findings {
if strings.HasSuffix(key, ".unreachable") && strings.HasPrefix(key, ScopeMachine+".") && !want[key] {
w.unfind(key)
}
}
}
@@ -240,6 +240,7 @@ func (w *Watchman) stuckFinding(sk string, s delivery.Stuck) {
if s.Bucket == delivery.PresenceBucket {
if node, part, _, err := delivery.ParsePresenceKey(s.Key); err == nil && part == "" && node != "" {
key := FindingKey(ScopeMachine, node, KindSilent)
defer w.unreachable() // a silent machine is said by its silence, never also as unreachable
if s.Moved || w.away[node] {
w.unfind(key)
w.unsignal("S1", node)
@@ -86,6 +86,9 @@ type Watchman struct {
recount func()
// waiting is, by `<consumer>.<node>`, each of its provisions in waits, with its words.
waiting map[string]map[string]string
// health is each node's own health report, and tunnels each node's tunnel key from its record.
health map[string]delivery.NodeHealth
tunnels map[string]string
// stepBound is how long a build step may run once a node took the ask; a test shortens it.
stepBound time.Duration
}
@@ -96,7 +99,8 @@ func New(out Out, now func() time.Time) *Watchman {
pulses: map[string]*pulse{}, stuck: map[string]delivery.Stuck{}, wants: map[string]string{}, haves: map[string]string{}, bound: delivery.PulseMoveBound, stepBound: StepBound,
findings: map[string][]byte{}, seededFindings: map[string]bool{}, signals: map[string]signalDue{}, away: map[string]bool{},
haveState: map[string]delivery.Have{}, unhealthy: map[string]int{}, advised: map[string]int{}, advisedSince: map[string]time.Time{},
control: map[string]bool{}, failedSince: map[string]time.Time{}, provFailed: map[string]map[string]ProvReport{}, closures: map[string]map[string]time.Time{}, waiting: map[string]map[string]string{}}
control: map[string]bool{}, failedSince: map[string]time.Time{}, provFailed: map[string]map[string]ProvReport{}, closures: map[string]map[string]time.Time{}, waiting: map[string]map[string]string{},
health: map[string]delivery.NodeHealth{}, tunnels: map[string]string{}}
}
// Logf sets where a write that failed is said; it is silent until set.
@@ -91,7 +91,7 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
reads := slices.Clone(m.Reads)
slices.Sort(reads)
if !reflect.DeepEqual(reads, []string{"mesh-build-go.steps", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-controller.lease",
"mesh-delivery-assigner.wants", "mesh-provisioner.provisions", "node-engine.have"}) {
"mesh-delivery-assigner.wants", "mesh-host.health", "mesh-provisioner.provisions", "node-engine.have"}) {
t.Errorf("reads = %v", reads)
}
// 1952/WM1: the watchman holds the seat mesh-watchdogs at mesh scope, its first seat (novox/hq ADR 0564 §1, §4).
@@ -21,6 +21,7 @@
"mesh-build-node.steps",
"mesh-controller.lease",
"mesh-delivery-assigner.wants",
"mesh-host.health",
"mesh-provisioner.provisions",
"node-engine.have"
],