jschoubben
bbd02004b8
mesh/delivery merged: source.main moved to novox/mesh-catalog@19714898986c
mesh/delivery-group group fix/l4-quick merged: every member is merged
mesh/check pass bbd02004b863 on 3d7830cfdc4f | unit 2729 run, 4570 kept, 16 skipped | integration 2975 run, 4493 kept, 25 skipped | over bound: unit, …
The self-check's D13 read every reading, good night, indexer module and manifest one key at a time through the runtime and never finished within its 30 s; it now keeps those states by watching them and judges from memory, and reads the setting `probes` once a run. Findings name their key's last part `ref`, since the runtime refuses a field named `token`; the self-check, the healers and the keeper are pinned to the mesh-sdk that reads it and lets a carried holder say its seat's events. The controller's own bus.nats.upgrade-failed ends with its lease, as the operator approved on 2026-10-31, so compare lists it apart. Quick vs thorough: quick, these fixes for L4's take-over now; the thorough fix (research 032 / ADR 0240's health view extension) is decided later. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
443 lines
16 KiB
Go
443 lines
16 KiB
Go
package main
|
|
|
|
// D13, the data the modules declare (novox/hq ADR 0233, ADR 0571 clause 3, ADR 0564 §6): every item of data a machine
|
|
// declares is measured, is there, holds what it held, and is backed up within its bound. The controller composed what
|
|
// every machine declares and asked each machine's backup holder what it measured; the facts now live on the bus: the
|
|
// readings disk-load keeps (its state `readings`, with the item's seven-day peak), the last good night per item restic
|
|
// keeps (its state `good`), and the manifests the indexer holds (for the bounds an item declares). Each kind, key and
|
|
// word is the controller's (mesh-controller cmd/mesh-controller/data.go), so the condition the keeper raises from it is
|
|
// the one the controller raised.
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"maps"
|
|
"slices"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"git.novox.be/novox/mesh-sdk/go/health"
|
|
)
|
|
|
|
// The states D13 reads (declared under `reads`).
|
|
const (
|
|
ReadingsRead = health.ReadingsRead
|
|
GoodRead = health.GoodRead
|
|
ModulesRead = delivery.IndexerModule + ".modules"
|
|
ManifestsRead = delivery.IndexerModule + ".manifests"
|
|
)
|
|
|
|
// The kinds D13 raises.
|
|
const (
|
|
kindDataShrank = "data-shrank"
|
|
kindEmptyReplacement = "empty-replacement"
|
|
kindDataHeldTwice = "data-held-twice"
|
|
kindDataQuiet = "data-quiet"
|
|
kindBackupStale = "backup-stale"
|
|
kindDataUnmeasured = "data-unmeasured"
|
|
kindDataMissing = "data-missing"
|
|
kindArrayDegraded = "array-degraded"
|
|
kindProtectionMissing = "protection-missing"
|
|
kindCleanupWaiting = "cleanup-waiting"
|
|
shrinkWindowDays = 7
|
|
cleanupAfter = 30 * 24 * time.Hour
|
|
shrinkFloor int64 = 16 << 20
|
|
)
|
|
|
|
// dataFacts is everything D13 judges: the readings, the last good nights, each module's manifest, and the machines whose
|
|
// backup holder was heard from.
|
|
type dataFacts struct {
|
|
readings []health.Reading
|
|
good map[string]time.Time // by health.GoodKey
|
|
manifests map[string]catalogue.Manifest
|
|
// unmeasured are the machines that hold node-backup, are heard from, and have no reading of anything.
|
|
unmeasured []string
|
|
now time.Time
|
|
}
|
|
|
|
// keptStates are the states D13 judges from memory (issue 1952): read one key at a time through the runtime, each
|
|
// read reopening its bucket, a run of D13 never finished within its bound. Each is watched once and kept as it changes.
|
|
var keptStates = []string{ReadingsRead, GoodRead, ModulesRead, ManifestsRead}
|
|
|
|
// view is one state kept in memory from its watch: ready once the watch handed over every current value, err when it
|
|
// was refused (then the next read asks again).
|
|
type view struct {
|
|
mu sync.Mutex
|
|
values map[string][]byte
|
|
ready chan struct{}
|
|
err error
|
|
}
|
|
|
|
// WatchKept starts watching every state D13 reads, so the first run judges from memory too. Called once the bundle serves.
|
|
func (c *Checker) WatchKept(ctx context.Context) {
|
|
for _, s := range keptStates {
|
|
go func(s string) { _, _ = c.kept(ctx, s) }(s)
|
|
}
|
|
}
|
|
|
|
// kept is every current value of the state, from its watch, which it starts on the first ask. It waits for the watch's
|
|
// current values within ctx; a state not yet kept, or one whose watch was refused, is an error, never an empty state.
|
|
func (c *Checker) kept(ctx context.Context, state string) (map[string][]byte, error) {
|
|
c.viewsMu.Lock()
|
|
v := c.views[state]
|
|
if v == nil {
|
|
v = &view{values: map[string][]byte{}, ready: make(chan struct{})}
|
|
if c.views == nil {
|
|
c.views = map[string]*view{}
|
|
}
|
|
c.views[state] = v
|
|
go func() {
|
|
err := c.Store.Watch(state, func(key string, value []byte, gone bool) {
|
|
v.mu.Lock()
|
|
defer v.mu.Unlock()
|
|
if gone {
|
|
delete(v.values, key)
|
|
} else {
|
|
v.values[key] = value
|
|
}
|
|
})
|
|
if err != nil {
|
|
c.viewsMu.Lock()
|
|
if c.views[state] == v {
|
|
delete(c.views, state)
|
|
}
|
|
c.viewsMu.Unlock()
|
|
v.err = err
|
|
}
|
|
close(v.ready)
|
|
}()
|
|
}
|
|
c.viewsMu.Unlock()
|
|
select {
|
|
case <-v.ready:
|
|
case <-ctx.Done():
|
|
return nil, fmt.Errorf("the state %s is not kept yet: its watch has not handed over every value (%v)", state, ctx.Err())
|
|
}
|
|
if v.err != nil {
|
|
return nil, fmt.Errorf("the state %s cannot be watched: %w", state, v.err)
|
|
}
|
|
v.mu.Lock()
|
|
defer v.mu.Unlock()
|
|
return maps.Clone(v.values), nil
|
|
}
|
|
|
|
// dataFacts reads them from the states the self-check keeps by watching them.
|
|
func (c *Checker) dataFacts(ctx context.Context) (dataFacts, error) {
|
|
f := dataFacts{good: map[string]time.Time{}, manifests: map[string]catalogue.Manifest{}, now: c.Now()}
|
|
states := map[string]map[string][]byte{}
|
|
for _, s := range keptStates {
|
|
values, err := c.kept(ctx, s)
|
|
if err != nil {
|
|
return f, err
|
|
}
|
|
states[s] = values
|
|
}
|
|
byNode := map[string]int{}
|
|
for _, k := range slices.Sorted(maps.Keys(states[ReadingsRead])) {
|
|
r, err := health.DecodeReading(states[ReadingsRead][k])
|
|
if err != nil {
|
|
continue
|
|
}
|
|
f.readings = append(f.readings, r)
|
|
byNode[r.Node]++
|
|
}
|
|
for _, raw := range states[GoodRead] {
|
|
if g, derr := health.DecodeGoodRow(raw); derr == nil && !g.LastGood.IsZero() {
|
|
f.good[health.GoodKey(g.Node, g.Module, g.Item)] = g.LastGood
|
|
}
|
|
}
|
|
for name, raw := range states[ModulesRead] {
|
|
// Only the manifest's digest of the indexer's module, and the manifest's text: what the bounds are read from.
|
|
var m struct {
|
|
Manifest string `json:"manifest"`
|
|
}
|
|
if json.Unmarshal(raw, &m) != nil || m.Manifest == "" {
|
|
continue
|
|
}
|
|
mraw, ok := states[ManifestsRead][m.Manifest]
|
|
if !ok {
|
|
continue
|
|
}
|
|
var doc struct {
|
|
JSON string `json:"manifest"`
|
|
}
|
|
if json.Unmarshal(mraw, &doc) != nil {
|
|
continue
|
|
}
|
|
var man catalogue.Manifest
|
|
if json.Unmarshal([]byte(doc.JSON), &man) == nil {
|
|
f.manifests[name] = man
|
|
}
|
|
}
|
|
heard, err := c.heard()
|
|
if err != nil {
|
|
return f, err
|
|
}
|
|
holders, err := c.holdersOf("node-backup")
|
|
if err != nil {
|
|
return f, err
|
|
}
|
|
for _, n := range holders {
|
|
if heard[n] && byNode[n] == 0 {
|
|
f.unmeasured = append(f.unmeasured, n)
|
|
}
|
|
}
|
|
return f, nil
|
|
}
|
|
|
|
// probeData is D13.
|
|
func probeData(ctx context.Context, c *Checker) ([]Observation, error) {
|
|
f, err := c.dataFacts(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return dataFindings(f), nil
|
|
}
|
|
|
|
func severityOf(class string) health.Severity {
|
|
if class == catalogue.ClassIrreplaceable {
|
|
return Urgent
|
|
}
|
|
return Warning
|
|
}
|
|
|
|
func keyOfItem(machine, module, item string) string { return machine + "." + module + "." + item }
|
|
|
|
func comparable(precision string) bool {
|
|
return precision == "" || precision == "exact" || strings.HasPrefix(precision, "dataset ")
|
|
}
|
|
|
|
func datasetOf(precision string) string {
|
|
rest, ok := strings.CutPrefix(precision, "dataset ")
|
|
if !ok {
|
|
return ""
|
|
}
|
|
name, _, _ := strings.Cut(rest, ":")
|
|
return name
|
|
}
|
|
|
|
func orUnknownPath(p string) string {
|
|
if p == "" {
|
|
return "a path its backup holder never named"
|
|
}
|
|
return p
|
|
}
|
|
|
|
func sizeWords(size *int64) string {
|
|
if size == nil || *size < 0 {
|
|
return "size unknown"
|
|
}
|
|
b := float64(*size)
|
|
for _, unit := range []string{"B", "KB", "MB", "GB"} {
|
|
if b < 1024 || unit == "GB" {
|
|
if unit == "B" {
|
|
return fmt.Sprintf("%d B", *size)
|
|
}
|
|
return fmt.Sprintf("%.1f %s", b, unit)
|
|
}
|
|
b /= 1024
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// firstSeen is when an item was first measured on its node (the reading keeps it); for a reading written before it did,
|
|
// the oldest day its seven-day peak is made of, else its measurement: a lower bound on how long it has been declared.
|
|
func firstSeen(r health.Reading, now time.Time) time.Time {
|
|
if !r.FirstSeen.IsZero() {
|
|
return r.FirstSeen
|
|
}
|
|
if len(r.Days) > 0 {
|
|
if t, err := time.Parse("2006-01-02", r.Days[0].Day); err == nil {
|
|
return t
|
|
}
|
|
}
|
|
if r.MeasuredAt != nil {
|
|
return *r.MeasuredAt
|
|
}
|
|
return now
|
|
}
|
|
|
|
func obs(kind string, id, machine string, sev health.Severity, summary, said string) Observation {
|
|
return Observation{Kind: kind, Ref: kind, Severity: sev, Summary: summary, Said: said,
|
|
Subject: Subject{Scope: ScopeMachine, ID: id, Machine: machine}}
|
|
}
|
|
|
|
// 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) []Observation {
|
|
var out []Observation
|
|
arrays := map[string][]health.Reading{}
|
|
type shrunk struct {
|
|
machine, dataset, class string
|
|
size, peak int64
|
|
items []string
|
|
}
|
|
shrunkDatasets := map[string]shrunk{}
|
|
byItem := map[string][]health.Reading{}
|
|
retiredCopies := map[string][]health.Reading{}
|
|
for _, m := range f.unmeasured {
|
|
o := obs(kindDataUnmeasured, m, m, Warning, fmt.Sprintf("%s's backup holder did not say what it measured of the data "+
|
|
"declared there, so nothing about that data is known", m), "disk-load keeps no reading of this machine's data")
|
|
o.Confirm = true
|
|
out = append(out, o)
|
|
}
|
|
rs := append([]health.Reading(nil), f.readings...)
|
|
sort.Slice(rs, func(i, j int) bool {
|
|
return keyOfItem(rs[i].Node, rs[i].Module, rs[i].ID) < keyOfItem(rs[j].Node, rs[j].Module, rs[j].ID)
|
|
})
|
|
for _, r := range rs {
|
|
class := r.Class
|
|
item, declared := f.manifests[r.Module].DataItem(r.ID)
|
|
id := keyOfItem(r.Node, r.Module, r.ID)
|
|
if !r.RetiredAt.IsZero() {
|
|
retiredCopies[r.Module+"/"+r.ID] = append(retiredCopies[r.Module+"/"+r.ID], r)
|
|
if f.now.Sub(r.RetiredAt) > cleanupAfter && catalogue.Watched(class) {
|
|
out = append(out, obs(kindCleanupWaiting, id, r.Node, Warning,
|
|
fmt.Sprintf("%s of %s on %s (%s, %s) has been retired %d days — kept where it was since %s; "+
|
|
"`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
|
|
}
|
|
byItem[r.Module+"/"+r.ID] = append(byItem[r.Module+"/"+r.ID], r)
|
|
if !catalogue.Watched(class) {
|
|
continue
|
|
}
|
|
severity := severityOf(class)
|
|
if r.Redundancy != nil {
|
|
where := r.Node + "/" + r.Redundancy.Kind + ":" + r.Redundancy.Where
|
|
arrays[where] = append(arrays[where], r)
|
|
} else if declared && item.Redundancy != "" && r.MeasuredAt != nil && r.Error == "" {
|
|
out = append(out, obs(kindProtectionMissing, id, r.Node, severity,
|
|
fmt.Sprintf("%s of %s on %s (%s) is said to be protected by the redundancy of the storage it is on, "+
|
|
"and it is on nothing the backup holder can read as redundant: it has no protection the mesh can see",
|
|
r.ID, r.Module, r.Node, class),
|
|
orUnknownPath(r.Path)+" is on no redundant storage the backup holder can read"))
|
|
}
|
|
peak := r.Peak
|
|
shrank := peak != nil && r.SizeBytes != nil && comparable(r.Precision) && *r.SizeBytes*2 < *peak && *peak-*r.SizeBytes >= shrinkFloor
|
|
switch {
|
|
case r.Error != "" && strings.Contains(r.Error, "does not exist"):
|
|
out = append(out, obs(kindDataMissing, id, r.Node, severity,
|
|
fmt.Sprintf("%s of %s on %s (%s) is gone: where it was declared, nothing exists any more", r.ID, r.Module, r.Node, class),
|
|
orUnknownPath(r.Path)+" does not exist: "+firstLine(r.Error)))
|
|
case shrank && datasetOf(r.Precision) != "":
|
|
// Several items on one dataset share its size: one condition for the dataset, as loud as the most
|
|
// precious item on it.
|
|
k := r.Node + "/" + datasetOf(r.Precision)
|
|
ds := shrunkDatasets[k]
|
|
ds.machine, ds.dataset, ds.size, ds.peak = r.Node, datasetOf(r.Precision), *r.SizeBytes, *peak
|
|
ds.class = catalogue.StricterClass(ds.class, class)
|
|
ds.items = append(ds.items, r.Module+"/"+r.ID)
|
|
shrunkDatasets[k] = ds
|
|
case shrank:
|
|
out = append(out, obs(kindDataShrank, id, r.Node, severity,
|
|
fmt.Sprintf("%s of %s on %s (%s) shrank to %s from %s within %d days — more than half of what it "+
|
|
"held is gone. If that was meant, silence this with why; if not, `node-backup.restore` puts the last "+
|
|
"good copy beside it", r.ID, r.Module, r.Node, class, sizeWords(r.SizeBytes), sizeWords(peak), shrinkWindowDays), ""))
|
|
}
|
|
if !declared {
|
|
continue
|
|
}
|
|
if within := item.ActiveWithin(); within > 0 && r.LastWrite != nil && f.now.Sub(*r.LastWrite) > within {
|
|
out = append(out, obs(kindDataQuiet, id, r.Node, severity,
|
|
fmt.Sprintf("%s of %s on %s is written all the time, and has not been since %s (its bound is %s): "+
|
|
"whatever writes it has stopped", r.ID, r.Module, r.Node, r.LastWrite.UTC().Format(time.RFC3339), within), ""))
|
|
}
|
|
// A backup is required of what is irreplaceable and copied; of what is valuable it is the standard
|
|
// plan, said only where the machine was measured — where a holder is there to take it.
|
|
if item.BackedUp() && (class == catalogue.ClassIrreplaceable || r.MeasuredAt != nil) {
|
|
within := item.BackupWithin()
|
|
last, has := f.good[health.GoodKey(r.Node, r.Module, r.ID)]
|
|
switch {
|
|
case !has && f.now.Sub(firstSeen(r, f.now)) > within:
|
|
out = append(out, obs(kindBackupStale, id, r.Node, severity,
|
|
fmt.Sprintf("%s of %s on %s is %s and has no good backup on record, %s after it was first "+
|
|
"declared — is node-backup held there, and do its nights succeed? (`node-backup.backed-up`)",
|
|
r.ID, r.Module, r.Node, class, f.now.Sub(firstSeen(r, f.now)).Round(time.Hour)), ""))
|
|
case has && f.now.Sub(last) > within:
|
|
out = append(out, obs(kindBackupStale, id, r.Node, severity,
|
|
fmt.Sprintf("%s of %s on %s is %s and its newest good backup is from %s, older than its bound "+
|
|
"of %s", r.ID, r.Module, r.Node, class, last.UTC().Format(time.RFC3339), within), ""))
|
|
}
|
|
}
|
|
}
|
|
keys := make([]string, 0, len(shrunkDatasets))
|
|
for k := range shrunkDatasets {
|
|
keys = append(keys, k)
|
|
}
|
|
sort.Strings(keys)
|
|
for _, k := range keys {
|
|
ds := shrunkDatasets[k]
|
|
out = append(out, obs(kindDataShrank, ds.machine+".dataset."+strings.ReplaceAll(ds.dataset, "/", "-"), ds.machine,
|
|
severityOf(ds.class), fmt.Sprintf("the dataset on %s that holds %s shrank to %s from %s within %d days — more than "+
|
|
"half of what it held is gone", ds.machine, strings.Join(ds.items, ", "), sizeWords(&ds.size), sizeWords(&ds.peak),
|
|
shrinkWindowDays), "the dataset "+ds.dataset))
|
|
}
|
|
// The redundant storage watched data is on: one condition per array, as loud as the most precious item on it —
|
|
// the array, not each item, is what degrades.
|
|
wheres := make([]string, 0, len(arrays))
|
|
for w := range arrays {
|
|
wheres = append(wheres, w)
|
|
}
|
|
sort.Strings(wheres)
|
|
for _, where := range wheres {
|
|
items := arrays[where]
|
|
red := items[0].Redundancy
|
|
if red.Healthy != nil && *red.Healthy {
|
|
continue
|
|
}
|
|
class, machine := catalogue.ClassValuable, items[0].Node
|
|
var names []string
|
|
for _, r := range items {
|
|
class = catalogue.StricterClass(class, r.Class)
|
|
names = append(names, r.Module+"/"+r.ID)
|
|
}
|
|
state := "could not be read"
|
|
if red.Healthy != nil {
|
|
state = "is NOT healthy"
|
|
}
|
|
out = append(out, obs(kindArrayDegraded, machine+".array."+strings.NewReplacer("/", "-", ":", "-").Replace(red.Kind+"-"+red.Where),
|
|
machine, severityOf(class), fmt.Sprintf("the %s storage on %s that protects %s %s", red.Kind, machine,
|
|
strings.Join(names, ", "), state), fmt.Sprintf("the %s storage %s %s: %s", red.Kind, red.Where, state, firstLine(red.Said))))
|
|
}
|
|
// The same item on several machines: a copy that is in use and far smaller than one kept retired elsewhere is an
|
|
// empty replacement. Only against a retired copy — a module running on two machines on purpose keeps two different
|
|
// sets of data.
|
|
items := make([]string, 0, len(byItem))
|
|
for k := range byItem {
|
|
items = append(items, k)
|
|
}
|
|
sort.Strings(items)
|
|
for _, key := range items {
|
|
for _, a := range byItem[key] {
|
|
if a.SizeBytes == nil || !catalogue.Watched(a.Class) || !comparable(a.Precision) {
|
|
continue
|
|
}
|
|
for _, o := range retiredCopies[key] {
|
|
if o.Node == a.Node || o.SizeBytes == nil || !comparable(o.Precision) || !replacedByLess(*a.SizeBytes, *o.SizeBytes) {
|
|
continue
|
|
}
|
|
ob := obs(kindEmptyReplacement, keyOfItem(a.Node, a.Module, a.ID), a.Node, severityOf(a.Class),
|
|
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.Subject.Also = []string{o.Node}
|
|
out = append(out, ob)
|
|
break
|
|
}
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// replacedByLess is whether a copy in use is an empty replacement of a copy kept elsewhere: less than half of it, and at
|
|
// least shrinkFloor less.
|
|
func replacedByLess(inUse, kept int64) bool { return inUse*2 < kept && kept-inUse >= shrinkFloor }
|