diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go index e1aa03d0..8dac5cfb 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go @@ -200,7 +200,7 @@ func recordSubject(w *watch.Watchman, subject string, body []byte) bool { return true } -// engineWaitSubject hands the node-engine seat's `wait-held` and `wait-ended`, on +// engineWaitSubject hands the node-engine seat's `wait-held`, `wait-ended` and `running`, on // mesh.seat.node-engine.event..., to the watchman (design 47 G §499, §541). It says whether the subject was one. func engineWaitSubject(w *watch.Watchman, subject string, body []byte) bool { parts := strings.Split(subject, ".") @@ -212,6 +212,8 @@ func engineWaitSubject(w *watch.Watchman, subject string, body []byte) bool { w.EngineWaitHeld(parts[5], parts[6], body) case nodeengine.EventWaitEnded: w.EngineWaitEnded(parts[5], parts[6]) + case nodeengine.EventRunning: + w.EngineRunning(parts[5], parts[6], body) default: return false } diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go index b5facb10..0b08b916 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go @@ -17,7 +17,6 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" - "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" @@ -150,7 +149,6 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 {Bucket: watch.StuckBucket, History: 20}, {Bucket: watch.PulseBucket, LimitMarkerTTL: time.Second}, {Bucket: watch.AssignerWantsBucket, History: 20}, - {Bucket: watch.HaveBucket, History: 20}, {Bucket: watch.FindingsBucket, History: 5}, {Bucket: firstViewsBucket, History: 1}, } { @@ -390,7 +388,6 @@ func anAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909 {Bucket: watch.StuckBucket, History: 20}, {Bucket: watch.PulseBucket, LimitMarkerTTL: time.Second}, {Bucket: watch.AssignerWantsBucket, History: 20}, - {Bucket: watch.HaveBucket, History: 20}, {Bucket: watch.FindingsBucket, History: 5}, {Bucket: firstViewsBucket, History: 1}, } { @@ -472,7 +469,6 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) { {Bucket: watch.StuckBucket, History: 20}, {Bucket: watch.PulseBucket, LimitMarkerTTL: time.Second}, {Bucket: watch.AssignerWantsBucket, History: 20}, - {Bucket: watch.HaveBucket, History: 20}, {Bucket: watch.FindingsBucket, History: 5}, {Bucket: firstViewsBucket, History: 1}, } { @@ -498,14 +494,11 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) { t.Fatal(err) } } + // The node-engine says `running` on its seat (design 47 G §541), which the watchman hears on its consumer of EVENTS. putHave := func(node, module, fp string) { t.Helper() - body, err := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}) - if err != nil { - t.Fatal(err) - } - key, _ := nodeengineclient.HaveKey(node, module) - if _, err := kvs[watch.HaveBucket].Put(ctx, key, body); err != nil { + body, _ := json.Marshal(nodeengine.Running{Fingerprint: fp, Module: module, Node: node, At: time.Now().UTC(), V: 1}) + if _, err := js.Publish(ctx, "mesh.seat.node-engine.event.running."+node+"."+module, body); err != nil { t.Fatal(err) } } diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go index f2ffe4be..ed02decb 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -221,7 +221,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // Its own presence, pulses and stuck rows are what it keeps: each stops the start, naming its grant. The assigner's wants // and the node-engine's haves only make the pulses of a move: without the haves the watchman watches presence and the // build seats' steps, and says that moves are not watched (issue 1945). - var kvs struct{ presence, pulse, stuck, haves jetstream.KeyValue } + var kvs struct{ presence, pulse, stuck jetstream.KeyValue } for _, b := range []struct { name string into *jetstream.KeyValue @@ -230,7 +230,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } } - kvs.haves = process.OptionalBucket(ctx, js, watch.HaveBucket, "a want makes no pulse, so a move that stops is not seen, until the watchman starts again", logger.Printf) // The assigner's wants and the indexer's modules are their owners' states, read by the first view of their read tools and // then their views families, never their buckets (ADR 0615 §1, DLV-M5). An owner that is down never holds the start: each // is read on a goroutine once its first view is answered, and until then the unanswered first view is the watchman's @@ -308,15 +307,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } - // What the node agents run (issue 1906, ADR 0539 §19): a want is a pulse owed by its node's agent until the have names its - // fingerprint, so the haves are read at the start and the wants after them, once their first view is answered. - var haveSeen jetstream.KeyWatcher - if kvs.haves != nil { - if haveSeen, err = settle(ctx, kvs.haves, func(e jetstream.KeyValueEntry) { w.Have(e.Key(), e.Value(), opOf(e)) }); err != nil { - process.NotWatched(logger.Printf, watch.HaveBucket, err, "moves are not watched until the watchman starts again") - haveSeen = nil - } - } // The events first, then the one request of the open connections, so no connection falls between them. onEvent := func(open bool) nats.MsgHandler { return func(m *nats.Msg) { @@ -377,25 +367,6 @@ 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, watch.NodeHealthBucket); err == nil { - if healthWatch, err = settle(ctx, kv, healthEach); err != nil { - process.NotWatched(logger.Printf, watch.NodeHealthBucket, err, "no node's own health is said until the watchman starts again") - } - } else { - process.AwaitBucket(ctx, js, conn, watch.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, watch.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() @@ -439,24 +410,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } errs := make(chan error, 16) - if healthWatch != nil { - go func() { errs <- follow(ctx, healthWatch, healthEach) }() - } // The first views of the wants and the modules, asked now and read each on its own goroutine once answered (ADR 0615). reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "watchman_first_views reads nothing and watchman_first_view_again is not heard", logger.Printf)) // Every part has registered: the hub pulls the module's durable, and the watchman's other consumers open. hearConsumed(ctx, js, cred, hub, w, logger.Printf) defer hub.Stop() - if haveSeen != nil { - go func() { - errs <- follow(ctx, haveSeen, func(e jetstream.KeyValueEntry) { w.Have(e.Key(), e.Value(), opOf(e)) }) - }() - go func() { - errs <- readLate(ctx, w, wants, watch.AssignerWantsBucket, func(e jetstream.KeyValueEntry) { - w.Want(e.Key(), e.Value(), opOf(e), createdOr(e, time.Now())) - }) - }() - } + go func() { + errs <- readLate(ctx, w, wants, watch.AssignerWantsBucket, func(e jetstream.KeyValueEntry) { + w.Want(e.Key(), e.Value(), opOf(e), createdOr(e, time.Now())) + }) + }() if provWatch != nil { go func() { errs <- follow(ctx, provWatch, func(e jetstream.KeyValueEntry) { w.Provision(e.Key(), e.Value(), opOf(e)) }) diff --git a/modules/mesh-delivery-watchman/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go index 0c6f3d11..6e3a2778 100644 --- a/modules/mesh-delivery-watchman/internal/watch/findings_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/findings_test.go @@ -3,14 +3,11 @@ package watch import ( "encoding/json" "errors" - "slices" "strings" "sync/atomic" "testing" "time" - "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" - "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient" @@ -30,15 +27,6 @@ func mustEncode(t *testing.T, f Finding) []byte { return v } -func haveOf(t *testing.T, phase string, health string, detail string) []byte { - t.Helper() - b, err := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fpNew, Phase: phase, Health: health, Detail: detail}) - if err != nil { - t.Fatal(err) - } - return b -} - // expire hears the expiry of the pulse at pk. func expire(w *Watchman, c *clock, pk string, after time.Duration) { c.t = c.t.Add(after) @@ -323,29 +311,6 @@ func TestTheWatchmansFindings(t *testing.T) { } } }, map[string]string{}}, - {"ADR 0601 §4: what a node keeps of a module as it runs, for a manifest that does not say it yet, is a warning until the have says none", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { - have := func(kept ...string) []byte { - b, err := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: "src2:" + strings.Repeat("a", 64), Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK, Kept: kept}) - if err != nil { - t.Fatal(err) - } - return b - } - k := FindingKey(ScopeModule, "runner-x.ace", "kept") - w.Have("ace.runner-x", have("runner-x runs the tool runner's process here"), Put) - f := o.findings[k] - if f.Kind != KindKeptAsRuns || f.Severity != Warning || !strings.Contains(f.Summary, "the tool runner's process") { - t.Fatalf("findings = %v", o.findings) - } - c.t = c.t.Add(time.Minute) - w.Have("ace.runner-x", have("runner-x runs the tool runner's process here"), Put) - if !o.findings[k].Since.Equal(f.Since) { - t.Errorf("a have written again moved the finding's since: %v, was %v", o.findings[k].Since, f.Since) - } - w.Have("ace.prov-y", have("prov-y carries its own secrets sealed"), Put) - w.Have("ace.runner-x", have(), Put) // its manifest says tool-runner now - w.Have("ace.prov-y", nil, Delete) - }, map[string]string{}}, {"WM4: a stuck key not moved is its finding, and its move deletes it", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { askProgress(w, "ask1", t0) expire(w, c, pulseKeyOf(t, checksKey, "ask1"), 120*time.Second) @@ -447,18 +412,6 @@ func TestTheWatchmansFindings(t *testing.T) { } expire(w, c, signalKey("S9", k), AdvisoryQuiet) }, map[string]string{}}, - {"WM3 S21: a heard machine's have failed for one apply interval is apply-failing, naming the package manager; applied again it ends", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { - w.Connect("node.ace") - w.Have("ace.web", haveOf(t, nodeengineclient.PhaseFailed, nodeengineclient.HealthUnknown, "refused: the package manager (pacman) does not answer"), Put) - if len(o.findings) != 0 { - t.Fatalf("raised before its bound: %v", o.findings) - } - expire(w, c, signalKey("S21", "ace"), ApplyEvery+time.Second) - if f := o.findings["machine.ace.apply-failing"]; !strings.Contains(f.Summary, "pacman does not answer") || f.Headline != "ace cannot install packages" { - t.Fatalf("findings = %v", o.findings) - } - w.Have("ace.web", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthOK, ""), Put) - }, map[string]string{}}, {"WM3 S21: a provisioner client's failed files are apply-failing past the bound; a silent machine's are not", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { w.Connect("node.ace") w.Provisioned(ProvReport{Node: "ace", Module: "web", Status: ProvFailed, Reason: "disk full", At: t0.Add(-10 * time.Minute)}) @@ -468,48 +421,6 @@ func TestTheWatchmansFindings(t *testing.T) { w.Provisioned(ProvReport{Node: "g14", Module: "web", Status: ProvFailed, Reason: "x", At: t0.Add(-10 * time.Minute)}) w.Provisioned(ProvReport{Node: "ace", Module: "web", Status: "applied", At: t0}) }, map[string]string{}}, - {"WM3: unhealthy once a failed have is confirmed after a minute (written once and steady); recovered within it none; urgent after four hours", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { - w.Have("ace.web", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthFailed, "the unit failed"), Put) - if len(o.findings) != 0 { - t.Fatalf("raised before its bound: %v", o.findings) - } - w.Have("ace.quick", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthFailed, "blip"), Put) - w.Have("ace.quick", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthOK, ""), Put) // recovered within the bound - expire(w, c, signalKey("unhealthy-confirm", "quick.ace"), UnhealthyConfirmAfter) - expire(w, c, signalKey("unhealthy-confirm", "web.ace"), UnhealthyConfirmAfter) // the one write, never repeated - if o.findings["module.web.ace.unhealthy"].Severity != Warning || len(o.findings) != 1 { - t.Fatalf("findings = %v", o.findings) - } - expire(w, c, signalKey("unhealthy", "web.ace"), UnhealthyUrgentAfter) - w.Have("ace.mail", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthWaits, "a part waits for the secret token"), Put) // needs-operator is not observed here - }, map[string]string{"module.web.ace.unhealthy": Urgent}}, - {"WM3: a warning unhealthy finding read back at the start is kept as it was, urgent when its four hours from Since are out", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { - seedUnhealthy(t, w, o, Warning, t0.Add(-time.Hour)) - w.Have("ace.web", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthFailed, "x"), Put) - w.Reconcile() - if o.findings["module.web.ace.unhealthy"].Severity != Warning { - t.Fatalf("cleared: findings %v", o.findings) - } - if ttl := o.ttls[signalKey("unhealthy", "web.ace")]; ttl != UnhealthyUrgentAfter-time.Hour { - t.Fatalf("urgent signal restarted from zero: %v", ttl) - } - if _, ok := w.signals[signalKey("unhealthy-confirm", "web.ace")]; ok { - t.Fatal("a standing finding is confirmed again") - } - expire(w, c, signalKey("unhealthy", "web.ace"), UnhealthyUrgentAfter-time.Hour) - }, map[string]string{"module.web.ace.unhealthy": Urgent}}, - {"WM3: an urgent unhealthy finding read back at the start stays urgent, with no signal restarted", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { - seedUnhealthy(t, w, o, Urgent, t0.Add(-5*time.Hour)) - w.Have("ace.web", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthFailed, "x"), Put) - w.Reconcile() - if o.findings["module.web.ace.unhealthy"].Severity != Urgent || len(w.signals) != 0 { - t.Fatalf("findings %v signals %v", o.findings, w.signals) - } - }, map[string]string{"module.web.ace.unhealthy": Urgent}}, - {"WM3: a restart replays a have already failed: it is confirmed after a minute like a new one", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { - w.Have("novox.fail2ban", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthFailed, "down since 19:40"), Put) - expire(w, c, signalKey("unhealthy-confirm", "fail2ban.novox"), UnhealthyConfirmAfter) - }, map[string]string{"module.fail2ban.novox.unhealthy": Warning}}, {"1954/D2: a provision in waits is its consumer's needs-operator; provided, it ends", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { enc := func(state, why string) []byte { b, err := meshprovisionerclient.Encode(meshprovisionerclient.Provision{Provision: "oidc", Consumer: meshprovisionerclient.End{Module: "photos", Node: "ace"}, @@ -528,141 +439,6 @@ func TestTheWatchmansFindings(t *testing.T) { w.Provision("ace.web.oidc", enc(meshprovisionerclient.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, a login fallen back", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { - report := func(h nodeengine.HealthByNodeValue) []byte { - b, err := nodeengineclient.EncodeHealth(h) - if err != nil { - t.Fatal(err) - } - return b - } - w.Have("novox.nats", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthOK, ""), Put) // the control node: the bus node - w.NodeHealth("novox", report(nodeengine.HealthByNodeValue{Node: "novox", Network: &nodeengine.NetworkHealth{State: "unhealthy", Parts: []nodeengine.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: &nodeengine.UnitsHealth{State: "degraded", Failed: []nodeengine.FailedUnit{{Unit: "cups.service", Scope: "system", Result: "exit-code"}}}, - Relogin: []nodeengine.NodeAccount{{Account: "jochen", Module: "docker", Reason: "a group"}}, - Found: []nodeengine.NodeDirectory{{Module: "photos", Resource: "data", Path: "/srv/photos", Reason: "owner differs"}}, - Fallback: &nodeengine.LoginFallback{Node: "novox", From: "node.novox--b", Why: "refused twice", At: time.Date(2026, 11, 2, 12, 0, 0, 0, time.UTC)}}), 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, - "machine.novox.node-account-fallback": 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(nodeengine.HealthByNodeValue{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, _ := nodeengineclient.EncodeHealth(nodeengine.HealthByNodeValue{Node: node, Network: &nodeengine.NetworkHealth{State: "unhealthy", - Parts: []nodeengine.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") - } - // design 47 G §706: the silence lists the machines failing toward it in `also`, as `unreachable` does. - if f := o.findings["machine.hub.silent"]; !slices.Equal(f.Also, []string{"lap"}) { - t.Errorf("the hub's silence lists %v in also", f.Also) - } - w.NodeHealth("lap", nil, Delete) - if f := o.findings["machine.hub.silent"]; len(f.Also) != 0 { - t.Errorf("the hub's silence still lists %v once nothing fails toward it", f.Also) - } - w.NodeHealth("lap", toward("lap", "tunnel", "hub"), Put) - // the hub's own network open, two machines failing toward it: its `network` finding lists both. - w.NodeHealth("hub", toward("hub", "bus"), Put) - w.NodeHealth("ace", toward("ace", "tunnel", "hub"), Put) - if f := o.findings["machine.hub.network"]; !slices.Equal(f.Also, []string{"ace", "lap"}) { - t.Errorf("the hub's network lists %v in also: %v", f.Also, o.findings) - } - w.NodeHealth("hub", nil, Delete) - w.NodeHealth("ace", nil, Delete) - 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, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: node, TunnelKey: key}}) - w.Have(node+".node", b, Put) - } - peers := func(from string, seen bool) []byte { - b, _ := nodeengineclient.EncodeHealth(nodeengine.HealthByNodeValue{Node: from, Peers: []nodeengine.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, nodeengineclient.PhaseRunning, nodeengineclient.HealthFailed, "x"), Put) - expire(w, c, signalKey("unhealthy-confirm", "mail.ace"), UnhealthyConfirmAfter) - w.Have("ace.mail", haveOf(t, nodeengineclient.PhaseRemoved, nodeengineclient.HealthUnknown, ""), Put) - }, map[string]string{}}, - {"WM2 S1: a control node silent is urgent after half an hour", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { - // 1126 (ADR 0590): the control node is the bus node, the one whose have holds the bus module; a have of the - // ended controller names none. - w.Have("g14.mesh-controller", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthOK, ""), Put) - w.Connect("node.g14") - w.Disconnect("node.g14") - expire(w, c, pulseKeyOf(t, presenceNS, "g14"), 120*time.Second) - if _, ok := w.signals[signalKey("S1", "g14")]; ok { - t.Fatalf("a node running the controller's have is held a control node") - } - w.Have("novox.nats", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthOK, ""), Put) - w.Connect("node.novox") - w.Disconnect("node.novox") - expire(w, c, pulseKeyOf(t, presenceNS, "novox"), 120*time.Second) - if o.findings["machine.novox.silent"].Severity != Warning { - t.Fatalf("findings = %v", o.findings) - } - expire(w, c, signalKey("S1", "novox"), ControlNodeUrgentAfter) - }, map[string]string{"machine.novox.silent": Urgent, "machine.g14.silent": Warning}}, {"WM3 S9: the bus module's closures, each counted once, ended an hour after the newest", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { cl := BusClosure{User: "ace.web", CID: 7, Reason: "Slow Consumer (Write Deadline)", Opened: t0.Add(-time.Hour), Closed: t0} w.Closed(KindSlowConsumer, cl) @@ -697,14 +473,6 @@ func TestTheWatchmansFindings(t *testing.T) { t.Fatal("still counted again with nothing held") } }, map[string]string{}}, - {"WM4: resync after a cleared bucket writes the same findings", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { - w.Have("ace.mail", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthFailed, "x"), Put) - expire(w, c, signalKey("unhealthy-confirm", "mail.ace"), UnhealthyConfirmAfter) - o.findings = map[string]Finding{} - if n := w.Rewrite(); n != 1 { - t.Fatalf("rewrote %d", n) - } - }, map[string]string{"module.mail.ace.unhealthy": Warning}}, {"WM4: a finding read at the start and not observed again is deleted; a signal's stays", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { for _, f := range []Finding{ {Scope: ScopeDelivery, ID: "x.y", Kind: KindStuck, Severity: Warning, Summary: "old", Source: "stuck"}, @@ -733,7 +501,7 @@ func TestTheWatchmansFindings(t *testing.T) { Provider: meshprovisionerclient.End{Module: "keycloak", Node: "novox"}, State: meshprovisionerclient.Asked, At: t0}) provided, _ := meshprovisionerclient.Encode(meshprovisionerclient.Provision{Provision: "oidc", Consumer: meshprovisionerclient.End{Module: "photos", Node: "ace"}, Provider: meshprovisionerclient.End{Module: "keycloak", Node: "novox"}, State: meshprovisionerclient.Provided, At: t0}) - w.Have("ace.web", haveValue(t, fpOld), Put) // the node runs what the held want names: the hold is still the wait + w.EngineRunning("ace", "web", []byte(`{"fingerprint":"`+fpOld+`"}`)) // the node runs what the held want names: the hold is still the wait w.Want("want.ace.web", held, Put, t0) w.Provision("ace.photos.oidc", asked, Put) waits := map[string]string{AssignerWantsBucket: "want.ace.web", ProvisionsBucket: "ace.photos.oidc"} @@ -783,14 +551,6 @@ func TestTheWatchmansFindings(t *testing.T) { } } -func seedUnhealthy(t *testing.T, w *Watchman, o *fakeOut, sev string, since time.Time) { - t.Helper() - f := Finding{Scope: ScopeModule, ID: "web.ace", Token: "unhealthy", Kind: KindUnhealthy, Machine: "ace", Severity: sev, Source: "health", Since: since, Summary: "web on ace is not healthy: x"} - v, _ := EncodeFinding(f) - o.findings[f.Key()], _ = DecodeFinding(v) - w.SeedFinding(f.Key(), v) -} - const ( revKey = "m.00000000000000000042" groupKey = "g.feat/take-in" @@ -827,9 +587,9 @@ func takeInBody(t *testing.T, kind, key string, since time.Time) []byte { func TestAFindingIsTheContractsWithItsRefInItsID(t *testing.T) { since := time.Date(2026, 11, 11, 20, 0, 0, 0, time.UTC) for _, f := range []Finding{ - {Scope: ScopeModule, ID: "web.ace", Token: "unhealthy", Kind: KindUnhealthy, Machine: "ace", Severity: Warning, Source: "health", Since: since, Summary: "s", Said: "said"}, + {Scope: ScopeModule, ID: "web.ace", Token: "unhealthy", Kind: KindStuck, Machine: "ace", Severity: Warning, Source: "stuck", Since: since, Summary: "s", Said: "said"}, {Scope: ScopeModule, ID: "web", Token: KindBusStateOverdue + "-state", Kind: KindBusStateOverdue, Severity: Warning, Source: "bus", Since: since, Summary: "s"}, - {Scope: ScopeMachine, ID: "ace", Kind: KindMachineUnits, Severity: Warning, Source: "units", Since: since, Summary: "s", Needs: "mend"}, + {Scope: ScopeMachine, ID: "ace", Kind: KindSilent, Severity: Warning, Source: "S1", Since: since, Summary: "s", Needs: "mend"}, } { c := f.Contract() wantID := f.ID diff --git a/modules/mesh-delivery-watchman/internal/watch/mirror.go b/modules/mesh-delivery-watchman/internal/watch/mirror.go index f20ceefb..a45fc04f 100644 --- a/modules/mesh-delivery-watchman/internal/watch/mirror.go +++ b/modules/mesh-delivery-watchman/internal/watch/mirror.go @@ -164,46 +164,30 @@ func (w *Watchman) Want(key string, value []byte, op Op, at time.Time) { describe("the node agent of %s to run %s at %s", node, module, shortFP(want.Fp)), want.Fp, at, 0) } -// Have is a change of a key of the node-engine's have bucket: the fingerprint a node runs of a module. One that names the -// 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 := nodeengineclient.ParseHaveKey(key) +// EngineRunning hears the node-engine of node saying `running` for module (design 47 G §541, A-192, A-197): the fingerprint it runs, or +// that it runs none (stopped). One that names the fingerprint of the node's want for the module ends the want's pulse. +func (w *Watchman) EngineRunning(node, module string, body []byte) { + var r nodeengine.Running + if err := json.Unmarshal(body, &r); err != nil { + w.say("a running of the node-engine of %s for %s does not read, left: %v", node, module, err) + return + } + key, err := nodeengineclient.HaveKey(node, module) 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 *nodeengine.NodeRecord - if op == Put { - if hn, err := nodeengineclient.DecodeHaveNode(value); err == nil { - record = hn.Record - } - } - w.tunnelKey(node, record) - return - } - switch op { - case Expired: - return - case Delete: + if r.Stopped || r.Fingerprint == "" { delete(w.haves, key) - w.haveFindings(key, node, module, nil) return } - have, err := nodeengineclient.DecodeHave(value) - if err != nil { - w.say("the have %s cannot be read, and is left: %v", key, err) - return - } - w.haves[key] = have.Fp - w.haveFindings(key, node, module, &have) + w.haves[key] = r.Fingerprint wk, err := meshwantsclient.WantKey(node, module) if err != nil { return } - if fp, ok := w.wants[wk]; ok && fp == have.Fp { + if fp, ok := w.wants[wk]; ok && fp == r.Fingerprint { w.moved(AssignerWantsBucket, wk) } } diff --git a/modules/mesh-delivery-watchman/internal/watch/names.go b/modules/mesh-delivery-watchman/internal/watch/names.go index d8e9121d..67060d94 100644 --- a/modules/mesh-delivery-watchman/internal/watch/names.go +++ b/modules/mesh-delivery-watchman/internal/watch/names.go @@ -20,9 +20,6 @@ const ( PulseBucket = WatchmanModule + "_" + meshwatchdogs.StatePulse StuckBucket = WatchmanModule + "_" + meshwatchdogs.StateStuck - // HaveBucket and NodeHealthBucket are the node-engine's states have and health (reads: node-engine.have, mesh-host.health). - HaveBucket = "node-engine_have" - NodeHealthBucket = "mesh-host_health" // AssignerWantsBucket is the assigner's state wants, followed by its views family (consumes: mesh-delivery-assigner.views.wants). AssignerWantsBucket = AssignerModule + "_wants" ) diff --git a/modules/mesh-delivery-watchman/internal/watch/nodehealth.go b/modules/mesh-delivery-watchman/internal/watch/nodehealth.go deleted file mode 100644 index 7295e1b5..00000000 --- a/modules/mesh-delivery-watchman/internal/watch/nodehealth.go +++ /dev/null @@ -1,485 +0,0 @@ -package watch - -import ( - "fmt" - "maps" - "slices" - "sort" - "strings" - "time" - - "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" - "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" -) - -// 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.` 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..network`: the machine's own unhealthy parts; urgent on the control node or for the bus; -// - network-rewritten `machine.[.].rewritten`: the resolver file rewritten by another program; -// - network-unreachable `machine..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..units`: failed units no module places; -// - relogin-needed `module...relogin-needed` and directory-used-as-found -// `module...directory-used-as-found`; -// - node-account-fallback `machine..node-account-fallback`: the engine fell back to its node's old login during a -// rotation (novox/hq ADR 0583 §6), raised while its report carries the fall back and cleared once it no longer does. -// -// 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 ( - KindMachineNetwork = "machine-network" - KindNetworkRewritten = "network-rewritten" - KindNetworkUnreachable = "network-unreachable" - KindMachineUnits = "machine-units" - KindReloginNeeded = "relogin-needed" - KindUsedAsFound = "directory-used-as-found" - KindAccountFallback = FindingNodeAccountFallback -) - -// 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 := nodeengineclient.DecodeHealth(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.networks() -} - -// 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 *nodeengine.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.networks() -} - -// 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 { - 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) - } - if f, raise := fallbackOf(node, h.Fallback); raise { - 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 && f.Kind != KindMachineNetwork && f.Kind != KindNetworkRewritten -} - -func (w *Watchman) rewrittenOf(m string, p nodeengine.NetworkPart, all []nodeengine.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 *nodeengine.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)}, true -} - -// reloginOf is one finding per module whose accounts wait for a new login. -func reloginOf(node string, accounts []nodeengine.NodeAccount) []Finding { - by := map[string][]nodeengine.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 []nodeengine.NodeDirectory) []Finding { - by := map[string][]nodeengine.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."}) - } - return out -} - -// fallbackOf is a node's fall back to its old login during a rotation of its account (ADR 0583 §6), one finding. -func fallbackOf(node string, f *nodeengine.LoginFallback) (Finding, bool) { - if f == nil || f.Node != node { - return Finding{}, false - } - said := fmt.Sprintf("%s fell back from its new login %s to its old one during a rotation of its bus account: %s", node, f.From, f.Why) - return Finding{Scope: ScopeMachine, ID: node, Token: KindAccountFallback, Kind: KindAccountFallback, Machine: node, Severity: Warning, - Source: "health-report", Since: f.At, Summary: said, Said: said, - Headline: fmt.Sprintf("%s fell back to its old bus login", node), - Needs: fmt.Sprintf("abandon the rotation of %s's account, or rotate it again once its new login authenticates.", node)}, true -} - -func orModule(m string) string { - if m == "" { - return "_" - } - return m -} - -// silentAlso sets the machines `also` lists on the open `silent` finding of x, when there is one. -func (w *Watchman) silentAlso(x string, also []string) { - value, ok := w.findings[FindingKey(ScopeMachine, x, KindSilent)] - if !ok { - return - } - f, err := DecodeFinding(value) - if err != nil || slices.Equal(f.Also, also) { - return - } - f.Also = also - w.find(f) -} - -// keepSilentAlso gives a `silent` finding about to be written the `also` the open one holds, so a write of its other -// words (its urgency after half an hour) never drops the machines listed there. -func (w *Watchman) keepSilentAlso(f Finding) Finding { - if f.Kind != KindSilent { - return f - } - if value, ok := w.findings[f.Key()]; ok { - if cur, err := DecodeFinding(value); err == nil { - f.Also = cur.Also - } - } - return f -} - -// pointed is one machine's failing part that points at another machine. -type pointed struct { - from string - part nodeengine.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) []nodeengine.NetworkPart { - h := w.health[m] - if h.Network == nil { - return nil - } - var out []nodeengine.NetworkPart - for _, p := range h.Network.Parts { - if p.State == stateUnhealthy { - out = append(out, p) - } - } - return out - } - targets := func(m string, p nodeengine.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 - } - 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: nodeengine.NetworkPart{Part: partTunnel, State: stateUnhealthy, - Said: "no handshake with " + x + " within five minutes", Targets: []string{x}}}) - } - } - } - 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 - } - // A machine silent on the record holds the failures toward it under its silence, so its finding lists in `also` the - // machines whose failing parts point at it, as `unreachable` does (design 47 G §706); a change of `also` rewrites it. - for key, value := range w.findings { - if f, err := DecodeFinding(value); err == nil && f.Kind == KindSilent && key == f.Key() { - w.silentAlso(f.ID, from(f.ID)) - } - } - want := map[string]bool{} - saysOwn := map[string]bool{} - for _, m := range machines { - parts := unhealthy(m) - if len(parts) == 0 { - continue - } - var own []nodeengine.NetworkPart - var rewritten *nodeengine.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 - } - 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 - } - 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, Also: from(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: %s", x, strings.Join(from(x), ", "), strings.Join(what, ", ")), - Said: strings.Join(said, " | ")} - want[f.Key()] = true - w.find(f) - } - 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 4485ffc1..24361bf1 100644 --- a/modules/mesh-delivery-watchman/internal/watch/signals.go +++ b/modules/mesh-delivery-watchman/internal/watch/signals.go @@ -8,8 +8,6 @@ import ( "strings" "time" - "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" - "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient" "git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient" ) @@ -45,14 +43,8 @@ const ( HeartbeatEvery = 60 * time.Second // ApplyEvery is one apply interval (S21): a failure still there after it is not one the next apply clears. ApplyEvery = 5 * time.Minute - // UnhealthyConfirmAfter is how long a have must keep saying failed before the module is said unhealthy. - UnhealthyConfirmAfter = time.Minute - // UnhealthyUrgentAfter is how long a module stays unhealthy before it is urgent. - UnhealthyUrgentAfter = 4 * time.Hour - // SilentBound is the bound S1 says, three of the intervals a machine has always used, and ControlNodeUrgentAfter - // how long a control node may be silent before it is urgent (the controller's silentBound, controlNodeUrgentAfter). - SilentBound = 3 * HeartbeatEvery - ControlNodeUrgentAfter = 30 * time.Minute + // SilentBound is the bound S1 says, three of the intervals a machine has always used (the controller's silentBound). + SilentBound = 3 * HeartbeatEvery ) // signalBucket is the pseudo-bucket of a signal's pulse key. @@ -179,27 +171,16 @@ func (w *Watchman) signalExpired(pk string) { if due.While != "" && !w.stillWrong(due.While, due.Finding.Kind) { return } - w.find(w.keepSilentAlso(due.Finding)) + w.find(due.Finding) } } -// stillWrong says a have's key still says what a row's finding was waited on: failed for apply-failing, unhealthy for -// unhealthy. -func (w *Watchman) stillWrong(haveKey, kind string) bool { - if node, ok := strings.CutPrefix(haveKey, "silent:"); ok { +// stillWrong says a silent machine's finding still stands, which a row's finding waited on. +func (w *Watchman) stillWrong(key, _ string) bool { + if node, ok := strings.CutPrefix(key, "silent:"); ok { _, open := w.findings[FindingKey(ScopeMachine, node, KindSilent)] return open } - h, ok := w.haveState[haveKey] - if !ok { - return false - } - switch kind { - case KindApplyFailing: - return h.Phase == nodeengineclient.PhaseFailed - case KindUnhealthy: - return h.Health == nodeengineclient.HealthFailed - } return false } @@ -212,10 +193,8 @@ const ( KindSelfCheckSilent = "self-check-silent" KindToolsSilent = "tools-silent" KindApplyFailing = "apply-failing" - KindUnhealthy = "module-unhealthy" KindNeedsOperator = "needs-operator" KindDeadLetters = "max-deliveries" - KindKeptAsRuns = "kept-as-it-runs" ) func stamp(t time.Time) string { return t.UTC().Format("2006-01-02 15:04 MST") } @@ -264,7 +243,6 @@ func (w *Watchman) stuckFinding(sk string, s meshwatchdogsclient.Stuck) { } else { w.machineAway(node, s.Since) } - 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) @@ -274,18 +252,7 @@ func (w *Watchman) stuckFinding(sk string, s meshwatchdogsclient.Stuck) { f := Finding{Scope: ScopeMachine, ID: node, Kind: KindSilent, Machine: node, Severity: Warning, Source: "S1", Summary: fmt.Sprintf("%s has not been heard from since %s (bound %s)", node, stamp(s.Since), SilentBound), Said: fmt.Sprintf("offline since %s", s.Since.UTC().Format(time.RFC3339)), Since: s.Since} - // A control node, the bus node (ControlNodeModule), is urgent after half an hour silent, as the controller - // says it (its watchHeartbeats): the expiry of a pulse from its silence's start, never a loop. - if w.control[node] { - if age := w.now().Sub(s.Since); age > ControlNodeUrgentAfter { - f.Severity = Urgent - } else { - u := f - u.Severity = Urgent - w.signal("S1", node, ControlNodeUrgentAfter-age+time.Second, signalDue{Do: dueFind, While: "silent:" + node, Finding: u}) - } - } - w.find(w.keepSilentAlso(f)) + w.find(f) w.applyCheck(node) return } @@ -478,114 +445,7 @@ func ToolsNode(subject string) string { return node } -// ---- S21 and unhealthy, from the haves and the provisioner clients' reports ---- - -// haveFindings reads a have's change for the rows the haves feed: S21 (the machine's applies failing, applyCheck), -// unhealthy (two writes in a row, urgent after four hours), and where the bus module runs (S1's control node). A have -// removed or deleted ends its part of them. -// -// needs-operator is not observed from the haves: a module's part that waits for the operator is a provision in the state -// `waits` of the seat mesh-provisioner (Provision, ADR 0570 §7). -func (w *Watchman) haveFindings(key string, node, module string, h *nodeengine.HaveByNodeModuleValue) { - unhealthyKey := FindingKey(ScopeModule, module+"."+node, "unhealthy") - if module == ControlNodeModule { - w.control[node] = h != nil && h.Phase != nodeengineclient.PhaseRemoved - } - if h == nil || h.Phase == nodeengineclient.PhaseRemoved { - delete(w.haveState, key) - delete(w.unhealthy, key) - delete(w.failedSince, key) - w.unsignal("unhealthy-confirm", module+"."+node) - w.unsignal("unhealthy", module+"."+node) - w.unfind(unhealthyKey) - w.keptFinding(key, node, module, nil) - w.applyCheck(node) - return - } - w.haveState[key] = *h - now := w.now() - w.keptFinding(key, node, module, h.Kept) - - // S21: a have refused or applied in part says phase failed, with the node's words in its detail. - if h.Phase == nodeengineclient.PhaseFailed { - if _, ok := w.failedSince[key]; !ok { - w.failedSince[key] = now - } - } else { - delete(w.failedSince, key) - } - w.applyCheck(node) - - // unhealthy: a have that says failed and still says it after UnhealthyConfirmAfter (ADR 0564 §4's confirmation, by time: - // a have written once and then unchanged never gets a second write). The confirm signal's expiry writes the warning; a - // signal already standing (kept across a restart) is not renewed, so a steady have does not push its own expiry out. - if h.Health == nodeengineclient.HealthFailed { - if w.unhealthy[key] == 0 { - w.unhealthy[key] = 1 - f := Finding{Scope: ScopeModule, ID: module + "." + node, Token: "unhealthy", Kind: KindUnhealthy, Machine: node, - Severity: Warning, Source: "health", Since: now, - Summary: fmt.Sprintf("%s on %s is not healthy: %s", module, node, orWords(h.Detail, "its check failed")), - Said: firstLine(orWords(h.Detail, "failed")), - Headline: fmt.Sprintf("%s not working on %s", module, node)} - id := module + "." + node - urgentIn := UnhealthyUrgentAfter - if w.seededFindings[unhealthyKey] { - // Read back at the start: it stands, so it is observed again, kept as it was (not cleared and raised again), and - // its four hours run on from its own Since, not from zero; one that already went urgent needs no signal. - delete(w.seededFindings, unhealthyKey) - old, err := DecodeFinding(w.findings[unhealthyKey]) - if err == nil && old.Severity == Urgent { - urgentIn = 0 - } else if err == nil && !old.Since.IsZero() { - urgentIn = max(time.Second, UnhealthyUrgentAfter-now.Sub(old.Since)) - } - } else if _, ok := w.signals[signalKey("unhealthy-confirm", id)]; !ok { - w.signal("unhealthy-confirm", id, UnhealthyConfirmAfter, signalDue{Do: dueFind, While: key, Finding: f}) - } - if _, ok := w.signals[signalKey("unhealthy", id)]; !ok && urgentIn > 0 { - f.Severity = Urgent - w.signal("unhealthy", id, urgentIn, signalDue{Do: dueFind, While: key, Finding: f}) - } - } - } else { - delete(w.unhealthy, key) - w.unsignal("unhealthy-confirm", module+"."+node) - w.unsignal("unhealthy", module+"."+node) - w.unfind(unhealthyKey) - } -} - -// keptFinding raises, as a warning, what a node keeps of a module as it runs because its manifest does not yet say what shared -// code no longer infers (its have's Kept, novox/hq ADR 0601 §4: the tool runner, the core's credentials), naming each; and -// clears it once the have says none, or is gone. Its since is the first have that said it, so a have written again moves -// nothing. -func (w *Watchman) keptFinding(key, node, module string, kept []string) { - id := module + "." + node - fkey := FindingKey(ScopeModule, id, "kept") - if len(kept) == 0 { - delete(w.keptSince, key) - w.unfind(fkey) - return - } - if w.keptSince == nil { - w.keptSince = map[string]time.Time{} - } - since, ok := w.keptSince[key] - if !ok { - since = w.now() - w.keptSince[key] = since - } - words := strings.Join(kept, "; ") - w.find(Finding{Scope: ScopeModule, ID: id, Token: "kept", Kind: KindKeptAsRuns, Machine: node, Severity: Warning, - Source: "node-engine", Since: since, Summary: fmt.Sprintf("%s on %s is kept as it runs: %s", module, node, words), - Said: firstLine(words), Headline: fmt.Sprintf("%s kept as it runs on %s", module, node), - Needs: "a build of " + module + " whose manifest says it (tool-runner, core-credentials)"}) -} - -// ControlNodeModule is the module whose have names a control node: the bus module, so the control node is the bus node, -// the one stably reachable node every part depends on (novox/hq ADR 0590, issue 1126: the controller that named it ended). -// The same name the bus module's events are heard under (SlowConsumerSubject). -const ControlNodeModule = "nats" +// ---- S21, from the provisioner clients' reports ---- // Provision is a change of a key of mesh-provisioner's `provisions` (ADR 0570 §7): a provision in `waits` is a part of its // consumer waiting for the operator, said as `module...needs-operator` naming each waiting provision and @@ -713,24 +573,12 @@ func (w *Watchman) heard(node string) bool { func (w *Watchman) applyCheck(node string) { key := FindingKey(ScopeMachine, node, KindApplyFailing) type failure struct{ module, words string } - var applied, placed []failure + var placed []failure var since time.Time - for hk, at := range w.failedSince { - n, module, _, err := nodeengineclient.ParseHaveKey(hk) - if err != nil || n != node { - continue - } - applied = append(applied, failure{module, firstLine(orWords(w.haveState[hk].Detail, "no reason given"))}) - if since.IsZero() || at.Before(since) { - since = at - } - } - if len(applied) == 0 { - for module, r := range w.provFailed[node] { - placed = append(placed, failure{module, firstLine(orWords(r.Reason, "no reason given"))}) - if since.IsZero() || r.At.Before(since) { - since = r.At - } + for module, r := range w.provFailed[node] { + placed = append(placed, failure{module, firstLine(orWords(r.Reason, "no reason given"))}) + if since.IsZero() || r.At.Before(since) { + since = r.At } } if since.IsZero() || !w.heard(node) { @@ -745,7 +593,7 @@ func (w *Watchman) applyCheck(node string) { } return } - failed := append(applied, placed...) + failed := placed sort.Slice(failed, func(i, j int) bool { return failed[i].module < failed[j].module }) var names, words []string tool := "" @@ -761,19 +609,14 @@ func (w *Watchman) applyCheck(node string) { } f := Finding{Scope: ScopeMachine, ID: node, Kind: KindApplyFailing, Machine: node, Severity: Warning, Source: "S21", Since: since, Said: strings.Join(words, "; "), Headline: node + " is not applying its update"} - switch { - case len(applied) == 0: - f.Summary = fmt.Sprintf("%s's provisioner has failed to place the files of %s since %s (bound %s)", node, - strings.Join(names, ", "), stamp(since), ApplyEvery) - f.Explanation = fmt.Sprintf("The provisioner on %s has not been able to write the files of %s. The details say why.", - node, strings.Join(names, ", ")) - case tool != "": - f.Summary = fmt.Sprintf("%s: %s does not answer, so no package reaches it; everything else it was sent is applied, "+ + f.Summary = fmt.Sprintf("%s's provisioner has failed to place the files of %s since %s (bound %s)", node, + strings.Join(names, ", "), stamp(since), ApplyEvery) + f.Explanation = fmt.Sprintf("The provisioner on %s has not been able to write the files of %s. The details say why.", + node, strings.Join(names, ", ")) + if tool != "" { + f.Summary = fmt.Sprintf("%s: %s does not answer, so no package reaches it; everything else it was sent is placed, "+ "since %s (bound %s)", node, tool, stamp(since), ApplyEvery) f.Headline = node + " cannot install packages" - default: - f.Summary = fmt.Sprintf("%s has failed to apply %s since %s (bound %s)", node, strings.Join(names, ", "), stamp(since), ApplyEvery) - f.Explanation = fmt.Sprintf("%s has failed to apply its updates, and trying again has not helped. The details say what fails.", node) } w.find(f) } @@ -931,13 +774,12 @@ func orWords(s, otherwise string) string { // SignalRows are the rows the watchman holds, in the table's order, for `signals`. var SignalRows = []struct{ Row, Signal, Bound, Kind string }{ - {"S1", "a machine's presence (not while it said it sleeps or shuts down); a control node urgent after " + ControlNodeUrgentAfter.String(), SilentBound.String(), KindSilent}, + {"S1", "a machine's presence (not while it said it sleeps or shuts down)", SilentBound.String(), KindSilent}, {"S8", "a provider's failing word repeated", ProviderSaysAgainWithin.String(), KindProviderSilent}, {"S9", "the bus's advisories", "any occurrence; clears after " + AdvisoryQuiet.String() + " without another", "slow-consumer, connection-closed, max-deliveries, refused, consumer-lost"}, {"S10", "the self-check's checked", (3 * SelfCheckEvery).String(), KindSelfCheckSilent}, {"S11", "a machine's tool runner heartbeat", "3 × its interval (" + (3 * HeartbeatEvery).String() + " when it says none)", KindToolsSilent}, - {"S21", "a machine's have failed or its provisioner client's files not placed", ApplyEvery.String(), KindApplyFailing}, - {"health", "a have unhealthy in two writes in a row; urgent after " + UnhealthyUrgentAfter.String(), "two writes", KindUnhealthy}, + {"S21", "a machine's provisioner client's files not placed", ApplyEvery.String(), KindApplyFailing}, {"take-in", "a source revision or a delivery group the indexer holds back (ADR 0633 §5)", TakeInRevisionBound.String() + " a revision, " + TakeInGroupBound.String() + " a group", KindTakeInWaits}, {actingRow, "an acting part's wait held, or a put's, a put without hold that stands (the seats' wait-held, ended by wait-ended)", "the wait's own bound", KindActingPartLate + ", " + KindPutStands}, {storeRow, "a store of the delivery's parts filling or full (the seats' filling, full and room)", "filling from " + fmt.Sprintf("%.0f", FillingFrom*100) + " %, full urgent", KindStoreFilling}, diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index 8bcf9ea9..4213a57f 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch.go @@ -8,7 +8,6 @@ import ( "sync" "time" - "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient" "git.novox.be/novox/mesh-sdk/go/keys" ) @@ -64,10 +63,8 @@ type Watchman struct { presence map[string]presence pulses map[string]*pulse // live pulses, by pulse key stuck map[string]meshwatchdogsclient.Stuck - // keptSince is, by have key, when its have first said what the node keeps of the module as it runs (keptFinding). - keptSince map[string]time.Time - wants map[string]string // the fingerprint of each want that stands, by want key - haves map[string]string // the fingerprint each node runs of a module, by have key + wants map[string]string // the fingerprint of each want that stands, by want key + haves map[string]string // the fingerprint each node runs of a module, by have key, from the node-engine's running // starts is, by pulse key, the first start of a wait whose mark still stands (mirrorKept). starts map[string]startMark held map[string]bool // each want key held by a rule (ADR 0603 §8), mirrored as the hold's wait @@ -85,24 +82,15 @@ type Watchman struct { takeInCaught bool // away is each machine that said it sleeps or shuts down, until it says it is back. away map[string]bool - // haveState is each module's have as last read, unhealthy its writes in a row that said it failed, advised each - // advisory's count while its finding stands and advisedSince since when. - haveState map[string]nodeengine.HaveByNodeModuleValue - unhealthy map[string]int + // advised is each advisory's count while its finding stands and advisedSince since when. advised map[string]int advisedSince map[string]time.Time - // control is each node running the controller; failedSince each have's first failure still standing; provFailed each - // machine's provisioner client's failed reports; closures each bus closure counted, by finding key and identity. - control map[string]bool - failedSince map[string]time.Time - provFailed map[string]map[string]ProvReport - closures map[string]map[string]time.Time - recount func() + // provFailed is each machine's provisioner client's failed reports; closures each bus closure counted, by finding key and identity. + provFailed map[string]map[string]ProvReport + closures map[string]map[string]time.Time + recount func() // waiting is, by `.`, 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]nodeengine.HealthByNodeValue - 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 // progressBound is how long a running ask may say nothing of its progress; a test shortens it. @@ -137,9 +125,9 @@ func New(out Out, now func() time.Time) *Watchman { return &Watchman{out: out, now: now, counts: map[string]int{}, presence: map[string]presence{}, pulses: map[string]*pulse{}, stuck: map[string]meshwatchdogsclient.Stuck{}, wants: map[string]string{}, haves: map[string]string{}, held: map[string]bool{}, starts: map[string]startMark{}, bound: meshwatchdogsclient.PulseMoveBound, stepBound: StepBound, progressBound: ProgressBound, findings: map[string][]byte{}, seededFindings: map[string]bool{}, signals: map[string]signalDue{}, busCreated: map[string]bool{}, takeInHeld: map[string]Finding{}, pendingBranch: map[string]string{}, takenEntries: map[string][]string{}, recordMilestone: map[string]string{}, away: map[string]bool{}, - haveState: map[string]nodeengine.HaveByNodeModuleValue{}, 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{}, - health: map[string]nodeengine.HealthByNodeValue{}, tunnels: map[string]string{}, + advised: map[string]int{}, advisedSince: map[string]time.Time{}, + provFailed: map[string]map[string]ProvReport{}, closures: map[string]map[string]time.Time{}, waiting: map[string]map[string]string{}, + standing: map[string]map[string]bool{}, readWhole: map[string]bool{}, awaited: map[string]bool{}, heads: map[string]head{}, queued: map[string]map[string]time.Time{}, awaySaid: map[string]time.Time{}, opened: map[string]Finding{}} } diff --git a/modules/mesh-delivery-watchman/internal/watch/watch_test.go b/modules/mesh-delivery-watchman/internal/watch/watch_test.go index 3862f84b..0a8e05ea 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch_test.go @@ -8,8 +8,6 @@ import ( "testing" "time" - "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" - "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient" @@ -614,14 +612,8 @@ func wantValue(t *testing.T, fp string, removed bool) []byte { return b } -func haveValue(t *testing.T, fp string) []byte { - t.Helper() - b, err := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}) - if err != nil { - t.Fatal(err) - } - return b -} +// runningBody is the node-engine's `running` for a build of fingerprint fp (design 47 G §541). +func runningBody(fp string) []byte { return []byte(`{"fingerprint":"` + fp + `"}`) } var ( fpOld = meshcatalogclient.SourcePrefix2 + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" @@ -637,11 +629,11 @@ func TestAWantIsAPulseOwedByTheNodeAndTheHaveNamingItEndsIt(t *testing.T) { if !ok || p.Key != "want.ace.web" || !strings.Contains(p.Owes, "ace") || !strings.Contains(p.Owes, "web") { t.Fatalf("pulses = %v", o.pulses) } - w.Have("ace.web", haveValue(t, fpOld), Put) // the node still runs the old one + w.EngineRunning("ace", "web", runningBody(fpOld)) // the node still runs the old one if _, ok := o.pulses[pk]; !ok { t.Fatalf("a have of another fingerprint ended the pulse") } - w.Have("ace.web", haveValue(t, fpNew), Put) + w.EngineRunning("ace", "web", runningBody(fpNew)) if _, ok := o.pulses[pk]; ok { t.Errorf("the have naming the want's fingerprint left the pulse standing: %v", o.pulses) } @@ -653,7 +645,7 @@ func TestAWantIsAPulseOwedByTheNodeAndTheHaveNamingItEndsIt(t *testing.T) { func TestAHaveThatCameBeforeTheWantEndsItAtOnceAndARemovedWantHasNoPulse(t *testing.T) { w, o, c := setup() c.t = t0.Add(time.Second) - w.Have("ace.web", haveValue(t, fpNew), Put) + w.EngineRunning("ace", "web", runningBody(fpNew)) w.Want("want.ace.web", wantValue(t, fpNew, false), Put, t0) w.Want("want.ace.mail", wantValue(t, "", true), Put, t0) if len(o.pulses) != 0 { diff --git a/modules/mesh-delivery-watchman/manifest_test.go b/modules/mesh-delivery-watchman/manifest_test.go index 5597b29e..d78bcf2e 100644 --- a/modules/mesh-delivery-watchman/manifest_test.go +++ b/modules/mesh-delivery-watchman/manifest_test.go @@ -103,7 +103,7 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6 reads := slices.Clone(m.Reads) slices.Sort(reads) // The build seats' steps are no read (design 47 G §483, A-136): the watchman times an ask from the seats' step-progress. - if !reflect.DeepEqual(reads, []string{"mesh-host.health", "mesh-provisioner.provisions", "node-engine.have"}) { + if !reflect.DeepEqual(reads, []string{"mesh-provisioner.provisions"}) { t.Errorf("reads = %v", reads) } // Design 47 G §368: the version is MAJOR.MINOR and nothing else. @@ -130,7 +130,7 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6 t.Errorf("invokes = %v, want %v", m.Invokes, want) } // DLV-M5 its first views, 1585/T20 the indexer's take-in events, and DLV-W1 (ADR 0595 §4): the accepted verdicts and the bus module's created and refused, durably. - if want := []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.take-in.pending", "mesh-delivery-indexer.take-in.taken", "mesh-catalog.manifest-accepted", "mesh-catalog.manifest-refused", "mesh-delivery.source-moved", "mesh-delivery.filling", "mesh-delivery.full", "mesh-delivery.room", "mesh-delivery.record-opened", "mesh-delivery.record-ended", "mesh-catalog.filling", "mesh-catalog.full", "mesh-catalog.room", "mesh-wants.filling", "mesh-wants.full", "mesh-wants.room", "mesh-grants.filling", "mesh-grants.full", "mesh-grants.room", "node-build-go.ask-taken", "node-build-go.step-progress", "node-build-go.step-ended", "node-build-image.ask-taken", "node-build-image.step-progress", "node-build-image.step-ended", "node-build-node.ask-taken", "node-build-node.step-progress", "node-build-node.step-ended", "mesh-pipeline-runner.pipeline-started", "mesh-pipeline-runner.step-started", "mesh-pipeline-runner.step-ended", "mesh-pipeline-runner.wait-held", "mesh-pipeline-runner.wait-ended", "mesh-wants.wait-held", "mesh-wants.wait-ended", "mesh-grants.wait-held", "mesh-grants.wait-ended", "mesh-delivery-builder.wait-held", "mesh-delivery-builder.wait-ended", "node-engine.wait-held", "node-engine.wait-ended", "mesh-broker.created", "mesh-broker.refused", "mesh-conditions.findings-asked"}; !reflect.DeepEqual(m.Consumes, want) { + if want := []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.take-in.pending", "mesh-delivery-indexer.take-in.taken", "mesh-catalog.manifest-accepted", "mesh-catalog.manifest-refused", "mesh-delivery.source-moved", "mesh-delivery.filling", "mesh-delivery.full", "mesh-delivery.room", "mesh-delivery.record-opened", "mesh-delivery.record-ended", "mesh-catalog.filling", "mesh-catalog.full", "mesh-catalog.room", "mesh-wants.filling", "mesh-wants.full", "mesh-wants.room", "mesh-grants.filling", "mesh-grants.full", "mesh-grants.room", "node-build-go.ask-taken", "node-build-go.step-progress", "node-build-go.step-ended", "node-build-image.ask-taken", "node-build-image.step-progress", "node-build-image.step-ended", "node-build-node.ask-taken", "node-build-node.step-progress", "node-build-node.step-ended", "mesh-pipeline-runner.pipeline-started", "mesh-pipeline-runner.step-started", "mesh-pipeline-runner.step-ended", "mesh-pipeline-runner.wait-held", "mesh-pipeline-runner.wait-ended", "mesh-wants.wait-held", "mesh-wants.wait-ended", "mesh-grants.wait-held", "mesh-grants.wait-ended", "mesh-delivery-builder.wait-held", "mesh-delivery-builder.wait-ended", "node-engine.wait-held", "node-engine.wait-ended", "node-engine.running", "mesh-broker.created", "mesh-broker.refused", "mesh-conditions.findings-asked"}; !reflect.DeepEqual(m.Consumes, want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } // 1952/WM1: the watchman holds the seat mesh-watchdogs at mesh scope, its first seat (novox/hq ADR 0564 §1, §4). @@ -172,8 +172,6 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6 "presence": {watch.PresenceBucket, m.Module + "_" + m.State[0].Name}, "pulse": {watch.PulseBucket, m.Module + "_" + m.State[1].Name}, "stuck": {watch.StuckBucket, m.Module + "_" + m.State[2].Name}, - "have": {watch.HaveBucket, bucket(m.Reads, "node-engine.have")}, - "health": {watch.NodeHealthBucket, bucket(m.Reads, "mesh-host.health")}, "wants": {watch.AssignerWantsBucket, bucket(m.Consumes, "mesh-delivery-assigner.views.wants")}, "provisions": {watch.ProvisionsBucket, bucket(m.Reads, "mesh-provisioner.provisions")}, "assigner": {watch.AssignerModule + ".assigner_read_wants", entry(m.Invokes, "mesh-delivery-assigner.assigner_read_wants")}, diff --git a/modules/mesh-delivery-watchman/module.json b/modules/mesh-delivery-watchman/module.json index eee55b48..e113c701 100644 --- a/modules/mesh-delivery-watchman/module.json +++ b/modules/mesh-delivery-watchman/module.json @@ -16,9 +16,7 @@ } ], "reads": [ - "mesh-host.health", - "mesh-provisioner.provisions", - "node-engine.have" + "mesh-provisioner.provisions" ], "invokes": [ "mesh-delivery-assigner.assigner_read_wants", @@ -70,6 +68,7 @@ "mesh-delivery-builder.wait-ended", "node-engine.wait-held", "node-engine.wait-ended", + "node-engine.running", "mesh-broker.created", "mesh-broker.refused", "mesh-conditions.findings-asked"