diff --git a/go/buscompose/buscompose_test.go b/go/buscompose/buscompose_test.go index 0f6b29f..5e356b4 100644 --- a/go/buscompose/buscompose_test.go +++ b/go/buscompose/buscompose_test.go @@ -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"}, diff --git a/go/catalogue/data_placed.go b/go/catalogue/data_placed.go new file mode 100644 index 0000000..0744fe6 --- /dev/null +++ b/go/catalogue/data_placed.go @@ -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 `# ` line, then one line per item, +// `item `) 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 +} diff --git a/go/catalogue/data_test.go b/go/catalogue/data_test.go index 9511cda..a013e9c 100644 --- a/go/catalogue/data_test.go +++ b/go/catalogue/data_test.go @@ -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") } diff --git a/go/compose/frombus/frombus.go b/go/compose/frombus/frombus.go index 8478333..2fedc4c 100644 --- a/go/compose/frombus/frombus.go +++ b/go/compose/frombus/frombus.go @@ -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) +} diff --git a/go/compose/frombus/frombus_test.go b/go/compose/frombus/frombus_test.go index 69d6333..1a4b562 100644 --- a/go/compose/frombus/frombus_test.go +++ b/go/compose/frombus/frombus_test.go @@ -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. diff --git a/go/compose/node.go b/go/compose/node.go index 858f77d..81e8557 100644 --- a/go/compose/node.go +++ b/go/compose/node.go @@ -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) +} diff --git a/go/health/condition_test.go b/go/health/condition_test.go index e62ea04..9c25aab 100644 --- a/go/health/condition_test.go +++ b/go/health/condition_test.go @@ -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} diff --git a/go/health/readings.go b/go/health/readings.go new file mode 100644 index 0000000..a0983c9 --- /dev/null +++ b/go/health/readings.go @@ -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 +// `..`, 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 : 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: `..`. +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) +}