disk-load: take the node's data items from the bus and the readings layout from the sdk, with no composed file (issue 1955, ADR 0571 clause 3)

This commit is contained in:
2026-10-29 23:46:56 +01:00
parent 45902cd728
commit f149768f5d
13 changed files with 309 additions and 288 deletions
+233
View File
@@ -0,0 +1,233 @@
package main
// The bus as disk-load composes a node's data items from it (frombus.Reader): placement, each node's record in its own
// have key, the indexer's modules and manifests, the builder's builds and mesh-config's resolved values, all through this
// module's runtime with its own `reads`. Items name no value, so no own secret and no credential is read. This is the
// reader mesh-provisioner has too; moving it into mesh-sdk's frombus is the way to keep one (code shared between
// modules belongs in the sdk).
import (
"context"
"crypto/sha256"
"encoding/hex"
"fmt"
"io"
"net/http"
"strings"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/compose"
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// The states this module reads to compose from.
const (
placementRead = delivery.PlacementRead
modulesRead = delivery.IndexerModule + ".modules"
manifestsRead = delivery.IndexerModule + ".manifests"
buildsRead = delivery.BuilderModule + ".builds"
haveRead = "node-engine.have"
resolvedRead = "mesh-config.resolved"
)
// entryReader is a read state as the runtime gives it.
type entryReader interface {
Get(state, key string) (*stdio.StateEntry, error)
Keys(state string) ([]string, error)
}
type busState struct{}
func (busState) Get(state, key string) (*stdio.StateEntry, error) { return stdio.State(state).Get(key) }
func (busState) Keys(state string) ([]string, error) { return stdio.State(state).Keys() }
// busReader reads the bus for frombus through a Store.
type busReader struct {
s entryReader
fetch func(ctx context.Context, url string) ([]byte, error)
}
func (r busReader) each(read string, visit func(key string, value []byte)) error {
keys, err := r.s.Keys(read)
if err != nil {
return fmt.Errorf("%s: %w", read, err)
}
for _, k := range keys {
e, err := r.s.Get(read, k)
if err != nil {
return fmt.Errorf("%s %s: %w", read, k, err)
}
if e != nil {
visit(k, e.Value)
}
}
return nil
}
func (r busReader) Placements(context.Context) (map[string]delivery.Placement, error) {
out := map[string]delivery.Placement{}
err := r.each(placementRead, func(k string, v []byte) {
node, err := delivery.ParsePlacementKey(k)
if err != nil {
return
}
if p, err := delivery.DecodePlacement(v); err == nil && p.Node == node {
out[node] = p
}
})
return out, err
}
func (r busReader) Records(context.Context) (map[string]delivery.NodeRecord, error) {
out := map[string]delivery.NodeRecord{}
err := r.each(haveRead, func(k string, v []byte) {
node, _, isNode, err := delivery.ParseHaveKey(k)
if err != nil || !isNode {
return
}
if h, err := delivery.DecodeHaveNode(v); err == nil && h.Record != nil {
out[node] = *h.Record
}
})
return out, err
}
func (r busReader) Modules(context.Context) (map[string]delivery.Module, error) {
out := map[string]delivery.Module{}
err := r.each(modulesRead, func(k string, v []byte) {
module, err := delivery.ParseModulesKey(k)
if err != nil {
return
}
if m, err := delivery.DecodeModule(v); err == nil {
out[module] = m
}
})
return out, err
}
func (r busReader) Manifest(_ context.Context, digest string) ([]byte, error) {
key, err := delivery.ManifestsKey(digest)
if err != nil {
return nil, err
}
e, err := r.s.Get(manifestsRead, key)
if err != nil {
return nil, err
}
if e == nil {
return nil, fmt.Errorf("the indexer holds no manifest %s", digest)
}
m, err := delivery.DecodeManifest(e.Value)
if err != nil {
return nil, err
}
return []byte(m.JSON), nil
}
func (r busReader) Values(_ context.Context, node, module string) (map[string]any, error) {
var out map[string]any
prefix := node + "." + module + "."
err := r.each(resolvedRead, func(k string, v []byte) {
name, ok := strings.CutPrefix(k, prefix)
if !ok || strings.Contains(name, ".") {
return
}
if value, laid := catalogue.LaidValue(v); laid {
if out == nil {
out = map[string]any{}
}
out[name] = value
}
})
return out, err
}
func (busReader) OwnSecrets(context.Context, string) ([]compose.OwnSecretRow, error) { return nil, nil }
func (busReader) Credentials(context.Context, string, []compose.CredentialKey, []compose.CredentialKey) ([]compose.Credential, []compose.Issued, error) {
return nil, nil, nil
}
func (r busReader) Build(_ context.Context, module, fp string) (*delivery.BuilderBuild, error) {
key, err := delivery.BuilderBuildKey(module, fp)
if err != nil {
return nil, err
}
e, err := r.s.Get(buildsRead, key)
if err != nil || e == nil {
return nil, err
}
b, err := delivery.DecodeBuilderBuild(e.Value)
if err != nil {
return nil, err
}
return &b, nil
}
// archiveWithin bounds one read of a templates archive.
const archiveWithin = 30 * time.Second
func (r busReader) Archive(ctx context.Context, url, digest string) ([]byte, error) {
if r.fetch == nil {
return nil, fmt.Errorf("this holder reads no archive (%s)", url)
}
fetching, cancel := context.WithTimeout(ctx, archiveWithin)
defer cancel()
body, err := r.fetch(fetching, url)
if err != nil {
return nil, err
}
sum := sha256.Sum256(body)
if got := "sha256:" + hex.EncodeToString(sum[:]); got != digest {
return nil, fmt.Errorf("%s was recorded as %s and what arrived is %s", url, digest, got)
}
return body, nil
}
func (r busReader) Served(ctx context.Context, node, module string, keys []string) (map[string]any, error) {
all, err := r.Values(ctx, node, module)
var out map[string]any
for _, k := range keys {
if v, ok := all[k]; ok {
if out == nil {
out = map[string]any{}
}
out[k] = v
}
}
return out, err
}
// httpFetch reads an archive from the artifact store over HTTP.
func httpFetch(ctx context.Context, url string) ([]byte, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
if err != nil {
return nil, err
}
resp, err := http.DefaultClient.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("%s answered %s", url, resp.Status)
}
return io.ReadAll(io.LimitReader(resp.Body, 256<<20))
}
// PlacedItems is every data item the modules placed on node declare, composed from the bus as the node composes itself.
func PlacedItems(ctx context.Context, s entryReader, node string) (map[string][]catalogue.PlacedDataItem, error) {
r := busReader{s: s, fetch: httpFetch}
items, unplaced, err := frombus.DataItems(ctx, r, frombus.Mesh{}, frombus.Scope{Node: node})
if err != nil {
return nil, err
}
if len(unplaced) > 0 {
logf("data items left out as not placeable: %s", strings.Join(unplaced, "; "))
}
return items, nil
}
+21 -18
View File
@@ -6,6 +6,8 @@ import (
"encoding/json"
"errors"
"fmt"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/health"
"os"
"os/exec"
"sort"
@@ -41,8 +43,9 @@ func (busStore) Keys() ([]string, error) { return stdio.State(StateReadings).Key
// Keeper measures the items of its node and keeps the readings.
type Keeper struct {
Node string
DataFile string
Node string
// Items is every data item the modules placed on this node declare, composed from the bus (frombus.DataItems).
Items func(ctx context.Context) (map[string][]catalogue.PlacedDataItem, error)
Measurer Measurer
Store Store
Say func(string, ...any)
@@ -56,7 +59,7 @@ type Keeper struct {
func (k *Keeper) Round(ctx context.Context, walks bool) (written, kept int, err error) {
k.mu.Lock()
defer k.mu.Unlock()
items, err := ReadItems(k.DataFile)
items, err := k.Items(ctx)
if err != nil {
return 0, 0, err
}
@@ -89,24 +92,24 @@ func (k *Keeper) Round(ctx context.Context, walks bool) (written, kept int, err
return written, kept, nil
}
func (k *Keeper) measureOne(ctx context.Context, it Item, walks bool) (bool, error) {
key := ReadingKey(k.Node, it.Module, it.ID)
var prev *Reading
func (k *Keeper) measureOne(ctx context.Context, it catalogue.PlacedDataItem, walks bool) (bool, error) {
key := health.ReadingKey(k.Node, it.Module, it.ID)
var prev *health.Reading
raw, err := k.Store.Get(key)
if err != nil {
return false, err
}
if len(raw) > 0 {
if p, err := DecodeReading(raw); err == nil {
if p, err := health.DecodeReading(raw); err == nil {
prev = &p
}
}
now := k.Measurer.Now()
if prev != nil && it.Measure == MeasureWalk && !walks && prev.MeasuredAt != nil && now.Sub(*prev.MeasuredAt) < WalkEvery {
if prev != nil && it.Measure == catalogue.MeasureWalk && !walks && prev.MeasuredAt != nil && now.Sub(*prev.MeasuredAt) < WalkEvery {
return false, nil
}
m := k.Measurer.Measure(ctx, it)
next, changed := Next(prev, k.Node, it, m, now)
next, changed := health.NextReading(prev, k.Node, it, m, now)
if !changed {
return false, nil
}
@@ -114,32 +117,32 @@ func (k *Keeper) measureOne(ctx context.Context, it Item, walks bool) (bool, err
}
// MeasureNow measures one item now, walk or not, and writes its reading when it changed.
func (k *Keeper) MeasureNow(ctx context.Context, module, item string) (Reading, bool, error) {
func (k *Keeper) MeasureNow(ctx context.Context, module, item string) (health.Reading, bool, error) {
k.mu.Lock()
defer k.mu.Unlock()
items, err := ReadItems(k.DataFile)
items, err := k.Items(ctx)
if err != nil {
return Reading{}, false, err
return health.Reading{}, false, err
}
for _, it := range items[module] {
if it.ID != item {
continue
}
if it.Class == ClassCache {
return Reading{}, false, fmt.Errorf("%s.%s is a cache, which is listed and never measured", module, item)
return health.Reading{}, false, fmt.Errorf("%s.%s is a cache, which is listed and never measured", module, item)
}
w, err := k.measureOne(ctx, it, true)
if err != nil {
return Reading{}, false, err
return health.Reading{}, false, err
}
raw, err := k.Store.Get(ReadingKey(k.Node, module, item))
raw, err := k.Store.Get(health.ReadingKey(k.Node, module, item))
if err != nil {
return Reading{}, w, err
return health.Reading{}, w, err
}
r, err := DecodeReading(raw)
r, err := health.DecodeReading(raw)
return r, w, err
}
return Reading{}, false, fmt.Errorf("%s declares no data item %q on this machine", module, item)
return health.Reading{}, false, fmt.Errorf("%s declares no data item %q on this machine", module, item)
}
// Loop runs a round after a few minutes and then every `every`, with the walks of the night's round when it is the
+8 -4
View File
@@ -4,7 +4,7 @@
// filesystems' space and the processes doing the I/O — and changes nothing of the machine.
//
// It also measures the data items the modules of its node declare (ADR 0233) and keeps each reading and the item's
// seven-day peak as its own state `readings` (ADR 0571 clause 3), when MESH_DISK_DATA names the composed data file.
// seven-day peak as its own state `readings` (ADR 0571 clause 3), composed from the bus (no file).
package main
import (
@@ -15,6 +15,7 @@ import (
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/announce"
"git.novox.be/novox/mesh-sdk/go/catalogue"
)
func logf(format string, args ...any) { fmt.Fprintf(os.Stderr, "[disk-load] "+format+"\n", args...) }
@@ -24,9 +25,12 @@ func main() {
logf("%v", err)
os.Exit(1)
}
k := &Keeper{Node: os.Getenv("MESH_NODE"), DataFile: os.Getenv("MESH_DISK_DATA"), Store: busStore{}, Say: logf,
Measurer: Measurer{Run: RootRunner, Now: time.Now}}
if k.DataFile != "" {
node := os.Getenv("MESH_NODE")
k := &Keeper{Node: node, Store: busStore{}, Say: logf, Measurer: Measurer{Run: RootRunner, Now: time.Now},
Items: func(ctx context.Context) (map[string][]catalogue.PlacedDataItem, error) {
return PlacedItems(ctx, busState{}, node)
}}
if node != "" {
every := time.Hour
if d, err := time.ParseDuration(os.Getenv("MESH_DISK_MEASURE_EVERY")); err == nil && d >= time.Minute {
every = d
+10 -68
View File
@@ -6,7 +6,8 @@ package main
import (
"context"
"fmt"
"os"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/health"
"regexp"
"strconv"
"strings"
@@ -20,65 +21,6 @@ var (
WalkEvery = 23 * time.Hour
)
var moduleHeader = regexp.MustCompile(`^#\s*([a-z0-9][a-z0-9-]*)$`)
var safePath = regexp.MustCompile(`^/[^\s'"\\$` + "`" + `]*$`)
// ParseItems reads the composed data file into each module's items. A line it does not read is refused, naming the
// module.
func ParseItems(text string) (map[string][]Item, error) {
out := map[string][]Item{}
module := ""
for _, raw := range strings.Split(text, "\n") {
line := strings.TrimSpace(raw)
if line == "" {
continue
}
if m := moduleHeader.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" || !safePath.MatchString(f[3]) || (f[4] != "-" && !safePath.MatchString(f[4])) {
return nil, fmt.Errorf("%s declares a data line this module does not read: %s", module, line)
}
it := Item{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]
}
switch it.Measure {
case MeasureWalk, MeasureDataset, MeasureShallow:
default:
return nil, fmt.Errorf("%s declares a data line this module does not read: %s", module, line)
}
if f[4] != "-" {
it.CoveredBy = f[4]
}
out[module] = append(out[module], it)
}
return out, nil
}
// ReadItems reads the composed data file; none where there is none.
func ReadItems(path string) (map[string][]Item, error) {
if path == "" {
return map[string][]Item{}, nil
}
raw, err := os.ReadFile(path)
if os.IsNotExist(err) {
return map[string][]Item{}, nil
}
if err != nil {
return nil, err
}
return ParseItems(string(raw))
}
// walkCommand sums the files under a path and finds its newest change, in one walk, bounded by time and by files.
// One line out: "<bytes> <epoch seconds> <files>", then the status of find.
func walkCommand(path string) string {
@@ -114,9 +56,9 @@ func (b Measurer) exists(ctx context.Context, path string) bool {
}
// Measure is one item, measured now, by its own method.
func (b Measurer) Measure(ctx context.Context, it Item) Measured {
func (b Measurer) Measure(ctx context.Context, it catalogue.PlacedDataItem) health.Measured {
at := b.Now().UTC()
m := Measured{MeasuredAt: &at}
m := health.Measured{MeasuredAt: &at}
if !b.exists(ctx, it.Path) {
m.Error = it.Path + " does not exist"
zero := int64(0)
@@ -125,7 +67,7 @@ func (b Measurer) Measure(ctx context.Context, it Item) Measured {
}
m.Redundancy = b.redundancyOf(ctx, it.Path)
switch it.Measure {
case MeasureDataset, MeasureShallow:
case catalogue.MeasureDataset, catalogue.MeasureShallow:
out, err := b.Run(ctx, "bash", "-c", topCommand(it.Path))
if f := strings.Fields(out); err == nil && len(f) == 2 {
m.LastWrite = epochTime(f[0])
@@ -133,7 +75,7 @@ func (b Measurer) Measure(ctx context.Context, it Item) Measured {
m.Error = err.Error()
}
m.Precision = "no size: the newest change among its top-level entries only"
if it.Measure == MeasureShallow {
if it.Measure == catalogue.MeasureShallow {
return m
}
out, err = b.Run(ctx, "zfs", "list", "-Hp", "-o", "name,used", it.Path)
@@ -178,7 +120,7 @@ func (b Measurer) Measure(ctx context.Context, it Item) Measured {
// redundancyOf is the redundant storage a path is on: a ZFS pool's health, an md array's members, a btrfs
// filesystem's error counters. Nil for storage with no redundancy it knows how to read.
func (b Measurer) redundancyOf(ctx context.Context, path string) *Redundancy {
func (b Measurer) redundancyOf(ctx context.Context, path string) *health.Redundancy {
out, err := b.Run(ctx, "findmnt", "-no", "SOURCE,FSTYPE", "--target", path)
if err != nil {
return nil
@@ -192,7 +134,7 @@ func (b Measurer) redundancyOf(ctx context.Context, path string) *Redundancy {
switch {
case fstype == "zfs":
pool, _, _ := strings.Cut(source, "/")
r := &Redundancy{Kind: "zfs", Where: pool}
r := &health.Redundancy{Kind: "zfs", Where: pool}
health, err := b.Run(ctx, "zpool", "list", "-H", "-o", "health", pool)
if err != nil {
r.Said = "zpool list: " + err.Error()
@@ -214,7 +156,7 @@ func (b Measurer) redundancyOf(ctx context.Context, path string) *Redundancy {
return r
case strings.HasPrefix(source, "/dev/md"):
device := strings.TrimPrefix(source, "/dev/")
r := &Redundancy{Kind: "md", Where: device}
r := &health.Redundancy{Kind: "md", Where: device}
mdstat, err := b.Run(ctx, "cat", "/proc/mdstat")
if err != nil {
r.Said = err.Error()
@@ -233,7 +175,7 @@ func (b Measurer) redundancyOf(ctx context.Context, path string) *Redundancy {
}
return r
case fstype == "btrfs":
r := &Redundancy{Kind: "btrfs", Where: source}
r := &health.Redundancy{Kind: "btrfs", Where: source}
stats, err := b.Run(ctx, "btrfs", "device", "stats", "--check", path)
if err != nil {
r.Healthy, r.Said = &no, "device errors: "+oneLineOf(err.Error())
+6 -133
View File
@@ -1,139 +1,12 @@
package main
// **Disk usage readings and peaks per data item** (novox/hq ADR 0571 clause 3, ADR 0233 §21, §27, §34): disk-load measures
// each data item the modules of its node declare 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.
// The readings layout, their key and the peak are mesh-sdk's (health.Reading, health.NextReading; novox/hq ADR 0571
// clause 3): disk-load measures each placed data item of its node and keeps the readings as its own state.
import (
"encoding/json"
"reflect"
"time"
)
// 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
import "git.novox.be/novox/mesh-sdk/go/health"
// StateReadings is the state disk-load keeps.
const StateReadings = "readings"
const StateReadings = health.ReadingsState
// The ways an item is measured (ADR 0233 §22 to §25).
const (
MeasureWalk = "walk"
MeasureDataset = "dataset"
MeasureShallow = "shallow"
ClassCache = "cache"
)
// Item is one data item a module declares.
type Item struct {
Measure string `json:"measure,omitempty"`
Module string `json:"module"`
ID string `json:"item"`
Class string `json:"class"`
Path string `json:"path"`
CoveredBy string `json:"covered_by,omitempty"`
Protection string `json:"protection,omitempty"`
}
// Redundancy is the redundant storage an item is on, as 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.
type Measured struct {
SizeBytes *int64 `json:"size_bytes,omitempty"`
LastWrite *time.Time `json:"last_write,omitempty"`
MeasuredAt *time.Time `json:"measured_at,omitempty"`
// Precision says what the size is: "exact", "dataset <name>: its whole size", "partial: ..." (a lower bound,
// never compared), or "no size: ...".
Precision string `json:"precision,omitempty"`
Error string `json:"error,omitempty"`
Redundancy *Redundancy `json:"redundancy,omitempty"`
}
// 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.
type Reading struct {
V int `json:"v"`
Node string `json:"node"`
Item
Measured
// Peak is the largest size measured in the seven days up to the reading, PeakAt the day it was measured; Days are
// the day maxima it is read from.
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.
func DecodeReading(raw []byte) (Reading, error) {
var r Reading
err := json.Unmarshal(raw, &r)
return r, err
}
// Next 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).
func Next(prev *Reading, node string, it Item, m Measured, now time.Time) (Reading, bool) {
r := Reading{V: ReadingVersion, Node: node, Item: it, Measured: m}
var days []DaySize
if prev != nil {
days = prev.Days
}
// A reading with no size (an error, a shallow item) carries the last size's days on, so the peak outlives one failed
// measurement.
today := now.UTC().Format("2006-01-02")
cut := now.UTC().Add(-PeakWindow).Format("2006-01-02")
var kept []DaySize
for _, d := range days {
if d.Day >= cut && d.Day != today {
kept = append(kept, d)
}
}
todays := int64(-1)
for _, d := range days {
if d.Day == today {
todays = d.Bytes
}
}
if m.SizeBytes != nil && !partial(m) && *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)
}
// partial says a size is only a lower bound, which a peak is never made of.
func partial(m Measured) bool {
return len(m.Precision) >= 7 && m.Precision[:7] == "partial"
}
// ClassCache is the class of an item that is listed and never measured.
const ClassCache = "cache"
@@ -4,10 +4,11 @@ import (
"context"
"encoding/json"
"os"
"path/filepath"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
)
type memStore struct{ m map[string]json.RawMessage }
@@ -26,54 +27,12 @@ func (s *memStore) Keys() ([]string, error) {
return out, nil
}
func size(n int64) *int64 { return &n }
// A reading is kept under <node>.<module>.<item> with the largest size of the last seven days; one that did not change
// is not written; a size older than seven days leaves the peak; a partial size is no peak.
func TestAReadingKeepsItsSevenDayPeakAndIsWrittenOnlyWhenItChanges(t *testing.T) {
it := Item{Module: "pg", ID: "data", Class: "valuable", Path: "/srv/pg", Measure: MeasureWalk}
day := func(d int, h int) time.Time { return time.Date(2026, 10, d, h, 0, 0, 0, time.UTC) }
at := func(tm time.Time) Measured { return Measured{MeasuredAt: &tm, Precision: "exact"} }
steps := []struct {
now time.Time
size *int64
partial bool
changed bool
peak int64
}{
{day(20, 1), size(100), false, true, 100},
{day(20, 2), size(100), false, false, 100}, // only the time differs
{day(21, 1), size(300), false, true, 300},
{day(22, 1), size(50), false, true, 300}, // the peak stays
{day(23, 1), size(9000), true, true, 300}, // a lower bound is no peak
{day(29, 1), size(10), false, true, 50}, // the 21st left seven days ago
{day(30, 1), size(10), false, true, 10}, // the 22nd left too
}
var prev *Reading
for i, s := range steps {
m := at(s.now)
m.SizeBytes = s.size
if s.partial {
m.Precision = "partial: stopped"
}
r, changed := Next(prev, "ace", it, m, s.now)
if changed != s.changed || r.Peak == nil || *r.Peak != s.peak {
t.Fatalf("step %d: changed %v peak %v, wanted %v and %d", i, changed, r.Peak, s.changed, s.peak)
}
if ReadingKey(r.Node, r.Module, r.ID) != "ace.pg.data" {
t.Fatalf("key %s", ReadingKey(r.Node, r.Module, r.ID))
}
prev = &r
}
}
// A round measures what the data file declares but caches, keeps the readings, writes nothing the second time, and a
// walk is not repeated within a day unless asked.
func TestARoundMeasuresTheDeclaredItemsAndWritesWhatChanged(t *testing.T) {
dir := t.TempDir()
file := filepath.Join(dir, "data.conf")
text := "# pg\nitem data valuable /srv/pg /srv/pg backup walk\nitem tmp cache /srv/tmp - none\n# bad-line-module\nitem logs rebuildable /srv/logs - none shallow\n"
if err := os.WriteFile(file, []byte(text), 0o600); err != nil {
placed, err := catalogue.ParseDataItems(text)
if err != nil {
t.Fatal(err)
}
walks := 0
@@ -93,7 +52,7 @@ func TestARoundMeasuresTheDeclaredItemsAndWritesWhatChanged(t *testing.T) {
}
now := time.Date(2026, 10, 29, 12, 0, 0, 0, time.UTC)
store := &memStore{m: map[string]json.RawMessage{}}
k := &Keeper{Node: "ace", DataFile: file, Store: store, Say: func(string, ...any) {},
k := &Keeper{Node: "ace", Items: func(context.Context) (map[string][]catalogue.PlacedDataItem, error) { return placed, nil }, Store: store, Say: func(string, ...any) {},
Measurer: Measurer{Run: run, Now: func() time.Time { return now }}}
w, kept, err := k.Round(context.Background(), false)
if err != nil || w != 2 || kept != 0 || len(store.m) != 2 {
@@ -109,7 +68,4 @@ func TestARoundMeasuresTheDeclaredItemsAndWritesWhatChanged(t *testing.T) {
if _, _, err := k.MeasureNow(context.Background(), "pg", "tmp"); err == nil {
t.Fatal("a cache was measured")
}
if _, err := ParseItems("# pg\nitem data valuable relative - none\n"); err == nil {
t.Fatal("a relative path was read")
}
}
@@ -7,6 +7,7 @@ import (
"strings"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/health"
)
// ReadingTools are the tools over the state `readings`: every key, one key, how full the state is, and one item
@@ -32,7 +33,7 @@ func ReadingTools(k *Keeper) []stdio.Tool {
if len(raw) == 0 {
return nil, fmt.Errorf("no reading is kept at %q", key)
}
r, err := DecodeReading(raw)
r, err := health.DecodeReading(raw)
return r, err
}
keys, err := k.Store.Keys()
@@ -42,12 +43,12 @@ func ReadingTools(k *Keeper) []stdio.Tool {
sort.Strings(keys)
module := text(args, "module")
type line struct {
Key string `json:"key"`
SizeBytes *int64 `json:"size_bytes,omitempty"`
PeakBytes *int64 `json:"peak_bytes,omitempty"`
Precision string `json:"precision,omitempty"`
Error string `json:"error,omitempty"`
Reading Reading `json:"-"`
Key string `json:"key"`
SizeBytes *int64 `json:"size_bytes,omitempty"`
PeakBytes *int64 `json:"peak_bytes,omitempty"`
Precision string `json:"precision,omitempty"`
Error string `json:"error,omitempty"`
Reading health.Reading `json:"-"`
}
var lines []line
total := 0
@@ -57,7 +58,7 @@ func ReadingTools(k *Keeper) []stdio.Tool {
return nil, err
}
total += len(raw)
r, err := DecodeReading(raw)
r, err := health.DecodeReading(raw)
if err != nil || (module != "" && r.Module != module) {
continue
}
@@ -84,7 +85,7 @@ func ReadingTools(k *Keeper) []stdio.Tool {
if err != nil {
return nil, err
}
return map[string]any{"written": written, "reading": r, "key": ReadingKey(k.Node, module, item),
return map[string]any{"written": written, "reading": r, "key": health.ReadingKey(k.Node, module, item),
"note": strings.TrimSpace(map[bool]string{true: "the reading changed and was written", false: "the reading had not changed"}[written])}, nil
},
},
@@ -3,6 +3,7 @@ package main
import (
"encoding/json"
"os"
"regexp"
"sort"
"strings"
"testing"
@@ -52,7 +53,7 @@ func TestTheManifestNamesExactlyTheToolsTheBundleServesAndNoMachine(t *testing.T
t.Errorf("the module only reads, and declares %d resources", len(m.Resources))
}
for _, banned := range []string{"g14", "novox", "ace", "shanks", "jochen", "/home/"} {
if strings.Contains(string(raw), banned) {
if regexp.MustCompile(`(^|[^a-z0-9-])` + regexp.QuoteMeta(banned) + `($|[^a-z0-9-])`).Match(raw) {
t.Errorf("the manifest says %q", banned)
}
}
+1 -1
View File
@@ -2,4 +2,4 @@ module diskload
go 1.26.0
require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029210043-32f5ce371268
require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029221656-196c5b6af277
+2 -2
View File
@@ -1,2 +1,2 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029210043-32f5ce371268 h1:+jH77IJmxUiFDdbJljhynQRY0Cc5/5jtwmMV7UUqRI4=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029210043-32f5ce371268/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029221656-196c5b6af277 h1:ajjsH4MWcx45MGYASjUSRQKhLzEringuk3TA0oxZRCU=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029221656-196c5b6af277/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+8
View File
@@ -28,5 +28,13 @@
},
"state": [
"readings"
],
"reads": [
"mesh-delivery-assigner.placement",
"mesh-delivery-indexer.modules",
"mesh-delivery-indexer.manifests",
"mesh-delivery-builder.builds",
"node-engine.have",
"mesh-config.resolved"
]
}