Files
mesh-controller/cmd/mesh-controller/data_test.go
T
mesh-tools 013dad4688 Take the catalogue, overlay and validator from mesh-sdk in cmd/mesh-builder and cmd/mesh-controller, a to g (issue 1381, 1 of 6 import swaps)
Swaps three import paths for mesh-sdk's go/catalogue, go/overlay and go/validate (the same packages, copied from this repository at 6c2a8d62) and lets gofmt order the groups. The six swap commits compile together once the pin commit (go.mod, modules.txt) is on the branch; each alone does not, because a file on the SDK's types cannot call one still on the old package.
2026-10-20 08:22:07 +00:00

370 lines
19 KiB
Go

package main
import (
"encoding/json"
"strings"
"sync"
"testing"
"time"
"github.com/nats-io/nats.go"
"github.com/novox/mesh-controller/internal/link"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-controller/internal/conditions"
"github.com/novox/mesh-controller/internal/inventory"
)
func bytesOf(n int64) *int64 { return &n }
func when(t time.Time) *time.Time { return &t }
func shelfFor(t *testing.T, manifests ...string) map[string]catalogue.Manifest {
t.Helper()
out := map[string]catalogue.Manifest{}
for _, raw := range manifests {
m, err := catalogue.ParseManifest([]byte(raw))
if err != nil {
t.Fatal(err)
}
out[m.Module] = m
}
return out
}
const houseManifest = `{"module":"house","version":"1",
"data":{"own":[{"id":"config","path":"${dir:config}","class":"irreplaceable","active":"1d"}]},
"resources":[{"id":"config","type":"directory","mode":"0700"}]}`
func findingsByKind(obs []conditions.Observation) map[string]conditions.Observation {
out := map[string]conditions.Observation{}
for _, o := range linted(obs) {
out[o.Kind] = o
}
return out
}
// THE INCIDENT (issue 273), replayed against what D13 reads: five applications on the home server bound,
// by one changed rule, to the store on the control node, which made each an empty database — while
// their real databases, hundreds of megabytes each, sat on the home server's own store, by then retired
// because the mesh no longer asked for them there. Each is an empty replacement, naming both machines
// and the pin back: a warning for the store's consumers, whose data is valuable; urgent for one that says
// its data there is irreplaceable (`kept-by`).
func TestAnEmptyReplacementOfAConsumersDataIsSaid(t *testing.T) {
var copies []consumerCopy
for _, app := range []string{"mesh_home_board", "mesh_home_flows", "mesh_home_agents", "mesh_home_game", "mesh_home_cars"} {
copies = append(copies,
consumerCopy{Node: "home", Module: "postgres", Consumer: app, Size: bytesOf(400 << 20), Retired: true, Class: "valuable"},
consumerCopy{Node: "anchor", Module: "postgres", Consumer: app, Size: bytesOf(9 << 20), Class: "valuable"})
}
copies[1].Class = "irreplaceable" // the photo site's own database says so
got := linted(dataFindings(nil, nil, nil, copies, nil, time.Now()))
if len(got) != 5 {
t.Fatalf("%d findings for five empty replacements: %+v", len(got), got)
}
for i, o := range got {
want := conditions.Warning
if strings.Contains(o.ID, "mesh_home_board") {
want = conditions.Urgent
}
_ = i
if o.Kind != kindEmptyReplacement || o.Severity != want || o.Machine != "anchor" ||
len(o.Also) != 1 || o.Also[0] != "home" || !strings.Contains(o.Summary, "Pin it back to home") {
t.Errorf("%+v", o)
}
}
// While the old copy is still active (the first ten minutes), it is the same finding.
copies[0].Retired = false
copies[1].Class = "valuable"
if got := dataFindings(nil, nil, nil, copies[:2], nil, time.Now()); len(got) != 1 || got[0].Kind != kindEmptyReplacement {
t.Fatalf("with the old copy still active: %+v", got)
}
}
// A move a person made — the data moved first, then pinned — leaves a full copy at the new provider and
// a retired one at the old: nothing to say here; `cleanup` covers the old one.
func TestADeliberateMoveIsNoEmptyReplacement(t *testing.T) {
copies := []consumerCopy{
{Node: "home", Module: "postgres", Consumer: "mesh_home_board", Size: bytesOf(400 << 20), Retired: true},
{Node: "anchor", Module: "postgres", Consumer: "mesh_home_board", Size: bytesOf(402 << 20)},
}
if got := dataFindings(nil, nil, nil, copies, nil, time.Now()); len(got) != 0 {
t.Fatalf("a deliberate move raised %+v", got)
}
// Two small databases differing by less than the floor are not a finding either.
copies[0].Size, copies[1].Size = bytesOf(12<<20), bytesOf(8<<20)
if got := dataFindings(nil, nil, nil, copies, nil, time.Now()); len(got) != 0 {
t.Fatalf("noise between two empty databases raised %+v", got)
}
}
// Where a provider cannot say sizes, a consumer active at two providers is still said — as a warning,
// since which one is empty cannot be told.
func TestConsumerDataActiveTwiceWithoutSizesIsAWarning(t *testing.T) {
copies := []consumerCopy{
{Node: "home", Module: "minio", Consumer: "mesh_home_photos"},
{Node: "anchor", Module: "minio", Consumer: "mesh_home_photos"},
}
got := linted(dataFindings(nil, nil, nil, copies, nil, time.Now()))
if len(got) != 1 || got[0].Kind != kindDataHeldTwice || got[0].Severity != conditions.Warning {
t.Fatalf("%+v", got)
}
}
// The same incident for a module's own data: a module unassigned from one machine and assigned on
// another starts over in an empty directory while its full one is kept, retired, where it was.
func TestAnEmptyReplacementOfAModulesOwnDataIsUrgent(t *testing.T) {
now := time.Now()
retired := now.Add(-time.Hour)
records := []inventory.DataRecord{
{Machine: "home", Module: "house", Item: "config", Class: "irreplaceable", Path: "/var/lib/house/config",
Size: bytesOf(2 << 30), RetiredAt: &retired, FirstSeen: now.Add(-90 * 24 * time.Hour)},
{Machine: "anchor", Module: "house", Item: "config", Class: "irreplaceable", Path: "/var/lib/house/config",
Size: bytesOf(1 << 20), FirstSeen: now.Add(-time.Hour), LastWrite: when(now)},
}
got := findingsByKind(dataFindings(records, nil, shelfFor(t, houseManifest), nil, nil, now))
o, ok := got[kindEmptyReplacement]
if !ok || o.Severity != conditions.Urgent || o.Machine != "anchor" || o.Also[0] != "home" {
t.Fatalf("%+v", got)
}
// The same of a valuable item is a warning.
records[0].Class, records[1].Class = "valuable", "valuable"
if o := findingsByKind(dataFindings(records, nil, shelfFor(t, houseManifest), nil, nil, now))[kindEmptyReplacement]; o.Severity != conditions.Warning {
t.Fatalf("a valuable empty replacement: %+v", o)
}
records[0].Class, records[1].Class = "irreplaceable", "irreplaceable"
// Two machines running a module on purpose, both active, keep two sets of data: nothing to say.
records[0].RetiredAt = nil
if got := findingsByKind(dataFindings(records, nil, shelfFor(t, houseManifest), nil, nil, now)); got[kindEmptyReplacement].Kind != "" {
t.Fatalf("two active copies were read as a replacement: %+v", got)
}
}
// An irreplaceable item that lost more than half of its largest size in a week is urgent; a smaller loss,
// or a loss under the floor, is not a finding.
func TestAShrinkOfMoreThanHalfIsUrgent(t *testing.T) {
now := time.Now()
r := inventory.DataRecord{Machine: "home", Module: "house", Item: "config", Class: "irreplaceable",
Size: bytesOf(300 << 20), FirstSeen: now.Add(-30 * 24 * time.Hour), LastWrite: when(now), LastBackup: when(now)}
shelf := shelfFor(t, houseManifest)
o := findingsByKind(dataFindings([]inventory.DataRecord{r}, map[string]int64{r.Key(): 1 << 30}, shelf, nil, nil, now))[kindDataShrank]
if o.Severity != conditions.Urgent || !strings.Contains(o.Summary, "shrank") {
t.Fatalf("%+v", o)
}
for _, peak := range []int64{500 << 20, 20 << 20} {
if got := findingsByKind(dataFindings([]inventory.DataRecord{r}, map[string]int64{r.Key(): peak}, shelf, nil, nil, now)); got[kindDataShrank].Kind != "" {
t.Errorf("a peak of %d raised a shrink", peak)
}
}
small := r
small.Size = bytesOf(1 << 20)
if got := findingsByKind(dataFindings([]inventory.DataRecord{small}, map[string]int64{r.Key(): 10 << 20}, shelf, nil, nil, now)); got[kindDataShrank].Kind != "" {
t.Error("a loss under the floor raised a shrink")
}
}
// THE FALSE ALARM (issue 368), replayed through the store D13 reads: an agent's home moved from the
// operator's own home (94.7 MB) to the agent account's fresh one (490 B), and `data-shrank` was raised
// for data that was never lost. A moved item is read against its new path only, so nothing is raised —
// and a genuine shrink at the new path, a week of history later, still is.
// Data said to be written all the time and not written; data with no backup or an old one — urgent when
// irreplaceable, a warning when valuable; and a new item given its bound before it is said.
func TestQuietDataAndMissingBackupsAreSaidByClass(t *testing.T) {
now := time.Now()
shelf := shelfFor(t, houseManifest)
r := inventory.DataRecord{Machine: "home", Module: "house", Item: "config", Class: "irreplaceable",
Size: bytesOf(1 << 30), FirstSeen: now.Add(-10 * 24 * time.Hour), LastWrite: when(now.Add(-3 * 24 * time.Hour)),
LastBackup: when(now.Add(-72 * time.Hour))}
got := findingsByKind(dataFindings([]inventory.DataRecord{r}, nil, shelf, nil, nil, now))
if got[kindDataQuiet].Severity != conditions.Urgent || got[kindBackupStale].Severity != conditions.Urgent {
t.Fatalf("irreplaceable: %+v", got)
}
valuable := r
valuable.Class, valuable.MeasuredAt = "valuable", when(now) // measured: a holder is there to take its backup
if got := findingsByKind(dataFindings([]inventory.DataRecord{valuable}, nil, shelf, nil, nil, now)); got[kindDataQuiet].Severity != conditions.Warning ||
got[kindBackupStale].Severity != conditions.Warning {
t.Fatalf("valuable: %+v", got)
}
never := r
never.LastBackup = nil
if o := findingsByKind(dataFindings([]inventory.DataRecord{never}, nil, shelf, nil, nil, now))[kindBackupStale]; !strings.Contains(o.Summary, "no good backup") {
t.Fatalf("never backed up: %+v", o)
}
fresh := never
fresh.FirstSeen, fresh.LastWrite = now.Add(-time.Hour), when(now)
if got := dataFindings([]inventory.DataRecord{fresh}, nil, shelf, nil, nil, now); len(got) != 0 {
t.Fatalf("an item declared an hour ago, before its first night, raised %+v", got)
}
}
// An item retired more than thirty days waits for a person; less, it is only listed.
func TestRetiredDataWaitingThirtyDaysIsSaid(t *testing.T) {
now := time.Now()
old, recent := now.Add(-31*24*time.Hour), now.Add(-2*24*time.Hour)
records := []inventory.DataRecord{
{Machine: "home", Module: "house", Item: "config", Class: "irreplaceable", Size: bytesOf(1 << 30), RetiredAt: &old},
{Machine: "home", Module: "attic", Item: "boxes", Class: "irreplaceable", Size: bytesOf(1 << 30), RetiredAt: &recent},
}
got := linted(dataFindings(records, nil, shelfFor(t, houseManifest), nil, nil, now))
if len(got) != 1 || got[0].Kind != kindCleanupWaiting || !strings.Contains(got[0].Summary, "cleanup delete home house config") {
t.Fatalf("%+v", got)
}
if rows := retiredData(records, now); len(rows) != 2 || rows[0].Kind != retiredDataKind {
t.Fatalf("cleanup list: %+v", rows)
}
}
// The holder's answer reads into measurements, by module and item.
func TestTheHoldersAnswerIsRead(t *testing.T) {
raw := []byte(`[{"module":"postgres","runs":1,"paths":["/var/lib/mesh-store/dumps"],"lastNight":null,"restorePoints":3,
"data":[{"item":"store","class":"irreplaceable","path":"/var/lib/mesh-store","covered_by":"/var/lib/mesh-store/dumps",
"size_bytes":1073741824,"last_write":"2026-10-06T10:00:00Z","measured_at":"2026-10-06T10:05:00Z","last_backup":"2026-10-06T03:10:00Z"}]}]`)
got, err := readHolder(raw)
if err != nil {
t.Fatal(err)
}
m := got["postgres"]["store"]
if m.Path != "/var/lib/mesh-store" || m.Size == nil || *m.Size != 1<<30 || m.LastBackup == nil || m.MeasuredAt == nil {
t.Fatalf("%+v", m)
}
// An older holder, which says no data, reads as nothing measured rather than a failure.
if got, err := readHolder([]byte(`[{"module":"postgres","runs":1,"paths":[]}]`)); err != nil || len(got) != 0 {
t.Fatalf("%v, %v", got, err)
}
}
// fakeHolder answers node-backup's `backed-up` on one machine over a real bus, with what it is told it
// measured.
type fakeHolder struct {
mu sync.Mutex
modules []map[string]any
}
func (f *fakeHolder) set(modules ...map[string]any) {
f.mu.Lock()
defer f.mu.Unlock()
f.modules = modules
}
func (f *fakeHolder) serve(t *testing.T, conn *nats.Conn, node string) {
t.Helper()
sub, err := conn.Subscribe(link.NodeSeatToolSubject(catalogue.BackupSeat, "backed-up", node), func(m *nats.Msg) {
f.mu.Lock()
defer f.mu.Unlock()
body, _ := json.Marshal(map[string]any{"result": f.modules, "error": "", "node": node})
_ = m.Respond(body)
})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = sub.Unsubscribe() })
if err := conn.Flush(); err != nil {
t.Fatal(err)
}
}
func measuredHouse(path string, size int64, at time.Time) map[string]any {
return map[string]any{"module": "house", "runs": 0, "paths": []string{path}, "data": []map[string]any{{
"item": "config", "class": "irreplaceable", "path": path, "covered_by": path, "size_bytes": size,
"last_write": at, "measured_at": at, "last_backup": at}}}
}
// UNASSIGNING A MODULE WITH IRREPLACEABLE DATA KEEPS THE DATA, and assigning it elsewhere onto an empty
// directory is an empty replacement — through the real stores and a real bus. The unassignment says the
// data stays; the self-check's next run retires it (kept, listed by cleanup), and when the module comes
// up on another machine with an empty directory while the full one waits retired, that is urgent.
// Data on redundant storage: the array it is on is watched, one condition per array as loud as the most
// precious item on it; an item said to be on redundancy and found on plain storage is said; an item
// whose path is gone is said.
func TestTheArrayUnderDataIsWatched(t *testing.T) {
now := time.Now()
media := `{"module":"media","version":"1","accesses":[{"id":"films","mode":"read"},{"id":"shows","mode":"read"}],
"data":{"own":[{"id":"films","path":"${access:films}","class":"irreplaceable","redundancy":"an array, no room to copy"},
{"id":"shows","path":"${access:shows}","class":"irreplaceable","redundancy":"an array, no room to copy"}]}}`
shelf := shelfFor(t, media)
sick := false
on := func(item string, healthy *bool) inventory.DataRecord {
return inventory.DataRecord{Machine: "home", Module: "media", Item: item, Class: "irreplaceable", Owned: false,
Path: "/tank/" + item, Size: bytesOf(40 << 40), FirstSeen: now.Add(-24 * time.Hour), MeasuredAt: when(now),
Redundancy: &inventory.Redundancy{Kind: "zfs", Where: "tank", Healthy: healthy, Said: "pool 'tank' is DEGRADED"}}
}
got := linted(dataFindings([]inventory.DataRecord{on("films", &sick), on("shows", &sick)}, nil, shelf, nil, nil, now))
if len(got) != 1 || got[0].Kind != kindArrayDegraded || got[0].Severity != conditions.Urgent ||
!strings.Contains(got[0].Summary, "media/films, media/shows") {
t.Fatalf("%+v", got)
}
well := true
if got := dataFindings([]inventory.DataRecord{on("films", &well)}, nil, shelf, nil, nil, now); len(got) != 0 {
t.Fatalf("a healthy array raised %+v", got)
}
plain := on("films", nil)
plain.Redundancy = nil
if o := findingsByKind(dataFindings([]inventory.DataRecord{plain}, nil, shelf, nil, nil, now))[kindProtectionMissing]; o.Severity != conditions.Urgent {
t.Fatalf("redundancy said and not found: %+v", o)
}
gone := on("films", &well)
gone.MeasureError, gone.Size = "/tank/films does not exist", bytesOf(0)
if o := findingsByKind(dataFindings([]inventory.DataRecord{gone}, map[string]int64{gone.Key(): 40 << 40}, shelf, nil, nil, now))[kindDataMissing]; o.Severity != conditions.Urgent {
t.Fatalf("a vanished library: %+v", o)
}
}
// What a consumer keeps with its provider as irreplaceable holds the provider's item to that class:
// the photo site's objects make the object store's data an urgent matter.
func TestKeptByHoldsTheProvidersItemToTheConsumersClass(t *testing.T) {
objects := `{"module":"objects","version":"1","provides":[{"name":"s3-bucket","scope":"mesh"}],"grants":{"s3-bucket":"${dir:g}"},
"data":{"own":[{"id":"data","path":"${dir:data}","class":"valuable"}],"consumers":{"s3-bucket":{"class":"valuable","in":"data"}}},
"resources":[{"id":"g","type":"directory","mode":"0700"},{"id":"data","type":"directory","mode":"0700"}]}`
photos := `{"module":"photos","version":"1","requires":["s3-bucket"],"data":{"kept-by":{"s3-bucket":{"class":"irreplaceable"}}}}`
shelf := shelfFor(t, objects, photos)
bindings := []inventory.Binding{{Machine: "anchor", Consumer: "photos", Provision: "s3-bucket",
Provider: catalogue.Chosen{Node: "anchor", Module: "objects"}}}
upgraded, keptBy := keptByClasses(bindings, shelf)
if upgraded["anchor/objects/data"] != "irreplaceable" || keptBy["objects/mesh_anchor_photos"] != "irreplaceable" {
t.Fatalf("upgraded %v, kept by %v", upgraded, keptBy)
}
now := time.Now()
r := inventory.DataRecord{Machine: "anchor", Module: "objects", Item: "data", Class: "valuable", Owned: true,
Size: bytesOf(10 << 30), FirstSeen: now.Add(-10 * 24 * time.Hour), MeasuredAt: when(now), LastBackup: when(now.Add(-72 * time.Hour))}
if o := findingsByKind(dataFindings([]inventory.DataRecord{r}, nil, shelf, nil, upgraded, now))[kindBackupStale]; o.Severity != conditions.Urgent {
t.Fatalf("the photos' store without a backup: %+v", o)
}
if o := findingsByKind(dataFindings([]inventory.DataRecord{r}, nil, shelf, nil, nil, now))[kindBackupStale]; o.Severity != conditions.Warning {
t.Fatalf("a valuable store without a backup: %+v", o)
}
}
// Items measured from one dataset's counters share its size: a shrink of the dataset is one condition
// naming every item on it, not one per item; and a partial walk's lower bound is never compared.
func TestADatasetShrinksOnceAndAPartialSizeIsNeverCompared(t *testing.T) {
now := time.Now()
media := `{"module":"media","version":"1","accesses":[{"id":"films","mode":"read"},{"id":"shows","mode":"read"}],
"data":{"own":[{"id":"films","path":"${access:films}","class":"irreplaceable","redundancy":"an array","measure":"dataset"},
{"id":"shows","path":"${access:shows}","class":"irreplaceable","redundancy":"an array","measure":"dataset"}]}}`
shelf := shelfFor(t, media, houseManifest)
well := true
on := func(item string, size int64) inventory.DataRecord {
return inventory.DataRecord{Machine: "home", Module: "media", Item: item, Class: "irreplaceable",
Size: bytesOf(size), FirstSeen: now.Add(-24 * time.Hour), MeasuredAt: when(now),
Precision: "dataset tank/media: its whole size",
Redundancy: &inventory.Redundancy{Kind: "zfs", Where: "tank", Healthy: &well}}
}
films, shows := on("films", 30<<40), on("shows", 30<<40)
peaks := map[string]int64{films.Key(): 90 << 40, shows.Key(): 90 << 40}
got := linted(dataFindings([]inventory.DataRecord{films, shows}, peaks, shelf, nil, nil, now))
if len(got) != 1 || got[0].Kind != kindDataShrank || got[0].Severity != conditions.Urgent ||
!strings.Contains(got[0].Summary, "media/films, media/shows") {
t.Fatalf("%+v", got)
}
partial := inventory.DataRecord{Machine: "home", Module: "house", Item: "config", Class: "irreplaceable",
Size: bytesOf(1 << 20), FirstSeen: now.Add(-24 * time.Hour), MeasuredAt: when(now), LastWrite: when(now),
LastBackup: when(now), Precision: "partial: measured partially"}
if got := dataFindings([]inventory.DataRecord{partial}, map[string]int64{partial.Key(): 1 << 30}, shelf, nil, nil, now); len(got) != 0 {
t.Fatalf("a partial size was compared: %+v", got)
}
}