diff --git a/modules/mesh-delivery-watchman/go.mod b/modules/mesh-delivery-watchman/go.mod index db64d6f..57dac09 100644 --- a/modules/mesh-delivery-watchman/go.mod +++ b/modules/mesh-delivery-watchman/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-watchman go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029215805-94d0de5c19b6 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220254-6eae0765f4d3 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-watchman/go.sum b/modules/mesh-delivery-watchman/go.sum index d016872..a9c7447 100644 --- a/modules/mesh-delivery-watchman/go.sum +++ b/modules/mesh-delivery-watchman/go.sum @@ -1,5 +1,5 @@ -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= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220254-6eae0765f4d3 h1:4qfwx2bv2vosFM6tEbBZUeaQ/pfRauXZuNI3whul30g= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220254-6eae0765f4d3/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= diff --git a/modules/mesh-delivery-watchman/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go index 4fbf380..03f8cac 100644 --- a/modules/mesh-delivery-watchman/internal/watch/findings_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/findings_test.go @@ -200,6 +200,45 @@ func TestTheWatchmansFindings(t *testing.T) { } w.NodeHealth("novox", report(delivery.NodeHealth{Node: "novox"}), Put) // all well again }, map[string]string{}}, + {"774: a failing part is held at the machines it points at, as the controller holds it", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + toward := func(node, part string, targets ...string) []byte { + b, _ := delivery.EncodeNodeHealth(delivery.NodeHealth{Node: node, Network: &delivery.NetworkHealth{State: "unhealthy", + Parts: []delivery.NetworkPart{{Part: part, State: "unhealthy", Reason: part + " fails", Said: "x", Targets: targets}}}}) + return b + } + // own part: one machine alone failing toward a healthy one is its own. + w.NodeHealth("ace", toward("ace", "names", "dns1"), Put) + if _, ok := o.findings["machine.ace.network"]; !ok || len(o.findings) != 1 { + t.Fatalf("own part: findings %v", o.findings) + } + // two resolvers' machines: a second machine failing toward dns1 holds both there, said once as unreachable. + w.NodeHealth("g14", toward("g14", "names", "dns1"), Put) + if f, ok := o.findings["machine.dns1.unreachable"]; !ok || len(f.Also) != 2 || len(o.findings) != 1 { + t.Fatalf("two toward one resolver: findings %v", o.findings) + } + w.NodeHealth("ace", toward("ace", "names", "dns1", "dns2"), Put) // toward two resolvers, one not down: its own + if _, ok := o.findings["machine.ace.network"]; !ok { + t.Fatalf("two resolvers, one up: findings %v", o.findings) + } + w.NodeHealth("ace", nil, Delete) + w.NodeHealth("g14", nil, Delete) + // the hub silent on the record: a tunnel toward it is held there, and its silence says it. + w.Connect("node.hub") + w.Disconnect("node.hub") + expire(w, c, pulseKeyOf(t, presenceNS, "hub"), 120*time.Second) + w.NodeHealth("lap", toward("lap", "tunnel", "hub"), Put) + if _, own := o.findings["machine.lap.network"]; own { + t.Errorf("a tunnel toward the silent hub is the reporter's own: %v", o.findings) + } + if _, ok := o.findings["machine.hub.unreachable"]; ok { + t.Error("the silent hub is said unreachable as well") + } + w.Connect("node.hub") // back: lap's tunnel failing toward a hub up is its own again + if _, own := o.findings["machine.lap.network"]; !own { + t.Errorf("hub back: findings %v", o.findings) + } + w.NodeHealth("lap", nil, Delete) + }, 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}}) diff --git a/modules/mesh-delivery-watchman/internal/watch/nodehealth.go b/modules/mesh-delivery-watchman/internal/watch/nodehealth.go index 7cb6f91..d3b6dd3 100644 --- a/modules/mesh-delivery-watchman/internal/watch/nodehealth.go +++ b/modules/mesh-delivery-watchman/internal/watch/nodehealth.go @@ -23,9 +23,10 @@ import ( // - relogin-needed `module...relogin-needed` and directory-used-as-found // `module...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. +// A failing part that points at other mesh machines (NetworkPart.Targets: the hub's node, each resolver's node) is held +// there, as the controller holds it: when every machine it points at is down on the record (silent, its own network +// unhealthy, or two machines failing toward it), it is that machine's `unreachable`, said once there, and no finding of +// the reporter's. A tunnel peer two or more nodes no longer see points at its node the same way. // The kinds a node's health report raises. const ( @@ -63,7 +64,7 @@ func (w *Watchman) NodeHealth(node string, value []byte, op Op) { delete(w.health, node) } w.nodeFindings(node) - w.unreachable() + w.networks() } // TunnelKey is a node's own tunnel key, from its record in its have: whose peers see it is read by it. @@ -80,7 +81,7 @@ func (w *Watchman) tunnelKey(node string, record *delivery.NodeRecord) { } else { w.tunnels[node] = key } - w.unreachable() + w.networks() } // nodeFindings writes and deletes one node's own findings from its report. @@ -88,10 +89,6 @@ 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) @@ -118,58 +115,7 @@ func (w *Watchman) ownsHealthKey(key, node string) bool { 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 + return f.Kind != KindNetworkUnreachable && f.Kind != KindMachineNetwork && f.Kind != KindNetworkRewritten } func (w *Watchman) rewrittenOf(m string, p delivery.NetworkPart, all []delivery.NetworkPart) Finding { @@ -310,48 +256,176 @@ func orModule(m string) string { 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() { +// pointed is one machine's failing part that points at another machine. +type pointed struct { + from string + part delivery.NetworkPart +} + +// networks is every machine's network findings from every report together (the controller's networkObservations): the +// resolver file rewritten, each machine's own unhealthy parts, and the machine others point at, held there. +func (w *Watchman) networks() { + machines := slices.Sorted(maps.Keys(w.health)) + unhealthy := func(m string) []delivery.NetworkPart { + h := w.health[m] + if h.Network == nil { + return nil + } + var out []delivery.NetworkPart + for _, p := range h.Network.Parts { + if p.State == stateUnhealthy { + out = append(out, p) + } + } + return out + } + targets := func(m string, p delivery.NetworkPart) ([]string, bool) { + var out []string + for _, x := range p.Targets { + if x == "" || x == m { + return nil, false + } + if !slices.Contains(out, x) { + out = append(out, x) + } + } + return out, len(out) > 0 + } + pointing := map[string][]pointed{} + for _, m := range machines { + for _, p := range unhealthy(m) { + if xs, ok := targets(m, p); ok { + for _, x := range xs { + pointing[x] = append(pointing[x], pointed{from: m, part: p}) + } + } + } + } + // A tunnel peer a node no longer sees points at that peer's node, as a tunnel part toward it would. 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 + for _, m := range machines { + for _, p := range w.health[m].Peers { + if x, ok := byKey[p.Key]; ok && x != m && !p.Seen { + pointing[x] = append(pointing[x], pointed{from: m, part: delivery.NetworkPart{Part: partTunnel, State: stateUnhealthy, + Said: "no handshake with " + x + " within five minutes", Targets: []string{x}}}) } - notSeen[x] = append(notSeen[x], from) } } + from := func(x string) []string { + var out []string + for _, pt := range pointing[x] { + if !slices.Contains(out, pt.from) { + out = append(out, pt.from) + } + } + sort.Strings(out) + return out + } + silent := func(x string) bool { + _, ok := w.findings[FindingKey(ScopeMachine, x, KindSilent)] + return ok + } + down := func(x string) bool { return silent(x) || len(unhealthy(x)) > 0 || len(from(x)) >= 2 } + allDown := func(xs []string) bool { + for _, x := range xs { + if !down(x) { + return false + } + } + return len(xs) > 0 + } want := map[string]bool{} - for x, from := range notSeen { - if len(from) < 2 { + saysOwn := map[string]bool{} + for _, m := range machines { + parts := unhealthy(m) + if len(parts) == 0 { continue } - if _, own := w.findings[FindingKey(ScopeMachine, x, "network")]; own { + var own []delivery.NetworkPart + var rewritten *delivery.NetworkPart + for i, p := range parts { + if p.Part == partResolvConf { + rewritten = &parts[i] + continue + } + if xs, ok := targets(m, p); ok && allDown(xs) { + continue // held at the machines it points at + } + own = append(own, p) + } + if rewritten != nil { + f := w.rewrittenOf(m, *rewritten, parts) + want[f.Key()] = true + w.find(f) + kept := own[:0] + for _, p := range own { + if p.Part != partNames { + kept = append(kept, p) + } + } + own = kept + } + if len(own) == 0 { continue } - if _, silent := w.findings[FindingKey(ScopeMachine, x, KindSilent)]; silent { - continue - } - sort.Strings(from) + var words, said []string severity := Warning - if w.control[x] { + 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 } - f := Finding{Scope: ScopeMachine, ID: x, Token: "unreachable", Kind: KindNetworkUnreachable, Machine: x, Also: from, + summary := fmt.Sprintf("%s's network is not healthy: %s", m, strings.Join(words, "; ")) + if waiting := from(m); len(waiting) > 0 { + severity = Urgent + summary += fmt.Sprintf("; %s cannot reach it", strings.Join(waiting, ", ")) + } + f := Finding{Scope: ScopeMachine, ID: m, Token: "network", Kind: KindMachineNetwork, Machine: m, Severity: severity, + Source: "health-report", Since: w.health[m].Network.Since, Summary: summary, Said: strings.Join(said, " | ")} + want[f.Key()] = true + saysOwn[m] = true + w.find(f) + } + for _, x := range slices.Sorted(maps.Keys(pointing)) { + if !down(x) || saysOwn[x] || silent(x) { + continue + } + var what, said []string + for _, pt := range pointing[x] { + wd := map[string]string{partTunnel: "the tunnel to it", partBus: "the bus on it", partNames: "its resolver"}[pt.part.Part] + if wd == "" { + wd = pt.part.Part + } + if !slices.Contains(what, wd) { + what = append(what, wd) + } + said = append(said, fmt.Sprintf("%s: %s: %s", pt.from, pt.part.Part, pt.part.Said)) + } + severity := Warning + if w.control[x] || slices.Contains(what, "the bus on it") { + severity = Urgent + } + f := Finding{Scope: ScopeMachine, ID: x, Token: "unreachable", Kind: KindNetworkUnreachable, Machine: x, Also: from(x), 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)} + Summary: fmt.Sprintf("%s cannot be reached from %s: %s", x, strings.Join(from(x), ", "), strings.Join(what, ", ")), + Said: strings.Join(said, " | ")} want[f.Key()] = true w.find(f) } - for key := range w.findings { - if strings.HasSuffix(key, ".unreachable") && strings.HasPrefix(key, ScopeMachine+".") && !want[key] { + for key, value := range w.findings { + f, err := DecodeFinding(value) + if err != nil || f.Source != "health-report" { + continue + } + if (f.Kind == KindMachineNetwork || f.Kind == KindNetworkRewritten || f.Kind == KindNetworkUnreachable) && !want[key] { w.unfind(key) } } diff --git a/modules/mesh-delivery-watchman/internal/watch/signals.go b/modules/mesh-delivery-watchman/internal/watch/signals.go index 217f896..214ee3f 100644 --- a/modules/mesh-delivery-watchman/internal/watch/signals.go +++ b/modules/mesh-delivery-watchman/internal/watch/signals.go @@ -240,7 +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 + defer w.networks() // a silent machine is down on the record: what points at it is held there, said by its silence if s.Moved || w.away[node] { w.unfind(key) w.unsignal("S1", node)