Compose a node's placed data items from the bus and move the readings layout into the sdk, so disk-load measures without a composed file and restic, mounts, nfs-server and the self-check read one layout (issue 1955, ADR 0571 clause 3)

2026-10-29 23:16:56 +01:00
parent dc78524e38
commit 196c5b6af2
8 changed files with 308 additions and 1 deletions
@@ -615,6 +615,9 @@ func healthChecks(t *testing.T) {
 		{Module: "artifact-store-tools", Holds: []Seat{{Name: "artifact-store", Emits: []string{"artifacts-gone"}}},
 			Reads: []string{delivery.SeatHoldersRead}},
 		{Module: delivery.AssignerModule, State: []Bucket{bucket(delivery.AssignerModule, delivery.SeatHoldersState)}},
+		// ADR 0571 clause 3: disk-load keeps the readings; the backup holder reads them.
+		{Module: "disk-load", State: []Bucket{bucket("disk-load", "readings")}},
+		{Module: "restic", Reads: []string{"disk-load.readings"}},
 		{Module: "telegram", Holds: []Seat{{Name: "telegram-sender", Accepts: []string{"send"}}},
 			Processes: []ProcessPrincipal{{Process: "w", Queues: []QueueOwner{{Process: "w", Stream: "M_WORK"}, {Process: "w", Stream: "KV_x"}}}}},
 		{Module: "nats", SnapshotsTheBus: true, Processes: []ProcessPrincipal{{Process: "nats-tools",
@@ -630,6 +633,8 @@ func healthChecks(t *testing.T) {
 		"a.mesh-vault": {"$JS.API.DIRECT.GET.KV_mesh-provisioner_provisions.$KV.mesh-provisioner_provisions.a.keycloak.postgres-database",
 			"$JS.API.DIRECT.GET.KV_mesh-provisioner_plans.$KV.mesh-provisioner_plans.a.keycloak"},
 		"a.mesh-hand-acts": {"$KV.mesh-hand-acts_warrants.mesh-config.ask-1", "mesh.log.mesh-hand-acts.acts.act-1"},
+		"a.disk-load":      {"$KV.disk-load_readings.a.pg.data"},
+		"a.restic":         {"$JS.API.DIRECT.GET.KV_disk-load_readings.$KV.disk-load_readings.a.pg.data"},
 		WatchmanModule: {"$JS.API.CONSUMER.CREATE.DEAD_LETTER_NOTICES." + WatchmanModule, "$JS.API.CONSUMER.MSG.NEXT.DEAD_LETTER_NOTICES." + WatchmanModule,
 			"$JS.API.CONSUMER.INFO.DEAD_LETTER_NOTICES.controller", "$JS.API.STREAM.MSG.GET.DEAD_LETTERS"},
 		WatchmanModule + " sources": {"$JS.API.STREAM.MSG.GET.EVENTS", "$JS.API.STREAM.INFO.SEAT_TELEGRAM_SENDER",
@@ -648,7 +653,9 @@ func healthChecks(t *testing.T) {
 			"mesh.seat.artifact-store.event.artifacts-gone"},
 		// Only the assigner writes seat-holders.
 		"a.artifact-store-tools": {"$KV.mesh-delivery-assigner_seat-holders.mesh-broker"},
-		"a.postgres":             {"$KV.mesh-provisioner_provisions.a.keycloak.postgres-database", "$KV.mesh-provisioner_plans.a.keycloak"},
+		// Only disk-load writes its readings.
+		"a.restic":   {"$KV.disk-load_readings.a.pg.data"},
+		"a.postgres": {"$KV.mesh-provisioner_provisions.a.keycloak.postgres-database", "$KV.mesh-provisioner_plans.a.keycloak"},
 		// Nor mesh-hand-acts' log and warrants, which it alone writes (ADR 0565 clause 10).
 		"a.mesh-healers": {"$KV.mesh-conditions_conditions.x", "$KV.mesh-conditions_takeover.switch",
 			"$KV.mesh-hand-acts_warrants.x", "mesh.log.mesh-hand-acts.acts.x"},
@@ -0,0 +1,98 @@
+package catalogue
+
+import (
+	"fmt"
+	"regexp"
+	"strings"
+)
+
+// **The data items of a node's modules, placed, read from the composition** (novox/hq ADR 0571 clause 3, ADR 0233 §22 to
+// §25): disk-load measures each item every module on its node declares, and the node-backup holder, mounts and nfs-server
+// read the readings. Each item is its module's declaration with its path resolved to this node's place, its protection
+// and how it is measured: the same lines the node-backup holder is handed as `${backup:data}`, composed by the same
+// function, so a reader that composes from the bus and a holder handed the file read one list.
+
+// PlacedDataItem is one data item a module on a node declares, as placed on that node.
+type PlacedDataItem struct {
+	Module string `json:"module"`
+	ID     string `json:"item"`
+	Class  string `json:"class"`
+	// Path is the item's place on the node; CoveredBy the path whose snapshot keeps it (the item, or the item its dump
+	// writes into), empty when it is not backed up.
+	Path      string `json:"path"`
+	CoveredBy string `json:"covered_by,omitempty"`
+	// Protection is backup, redundancy, both or none; Measure walk, dataset or shallow.
+	Protection string `json:"protection,omitempty"`
+	Measure    string `json:"measure,omitempty"`
+}
+
+var (
+	dataModuleHeader = regexp.MustCompile(`^#\s*([a-z0-9][a-z0-9-]*)$`)
+	dataSafePath     = regexp.MustCompile(`^/[^\s'"\\$` + "`" + `]*$`)
+)
+
+// ParseDataItems reads the node-backup holder's data text (`${backup:data}`: a `# <module>` line, then one line per item,
+// `item <id> <class> <path> <covered-by or -> <protection> <measure>`) into each module's items, in order. A line it does
+// not read is refused, naming the module: a path with a space or a quote is never handed to a measuring shell.
+func ParseDataItems(text string) (map[string][]PlacedDataItem, error) {
+	out := map[string][]PlacedDataItem{}
+	module := ""
+	for _, raw := range strings.Split(text, "\n") {
+		line := strings.TrimSpace(raw)
+		if line == "" {
+			continue
+		}
+		if m := dataModuleHeader.FindStringSubmatch(line); m != nil {
+			module = m[1]
+			continue
+		}
+		if strings.HasPrefix(line, "#") || module == "" {
+			continue
+		}
+		f := strings.Fields(line)
+		if len(f) < 5 || len(f) > 7 || f[0] != "item" || !dataSafePath.MatchString(f[3]) || (f[4] != "-" && !dataSafePath.MatchString(f[4])) {
+			return nil, fmt.Errorf("%s declares a data line that is not read: %s", module, line)
+		}
+		it := PlacedDataItem{Module: module, ID: f[1], Class: f[2], Path: f[3], Measure: MeasureWalk}
+		if len(f) >= 6 {
+			it.Protection = f[5]
+		}
+		if len(f) == 7 {
+			it.Measure = f[6]
+		}
+		if it.Measure != MeasureWalk && it.Measure != MeasureDataset && it.Measure != MeasureShallow {
+			return nil, fmt.Errorf("%s declares a data line that is not read: %s", module, line)
+		}
+		if f[4] != "-" {
+			it.CoveredBy = f[4]
+		}
+		out[module] = append(out[module], it)
+	}
+	return out, nil
+}
+
+// DataItems is every data item the resolution's modules declare, placed on its node with the rendering given: the text
+// the node-backup holder is handed, read. A module left out of the node's declaration keeps its items, as its data is
+// still on the node (ADR 0262); an item that cannot be placed is said in unplaced and left out.
+func (r Resolution) DataItems(with Rendering) (items map[string][]PlacedDataItem, unplaced []string, err error) {
+	left := r.LeftOut(with.Settings, with.Adopted)
+	for module, why := range with.NotPlaced {
+		if _, already := left[module]; !already {
+			left[module] = why
+		}
+	}
+	modules := make([]Manifest, 0, len(r.Modules))
+	for _, m := range r.Modules {
+		if _, isLeft := left[m.Module]; isLeft {
+			m = backupView(m)
+		}
+		modules = append(modules, m)
+	}
+	text, unplaced, err := seatContributions(modules, Manifest{}, BackupSeat+":"+BackupKindData, with,
+		machineFacts(r, with.Names, with.MeshRange), r.Capabilities)
+	if err != nil {
+		return nil, unplaced, err
+	}
+	items, err = ParseDataItems(text)
+	return items, unplaced, err
+}
@@ -192,6 +192,16 @@ func TestTheBackupHoldersLinesAreDerivedFromTheData(t *testing.T) {
 	if data != want {
 		t.Fatalf("data lines:\n%s\nwant:\n%s", data, want)
 	}
+	// ADR 0571 clause 3: the same text, read into items, as disk-load and the backup holder read it; a path a shell
+	// would split is refused.
+	items, err := ParseDataItems(data)
+	if err != nil || len(items) != 3 || items["media"][0] != (PlacedDataItem{Module: "media", ID: "films", Class: "irreplaceable",
+		Path: "/tank/films", Protection: "redundancy", Measure: MeasureDataset}) || items["pg"][0].CoveredBy != "/srv/store/dumps" {
+		t.Fatalf("the data lines read as %+v: %v", items, err)
+	}
+	if _, err := ParseDataItems("# a\nitem x cache /a b - none walk\n"); err == nil {
+		t.Fatal("a path with a space was read")
+	}
 	if _, _, err := seatContributions([]Manifest{agent}, Manifest{}, BackupSeat+":"+BackupKindData, Rendering{}, nil, nil); err == nil {
 		t.Fatal("a home directory with no account on the machine reached the holder as a placeholder")
 	}
@@ -853,3 +853,13 @@ func membershipOf(own delivery.Composed) string {
 	}
 	return ""
 }
+
+// DataItems is every data item the modules placed on scope's node declare, placed there, composed from the bus as Plans
+// composes (novox/hq ADR 0571 clause 3): disk-load measures these.
+func DataItems(ctx context.Context, r Reader, mesh Mesh, s Scope) (map[string][]catalogue.PlacedDataItem, []string, error) {
+	in, err := Input(ctx, r, mesh, s)
+	if err != nil {
+		return nil, nil, err
+	}
+	return compose.DataItems(in, s.Node)
+}
@@ -712,6 +712,18 @@ func TestInput(t *testing.T) {
 		wj, _ := json.MarshalIndent(controllers, "", " ")
 		t.Errorf("the seat's plans differ from the controller's:\n got %s\nwant %s", gj, wj)
 	}
+	// ADR 0571 clause 3: the data items of n1's modules, placed, from the bus alone: what disk-load measures.
+	items, unplaced, err := DataItems(ctx, noSecrets, Mesh{}, Scope{Node: "n1"})
+	if err != nil || len(unplaced) > 0 || len(items) == 0 {
+		t.Errorf("the data items from the bus are %v, unplaced %v: %v", items, unplaced, err)
+	}
+	for module, its := range items {
+		for _, it := range its {
+			if it.Module != module || !strings.HasPrefix(it.Path, "/") || it.Measure == "" {
+				t.Errorf("%s's item %+v is not placed", module, it)
+			}
+		}
+	}

 	// And what a provider on another node serves from its settings (mailu's smtp domain, issue 1126: shanks refused
 	// '${setting:domain} ... nothing sets "domain"'): read for that provider's served keys alone, through Served.
@@ -333,3 +333,19 @@ func namedModules(plan catalogue.Resolution, leftOut map[string]string) []string
 	sort.Strings(out)
 	return out
 }
+
+// DataItems is every data item the modules placed on a node declare, placed there (novox/hq ADR 0571 clause 3): what the
+// node-backup holder is handed as `${backup:data}`, composed over the Input as Compose composes the node, so disk-load
+// reads the list from the bus with no composed file. Unplaced names each item left out and why.
+func DataItems(in Input, node string) (map[string][]catalogue.PlacedDataItem, []string, error) {
+	c := newComposer(in)
+	resolved, err := c.planFor(node)
+	if err != nil {
+		return nil, nil, err
+	}
+	rendered, err := c.render(node, Planning)
+	if err != nil {
+		return nil, nil, err
+	}
+	return resolved.Plan.DataItems(rendered.Rendering)
+}
@@ -7,6 +7,8 @@ import (
 	"strings"
 	"testing"
 	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
 )

 // ADR 0564 clause 3 (issue 1952 S2): the bodies of condition-raised, condition-changed and condition-cleared read the
@@ -65,6 +67,31 @@ func TestTheControllersConditionEventsReadUnchanged(t *testing.T) {
 // take-over key the readers follow.
 func TestTheHealthLayoutsHoldTheKeepersRules(t *testing.T) {
 	at := time.Date(2026, 10, 29, 12, 0, 0, 0, time.UTC)
+	// ADR 0571 clause 3: a reading keeps its seven-day peak and is written only when it changes; a partial size is no peak.
+	var prev *Reading
+	for i, step := range []struct {
+		day, size      int64
+		partial, wrote bool
+		peak           int64
+	}{{20, 100, false, true, 100}, {20, 100, false, false, 100}, {21, 300, false, true, 300}, {22, 50, false, true, 300},
+		{23, 9000, true, true, 300}, {29, 10, false, true, 50}, {30, 10, false, true, 10}} {
+		now := time.Date(2026, 10, int(step.day), 1, 0, 0, 0, time.UTC)
+		size := step.size
+		m := Measured{SizeBytes: &size, MeasuredAt: &now, Precision: "exact"}
+		if step.partial {
+			m.Precision = "partial: stopped"
+		}
+		r, wrote := NextReading(prev, "ace", catalogue.PlacedDataItem{Module: "pg", ID: "data", Path: "/srv/pg"}, m, now)
+		raw, _ := json.Marshal(r)
+		back, err := DecodeReading(raw)
+		if wrote != step.wrote || r.Peak == nil || *r.Peak != step.peak || err != nil || back.ID != "data" {
+			t.Fatalf("reading %d: wrote %v, peak %v, read back %v", i, wrote, r.Peak, err)
+		}
+		prev = &r
+	}
+	if ReadingKey("ace", "pg", "data") != "ace.pg.data" || ReadingsBucket != "disk-load_readings" {
+		t.Fatal("the readings' key or bucket moved")
+	}
 	good := Finding{Key: "provider.keycloak.ace.mesh-mail.failing", Kind: "provider-failing", Token: "failing",
 		Subject:  Subject{Scope: ScopeProvider, ID: "keycloak.ace.mesh-mail", Machine: "ace"},
 		Severity: Warning, Summary: "keycloak on ace fails mesh-mail", Source: "S8", Observer: WatchmanModule, At: at}
@@ -0,0 +1,127 @@
+package health
+
+import (
+	"encoding/json"
+	"fmt"
+	"reflect"
+	"strings"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+)
+
+// **Disk usage readings and peaks per data item** (novox/hq ADR 0571 clause 3, ADR 0233 §21, §27, §34): disk-load, on every
+// node, measures each data item the modules of its node declare (catalogue.PlacedDataItem, from compose/frombus.DataItems)
+// and keeps the newest reading and the item's largest size in seven days as its own state `readings`, one key
+// `<node>.<module>.<item>`, written only when it changes. The node-backup holder, mounts, nfs-server and the self-check's
+// probes read it there; the controller keeps none of it. Moved here from mesh-catalog modules/disk-load (b4c237d2).
+const (
+	DiskLoadModule = "disk-load"
+	ReadingsState  = "readings"
+	ReadingsBucket = DiskLoadModule + "_" + ReadingsState
+	ReadingsRead   = DiskLoadModule + "." + ReadingsState
+)
+
+// ReadingVersion is the version of a reading's layout.
+const ReadingVersion = 1
+
+// PeakWindow is how far back the peak looks (ADR 0233 §34).
+const PeakWindow = 7 * 24 * time.Hour
+
+// Redundancy is the redundant storage an item is on, as read: zfs, md or btrfs, its pool, array or filesystem, and its
+// health (nil when it could not be read).
+type Redundancy struct {
+	Kind    string `json:"kind"`
+	Where   string `json:"where"`
+	Healthy *bool  `json:"healthy"`
+	Said    string `json:"said"`
+}
+
+// Measured is what one measurement found. Precision says what the size is: "exact", "dataset <name>: its whole size",
+// "partial: ..." (a lower bound, never compared), or "no size: ...".
+type Measured struct {
+	SizeBytes  *int64      `json:"size_bytes,omitempty"`
+	LastWrite  *time.Time  `json:"last_write,omitempty"`
+	MeasuredAt *time.Time  `json:"measured_at,omitempty"`
+	Precision  string      `json:"precision,omitempty"`
+	Error      string      `json:"error,omitempty"`
+	Redundancy *Redundancy `json:"redundancy,omitempty"`
+}
+
+// Partial says the size is only a lower bound, which a peak is never made of.
+func (m Measured) Partial() bool { return strings.HasPrefix(m.Precision, "partial") }
+
+// DaySize is the largest size measured on one UTC day.
+type DaySize struct {
+	Day   string `json:"day"`
+	Bytes int64  `json:"bytes"`
+}
+
+// Reading is the value of one key: the item as placed, what was measured, and the largest size in the seven days up to
+// it (Peak, measured on PeakAt), read from the day maxima Days.
+type Reading struct {
+	V    int    `json:"v"`
+	Node string `json:"node"`
+	catalogue.PlacedDataItem
+	Measured
+	Peak   *int64    `json:"peak_bytes,omitempty"`
+	PeakAt string    `json:"peak_day,omitempty"`
+	Days   []DaySize `json:"days,omitempty"`
+}
+
+// ReadingKey is the key of an item on a node: `<node>.<module>.<item>`.
+func ReadingKey(node, module, item string) string { return node + "." + module + "." + item }
+
+// DecodeReading reads a key's value; it refuses another version.
+func DecodeReading(raw []byte) (Reading, error) {
+	var r Reading
+	if err := json.Unmarshal(raw, &r); err != nil {
+		return Reading{}, fmt.Errorf("health: a reading is not readable: %w", err)
+	}
+	if r.V != ReadingVersion {
+		return Reading{}, fmt.Errorf("health: a reading of version %d, and this reads %d", r.V, ReadingVersion)
+	}
+	return r, nil
+}
+
+// NextReading is the reading after a measurement, and whether it differs from the one before in anything but the time it
+// was taken: a reading that did not change is not written. A measurement with no size (an error, a shallow item) carries
+// the last size's days on, so the peak outlives one failed measurement.
+func NextReading(prev *Reading, node string, it catalogue.PlacedDataItem, m Measured, now time.Time) (Reading, bool) {
+	r := Reading{V: ReadingVersion, Node: node, PlacedDataItem: it, Measured: m}
+	var days []DaySize
+	if prev != nil {
+		days = prev.Days
+	}
+	today := now.UTC().Format("2006-01-02")
+	cut := now.UTC().Add(-PeakWindow).Format("2006-01-02")
+	var kept []DaySize
+	todays := int64(-1)
+	for _, d := range days {
+		switch {
+		case d.Day == today:
+			todays = d.Bytes
+		case d.Day >= cut:
+			kept = append(kept, d)
+		}
+	}
+	if m.SizeBytes != nil && !m.Partial() && *m.SizeBytes > todays {
+		todays = *m.SizeBytes
+	}
+	if todays >= 0 {
+		kept = append(kept, DaySize{Day: today, Bytes: todays})
+	}
+	r.Days = kept
+	for _, d := range kept {
+		if r.Peak == nil || d.Bytes >= *r.Peak {
+			b := d.Bytes
+			r.Peak, r.PeakAt = &b, d.Day
+		}
+	}
+	if prev == nil {
+		return r, true
+	}
+	a, b := *prev, r
+	a.MeasuredAt, b.MeasuredAt = nil, nil
+	return r, !reflect.DeepEqual(a, b)
+}