novox/mesh-catalog
9
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'feat/47-G-findings4b' into feat/47-G-dev4-join

Browse Source
This commit is contained in:
jschoubben
2026-11-11 22:25:53 +01:00
parent 9aae6244d5 cd24677756
commit 469763547c
18 changed files with 546 additions and 288 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-healers/cmd/mesh-healers/findings.go
+96
View File
@@ -0,0 +1,96 @@
package main
// The healers' findings are the generated finding of the seat mesh-conditions, written through mesh-sdk's finding writer
// (design 47 G §461, §689).
import (
"context"
"encoding/json"
"fmt"
"sort"
"time"
meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
"git.novox.be/novox/mesh-sdk/go/keys"
"git.novox.be/novox/mesh-sdk/go/views"
)
// Finding, and the writer's other three bodies, are the seat mesh-conditions' own types.
type (
Finding = meshconditions.Finding
FindingCleared = meshconditions.FindingCleared
FindingsStated = meshconditions.FindingsStated
FindingsAsked = meshconditions.FindingsAskedEvent
FindingsWriter = views.Writer[Finding, FindingCleared, FindingsStated, FindingsAsked]
)
// finding is f about scope and id: it sets the key's scope and id and the subject's scope and id equal (design 47 G A-14).
func finding(scope, id string, f Finding) Finding {
f.Scope, f.ID = scope, keys.EscapeName(id)
f.Subject = meshconditions.Subject{Scope: scope, ID: f.ID}
return f
}
// evidence is the one observation a finding keeps as evidence; its moment is the finding's own, so that a finding that says
// the same is not written again.
func evidence(s string) []meshconditions.Evidence { return []meshconditions.Evidence{{Said: s}} }
// keyOf is where f is kept: `<scope>.<id>.<kind>`.
func keyOf(f Finding) string { return views.FindingKey(f.Scope, f.ID, f.Kind) }
// PendingBound is how long a finding the healers cannot judge yet stays open (design 47 G §384).
const PendingBound = 5 * time.Minute
// stored are the findings in the state `findings`, by key.
func (h *Healers) stored() (map[string]Finding, error) {
keys, err := h.Store.Keys(FindingsState)
if err != nil {
return nil, err
}
out := map[string]Finding{}
for _, k := range keys {
raw, found, err := h.Store.Get(FindingsState, k)
if err != nil {
return nil, err
}
if !found {
continue
}
var f Finding
if json.Unmarshal(raw, &f) == nil {
out[k] = f
}
}
return out, nil
}
// Observe is the writer's observe (design 47 G §384): every finding the healers observe now, judged from what they
// keep, and the stored findings they cannot judge yet.
func (h *Healers) Observe(ctx context.Context) (views.Observed[Finding], error) {
stored, err := h.stored()
if err != nil {
return views.Observed[Finding]{}, fmt.Errorf("its own findings cannot be read: %w", err)
}
now := h.Now()
var out views.Observed[Finding]
pend := func(source string) {
for k, f := range stored {
if f.Source == source {
out.Pending = append(out.Pending, views.PendingKey{Key: k, Since: f.Since, Bound: PendingBound.String()})
}
}
}
if all, err := h.attempts(); err != nil {
out.Findings = append(out.Findings, blindFinding(err))
pend("brake")
} else if braked, said := h.brakeHolds(actsWithin(all, now.Add(-BrakeWindow)), now); braked {
out.Findings = append(out.Findings, brakeFinding(actsWithin(all, now.Add(-BrakeWindow)), said))
}
if wanted, _, seeded := h.wantedNow(now); seeded {
out.Findings = append(out.Findings, wanted...)
} else {
pend("S15")
}
sort.Slice(out.Pending, func(i, j int) bool { return out.Pending[i].Key < out.Pending[j].Key })
return out, nil
}
modules/mesh-healers/cmd/mesh-healers/healers.go
+43 -47
View File
@@ -14,6 +14,7 @@ package main
// bucket's age); nothing writes there any more.
import (
"context"
"encoding/json"
"errors"
"fmt"
@@ -29,6 +30,7 @@ import (
handacts "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhandacts"
meshhealers "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhealers"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/views"
)
// Module and Seat are this module and the seat it holds once for the mesh.
@@ -181,12 +183,19 @@ type Healers struct {
Now func() time.Time
Logf func(format string, args ...any)
Registry []Healer
// Writer is the finding writer over this module's state `findings` (views.NewStdioWriter; design 47 G A-16), whose
// observe is Observe.
Writer *FindingsWriter
mu sync.Mutex
// declined is why each condition was last passed over, said once.
declined map[string]string
// acts are the hand-act log as watched, for S15.
acts map[string]HandAct
// actsMu guards acts and seeded, which the writer's observe reads while it holds the writer's lock; it is never held
// while a finding is written.
actsMu sync.Mutex
// acts are the hand-act log as watched, for S15; seeded is whether the seat's answer at the start was taken.
acts map[string]HandAct
seeded bool
}
// actingWhy is why the healers act: what the tool `healers` says.
@@ -240,7 +249,7 @@ func (h *Healers) Judge(c Condition, by string) (string, error) {
h.see("blind", nil)
acts := actsWithin(all, now.Add(-BrakeWindow))
if braked, said := h.brakeHolds(acts, now); braked {
h.see("brake", []meshconditionsclient.Observation{brakeFinding(acts, said)})
h.see("brake", []Finding{brakeFinding(acts, said)})
return "every healer has stopped: " + said, nil
}
h.see("brake", nil)
@@ -483,10 +492,11 @@ func (h *Healers) brakeHolds(acts []Attempt, now time.Time) (bool, string) {
return false, said
}
var brakeKey = meshconditionsclient.Key(meshconditionsclient.ScopeMesh, "healers", "braked")
// brakeKey is where the brake's finding is kept.
var brakeKey = views.FindingKey(meshconditionsclient.ScopeMesh, "healers", "healers-braked")
// brakeFinding is the brake, as the urgent finding that says it.
func brakeFinding(acts []Attempt, said string) meshconditionsclient.Observation {
func brakeFinding(acts []Attempt, said string) Finding {
counts := map[string]int{}
for _, a := range acts {
counts[a.Healer+" on "+a.Condition]++
@@ -496,67 +506,53 @@ func brakeFinding(acts []Attempt, said string) meshconditionsclient.Observation
most = append(most, fmt.Sprintf("%s ×%d", what, n))
}
sort.Strings(most)
return meshconditionsclient.Observation{Kind: "healers-braked", Ref: "braked", Severity: meshconditionsclient.Urgent, Scope: meshconditionsclient.ScopeMesh, ID: "healers",
return finding(meshconditionsclient.ScopeMesh, "healers", Finding{Kind: "healers-braked", Severity: meshconditionsclient.Urgent,
Source: "brake",
Summary: "every healer has stopped: the healers acted more often in an hour than the mesh allows, which is a " +
"healer looping, not repairing — " + said,
Said: strings.Join(most, ", "),
Evidence: evidence(strings.Join(most, ", ")),
Headline: "All automatic repairs stopped",
Needs: "read the details and decide whether the repairs may run again.",
Explanation: "The healers (the mesh's automatic repairs) acted more often in an hour than allowed, which looks " +
"like a loop, so all of them stopped.",
Resolved: "Automatic repairs run again"}
"like a loop, so all of them stopped."})
}
// blindFinding is the healers' own blindness: they cannot read what they did.
func blindFinding(err error) Finding {
return finding(meshconditionsclient.ScopeProbe, "healers", Finding{Kind: "probe-failed", Severity: meshconditionsclient.Warning,
Source: "blind",
Summary: "the healers cannot read what they did, so they count no budget and act on nothing",
Evidence: evidence(firstLine(err.Error())), Headline: "A check of the mesh could not run",
Explanation: "The automatic repairs cannot read what they did, so they act on nothing until they can."})
}
// blind writes the healers' own blindness: they cannot read what they did.
func (h *Healers) blind(err error) {
h.see("blind", []meshconditionsclient.Observation{{Kind: "probe-failed", Ref: "failed", Severity: meshconditionsclient.Warning,
Scope: meshconditionsclient.ScopeProbe, ID: "healers",
Summary: "the healers cannot read what they did, so they count no budget and act on nothing",
Said: firstLine(err.Error()), Headline: "A check of the mesh could not run",
Explanation: "The automatic repairs cannot read what they did, so they act on nothing until they can.",
Resolved: "The automatic repairs read what they did again"}})
}
func (h *Healers) blind(err error) { h.see("blind", []Finding{blindFinding(err)}) }
// see writes the healers' own findings of one source, each only when it changes, and deletes those of the source it
// no longer sees (ADR 0564 §2).
func (h *Healers) see(source string, found []meshconditionsclient.Observation) {
now := h.Now()
want := map[string]meshconditions.FindingsByKeyValue{}
for _, o := range found {
o.Source = source
want[o.Key()] = o.Finding(Module, "", now)
// see writes the healers' own findings of one source through the finding writer, each only when it changes, and clears
// those of the source it no longer sees (ADR 0564 §2; design 47 G §461).
func (h *Healers) see(source string, found []Finding) {
ctx := context.Background()
want := map[string]Finding{}
for _, f := range found {
f.Source = source
want[keyOf(f)] = f
}
keys, err := h.Store.Keys(FindingsState)
stored, err := h.stored()
if err != nil {
h.Logf("[%s] its own findings cannot be read, so none is written or cleared: %v", Module, err)
return
}
for _, k := range keys {
raw, found, err := h.Store.Get(FindingsState, k)
if err != nil || !found {
for k, was := range stored {
if _, still := want[k]; still || was.Source != source {
continue
}
var was meshconditions.FindingsByKeyValue
err = json.Unmarshal(raw, &was)
if err != nil || was.Source != source {
continue
}
if f, still := want[k]; still {
if meshconditionsclient.Says(was) == meshconditionsclient.Says(f) {
delete(want, k)
}
continue
}
if err := h.Store.Delete(FindingsState, k); err != nil {
h.Logf("[%s] the finding %s no longer seen is not deleted: %v", Module, k, err)
if err := h.Writer.Clear(ctx, was.Scope, was.ID, was.Kind, "the healers no longer see it"); err != nil {
h.Logf("[%s] the finding %s no longer seen is not cleared: %v", Module, k, err)
}
}
for k, f := range want {
raw, err := json.Marshal(f)
if err == nil {
err = h.Store.Put(FindingsState, k, json.RawMessage(raw))
}
if err != nil {
if err := h.Writer.Write(ctx, f); err != nil {
h.Logf("[%s] the finding %s is not written: %v", Module, k, err)
}
}
modules/mesh-healers/cmd/mesh-healers/healers_test.go
+43 -4
View File
@@ -1,6 +1,7 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
@@ -15,6 +16,8 @@ import (
meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
handacts "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhandacts"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/views"
)
type memStore struct {
@@ -131,6 +134,8 @@ type world struct {
now time.Time
asked []string
said []Acted
// events are the subjects the finding writer said its events on.
events []string
}
func newWorld() *world {
@@ -155,8 +160,35 @@ func (w *world) fresh() {
return nil
},
}
wr, err := views.NewWriter(views.WriterConfig[Finding, FindingCleared, FindingsStated, FindingsAsked]{
Module: Module, Node: "ace", Asked: "mesh.seat.mesh-conditions.event.findings-asked", State: findingState{w.store}, Observe: w.h.Observe, Now: func() time.Time { return w.now },
Publish: func(_ context.Context, subject string, _ []byte, _ string) error {
w.events = append(w.events, subject)
return nil
},
Listen: func(context.Context, string, func([]byte)) (func(), error) { return func() {}, nil }})
if err != nil {
panic(err)
}
w.h.Writer = wr
}
// findingState is the writer's view of the store's state `findings`.
type findingState struct{ s *memStore }
func (f findingState) Get(key string) (*stdio.StateEntry, error) {
raw, ok, err := f.s.Get(FindingsState, key)
if err != nil || !ok {
return nil, err
}
return &stdio.StateEntry{Key: key, Value: raw}, nil
}
func (f findingState) Put(key string, value any) (uint64, error) {
return 0, f.s.Put(FindingsState, key, value)
}
func (f findingState) Delete(key string) error { return f.s.Delete(FindingsState, key) }
func (f findingState) Keys() ([]string, error) { return f.s.Keys(FindingsState) }
// drill is a healer of the tests alone: no healer acts here since H3 ended with the controller (issue 1126), and the
// budget, the settle, the escalation and the brake are every healer's, judged on it.
var drill = Healer{ID: "HD", Kinds: []string{"drill-broken"}, Condition: "a drill", Repair: "ask the drill again",
@@ -288,11 +320,18 @@ func TestBudgetSettleEscalationAndBrake(t *testing.T) {
t.Errorf("past the brake: %q, asked %v", said, b.asked)
}
raw, braked, _ := b.store.Get(FindingsState, brakeKey)
var f meshconditions.FindingsByKeyValue
var f Finding
_ = json.Unmarshal(raw, &f)
if !braked || f.Severity != meshconditionsclient.Urgent || f.Kind != "healers-braked" || f.Observer != Module {
if !braked || f.Severity != meshconditionsclient.Urgent || f.Kind != "healers-braked" || f.Observer != Module || f.Since.IsZero() {
t.Errorf("the brake is written as %s", raw)
}
if !slices.Contains(b.events, "mesh.stream.mesh-healers.event.finding-written.ace") {
t.Errorf("the brake was not said as finding-written on the writer's subject: %v", b.events)
}
// Judged anew at a start, from what the healers keep: the brake still holds, and states the same finding.
if obs, err := b.h.Observe(context.Background()); err != nil || len(obs.Findings) != 1 || obs.Findings[0].Kind != "healers-braked" {
t.Errorf("a statement of the brake: %+v, %v", obs, err)
}
b.now = b.now.Add(55 * time.Minute) // fewer than twelve within the hour, but the brake is said: held
if said, _ := b.h.Judge(lost("mail"), ""); !strings.Contains(said, "held until") {
t.Errorf("the brake let go before an hour after the last act: %q", said)
@@ -310,7 +349,7 @@ func TestBudgetSettleEscalationAndBrake(t *testing.T) {
if _, err := z.h.Judge(lost("mail"), ""); err == nil || len(z.asked) != 0 {
t.Errorf("blind, the healers acted: %v", err)
}
if keys := z.store.keys(FindingsState); !slices.Equal(keys, []string{"probe.healers.failed"}) {
if keys := z.store.keys(FindingsState); !slices.Equal(keys, []string{"probe.healers.probe-failed"}) {
t.Errorf("blind, the findings are %v", keys)
}
}
@@ -381,7 +420,7 @@ func TestACauseRepairedByHandTwiceIsAHealerWanted(t *testing.T) {
"mesh-vault.secret_vault_export"} {
act(fmt.Sprintf("n%d", i), verb, "operator", time.Hour)
}
want := []string{"mesh.hand-acts.consumer-behind.healer-wanted"}
want := []string{"mesh.hand-acts=2econsumer-behind.healer-wanted"}
if keys := w.store.keys(FindingsState); !slices.Equal(keys, want) {
t.Fatalf("findings %v, want %v", keys, want)
}
modules/mesh-healers/cmd/mesh-healers/main.go
+13
View File
@@ -7,13 +7,16 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"os"
"time"
meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/views"
)
func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) }
@@ -55,6 +58,13 @@ func main() {
h := &Healers{Store: runtimeStore{}, Registry: Registry(), Now: time.Now, Logf: logf,
Ask: func(key string, body any) (json.RawMessage, error) { return stdio.Ask(key, body) },
Publish: stdio.SeatPublish}
w, err := views.NewStdioWriter(views.WriterConfig[Finding, FindingCleared, FindingsStated, FindingsAsked]{
Module: Module, State: stdio.State(FindingsState), Observe: h.Observe, Asked: meshconditions.SubjectFindingsAsked, Logf: logf})
if err != nil {
logf("%v", err)
os.Exit(1)
}
h.Writer = w
// At start, a settle that ended while no instance watched is judged (Rearm); the waits still running are keys on
// the bus, heard on their expiry.
go h.Rearm()
@@ -66,6 +76,9 @@ func main() {
// durable consumer (the controller's bucket is no longer read, issue 774).
go func() {
<-stdio.Served()
if err := h.Writer.Start(context.Background()); err != nil {
logf("[%s] the finding writer did not start: %v", Module, err)
}
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
answer, err := stdio.Ask("seat:mesh-hand-acts.hand-acts", map[string]any{"days": int(Within.Hours() / 24)})
if err == nil {
modules/mesh-healers/cmd/mesh-healers/wanted.go
+29 -21
View File
@@ -82,12 +82,12 @@ func (h *Healers) ActRecorded(body []byte) {
if json.Unmarshal(body, &a) != nil || a.ID == "" {
return
}
h.mu.Lock()
h.actsMu.Lock()
if h.acts == nil {
h.acts = map[string]HandAct{}
}
h.acts[handacts.Seat+":"+a.ID] = a
h.mu.Unlock()
h.actsMu.Unlock()
h.Wanted()
}
@@ -106,6 +106,9 @@ func (h *Healers) Seeded(answer []byte) (int, error) {
for _, raw := range l.Acts {
h.ActRecorded(raw)
}
h.actsMu.Lock()
h.seeded = true
h.actsMu.Unlock()
// A partial answer (the seat bounds it, issue 1966) seeds all the same: the newest acts are what a fortnight's
// count needs most, and the older ones are heard no more, so asking again would only repeat it.
if l.Complete != nil && !*l.Complete {
@@ -115,11 +118,25 @@ func (h *Healers) Seeded(answer []byte) (int, error) {
}
// Wanted counts every cause repaired by hand within the fortnight and writes a finding for each counted twice or more,
// deleting the rest; its wait `fortnight` expires when the oldest counted act of a wanted cause leaves the fortnight.
func (h *Healers) Wanted() []meshconditionsclient.Observation {
h.mu.Lock()
defer h.mu.Unlock()
// clearing the rest; its wait `fortnight` expires when the oldest counted act of a wanted cause leaves the fortnight.
func (h *Healers) Wanted() []Finding {
now := h.Now()
found, next, _ := h.wantedNow(now)
h.see("S15", found)
if !next.IsZero() {
ttl := next.Sub(now).Round(time.Second) + time.Second
if err := h.Store.PutTTL(WakesState, fortnightKey, map[string]any{"at": next.UTC()}, ttl); err != nil {
h.Logf("[%s] the fortnight's end is not written as a wait, so it is counted at the next act: %v", Module, err)
}
}
return found
}
// wantedNow is the finding of every cause repaired by hand twice or more within the fortnight, and when the oldest act
// of a wanted cause leaves it; seeded says the seat's acts of the start were taken, so that none is judged before.
func (h *Healers) wantedNow(now time.Time) (found []Finding, next time.Time, seeded bool) {
h.actsMu.Lock()
defer h.actsMu.Unlock()
byCause := map[string][]HandAct{}
for _, a := range h.acts {
if personsDecision(a) || now.Sub(a.At) > Within {
@@ -134,8 +151,6 @@ func (h *Healers) Wanted() []meshconditionsclient.Observation {
}
}
sort.Strings(causes)
var found []meshconditionsclient.Observation
var next time.Time
for _, cause := range causes {
acts := byCause[cause]
sort.Slice(acts, func(i, j int) bool { return acts[i].At.Before(acts[j].At) })
@@ -153,27 +168,20 @@ func (h *Healers) Wanted() []meshconditionsclient.Observation {
}
}
words := strings.NewReplacer(".", " ", "_", " ").Replace(cause)
found = append(found, meshconditionsclient.Observation{Kind: "healer-wanted", Ref: "healer-wanted", Severity: meshconditionsclient.Warning,
Scope: meshconditionsclient.ScopeMesh, ID: "hand-acts." + cause,
f := finding(meshconditionsclient.ScopeMesh, "hand-acts."+cause, Finding{Kind: "healer-wanted", Severity: meshconditionsclient.Warning,
Source: "S15",
Summary: fmt.Sprintf("%q was repaired by hand %d times in %d days, the last by %s: %s", cause, len(acts),
int(Within.Hours()/24), newest.By, wanted),
Said: strings.Join(lines, "; "),
Evidence: evidence(strings.Join(lines, "; ")),
Headline: capital(words) + " keeps being fixed by hand",
Explanation: fmt.Sprintf("A person repaired %s by hand %d times in %d days, so an automatic repair is wanted "+
"for it. Nothing is broken now.", words, len(acts), int(Within.Hours()/24)),
Resolved: "Resolved: no more hand repairs of " + words})
"for it. Nothing is broken now.", words, len(acts), int(Within.Hours()/24))})
found = append(found, f)
if leaves := acts[0].At.Add(Within); next.IsZero() || leaves.Before(next) {
next = leaves
}
}
h.see("S15", found)
if !next.IsZero() {
ttl := next.Sub(now).Round(time.Second) + time.Second
if err := h.Store.PutTTL(WakesState, fortnightKey, map[string]any{"at": next.UTC()}, ttl); err != nil {
h.Logf("[%s] the fortnight's end is not written as a wait, so it is counted at the next act: %v", Module, err)
}
}
return found
return found, next, h.seeded
}
func capital(s string) string {
modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go
+9
View File
@@ -360,6 +360,10 @@ func TestTheHealersWholeHearTheKeeperAndPassOverWhatEnded(t *testing.T) {
if err != nil {
t.Fatal(err)
}
stated, err := nc.SubscribeSync("mesh.mod." + Module + ".event.findings-stated")
if err != nil {
t.Fatal(err)
}
cmd := exec.CommandContext(ctx, bin)
cmd.Env = append(os.Environ(), "MESH_NODE=ace", "MESH_MODULE="+Module)
stdin, _ := cmd.StdinPipe()
@@ -384,6 +388,11 @@ func TestTheHealersWholeHearTheKeeperAndPassOverWhatEnded(t *testing.T) {
// The runtime's consumer holds an event until the bundle subscribes: delivered once it hears them.
eventually(t, "the healers hearing the keeper's conditions", func() bool { return p.hears(RaisedEvent) && p.hears(ChangedEvent) })
// The writer hears the keeper's asks for findings, and states at its start the findings it observes: none.
eventually(t, "the writer hearing findings-asked", func() bool { return p.hears("mesh-conditions.findings-asked") })
if msg, err := stated.NextMsg(10 * time.Second); err != nil || !strings.Contains(string(msg.Data), `"findings":[]`) {
t.Errorf("findings-stated at the start: %v", err)
}
// What ended with the controller is passed over: no act begun, nothing asked, nothing said.
p.deliver(RaisedEvent, raised("bus.EVENTS.records.missing", "records"))
if msg, err := acted.NextMsg(3 * time.Second); err == nil {
modules/mesh-self-check/cmd/mesh-self-check/findings.go
+52
View File
@@ -0,0 +1,52 @@
package main
// The self-check's findings are the generated finding of the seat mesh-conditions, written through mesh-sdk's finding
// writer (design 47 G §461, §689); the probes build them with these two helpers.
import (
"strings"
meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
"git.novox.be/novox/mesh-sdk/go/keys"
"git.novox.be/novox/mesh-sdk/go/views"
)
// Finding, and the writer's other three bodies, are the seat mesh-conditions' own types.
type (
Finding = meshconditions.Finding
FindingCleared = meshconditions.FindingCleared
FindingsStated = meshconditions.FindingsStated
FindingsAsked = meshconditions.FindingsAskedEvent
FindingsWriter = views.Writer[Finding, FindingCleared, FindingsStated, FindingsAsked]
)
// idOf is a finding's id: the id of what it is about, and a ref that tells findings of one kind on one id apart after it
// (`<id>.<ref>`, each escaped as a name of a key is; design 47 G A-14).
func idOf(id, ref string) string {
if ref == "" {
return keys.EscapeName(id)
}
return keys.EscapeName(id) + "." + keys.EscapeName(ref)
}
// finding is f about scope and id (with ref when findings of one kind on one id need telling apart), on machine: it sets
// the key's scope and id and the subject's scope and id equal (design 47 G A-14).
func finding(scope, id, ref, machine string, f Finding) Finding {
f.Scope, f.ID = scope, idOf(id, ref)
f.Subject = meshconditions.Subject{Scope: scope, ID: f.ID, Machine: machine}
return f
}
// said is the one observation a finding keeps as evidence; its moment is the finding's own (`at`), so that a finding
// that says the same is not written again.
func said(s string) []meshconditions.Evidence {
return []meshconditions.Evidence{{Said: strings.TrimSpace(s)}}
}
// keyOf is where f is kept: `<scope>.<id>.<kind>`.
func keyOf(f Finding) string { return views.FindingKey(f.Scope, f.ID, f.Kind) }
// findingKey is the key of the finding of kind about scope and id with ref, as finding makes it.
func findingKey(scope, id, ref, kind string) string {
return views.FindingKey(scope, idOf(id, ref), kind)
}
modules/mesh-self-check/cmd/mesh-self-check/main.go
+16 -5
View File
@@ -14,10 +14,10 @@ import (
placing "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
"git.novox.be/novox/mesh-sdk/go/views"
)
@@ -65,18 +65,26 @@ func (r runtimeStore) Watch(state string, handle func(key string, value []byte,
// NewChecker is the self-check over the runtime.
func NewChecker() *Checker {
store := runtimeStore{views: newViews()}
return &Checker{Store: store, Registry: Registry(), Now: time.Now, Logf: logf, Host: os.Getenv("MESH_NODE"),
c := &Checker{Registry: Registry(), Now: time.Now, Logf: logf, Host: os.Getenv("MESH_NODE"),
Enabled: enabledProbes,
Ask: func(key string, body any) (json.RawMessage, error) { return stdio.Ask(key, body) },
Publish: stdio.SeatPublish}
w, err := views.NewStdioWriter(views.WriterConfig[Finding, FindingCleared, FindingsStated, FindingsAsked]{
Module: Module, State: stdio.State(FindingsState), Observe: c.Observe, Asked: meshconditions.SubjectFindingsAsked, Logf: logf})
if err != nil {
logf("%v", err)
os.Exit(1)
}
c.Writer = w
c.Store = runtimeStore{views: newViews(w)}
return c
}
// newViews is the self-check's pictures of the indexer's and the assigner's states it probes (ADR 0615): each asked once of
// its owner's read tool, then followed by the owner's events. A first view unanswered past its bound is a finding of the
// self-check's own (first-view-unanswered), written into its state findings.
func newViews() *views.Reader {
r := &views.Reader{Module: Module, Logf: logf, Findings: meshconditionsclient.FindingWriter{State: stdio.State("findings")}, Acts: meshhandactsclient.Recorder{Ask: stdio.Ask}}
func newViews(w *FindingsWriter) *views.Reader {
r := &views.Reader{Module: Module, Logf: logf, Findings: w, Acts: meshhandactsclient.Recorder{Ask: stdio.Ask}}
r.Follow(assignerModule, placing.StateAssignments, AssignmentsTool)
r.Follow(assignerModule, placing.StateSeatHolders, SeatHoldersTool)
r.Follow(indexerModule, "modules", ModulesTool)
@@ -90,6 +98,9 @@ func main() {
<-stdio.Served()
c.Store.(runtimeStore).views.Start(context.Background())
c.WatchKept(context.Background())
if err := c.Writer.Start(context.Background()); err != nil {
logf("the finding writer did not start: %v", err)
}
c.Keep(context.Background())
}()
if err := stdio.Serve("", append(Tools(c), c.Store.(runtimeStore).views.Tools("mesh_self_check")...)); err != nil {
modules/mesh-self-check/cmd/mesh-self-check/probes.go
+22 -24
View File
@@ -140,7 +140,7 @@ func (c *Checker) streams(stream string) ([]streamRow, error) {
}
// probeStreams is D7: every stream the mesh defines exists as defined.
func probeStreams(ctx context.Context, c *Checker) ([]meshconditionsclient.Observation, error) {
func probeStreams(ctx context.Context, c *Checker) ([]Finding, error) {
rows, err := c.streams("")
if err != nil {
return nil, err
@@ -149,19 +149,19 @@ func probeStreams(ctx context.Context, c *Checker) ([]meshconditionsclient.Obser
for _, r := range rows {
held[r.Name] = r
}
var out []meshconditionsclient.Observation
var out []Finding
for _, s := range buscompose.MeshStreams() {
have, ok := held[s.Name]
if !ok {
out = append(out, meshconditionsclient.Observation{Ref: "stream-missing", Severity: meshconditionsclient.Urgent, Scope: meshconditionsclient.ScopeBus, ID: s.Name,
Summary: fmt.Sprintf("the stream %s is not on the bus: %s", s.Name, firstLine(s.Why)),
Said: "no stream " + s.Name})
out = append(out, finding(meshconditionsclient.ScopeBus, s.Name, "stream-missing", "", Finding{Kind: "stream-wrong", Severity: meshconditionsclient.Urgent,
Summary: fmt.Sprintf("the stream %s is not on the bus: %s", s.Name, firstLine(s.Why)),
Evidence: said("no stream " + s.Name)}))
continue
}
if differs := streamDiffers(s, have); differs != "" {
out = append(out, meshconditionsclient.Observation{Ref: "stream-redefined", Severity: meshconditionsclient.Warning, Scope: meshconditionsclient.ScopeBus, ID: s.Name,
Summary: fmt.Sprintf("the stream %s is not as the mesh defines it: %s", s.Name, differs),
Said: differs})
out = append(out, finding(meshconditionsclient.ScopeBus, s.Name, "stream-redefined", "", Finding{Kind: "stream-wrong", Severity: meshconditionsclient.Warning,
Summary: fmt.Sprintf("the stream %s is not as the mesh defines it: %s", s.Name, differs),
Evidence: said(differs)}))
}
}
return sorted(out), nil
@@ -213,7 +213,7 @@ type closedAnswer struct {
}
// probeReconnects is D15: each user of the bus whose connections were dropped more than twelve times an hour.
func probeReconnects(_ context.Context, c *Checker) ([]meshconditionsclient.Observation, error) {
func probeReconnects(_ context.Context, c *Checker) ([]Finding, error) {
raw, err := c.Ask(ClosedTool, map[string]any{"hours": 1})
if err != nil {
return nil, fmt.Errorf("the bus module did not answer %s: %w", ClosedTool, err)
@@ -226,7 +226,7 @@ func probeReconnects(_ context.Context, c *Checker) ([]meshconditionsclient.Obse
if hours <= 0 {
hours = 1
}
var out []meshconditionsclient.Observation
var out []Finding
for _, u := range read.Users {
if u.User == "" || strings.HasPrefix(u.User, "(") || float64(u.Dropped)/hours <= reconnectMax {
continue // refused before it logged in is nobody's reconnect loop
@@ -254,16 +254,14 @@ func probeReconnects(_ context.Context, c *Checker) ([]meshconditionsclient.Obse
partial = " (at least: the server's record of closed connections does not reach back the whole hour)"
}
who, machine := busUserWords(u.User)
out = append(out, meshconditionsclient.Observation{Kind: "bus-reconnects", Severity: meshconditionsclient.Warning,
Scope: meshconditionsclient.ScopeBus, ID: u.User, Machine: machine,
out = append(out, finding(meshconditionsclient.ScopeBus, u.User, "", machine, Finding{Kind: "bus-reconnects", Severity: meshconditionsclient.Warning,
Summary: fmt.Sprintf("the bus dropped %s's connection %d times in the last hour%s, more than %d: a client "+
"reconnecting in a loop", u.User, u.Dropped, partial, reconnectMax),
Said: fmt.Sprintf("%d of %d closed connections dropped in %.0f h; by name %s; why %s", u.Dropped, u.Closed,
hours, strings.Join(names, ", "), strings.Join(why, ", ")),
Evidence: said(fmt.Sprintf("%d of %d closed connections dropped in %.0f h; by name %s; why %s", u.Dropped, u.Closed,
hours, strings.Join(names, ", "), strings.Join(why, ", "))),
Headline: capital(who) + " keeps losing the bus",
Explanation: capital(fmt.Sprintf("%s lost its connection to the bus %d times in the last hour and connected "+
"again each time. While it reconnects, what it says and what it is asked waits.", who, u.Dropped)),
Resolved: "Resolved: " + who + " stays connected"})
"again each time. While it reconnects, what it says and what it is asked waits.", who, u.Dropped))}))
}
return sorted(out), nil
}
@@ -290,12 +288,12 @@ func busUserWords(user string) (string, string) {
// probeWatchdogs is DW: the watchdogs answer. The self-check and the watchdogs watch each other: S10, the
// watchman's pulse on `checked`, is the other half.
func probeWatchdogs(_ context.Context, c *Checker) ([]meshconditionsclient.Observation, error) {
func probeWatchdogs(_ context.Context, c *Checker) ([]Finding, error) {
if _, err := c.Ask(SignalsVerb, map[string]any{}); err != nil {
return []meshconditionsclient.Observation{{Ref: "silent", Severity: meshconditionsclient.Urgent, Scope: meshconditionsclient.ScopeCore, ID: "watchdogs",
return []Finding{finding(meshconditionsclient.ScopeCore, "watchdogs", "", "", Finding{Kind: "watchdogs-silent", Severity: meshconditionsclient.Urgent,
Summary: "the watchdogs do not answer: the watchman holding mesh-watchdogs did not answer signals, so no " +
"late signal is being said",
Said: firstLine(err.Error())}}, nil
Evidence: said(firstLine(err.Error()))})}, nil
}
return nil, nil
}
@@ -304,7 +302,7 @@ func probeWatchdogs(_ context.Context, c *Checker) ([]meshconditionsclient.Obser
// machine and back. The mesh defines no durable consumer of its own since the controller ended: the account consumers
// are the grants part's to make and judge (issue 1126). The question to the bus module is that request: an answer is its round trip,
// and no answer is no pass.
func probeBusHealth(ctx context.Context, c *Checker) ([]meshconditionsclient.Observation, error) {
func probeBusHealth(ctx context.Context, c *Checker) ([]Finding, error) {
rows, err := c.streams("")
if err != nil {
return nil, fmt.Errorf("no request crossed the bus to its module and came back: %w", err)
@@ -322,13 +320,13 @@ func probeBusHealth(ctx context.Context, c *Checker) ([]meshconditionsclient.Obs
if len(problems) == 0 {
return nil, nil
}
return []meshconditionsclient.Observation{{Ref: "unhealthy", Severity: meshconditionsclient.Urgent, Scope: meshconditionsclient.ScopeBus, ID: "mesh",
Summary: "the bus is not healthy: " + strings.Join(problems, "; "), Said: strings.Join(problems, "; ")}}, nil
return []Finding{finding(meshconditionsclient.ScopeBus, "mesh", "", "", Finding{Kind: "core-unhealthy", Severity: meshconditionsclient.Urgent,
Summary: "the bus is not healthy: " + strings.Join(problems, "; "), Evidence: said(strings.Join(problems, "; "))})}, nil
}
// sorted is findings in a stated order, so two runs over one mesh say the same.
func sorted(out []meshconditionsclient.Observation) []meshconditionsclient.Observation {
sort.SliceStable(out, func(i, j int) bool { return out[i].Key() < out[j].Key() })
func sorted(out []Finding) []Finding {
sort.SliceStable(out, func(i, j int) bool { return keyOf(out[i]) < keyOf(out[j]) })
return out
}
modules/mesh-self-check/cmd/mesh-self-check/probes_data.go
+6 -8
View File
@@ -192,7 +192,7 @@ func (c *Checker) dataFacts(ctx context.Context) (dataFacts, error) {
}
// probeData is D13.
func probeData(ctx context.Context, c *Checker) ([]meshconditionsclient.Observation, error) {
func probeData(ctx context.Context, c *Checker) ([]Finding, error) {
f, err := c.dataFacts(ctx)
if err != nil {
return nil, err
@@ -263,14 +263,13 @@ func firstSeen(r diskreadings.Reading, now time.Time) time.Time {
return now
}
func obs(kind string, id, machine string, sev string, summary, said string) meshconditionsclient.Observation {
return meshconditionsclient.Observation{Kind: kind, Ref: kind, Severity: sev, Summary: summary, Said: said,
Scope: ScopeMachine, ID: id, Machine: machine}
func obs(kind string, id, machine string, sev string, summary, saying string) Finding {
return finding(ScopeMachine, id, "", machine, Finding{Kind: kind, Severity: sev, Summary: summary, Evidence: said(saying)})
}
// dataFindings is every condition the data on record raises now: a function of what is known, as the controller's was.
func dataFindings(f dataFacts) []meshconditionsclient.Observation {
var out []meshconditionsclient.Observation
func dataFindings(f dataFacts) []Finding {
var out []Finding
arrays := map[string][]diskreadings.Reading{}
type shrunk struct {
machine, dataset, class string
@@ -301,7 +300,6 @@ func dataFindings(f dataFacts) []meshconditionsclient.Observation {
"`cleanup delete %s %s %s --why …` once a person has decided, or assign %s there again",
r.ID, r.Module, r.Node, r.Class, sizeWords(r.SizeBytes), int(f.now.Sub(r.RetiredAt).Hours()/24),
r.RetiredWhy, r.Node, r.Module, r.ID, r.Module), "kept at "+orUnknownPath(r.Path)))
out[len(out)-1].Ref = "cleanup"
}
continue
}
@@ -428,7 +426,7 @@ func dataFindings(f dataFacts) []meshconditionsclient.Observation {
fmt.Sprintf("%s of %s on %s holds %s, and the copy %s kept on %s holds %s: %s is running on "+
"an empty replacement of its data. Move the data, or assign it back where its data is",
a.ID, a.Module, a.Node, sizeWords(a.SizeBytes), o.Module, o.Node, sizeWords(o.SizeBytes), a.Module), "")
ob.Also = []string{o.Node}
ob.Subject.Also = []string{o.Node}
out = append(out, ob)
break
}
modules/mesh-self-check/cmd/mesh-self-check/probes_fill.go
+10 -12
View File
@@ -55,39 +55,37 @@ func mibWords(b uint64) string { return fmt.Sprintf("%.1f MiB", float64(b)/(1024
// fillKey is where a bucket's or log's fill finding is kept.
func fillKey(stream string) string {
return meshconditionsclient.Observation{Scope: meshconditionsclient.ScopeBus, ID: stream, Ref: "filling"}.Key()
return findingKey(meshconditionsclient.ScopeBus, stream, "", "bucket-or-log-filling")
}
// fillObservation is what one bucket's or log's fill says: a finding at fillRaiseAt or above, and, while its finding
// is written, at fillClearBelow or above too; nothing otherwise, which is what clears it.
func fillObservation(stream string, r fillRow, open bool) (meshconditionsclient.Observation, bool) {
func fillObservation(stream string, r fillRow, open bool) (Finding, bool) {
module, kind, name, ok := bucketOrLog(stream)
if !ok || r.MaxBytes == 0 {
return meshconditionsclient.Observation{}, false // no module's, or with no cap: it cannot fill
return Finding{}, false // no module's, or with no cap: it cannot fill
}
// Compared in bytes, not rounded percent: 74.9% is not 75%.
atRaise := r.Bytes*100 >= r.MaxBytes*fillRaiseAt
aboveClear := r.Bytes*100 >= r.MaxBytes*fillClearBelow
if !atRaise && !(open && aboveClear) {
return meshconditionsclient.Observation{}, false
return Finding{}, false
}
pct := r.Bytes * 100 / r.MaxBytes
return meshconditionsclient.Observation{Kind: "bucket-or-log-filling", Ref: "filling", Severity: meshconditionsclient.Warning,
Scope: meshconditionsclient.ScopeBus, ID: stream,
return finding(meshconditionsclient.ScopeBus, stream, "", "", Finding{Kind: "bucket-or-log-filling", Severity: meshconditionsclient.Warning,
Summary: fmt.Sprintf("%s's %s %s holds %s of %s, %d%%: at its cap the bus refuses the module's next write",
module, kind, name, mibWords(r.Bytes), mibWords(r.MaxBytes), pct),
Said: fmt.Sprintf("module %s, %s %s (stream %s): %d of %d bytes, %d%%", module, kind, name, stream, r.Bytes,
r.MaxBytes, pct),
Evidence: said(fmt.Sprintf("module %s, %s %s (stream %s): %d of %d bytes, %d%%", module, kind, name, stream, r.Bytes,
r.MaxBytes, pct)),
Headline: "A module's bucket or log on the bus is filling up",
Needs: "decide whether to raise its cap or have the module keep less.",
Explanation: "A bucket or log a module keeps on the bus holds three quarters of its cap or more. When it is " +
"full, the bus refuses what the module writes there.",
Resolved: "It has room again"}, true
"full, the bus refuses what the module writes there."}), true
}
// probeFill is D-fill: every module's bucket and log on the bus judged against its cap. The bus module not answering
// is the probe failing to run, which clears nothing; a finding already written counts as open.
func probeFill(_ context.Context, c *Checker) ([]meshconditionsclient.Observation, error) {
func probeFill(_ context.Context, c *Checker) ([]Finding, error) {
raw, err := c.Ask(StreamsTool, map[string]any{})
if err != nil {
return nil, fmt.Errorf("the bus module did not answer %s: %w", StreamsTool, err)
@@ -103,7 +101,7 @@ func probeFill(_ context.Context, c *Checker) ([]meshconditionsclient.Observatio
return nil, fmt.Errorf("its own findings cannot be read, so a fill between the thresholds cannot be judged: %w", err)
}
sort.Slice(a.Detail, func(i, j int) bool { return a.Detail[i].Name < a.Detail[j].Name })
var out []meshconditionsclient.Observation
var out []Finding
for _, r := range a.Detail {
if o, said := fillObservation(r.Name, r, written.has(fillKey(r.Name))); said {
out = append(out, o)
modules/mesh-self-check/cmd/mesh-self-check/probes_mesh.go
+5 -7
View File
@@ -128,7 +128,7 @@ var lookupHost = func(ctx context.Context, host string) ([]string, error) {
// probeBans is D8: no address the mesh owns is in any machine's ban list. The mesh's own addresses are every machine's
// private address and the address its endpoint names, from assignments; the ban lists are asked of each machine's holder
// of node-intrusion-prevention that is heard from.
func probeBans(ctx context.Context, c *Checker) ([]meshconditionsclient.Observation, error) {
func probeBans(ctx context.Context, c *Checker) ([]Finding, error) {
places, err := c.assignments()
if err != nil {
return nil, err
@@ -183,7 +183,7 @@ func probeBans(ctx context.Context, c *Checker) ([]meshconditionsclient.Observat
}(i, node)
}
wg.Wait()
var out []meshconditionsclient.Observation
var out []Finding
var unasked []string
for i, node := range holders {
if errs[i] != nil {
@@ -206,16 +206,14 @@ func probeBans(ctx context.Context, c *Checker) ([]meshconditionsclient.Observat
}
}
if len(banned) > 0 {
out = append(out, meshconditionsclient.Observation{Kind: "own-address-banned", Ref: "bans-the-mesh", Severity: meshconditionsclient.Urgent,
Scope: ScopeMachine, ID: node, Machine: node,
out = append(out, finding(ScopeMachine, node, "", node, Finding{Kind: "own-address-banned", Severity: meshconditionsclient.Urgent,
Summary: fmt.Sprintf("%s's ban list holds %d of the mesh's own address(es) — %s: the mesh is locked out "+
"of itself there (ADR 0186)", node, len(banned), strings.Join(whose, ", ")),
Said: strings.Join(banned, ", "),
Evidence: said(strings.Join(banned, ", ")),
Headline: node + " has banned the mesh",
Needs: fmt.Sprintf("lift the ban on %s, or wait for it to expire.", node),
Explanation: fmt.Sprintf("The intrusion protection on %s banned one of the mesh's own addresses, so the "+
"mesh is locked out of it.", node),
Resolved: node + " no longer bans the mesh"})
"mesh is locked out of it.", node)}))
}
}
if len(unasked) > 0 {
modules/mesh-self-check/cmd/mesh-self-check/probes_root.go
+5 -7
View File
@@ -205,7 +205,7 @@ func (c *Checker) confined(node string, place placing.Assignments, heard bool) (
// probeAgentAccounts is DA: every machine that names an agent account has it judged, on its node-engine's newest
// statement, unable to become root without a person (ADR 0266).
func probeAgentAccounts(_ context.Context, c *Checker) ([]meshconditionsclient.Observation, error) {
func probeAgentAccounts(_ context.Context, c *Checker) ([]Finding, error) {
places, err := c.assignments()
if err != nil {
return nil, err
@@ -219,7 +219,7 @@ func probeAgentAccounts(_ context.Context, c *Checker) ([]meshconditionsclient.O
nodes = append(nodes, n)
}
sort.Strings(nodes)
var out []meshconditionsclient.Observation
var out []Finding
for _, n := range nodes {
if places[n].Decided.AgentAccount == "" {
continue
@@ -233,17 +233,15 @@ func probeAgentAccounts(_ context.Context, c *Checker) ([]meshconditionsclient.O
if v.Confined || v.Quiet {
continue
}
out = append(out, meshconditionsclient.Observation{Kind: "agent-can-become-root", Ref: "agent-root", Severity: meshconditionsclient.Urgent,
Scope: ScopeMachine, ID: n, Machine: n,
out = append(out, finding(ScopeMachine, n, "", n, Finding{Kind: "agent-can-become-root", Severity: meshconditionsclient.Urgent,
Summary: fmt.Sprintf("on %s, %s (ADR 0266): an agent there may become root without a person, and no "+
"answer from a channel authorises an act there (ADR 0259 §8)", n, v.Why),
Said: v.Why,
Evidence: said(v.Why),
Headline: "Sessions on " + n + " could become root",
Needs: "take the sessions' own account out of every group and rule that grants root; the details say which.",
Explanation: fmt.Sprintf("Assistant sessions on %s run under an account of their own, so that none "+
"can take over the machine without you. The machine cannot show that this holds now, so an answer "+
"from your phone authorises nothing there until it does.", n),
Resolved: "Sessions on " + n + " cannot become root again"})
"from your phone authorises nothing there until it does.", n)}))
}
return sorted(out), nil
}
modules/mesh-self-check/cmd/mesh-self-check/probes_test.go
+25 -25
View File
@@ -107,10 +107,10 @@ func (b bus) ask(key string, body any) (json.RawMessage, error) {
return nil, errors.New("nats: no responders available for request")
}
func keysOf(found []meshconditionsclient.Observation) []string {
func keysOf(found []Finding) []string {
var out []string
for _, f := range found {
out = append(out, f.Key())
out = append(out, keyOf(f))
}
return out
}
@@ -119,7 +119,7 @@ func keysOf(found []meshconditionsclient.Observation) []string {
func TestTheBusProbes(t *testing.T) {
cases := []struct {
name string
probe func(context.Context, *Checker) ([]meshconditionsclient.Observation, error)
probe func(context.Context, *Checker) ([]Finding, error)
bus bus
want []string
err string
@@ -128,24 +128,24 @@ func TestTheBusProbes(t *testing.T) {
{"D7 all as defined", probeStreams, bus{}, nil, "", nil},
{"D7 missing and redefined", probeStreams, bus{drop: map[string]bool{"NODES": true}, change: func(rows map[string]*streamRow) {
rows["EVENTS"].Retention = "workqueue"
}}, []string{"bus.EVENTS.stream-redefined", "bus.NODES.stream-missing"}, "", nil},
}}, []string{"bus.EVENTS.stream-redefined.stream-wrong", "bus.NODES.stream-missing.stream-wrong"}, "", nil},
{"D15 quiet", probeReconnects, bus{}, nil, "", nil},
{"D15 a loop", probeReconnects, bus{closed: `{"hours":1,"reaches":true,"users":[
{"user":"g14.node-tools","closed":30,"dropped":20,"names":[{"name":"node-tools","dropped":20,"reasons":{"Stale Connection":20}}]},
{"user":"ace.records","closed":5,"dropped":3},{"user":"(unauthenticated)","closed":90,"dropped":90}]}`},
[]string{"bus.g14.node-tools.bus-reconnects"}, "", nil},
[]string{"bus.g14=2enode-tools.bus-reconnects"}, "", nil},
{"DW answers", probeWatchdogs, bus{}, nil, "", nil},
{"DW silent", probeWatchdogs, bus{signals: errors.New("no responders")}, []string{"core.watchdogs.silent"}, "", nil},
{"DW silent", probeWatchdogs, bus{signals: errors.New("no responders")}, []string{"core.watchdogs.watchdogs-silent"}, "", nil},
{"H-bus healthy", probeBusHealth, bus{}, nil, "", nil},
{"H-bus a stream missing", probeBusHealth,
bus{drop: map[string]bool{"NODES": true}}, []string{"bus.mesh.unhealthy"}, "", nil},
bus{drop: map[string]bool{"NODES": true}}, []string{"bus.mesh.core-unhealthy"}, "", nil},
{"H-bus no round trip", probeBusHealth, bus{asksFail: errors.New("timeout")}, nil, "no request crossed", nil},
// D8 (issue 238): the assignments' addresses and endpoints against each heard holder's ban list.
{"D8 nothing of the mesh banned", probeBans, bus{banned: map[string]string{"ace": `{"banned":["198.51.100.7"]}`,
"novox": `{"banned":[]}`}}, nil, "", machines()},
{"D8 a private address and an endpoint banned", probeBans, bus{banned: map[string]string{
"ace": `{"jails":{"sshd":["10.10.0.1","198.51.100.7"]}}`, "novox": `{"banned":["203.0.113.5"]}`}},
[]string{"machine.ace.bans-the-mesh", "machine.novox.bans-the-mesh"}, "", machines()},
[]string{"machine.ace.own-address-banned", "machine.novox.own-address-banned"}, "", machines()},
{"D8 a machine not heard from is not asked", probeBans, bus{banned: map[string]string{"novox": `{"banned":[]}`,
"ace": `{"banned":[]}`}}, nil, "", machines()},
{"D8 a heard holder that does not answer is no pass", probeBans, bus{banned: map[string]string{}}, nil,
@@ -154,9 +154,9 @@ func TestTheBusProbes(t *testing.T) {
// the manifests declare; the kinds, keys and words are the controller's.
{"D13 all held", probeData, bus{}, nil, "", dataMesh(func(*dataRig) {})},
{"D13 shrank, gone, quiet, unprotected and stale", probeData, bus{}, []string{
"machine.ace.pg.data.data-quiet", "machine.ace.pg.data.backup-stale", "machine.ace.pg.dumps.data-missing",
"machine.ace.plex.movies.protection-missing", "machine.ace.plex.thumbs.data-shrank",
"machine.ace.array.md-md127.array-degraded"},
"machine.ace=2epg=2edata.data-quiet", "machine.ace=2epg=2edata.backup-stale", "machine.ace=2epg=2edumps.data-missing",
"machine.ace=2eplex=2emovies.protection-missing", "machine.ace=2eplex=2ethumbs.data-shrank",
"machine.ace=2earray=2emd-md127.array-degraded"},
"", dataMesh(func(r *dataRig) {
r.size("ace", "pg", "data", 100<<20, 100<<20).write("ace", "pg", "data", -80*time.Hour)
r.gone("ace", "pg", "dumps")
@@ -165,23 +165,23 @@ func TestTheBusProbes(t *testing.T) {
r.size("ace", "plex", "library", 1<<30, 1<<30).array("ace", "plex", "library", "md", "md127", false)
r.lastGood("ace", "pg", "data", -100*time.Hour)
})},
{"D13 a dataset shrank once for its items", probeData, bus{}, []string{"machine.ace.dataset.storage-media.data-shrank"},
{"D13 a dataset shrank once for its items", probeData, bus{}, []string{"machine.ace=2edataset=2estorage-media.data-shrank"},
"", dataMesh(func(r *dataRig) {
r.size("ace", "plex", "thumbs", 10<<20, 600<<20).dataset("ace", "plex", "thumbs", "storage/media")
r.size("ace", "plex", "meta", 10<<20, 600<<20).dataset("ace", "plex", "meta", "storage/media")
})},
{"D13 a retired item is kept, not judged, and cleanup waits after thirty days", probeData, bus{}, []string{
"machine.ace.pg.dumps.cleanup"}, "", dataMesh(func(r *dataRig) {
"machine.ace=2epg=2edumps.cleanup-waiting"}, "", dataMesh(func(r *dataRig) {
r.size("ace", "pg", "dumps", 1<<20, 1<<30).retired("ace", "pg", "dumps", -31*24*time.Hour)
r.size("ace", "pg", "old", 1<<20, 1<<30).retired("ace", "pg", "old", -2*24*time.Hour)
})},
{"D13 a copy far smaller than a retired one elsewhere is an empty replacement", probeData, bus{},
[]string{"machine.ace.plex.thumbs.empty-replacement"}, "", dataMesh(func(r *dataRig) {
[]string{"machine.ace=2eplex=2ethumbs.empty-replacement"}, "", dataMesh(func(r *dataRig) {
r.size("ace", "plex", "thumbs", 10<<20, 10<<20)
r.size("g14", "plex", "thumbs", 600<<20, 600<<20).retired("g14", "plex", "thumbs", -24*time.Hour)
})},
{"D13 no good backup is counted from when the item was first measured", probeData, bus{},
[]string{"machine.ace.pg.data.backup-stale"}, "", dataMesh(func(r *dataRig) {
[]string{"machine.ace=2epg=2edata.backup-stale"}, "", dataMesh(func(r *dataRig) {
delete(r.good, diskreadings.ReadingKey("ace", "pg", "data"))
delete(r.good, diskreadings.ReadingKey("ace", "plex", "meta"))
r.firstSeen("ace", "pg", "data", -100*time.Hour).firstSeen("ace", "plex", "meta", -time.Hour)
@@ -193,24 +193,24 @@ func TestTheBusProbes(t *testing.T) {
{Name: "LOG_mesh-delivery-builder_build-log", Bytes: 10, MaxBytes: 100}, {Name: "KV_nocap_x", Bytes: 999}}}, nil, "", nil},
{"D-fill a bucket and a log filling", probeFill, bus{fills: []fillRow{{Name: "KV_mesh-issues_issues", Bytes: 75, MaxBytes: 100},
{Name: "LOG_mesh-delivery-builder_build-log", Bytes: 99, MaxBytes: 100}}},
[]string{"bus.KV_mesh-issues_issues.filling", "bus.LOG_mesh-delivery-builder_build-log.filling"}, "", nil},
[]string{"bus.KV_mesh-issues_issues.bucket-or-log-filling", "bus.LOG_mesh-delivery-builder_build-log.bucket-or-log-filling"}, "", nil},
{"D-fill between the thresholds raises nothing new", probeFill,
bus{fills: []fillRow{{Name: "KV_mesh-issues_issues", Bytes: 72, MaxBytes: 100}}}, nil, "", nil},
{"D-fill between the thresholds keeps what is said", probeFill,
bus{fills: []fillRow{{Name: "KV_mesh-issues_issues", Bytes: 72, MaxBytes: 100}}},
[]string{"bus.KV_mesh-issues_issues.filling"}, "", filling("KV_mesh-issues_issues")},
[]string{"bus.KV_mesh-issues_issues.bucket-or-log-filling"}, "", filling("KV_mesh-issues_issues")},
{"D-fill the bus module silent", probeFill, bus{asksFail: errors.New("no responders")}, nil, "did not answer", nil},
// DA (ADR 0266): the agent account each machine names, judged on its engine's report while it is heard from.
{"DA confined", probeAgentAccounts, bus{}, nil, "", agents(map[string]string{"novox": stateHealthy})},
{"DA a way to root found", probeAgentAccounts, bus{}, []string{"machine.novox.agent-root"}, "",
{"DA a way to root found", probeAgentAccounts, bus{}, []string{"machine.novox.agent-can-become-root"}, "",
agents(map[string]string{"novox": stateUnhealthy})},
{"DA no verdict on it", probeAgentAccounts, bus{}, []string{"machine.novox.agent-root"}, "",
{"DA no verdict on it", probeAgentAccounts, bus{}, []string{"machine.novox.agent-can-become-root"}, "",
agents(map[string]string{"novox": ""})},
{"DA the search runs, within its quiet", probeAgentAccounts, bus{}, nil, "",
agents(map[string]string{"novox": "pending"})},
{"DA a machine not heard from has no verdict on now", probeAgentAccounts, bus{}, []string{"machine.g14.agent-root"},
{"DA a machine not heard from has no verdict on now", probeAgentAccounts, bus{}, []string{"machine.g14.agent-can-become-root"},
"", agents(map[string]string{"novox": stateHealthy, "g14": stateHealthy})},
{"DA an engine older than the judging", probeAgentAccounts, bus{}, []string{"machine.novox.agent-root"}, "",
{"DA an engine older than the judging", probeAgentAccounts, bus{}, []string{"machine.novox.agent-can-become-root"}, "",
agents(map[string]string{"novox": "old"})},
}
for _, tc := range cases {
@@ -301,10 +301,10 @@ func placed(s *memStore) map[string]placing.Assignments {
// filling is a store holding the self-check's written fill finding of a stream: the condition open.
func filling(stream string) *memStore {
s := newStore()
o := meshconditionsclient.Observation{Kind: "bucket-or-log-filling", Ref: "filling", Severity: meshconditionsclient.Warning, Scope: meshconditionsclient.ScopeBus, ID: stream,
Summary: "filling", Source: "D-fill"}
raw, _ := json.Marshal(o.Finding(Module, "run", time.Now()))
_ = s.Put(FindingsState, o.Key(), json.RawMessage(raw))
o := finding(meshconditionsclient.ScopeBus, stream, "", "", Finding{Kind: "bucket-or-log-filling", Severity: meshconditionsclient.Warning,
Summary: "filling", Source: "D-fill", Observer: Module, Run: "run", At: time.Now(), Since: time.Now()})
raw, _ := json.Marshal(o)
_ = s.Put(FindingsState, keyOf(o), json.RawMessage(raw))
return s
}
modules/mesh-self-check/cmd/mesh-self-check/selfcheck.go
+69 -61
View File
@@ -2,7 +2,7 @@ package main
// The self-check (novox/hq ADR 0564 §6): the registry of probes, each a live invariant of a design, run every five
// minutes, each probe within thirty seconds. What a probe finds is written as a finding in this module's bucket
// `findings`, which the condition keeper raises and clears conditions from (ADR 0564 §2); every run is kept in the
// `findings`, through mesh-sdk's finding writer, which the condition keeper raises and clears conditions from (ADR 0564 §2); every run is kept in the
// bucket `verdicts` for a week and the last as the key `last` of `verdict`, and is said as `checked`, which the
// watchman and mesh-watcher hear (S10, ADR 0227 §6). A probe that errors or does not finish is never a pass: it is
// said in the verdict at once and written as a finding of its own when the next run cannot run it either.
@@ -20,6 +20,7 @@ import (
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
meshselfcheck "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshselfcheck"
"git.novox.be/novox/mesh-sdk/go/views"
)
// Module and Seat are this module and the seat it holds once for the mesh.
@@ -95,7 +96,7 @@ type Probe struct {
// Deferred names the facts it lacks on the bus, for a probe whose facts have no home there yet (ADR 0564 §6):
// answered deferred, never a pass.
Deferred string
Run func(ctx context.Context, c *Checker) ([]meshconditionsclient.Observation, error)
Run func(ctx context.Context, c *Checker) ([]Finding, error)
}
// ProbeVerdict, Counts and Run are a run as the seat mesh-self-check's contract holds it: the body of `checked`, the controller's
@@ -119,6 +120,9 @@ type Checker struct {
Logf func(format string, args ...any)
Host string
Registry []Probe
// Writer is the finding writer over this module's state `findings` (views.NewStdioWriter; design 47 G A-16), whose
// observe is Observe.
Writer *FindingsWriter
running sync.Mutex
mu sync.Mutex
@@ -144,14 +148,36 @@ func (c *Checker) LastRun() *Run {
}
type result struct {
found []meshconditionsclient.Observation
found []Finding
err error
took time.Duration
}
// RunOnce runs every probe of the registry, writes what changed in `findings`, keeps the verdict and says `checked`.
// One run at a time: a person's run during a scheduled one waits for it.
// The context a run is asked with: why it runs, and where its verdict is answered.
type (
whyKey struct{}
runOut struct{}
)
// RunOnce runs every probe of the registry through the findings writer, which writes what changed in `findings` and says
// `findings-stated` (design 47 G §461, §485), keeps the verdict and says `checked`. One run at a time: a person's run
// during a scheduled one waits for it.
func (c *Checker) RunOnce(ctx context.Context, why string) Run {
var out Run
ctx = context.WithValue(context.WithValue(ctx, whyKey{}, why), runOut{}, &out)
if err := c.Writer.State(ctx, ""); err != nil {
c.Logf("[%s] what this run found is not all written: %v", Module, err)
}
return out
}
// Observe is the writer's observe: it runs every probe, keeps the verdict and says `checked`, and answers the findings
// this run observes and the keys it cannot judge (those of a probe that could not run).
func (c *Checker) Observe(ctx context.Context) (views.Observed[Finding], error) {
why, _ := ctx.Value(whyKey{}).(string)
if why == "" {
why = "a statement of the findings"
}
c.running.Lock()
defer c.running.Unlock()
c.mu.Lock()
@@ -179,9 +205,9 @@ func (c *Checker) RunOnce(ctx context.Context, why string) Run {
written, readErr := c.written()
seen := map[string]bool{}
want := map[string]meshconditionsclient.Observation{} // every finding this run writes, by key
ran := map[string]bool{} // the sources whose findings this run judged whole
var blind []meshconditionsclient.Observation
want := map[string]Finding{} // every finding this run observes, by key
ran := map[string]bool{} // the sources whose findings this run judged whole
var blind []Finding
for i, p := range c.Registry {
if !on(p.ID) {
// Off by the setting: it ran nothing and sees nothing, so what it raised before is cleared, and it has no
@@ -199,10 +225,10 @@ func (c *Checker) RunOnce(ctx context.Context, why string) Run {
case r.err != nil:
v.Verdict, v.Error = VerdictFailedToRun, r.err.Error()
run.Counts.FailedToRun++
blind = append(blind, meshconditionsclient.Observation{Kind: "probe-failed", Ref: "failed", Severity: meshconditionsclient.Warning,
Scope: meshconditionsclient.ScopeProbe, ID: p.ID, Source: sourceSelf,
Summary: fmt.Sprintf("the probe %s (%s) could not run: what it checks is not known — never a pass", p.ID, p.Asserts),
Said: firstLine(r.err.Error())})
blind = append(blind, finding(meshconditionsclient.ScopeProbe, p.ID, "", "", Finding{Kind: "probe-failed",
Severity: meshconditionsclient.Warning, Source: sourceSelf,
Summary: fmt.Sprintf("the probe %s (%s) could not run: what it checks is not known — never a pass", p.ID, p.Asserts),
Evidence: said(firstLine(r.err.Error()))}))
default:
ran[p.ID] = true
for _, f := range r.found {
@@ -211,7 +237,7 @@ func (c *Checker) RunOnce(ctx context.Context, why string) Run {
}
f.Source = p.ID
f = worded(f)
key := f.Key()
key := keyOf(f)
seen[key] = true
if confirm(f) && !c.seenBefore(key) && !written.has(key) {
v.Unconfirmed = append(v.Unconfirmed, f.Summary)
@@ -236,18 +262,12 @@ func (c *Checker) RunOnce(ctx context.Context, why string) Run {
ran[sourceSelf] = true
for _, f := range blind {
f = worded(f)
key := f.Key()
key := keyOf(f)
seen[key] = true
if c.seenBefore(key) || written.has(key) {
want[key] = f
}
}
if readErr != nil {
// Blind to its own bucket: nothing is written or deleted (ADR 0227 §4), and the log says so.
c.Logf("[%s] its findings cannot be read, so none is written or cleared this run: %v", Module, readErr)
} else if err := c.write(written, want, ran, run.Run); err != nil {
c.Logf("[%s] what this run found is not all written: %v", Module, err)
}
c.mu.Lock()
c.seen = seen
c.mu.Unlock()
@@ -258,7 +278,32 @@ func (c *Checker) RunOnce(ctx context.Context, why string) Run {
run.Unsaid = int64(c.unsaid)
c.mu.Unlock()
c.keep(run)
return run
if out, ok := ctx.Value(runOut{}).(*Run); ok {
*out = run
}
if readErr != nil {
// Blind to its own findings: no set is said, so nothing is written or cleared (ADR 0227 §4).
return views.Observed[Finding]{}, fmt.Errorf("its findings cannot be read, so none is written or cleared this run: %w", readErr)
}
keys := make([]string, 0, len(want))
for k := range want {
keys = append(keys, k)
}
sort.Strings(keys)
obs := views.Observed[Finding]{}
for _, k := range keys {
f := want[k]
f.Run = run.Run
obs.Findings = append(obs.Findings, f)
}
// A finding of a source that could not run is left as it is: a blind observer clears nothing (ADR 0227 §4).
for k, f := range written {
if _, still := want[k]; !still && !ran[f.Source] {
obs.Pending = append(obs.Pending, views.PendingKey{Key: k, Since: f.Since, Bound: Every.String()})
}
}
sort.Slice(obs.Pending, func(i, j int) bool { return obs.Pending[i].Key < obs.Pending[j].Key })
return obs, nil
}
// probe runs one probe within its bound; a panic or an overrun is an error, never a pass.
@@ -291,8 +336,8 @@ func (c *Checker) seenBefore(key string) bool {
return c.seen[key]
}
// writtenFindings are the findings in the bucket, by key.
type writtenFindings map[string]meshconditions.FindingsByKeyValue
// writtenFindings are the findings in the state, by key.
type writtenFindings map[string]Finding
func (w writtenFindings) has(key string) bool { _, ok := w[key]; return ok }
@@ -311,7 +356,7 @@ func (c *Checker) written() (writtenFindings, error) {
if !found {
continue
}
var f meshconditions.FindingsByKeyValue
var f Finding
if err := json.Unmarshal(raw, &f); err != nil {
return out, fmt.Errorf("the finding %s is not readable: %w", k, err)
}
@@ -320,43 +365,6 @@ func (c *Checker) written() (writtenFindings, error) {
return out, nil
}
// write puts each finding whose words changed and deletes each one a source that ran whole no longer sees. A finding
// of a source that could not run is left as it is: a blind observer clears nothing (ADR 0227 §4).
func (c *Checker) write(written writtenFindings, want map[string]meshconditionsclient.Observation, ran map[string]bool, run string) error {
var errs []string
keys := make([]string, 0, len(want))
for k := range want {
keys = append(keys, k)
}
sort.Strings(keys)
now := c.Now()
for _, k := range keys {
f := want[k].Finding(Module, run, now)
if was, ok := written[k]; ok && meshconditionsclient.Says(was) == meshconditionsclient.Says(f) {
continue
}
raw, err := json.Marshal(f)
if err == nil {
err = c.Store.Put(FindingsState, k, json.RawMessage(raw))
}
if err != nil {
errs = append(errs, k+": "+err.Error())
}
}
for k, f := range written {
if _, still := want[k]; still || !ran[f.Source] {
continue
}
if err := c.Store.Delete(FindingsState, k); err != nil {
errs = append(errs, k+": "+err.Error())
}
}
if len(errs) > 0 {
return fmt.Errorf("%s", strings.Join(errs, "; "))
}
return nil
}
// keep keeps the run as the last verdict and in the week's verdicts, and says it as `checked`. Not said is said in
// the log: the watchman's S10 says the self-check silent when `checked` stops.
func (c *Checker) keep(run Run) {
modules/mesh-self-check/cmd/mesh-self-check/selfcheck_test.go
+69 -40
View File
@@ -13,7 +13,8 @@ import (
"time"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/views"
)
// memStore is the bus's buckets in memory, counting the writes and the reads one key at a time, and handing every
@@ -25,6 +26,7 @@ type memStore struct {
gets map[string]int
watches map[string][]func(key string, value []byte, gone bool)
keysErr error
events []string // the subjects the finding writer said its events on
}
func newStore() *memStore {
@@ -97,6 +99,22 @@ func (s *memStore) findingKeys() []string {
return keys
}
// findingState is the writer's view of the store's state `findings`.
type findingState struct{ s *memStore }
func (f findingState) Get(key string) (*stdio.StateEntry, error) {
raw, ok, err := f.s.Get(FindingsState, key)
if err != nil || !ok {
return nil, err
}
return &stdio.StateEntry{Key: key, Value: raw}, nil
}
func (f findingState) Put(key string, value any) (uint64, error) {
return 0, f.s.Put(FindingsState, key, value)
}
func (f findingState) Delete(key string) error { return f.s.Delete(FindingsState, key) }
func (f findingState) Keys() ([]string, error) { return f.s.Keys(FindingsState) }
// checker is a self-check over memory, with a registry the test drives.
func checker(store *memStore, registry ...Probe) (*Checker, *[]string) {
var said []string
@@ -108,6 +126,17 @@ func checker(store *memStore, registry ...Probe) (*Checker, *[]string) {
said = append(said, subject+" "+id+" "+string(raw))
return nil
}}
w, err := views.NewWriter(views.WriterConfig[Finding, FindingCleared, FindingsStated, FindingsAsked]{
Module: Module, Node: "novox", Asked: "mesh.seat.mesh-conditions.event.findings-asked", State: findingState{store}, Observe: c.Observe, Logf: c.Logf, Now: c.Now,
Publish: func(_ context.Context, subject string, _ []byte, _ string) error {
store.events = append(store.events, subject)
return nil
},
Listen: func(context.Context, string, func([]byte)) (func(), error) { return func() {}, nil }})
if err != nil {
panic(err)
}
c.Writer = w
return c, &said
}
@@ -115,37 +144,36 @@ func checker(store *memStore, registry ...Probe) (*Checker, *[]string) {
// could not run clears nothing and is written as its own finding at the second run in a row; a finding one look can
// be wrong about waits for the second run; a deferred probe is never a pass (ADR 0564 §2, §6; issue 1952 SC1).
func TestARunWritesFindingsAsTheyChange(t *testing.T) {
var found []meshconditionsclient.Observation
var found []Finding
var fail error
steady := Probe{ID: "P1", Kind: "stream-wrong", Run: func(context.Context, *Checker) ([]meshconditionsclient.Observation, error) { return found, fail }}
steady := Probe{ID: "P1", Kind: "stream-wrong", Run: func(context.Context, *Checker) ([]Finding, error) { return found, fail }}
deferred := Probe{ID: "P2", Kind: "resolver-wrong", Deferred: "the overlay's addresses"}
missing := meshconditionsclient.Observation{Ref: "stream-missing", Severity: meshconditionsclient.Urgent, Scope: meshconditionsclient.ScopeBus, ID: "EVENTS",
Summary: "the stream EVENTS is not on the bus"}
behind := meshconditionsclient.Observation{Kind: kindDataUnmeasured, Ref: "behind", Severity: meshconditionsclient.Warning,
Scope: meshconditionsclient.ScopeBus, ID: "EVENTS.controller", Summary: "far behind"}
missing := finding(meshconditionsclient.ScopeBus, "EVENTS", "stream-missing", "", Finding{Severity: meshconditionsclient.Urgent,
Summary: "the stream EVENTS is not on the bus", Evidence: said("no stream EVENTS")})
behind := finding(meshconditionsclient.ScopeBus, "EVENTS.controller", "", "", Finding{Kind: kindDataUnmeasured, Severity: meshconditionsclient.Warning,
Summary: "far behind"})
store := newStore()
c, said := checker(store, steady, deferred)
steps := []struct {
name string
found []meshconditionsclient.Observation
fail error
want []string // finding keys written after the run
verdict string
missingPs int // puts of the missing finding so far
name string
found []Finding
fail error
want []string // finding keys written after the run
verdict string
}{
{"a finding is written", []meshconditionsclient.Observation{missing}, nil, []string{"bus.EVENTS.stream-missing"}, VerdictFail, 1},
{"the same again is no write", []meshconditionsclient.Observation{missing}, nil, []string{"bus.EVENTS.stream-missing"}, VerdictFail, 1},
{"one look can be wrong: held", []meshconditionsclient.Observation{missing, behind}, nil, []string{"bus.EVENTS.stream-missing"}, VerdictFail, 1},
{"seen twice: written", []meshconditionsclient.Observation{missing, behind}, nil,
[]string{"bus.EVENTS.controller.behind", "bus.EVENTS.stream-missing"}, VerdictFail, 1},
{"a finding is written", []Finding{missing}, nil, []string{"bus.EVENTS.stream-missing.stream-wrong"}, VerdictFail},
{"the same again is stated once more", []Finding{missing}, nil, []string{"bus.EVENTS.stream-missing.stream-wrong"}, VerdictFail},
{"one look can be wrong: held", []Finding{missing, behind}, nil, []string{"bus.EVENTS.stream-missing.stream-wrong"}, VerdictFail},
{"seen twice: written", []Finding{missing, behind}, nil,
[]string{"bus.EVENTS.stream-missing.stream-wrong", "bus.EVENTS=2econtroller." + kindDataUnmeasured}, VerdictFail},
{"blind once: clears nothing, held", nil, errors.New("timed out"),
[]string{"bus.EVENTS.controller.behind", "bus.EVENTS.stream-missing"}, VerdictFailedToRun, 1},
[]string{"bus.EVENTS.stream-missing.stream-wrong", "bus.EVENTS=2econtroller." + kindDataUnmeasured}, VerdictFailedToRun},
{"blind twice: its own finding", nil, errors.New("timed out"),
[]string{"bus.EVENTS.controller.behind", "bus.EVENTS.stream-missing", "probe.P1.failed"}, VerdictFailedToRun, 1},
{"no longer seen: cleared", nil, nil, nil, VerdictPass, 1},
[]string{"bus.EVENTS.stream-missing.stream-wrong", "bus.EVENTS=2econtroller." + kindDataUnmeasured, "probe.P1.probe-failed"}, VerdictFailedToRun},
{"no longer seen: cleared", nil, nil, nil, VerdictPass},
}
for _, s := range steps {
for i, s := range steps {
found, fail = s.found, s.fail
run := c.RunOnce(context.Background(), "the test")
if got := store.findingKeys(); !slices.Equal(got, s.want) {
@@ -157,11 +185,12 @@ func TestARunWritesFindingsAsTheyChange(t *testing.T) {
if run.Probes[1].Verdict != VerdictDeferred || !strings.Contains(run.Probes[1].Error, "overlay") {
t.Errorf("%s: a deferred probe answered %+v", s.name, run.Probes[1])
}
if n := store.puts[FindingsState+"/bus.EVENTS.stream-missing"]; n != s.missingPs {
t.Errorf("%s: the unchanged finding was written %d times", s.name, n)
// The writer says one `findings-stated` a run, on the subject it forms itself (design 47 G A-13).
if n := len(store.events); n != i+1 || store.events[i] != "mesh.stream.mesh-self-check.event.findings-stated.novox" {
t.Errorf("%s: the writer said %v", s.name, store.events)
}
}
raw, _, _ := store.Get(FindingsState, "probe.P1.failed")
raw, _, _ := store.Get(FindingsState, "probe.P1.probe-failed")
if raw != nil {
t.Errorf("the blind finding outlived the probe running again: %s", raw)
}
@@ -189,12 +218,12 @@ func TestARunWritesFindingsAsTheyChange(t *testing.T) {
t.Errorf("the last verdict counts %+v", r.Counts)
}
// A finding carries the operator's words of its kind, the probe that found it and who observed it.
found, fail = []meshconditionsclient.Observation{missing}, nil
found, fail = []Finding{missing}, nil
c.RunOnce(context.Background(), "the test")
raw, _, _ = store.Get(FindingsState, "bus.EVENTS.stream-missing")
raw, _, _ = store.Get(FindingsState, "bus.EVENTS.stream-missing.stream-wrong")
f, err := decodeFinding(raw)
if err != nil || f.Kind != "stream-wrong" || f.Source != "P1" || f.Observer != Module || f.Headline == "" ||
f.Explanation == "" || f.Ref != "stream-missing" || len(f.Evidence) != 1 || f.Run == "" {
f.Explanation == "" || f.ID != "EVENTS.stream-missing" || f.Subject.ID != f.ID || f.Subject.Scope != f.Scope || len(f.Evidence) != 1 || f.Run == "" || f.Since.IsZero() {
t.Errorf("the written finding is %+v", f)
}
// Blind to its own bucket, a run writes and clears nothing.
@@ -202,7 +231,7 @@ func TestARunWritesFindingsAsTheyChange(t *testing.T) {
found = nil
c.RunOnce(context.Background(), "the test")
store.keysErr = nil
if got := store.findingKeys(); !slices.Equal(got, []string{"bus.EVENTS.stream-missing"}) {
if got := store.findingKeys(); !slices.Equal(got, []string{"bus.EVENTS.stream-missing.stream-wrong"}) {
t.Errorf("a run blind to its findings changed them: %v", got)
}
}
@@ -212,9 +241,9 @@ func TestAProbeThatPanicsOrOverrunsCouldNotRun(t *testing.T) {
within := Within
Within = 50 * time.Millisecond
t.Cleanup(func() { Within = within })
for name, run := range map[string]func(context.Context, *Checker) ([]meshconditionsclient.Observation, error){
"panic": func(context.Context, *Checker) ([]meshconditionsclient.Observation, error) { panic("boom") },
"overrun": func(ctx context.Context, _ *Checker) ([]meshconditionsclient.Observation, error) {
for name, run := range map[string]func(context.Context, *Checker) ([]Finding, error){
"panic": func(context.Context, *Checker) ([]Finding, error) { panic("boom") },
"overrun": func(ctx context.Context, _ *Checker) ([]Finding, error) {
time.Sleep(time.Second)
return nil, nil
},
@@ -232,12 +261,12 @@ func TestAProbeThatPanicsOrOverrunsCouldNotRun(t *testing.T) {
// every state the module keeps (ADR 0564 §12).
func TestTheVerbsAndReads(t *testing.T) {
store := newStore()
many := make([]meshconditionsclient.Observation, 25)
many := make([]Finding, 25)
for i := range many {
many[i] = meshconditionsclient.Observation{Severity: meshconditionsclient.Warning, Scope: meshconditionsclient.ScopeBus, ID: fmt.Sprintf("S%02d", i), Summary: "x"}
many[i] = finding(meshconditionsclient.ScopeBus, fmt.Sprintf("S%02d", i), "", "", Finding{Severity: meshconditionsclient.Warning, Summary: "x"})
}
c, _ := checker(store, Probe{ID: "D7", Kind: "stream-wrong",
Run: func(context.Context, *Checker) ([]meshconditionsclient.Observation, error) { return many, nil }})
Run: func(context.Context, *Checker) ([]Finding, error) { return many, nil }})
if _, err := c.Doctor(false, "", "x"); err == nil || !strings.Contains(err.Error(), "first run") {
t.Fatalf("before a run: %v", err)
}
@@ -283,11 +312,11 @@ func TestTheVerbsAndReads(t *testing.T) {
// and is reported by the verb as off by setting; the default list is D13 and D-fill (the operator, 2026-10-30).
func TestOnlyTheProbesTheSettingListsRun(t *testing.T) {
ran := map[string]int{}
finds := meshconditionsclient.Observation{Ref: "stream-missing", Severity: meshconditionsclient.Urgent, Scope: meshconditionsclient.ScopeBus, ID: "EVENTS", Summary: "gone"}
finds := finding(meshconditionsclient.ScopeBus, "EVENTS", "stream-missing", "", Finding{Severity: meshconditionsclient.Urgent, Summary: "gone"})
probe := func(id string) Probe {
return Probe{ID: id, Kind: "stream-wrong", Run: func(context.Context, *Checker) ([]meshconditionsclient.Observation, error) {
return Probe{ID: id, Kind: "stream-wrong", Run: func(context.Context, *Checker) ([]Finding, error) {
ran[id]++
return []meshconditionsclient.Observation{finds}, nil
return []Finding{finds}, nil
}}
}
store := newStore()
@@ -325,8 +354,8 @@ func TestOnlyTheProbesTheSettingListsRun(t *testing.T) {
}
// decodeFinding reads a finding as the seat mesh-conditions lays it out.
func decodeFinding(raw []byte) (meshconditions.FindingsByKeyValue, error) {
var f meshconditions.FindingsByKeyValue
func decodeFinding(raw []byte) (Finding, error) {
var f Finding
err := json.Unmarshal(raw, &f)
return f, err
}
modules/mesh-self-check/cmd/mesh-self-check/whole_integration_test.go
+24 -7
View File
@@ -157,6 +157,9 @@ func (p *player) handle(m wire) {
err = kv.Delete(p.ctx, a.Key)
}
p.answer(m.ID, map[string]any{}, err)
case "mesh/subscribe":
// The finding writer's consumer of the asks for findings (design 47 G A-16); no ask comes in this test.
p.answer(m.ID, map[string]any{}, nil)
case "mesh/seat.publish":
var e struct {
Subject string `json:"subject"`
@@ -252,6 +255,10 @@ func TestTheSelfCheckWholeRunsOnTheBus(t *testing.T) {
if err != nil {
t.Fatal(err)
}
statements, err := nc.SubscribeSync("mesh.mod." + Module + ".event.findings-stated")
if err != nil {
t.Fatal(err)
}
cmd := exec.CommandContext(ctx, bin)
cmd.Env = append(os.Environ(), "MESH_NODE=ace", "MESH_MODULE="+Module)
stdin, _ := cmd.StdinPipe()
@@ -282,16 +289,26 @@ func TestTheSelfCheckWholeRunsOnTheBus(t *testing.T) {
verdicts["D2"] != VerdictDeferred || answer.Run.Controller != "ace" || !strings.Contains(answer.Run.Why, "person.jochen") {
t.Errorf("the run answered %+v", answer.Run)
}
msg, err := checked.NextMsg(10 * time.Second)
if err != nil {
t.Fatalf("checked was not said: %v", err)
}
// The writer's statement at the start says checked too: the run of the person's ask is the one this waits for.
var said Run
if err := json.Unmarshal(msg.Data, &said); err != nil || said.Run != answer.Run.Run {
t.Errorf("checked said %s", msg.Data)
for said.Run != answer.Run.Run {
msg, err := checked.NextMsg(10 * time.Second)
if err != nil {
t.Fatalf("checked of %s was not said: %v", answer.Run.Run, err)
}
if err := json.Unmarshal(msg.Data, &said); err != nil {
t.Fatalf("checked said %s", msg.Data)
}
}
for stated := ""; !strings.Contains(stated, `"id":"EVENTS.stream-missing"`); {
msg, err := statements.NextMsg(10 * time.Second)
if err != nil {
t.Fatalf("the writer did not say findings-stated with the missing stream (last %s): %v", stated, err)
}
stated = string(msg.Data)
}
findings, _ := js.KeyValue(ctx, Module+"_"+FindingsState)
e, err := findings.Get(ctx, "bus.EVENTS.stream-missing")
e, err := findings.Get(ctx, "bus.EVENTS.stream-missing.stream-wrong")
if err != nil {
t.Fatalf("the missing stream is not a finding on the bus: %v", err)
}
modules/mesh-self-check/cmd/mesh-self-check/words.go
+10 -20
View File
@@ -1,12 +1,10 @@
package main
// The operator's words for the kinds the self-check raises; the observation and its finding are the conditions
// client's (novox/hq ADR 0564 §2, ADR 0634 §4).
import "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
// The operator's words for the kinds the self-check raises, written into a finding's headline and explanation
// (novox/hq ADR 0564 §2; design 47 G §689).
// worded gives a finding the operator's words registered for its kind, where it says none of its own.
func worded(f meshconditionsclient.Observation) meshconditionsclient.Observation {
func worded(f Finding) Finding {
w, ok := words[f.Kind]
if !ok {
return f
@@ -17,46 +15,38 @@ func worded(f meshconditionsclient.Observation) meshconditionsclient.Observation
if f.Explanation == "" {
f.Explanation = w.Explanation
}
if f.Resolved == "" {
f.Resolved = w.Resolved
}
if f.Needs == "" {
f.Needs = w.Needs
}
return f
}
type plain struct{ Headline, Explanation, Resolved, Needs string }
type plain struct{ Headline, Explanation, Needs string }
// words are the controller's words for each kind the self-check raises (mesh-controller plain_words.go), the
// controller's name for itself replaced by the self-check's where the words spoke of it.
var words = map[string]plain{
"stream-wrong": {Headline: "Part of the bus's storage is wrong",
Needs: "check the machine the bus runs on; the details say what is missing.",
Explanation: "A store the mesh keeps on the bus is missing or not as defined, so what it keeps there may be lost.",
Resolved: "The bus's storage is right again"},
Explanation: "A store the mesh keeps on the bus is missing or not as defined, so what it keeps there may be lost."},
"bus-reconnects": {Headline: "A client keeps losing the bus",
Explanation: "One of the mesh's clients lost its connection to the bus again and again in the last hour. While " +
"it reconnects, what it says and what it is asked waits.",
Resolved: "Resolved: the client stays connected"},
"it reconnects, what it says and what it is asked waits."},
"core-unhealthy": {Headline: "The bus is not healthy",
Needs: "check the machine the bus runs on; the details say what fails.",
Explanation: "The bus, which carries every message in the mesh, is not healthy. Instructions, builds and tool " +
"calls may fail until it recovers.",
Resolved: "The bus is healthy again"},
"calls may fail until it recovers."},
"watchdogs-silent": {Headline: "The mesh's watchdogs stopped",
Needs: "restart the watchman if this stays, from the mesh MCP server.",
Explanation: "The watchdogs, which notice when something expected does not happen, do not answer, so missed " +
"signals are not noticed.",
Resolved: "The watchdogs run again"},
"signals are not noticed."},
"probe-failed": {Headline: "A check of the mesh could not run",
Explanation: "One of the mesh's own checks could not run, so what it watches is unknown for now. It runs again " +
"every few minutes.",
Resolved: "The check runs again"},
"every few minutes."},
}
// confirm says one look can be wrong about the finding: it is written only when two runs in a row see it (issue 277).
// The kinds are the self-check's own: a probe that could not run, and a holder that measured nothing of the data.
func confirm(o meshconditionsclient.Observation) bool {
func confirm(o Finding) bool {
return o.Kind == "probe-failed" || o.Kind == kindDataUnmeasured
}