jschoubben
3ed3266dde
Nobody raises a lost or a far-behind consumer since the self-check's bus-consumer probe ended (issue 1126): the grants part makes a missing account consumer itself and says one far behind as mesh-grants.consumer-behind, and H4's repair was the controller's. The mesh defines no consumer of its own, so H-bus's consumer half judged nothing. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
325 lines
14 KiB
Go
325 lines
14 KiB
Go
package main
|
||
|
||
// The registry (novox/hq ADR 0564 §6 and the probe table of its Context): the probes that move from the controller,
|
||
// each with its assertion unchanged, reading its facts where they live on the bus. A probe whose facts have no home
|
||
// on the bus yet answers deferred, naming them, and is never a pass (tracked by issue 1952). Not registered, as the
|
||
// table says: D1 (nodes compose their own; a refusal is the watchman's S21), D10 (a want ahead of its have is the
|
||
// watchman's pulse), D5, D9 and H-controller (they end with the controller), stuck (the watchman's own finding), the
|
||
// bus step (its home is issue 1953's), and D6: the consumers it judged were the controller's, and the account consumers
|
||
// are the grants part's, which raises mesh-grants.consumer-differs and mesh-grants.consumer-behind (issue 1126).
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"slices"
|
||
"sort"
|
||
"strings"
|
||
|
||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||
)
|
||
|
||
// The tools a probe asks: the bus module's reads of its own server, and the watchdogs' seat.
|
||
const (
|
||
StreamsTool = "nats.nats_streams"
|
||
ClosedTool = "nats.nats_closed_connections"
|
||
SignalsVerb = "seat:mesh-watchdogs.signals"
|
||
notOnTheBus = "the bus module's tools (nats_streams) answer this only once asked; "
|
||
storeOnly = "held only in the controller's store today"
|
||
reconnectMax = 12
|
||
)
|
||
|
||
// Invokes are the tools and seat verbs the probes ask, and Reads the states they read: the manifest's `invokes` and
|
||
// `reads`, which its test holds to these.
|
||
var (
|
||
Invokes = []string{StreamsTool, ClosedTool, SignalsVerb, BannedVerb}
|
||
Reads = []string{PlacementRead, SeatHoldersRead, PresenceRead, HealthRead, ReadingsRead, GoodRead, ModulesRead, ManifestsRead}
|
||
)
|
||
|
||
// Registry is the self-check's registry, in to-be 45's order. **The registry is the design's live form**: a probe
|
||
// that moves here is a row added, and the registry test holds it to ADR 0564's table.
|
||
func Registry() []Probe {
|
||
return []Probe{
|
||
{ID: "D2", Asserts: "every holder of the mesh's resolver answers a machine name for IPv4, and NODATA for IPv6",
|
||
From: "issue 262", Kind: "resolver-wrong",
|
||
Deferred: "the machines' names and the resolver holders' addresses to ask: the overlay's addresses are " + storeOnly},
|
||
{ID: "D3", Asserts: "every seat on record that serves verbs has a live holder that answers, on every machine that " +
|
||
"is heard from", From: "issues 208, 218", Kind: "holder-silent",
|
||
Deferred: "the seats on record with their holders per machine, and which machines are heard from: the seat " +
|
||
"records are " + storeOnly + ", and the watchman's presence has no read granted to this module"},
|
||
{ID: "D4", Asserts: "every kept archive is held by a manifest", From: "issue 253", Kind: "archives-unheld",
|
||
Deferred: "the artifact store's kept archives with the manifests that hold them: the archive index is " + storeOnly},
|
||
{ID: "D7", Asserts: "every stream the mesh defines exists with its definition", From: "issue 208",
|
||
Kind: "stream-wrong", Run: probeStreams},
|
||
{ID: "D8", Asserts: "no address the mesh owns — a machine's private address or its endpoint — is in a ban list",
|
||
From: "issue 238", Kind: "own-address-banned", Run: probeBans},
|
||
{ID: "D11", Asserts: "no provider holds a consumer retired more than thirty days without a person deciding its " +
|
||
"cleanup", From: "ADR 0230", Kind: "cleanup-waiting",
|
||
Deferred: "the retirement records of providers' consumers: " + storeOnly},
|
||
{ID: "D12", Asserts: "every consumer of a provision that keeps its data is bound where it was last sent, or moves " +
|
||
"by a pin", From: "issue 273, ADR 0232", Kind: "binding-moved",
|
||
Deferred: "the bindings of provisions to their consumers: " + storeOnly},
|
||
{ID: "D13", Asserts: "every item of data a machine declares is measured, is there, holds what it held, and is " +
|
||
"backed up within its bound", From: "issue 273, ADR 0233", Kind: "data-shrank",
|
||
Raises: []string{kindDataMissing, kindDataQuiet, kindBackupStale, kindArrayDegraded, kindProtectionMissing,
|
||
kindDataUnmeasured},
|
||
Run: probeData},
|
||
{ID: "D15", Asserts: "no user of the bus had its connection dropped more than twelve times in the last hour: the " +
|
||
"bus module's nats_closed_connections", From: "issue 327", Kind: "bus-reconnects", Run: probeReconnects},
|
||
{ID: "DA", Asserts: "every machine that names an agent account has it judged, on its node-engine's newest " +
|
||
"statement, unable to become root without a person", From: "ADR 0266, ADR 0259 §8",
|
||
Kind: "agent-can-become-root", Run: probeAgentAccounts},
|
||
{ID: "D-root", Asserts: "no agent can become root without a person on a machine where the router or a channel " +
|
||
"proving its sender runs", From: "ADR 0259 §8", Kind: "root-not-free",
|
||
Deferred: "the node-engines' statements of root and the tools that run as root: " + storeOnly},
|
||
{ID: "D-fill", Asserts: "every module's bucket and log holds less than three quarters of its cap; one said " +
|
||
"filling is cleared once it holds less than seven tenths", From: "ADR 0297, issue 501",
|
||
Kind: "bucket-or-log-filling", Run: probeFill},
|
||
{ID: "D-heard", Asserts: "no grant of a restricted stream names a module that composes nothing for it",
|
||
From: "ADR 0376 §2", Kind: "grant-composes-nothing",
|
||
Deferred: "the grants of restricted streams with what each machine composed for them: " + storeOnly},
|
||
{ID: "DW", Asserts: "the watchdogs answer: the watchman holds the seat mesh-watchdogs and answers signals",
|
||
From: "ADR 0227 rule 6: the watchers are watched", Kind: "watchdogs-silent", Run: probeWatchdogs},
|
||
{ID: "H-bus", Asserts: "every stream the mesh defines is on the bus, and a request crosses it " +
|
||
"to the bus's machine and back", From: "ADR 0236, to-be 45 §8", Kind: "core-unhealthy", Run: probeBusHealth},
|
||
}
|
||
}
|
||
|
||
// --- the bus's own state, as its module answers it ---------------------------------------------------------------
|
||
|
||
// streamsAnswer is what nats_streams answers.
|
||
type streamsAnswer struct {
|
||
Detail []streamRow `json:"detail"`
|
||
}
|
||
|
||
type streamRow struct {
|
||
Name string `json:"name"`
|
||
Subjects []string `json:"subjects"`
|
||
Retention string `json:"retention"`
|
||
MaxPerSubject int64 `json:"max_per_subject"`
|
||
}
|
||
|
||
// errNoStream is a stream the bus module says it does not hold.
|
||
var errNoStream = errors.New("no such stream")
|
||
|
||
// streams asks the bus module for every stream, or for one with its consumers.
|
||
func (c *Checker) streams(stream string) ([]streamRow, error) {
|
||
args := map[string]any{}
|
||
if stream != "" {
|
||
args["stream"] = stream
|
||
}
|
||
raw, err := c.Ask(StreamsTool, args)
|
||
if err != nil {
|
||
if stream != "" && strings.Contains(err.Error(), "no stream called") {
|
||
return nil, errNoStream
|
||
}
|
||
return nil, fmt.Errorf("the bus module did not answer %s: %w", StreamsTool, err)
|
||
}
|
||
var a streamsAnswer
|
||
if err := json.Unmarshal(raw, &a); err != nil {
|
||
return nil, fmt.Errorf("the bus module answered %s with something unreadable: %w", StreamsTool, err)
|
||
}
|
||
return a.Detail, nil
|
||
}
|
||
|
||
// probeStreams is D7: every stream the mesh defines exists as defined.
|
||
func probeStreams(ctx context.Context, c *Checker) ([]Observation, error) {
|
||
rows, err := c.streams("")
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
held := map[string]streamRow{}
|
||
for _, r := range rows {
|
||
held[r.Name] = r
|
||
}
|
||
var out []Observation
|
||
for _, s := range buscompose.MeshStreams() {
|
||
have, ok := held[s.Name]
|
||
if !ok {
|
||
out = append(out, Observation{Ref: "stream-missing", Severity: Urgent, Subject: Subject{Scope: 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})
|
||
continue
|
||
}
|
||
if differs := streamDiffers(s, have); differs != "" {
|
||
out = append(out, Observation{Ref: "stream-redefined", Severity: Warning, Subject: Subject{Scope: ScopeBus, ID: s.Name},
|
||
Summary: fmt.Sprintf("the stream %s is not as the mesh defines it: %s", s.Name, differs),
|
||
Said: differs})
|
||
}
|
||
}
|
||
return sorted(out), nil
|
||
}
|
||
|
||
// streamDiffers is what about a stream on the bus is not as defined, of what the bus module says of it: its
|
||
// subjects, its retention and how many it keeps per subject. Empty when nothing is.
|
||
func streamDiffers(want buscompose.Stream, have streamRow) string {
|
||
var differs []string
|
||
haveSubjects := slices.Clone(have.Subjects)
|
||
wantSubjects := slices.Clone(want.Subjects)
|
||
sort.Strings(haveSubjects)
|
||
sort.Strings(wantSubjects)
|
||
if !slices.Equal(haveSubjects, wantSubjects) {
|
||
differs = append(differs, fmt.Sprintf("subjects %v, defined %v", haveSubjects, wantSubjects))
|
||
}
|
||
retention, perSubject := "limits", int64(want.MaxMsgsPerSubject)
|
||
switch want.Retention {
|
||
case buscompose.RetentionWorkQueue:
|
||
retention = "workqueue"
|
||
case buscompose.RetentionLastPerSubject:
|
||
perSubject = 1
|
||
}
|
||
if have.Retention != "" && have.Retention != retention {
|
||
differs = append(differs, fmt.Sprintf("keeps by %s, defined %s", have.Retention, retention))
|
||
}
|
||
if perSubject != 0 && have.MaxPerSubject != perSubject {
|
||
differs = append(differs, fmt.Sprintf("keeps %d per subject, defined %d", have.MaxPerSubject, perSubject))
|
||
}
|
||
return strings.Join(differs, "; ")
|
||
}
|
||
|
||
// --- D15 -----------------------------------------------------------------------------------------------------
|
||
|
||
// closedAnswer is what nats_closed_connections answers.
|
||
type closedAnswer struct {
|
||
Hours float64 `json:"hours"`
|
||
Reaches bool `json:"reaches"`
|
||
Users []struct {
|
||
User string `json:"user"`
|
||
Closed int `json:"closed"`
|
||
Dropped int `json:"dropped"`
|
||
Names []struct {
|
||
Name string `json:"name"`
|
||
Dropped int `json:"dropped"`
|
||
Reasons map[string]int `json:"reasons"`
|
||
} `json:"names"`
|
||
} `json:"users"`
|
||
}
|
||
|
||
// probeReconnects is D15: each user of the bus whose connections were dropped more than twelve times an hour.
|
||
func probeReconnects(_ context.Context, c *Checker) ([]Observation, 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)
|
||
}
|
||
var read closedAnswer
|
||
if err := json.Unmarshal(raw, &read); err != nil {
|
||
return nil, fmt.Errorf("the bus module answered %s with something unreadable: %w", ClosedTool, err)
|
||
}
|
||
hours := read.Hours
|
||
if hours <= 0 {
|
||
hours = 1
|
||
}
|
||
var out []Observation
|
||
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
|
||
}
|
||
reasons := map[string]int{}
|
||
var names []string
|
||
for _, n := range u.Names {
|
||
if n.Dropped == 0 {
|
||
continue
|
||
}
|
||
names = append(names, fmt.Sprintf("%q ×%d", n.Name, n.Dropped))
|
||
for r, k := range n.Reasons {
|
||
if r != "Client Closed" {
|
||
reasons[r] += k
|
||
}
|
||
}
|
||
}
|
||
var why []string
|
||
for r, k := range reasons {
|
||
why = append(why, fmt.Sprintf("%s ×%d", r, k))
|
||
}
|
||
sort.Strings(why)
|
||
partial := ""
|
||
if !read.Reaches {
|
||
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, Observation{Kind: "bus-reconnects", Severity: Warning,
|
||
Subject: Subject{Scope: ScopeBus, ID: u.User, Machine: machine},
|
||
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, ", ")),
|
||
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"})
|
||
}
|
||
return sorted(out), nil
|
||
}
|
||
|
||
// busUserWords is a bus user as the operator says it, and the machine it is on.
|
||
func busUserWords(user string) (string, string) {
|
||
switch {
|
||
case user == "controller":
|
||
return "the controller", ""
|
||
case strings.HasPrefix(user, "node."):
|
||
m := strings.TrimPrefix(user, "node.")
|
||
return "the node-engine on " + m, m
|
||
}
|
||
if m, module, ok := strings.Cut(user, "."); ok {
|
||
if module == "node-tools" {
|
||
return "the tool runner on " + m, m
|
||
}
|
||
return module + " on " + m, m
|
||
}
|
||
return "a client of the bus", ""
|
||
}
|
||
|
||
// --- DW and H-bus --------------------------------------------------------------------------------------------
|
||
|
||
// 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) ([]Observation, error) {
|
||
if _, err := c.Ask(SignalsVerb, map[string]any{}); err != nil {
|
||
return []Observation{{Ref: "silent", Severity: Urgent, Subject: Subject{Scope: ScopeCore, ID: "watchdogs"},
|
||
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
|
||
}
|
||
return nil, nil
|
||
}
|
||
|
||
// probeBusHealth is H-bus: every stream the mesh defines is on the bus, and a request crosses it to the bus module's
|
||
// 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) ([]Observation, 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)
|
||
}
|
||
held := map[string]bool{}
|
||
for _, r := range rows {
|
||
held[r.Name] = true
|
||
}
|
||
var problems []string
|
||
for _, s := range buscompose.MeshStreams() {
|
||
if !held[s.Name] {
|
||
problems = append(problems, "the stream "+s.Name+" is missing")
|
||
}
|
||
}
|
||
if len(problems) == 0 {
|
||
return nil, nil
|
||
}
|
||
return []Observation{{Ref: "unhealthy", Severity: Urgent, Subject: Subject{Scope: ScopeBus, ID: "mesh"},
|
||
Summary: "the bus is not healthy: " + strings.Join(problems, "; "), Said: strings.Join(problems, "; ")}}, nil
|
||
}
|
||
|
||
// sorted is findings in a stated order, so two runs over one mesh say the same.
|
||
func sorted(out []Observation) []Observation {
|
||
sort.SliceStable(out, func(i, j int) bool { return out[i].Key() < out[j].Key() })
|
||
return out
|
||
}
|
||
|
||
func capital(s string) string {
|
||
if s == "" {
|
||
return s
|
||
}
|
||
return strings.ToUpper(s[:1]) + s[1:]
|
||
}
|