diff --git a/modules/disk-load/cmd/disk-load/keeper.go b/modules/disk-load/cmd/disk-load/keeper.go new file mode 100644 index 00000000..3c8cd7bb --- /dev/null +++ b/modules/disk-load/cmd/disk-load/keeper.go @@ -0,0 +1,198 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "os" + "os/exec" + "sort" + "strings" + "sync" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// Store is the state `readings` as the keeper uses it. +type Store interface { + Get(key string) (json.RawMessage, error) + Put(key string, value any) error + Keys() ([]string, error) +} + +// busStore is the state through the runtime. +type busStore struct{} + +func (busStore) Get(key string) (json.RawMessage, error) { + e, err := stdio.State(StateReadings).Get(key) + if err != nil || e == nil { + return nil, err + } + return e.Value, nil +} +func (busStore) Put(key string, v any) error { + _, err := stdio.State(StateReadings).Put(key, v) + return err +} +func (busStore) Keys() ([]string, error) { return stdio.State(StateReadings).Keys() } + +// Keeper measures the items of its node and keeps the readings. +type Keeper struct { + Node string + DataFile string + Measurer Measurer + Store Store + Say func(string, ...any) + + mu sync.Mutex // one round at a time +} + +// Round measures every item that is not a cache, one at a time, and writes the readings that changed. An item +// measured by a walk is walked only when walks is true or its last walk is older than WalkEvery: the hourly round +// reads counters, and a walk runs at most once a day (ADR 0233 §22). +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) + if err != nil { + return 0, 0, err + } + modules := make([]string, 0, len(items)) + for m := range items { + modules = append(modules, m) + } + sort.Strings(modules) + var failed []string + for _, module := range modules { + for _, it := range items[module] { + if it.Class == ClassCache { + continue + } + w, err := k.measureOne(ctx, it, walks) + if err != nil { + failed = append(failed, fmt.Sprintf("%s.%s: %v", module, it.ID, err)) + continue + } + if w { + written++ + } else { + kept++ + } + } + } + if len(failed) > 0 { + return written, kept, errors.New("readings not kept: " + strings.Join(failed, "; ")) + } + 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 + raw, err := k.Store.Get(key) + if err != nil { + return false, err + } + if len(raw) > 0 { + if p, err := 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 { + return false, nil + } + m := k.Measurer.Measure(ctx, it) + next, changed := Next(prev, k.Node, it, m, now) + if !changed { + return false, nil + } + return true, k.Store.Put(key, next) +} + +// 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) { + k.mu.Lock() + defer k.mu.Unlock() + items, err := ReadItems(k.DataFile) + if err != nil { + return 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) + } + w, err := k.measureOne(ctx, it, true) + if err != nil { + return Reading{}, false, err + } + raw, err := k.Store.Get(ReadingKey(k.Node, module, item)) + if err != nil { + return Reading{}, w, err + } + r, err := DecodeReading(raw) + return r, w, err + } + return 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 +// hour after the night; ctx ends it. +func (k *Keeper) Loop(ctx context.Context, every time.Duration, first time.Duration) { + select { + case <-ctx.Done(): + return + case <-time.After(first): + } + for { + w, kept, err := k.Round(ctx, false) + if err != nil { + k.Say("measuring the data declared here: %v", err) + } else { + k.Say("measured the data declared here: %d readings changed, %d unchanged", w, kept) + } + select { + case <-ctx.Done(): + return + case <-time.After(every): + } + } +} + +// RootRunner runs a command as root, through sudo without a prompt when this process is not root. A walk is bounded +// by WalkFor itself; the runner's own bound is a little longer. +func RootRunner(ctx context.Context, name string, args ...string) (string, error) { + ctx, cancel := context.WithTimeout(ctx, WalkFor+time.Minute) + defer cancel() + program, argv := name, args + if os.Getuid() != 0 { + program, argv = "sudo", append([]string{"-n", name}, args...) + } + var stdout, stderr bytes.Buffer + cmd := exec.CommandContext(ctx, program, argv...) + cmd.Stdout, cmd.Stderr = &stdout, &stderr + if err := cmd.Run(); err != nil { + said := strings.TrimSpace(stderr.String()) + if said == "" { + said = strings.TrimSpace(stdout.String()) + } + if program == "sudo" && strings.HasPrefix(said, "sudo:") { + return "", fmt.Errorf("%s needs root and the runtime's account may not run it without a prompt: %s", name, said) + } + lines := strings.Split(said, "\n") + if len(lines) > 3 { + lines = lines[len(lines)-3:] + } + if said == "" { + return "", fmt.Errorf("%s: %w", name, err) + } + return "", errors.New(strings.Join(lines, " / ")) + } + return stdout.String(), nil +} diff --git a/modules/disk-load/cmd/disk-load/main.go b/modules/disk-load/cmd/disk-load/main.go index 92fd2ef4..9e4a19ff 100644 --- a/modules/disk-load/cmd/disk-load/main.go +++ b/modules/disk-load/cmd/disk-load/main.go @@ -1,19 +1,41 @@ // The disk-load module's Go bundle (novox/hq ADR 0188, ADR 0193, issue 312): a process the node's // runtime launches and speaks MCP over stdio to. It reads the machine's storage load from /proc and // /sys — iowait, pressure stall, each disk's busy share, throughput, IOPS, wait and queue depth, the -// filesystems' space and the processes doing the I/O — and changes nothing. +// 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. package main import ( + "context" "fmt" "os" + "time" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/announce" ) +func logf(format string, args ...any) { fmt.Fprintf(os.Stderr, "[disk-load] "+format+"\n", args...) } + func main() { + if err := stdio.Keeps(announce.State{Name: StateReadings}); err != nil { + 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 != "" { + every := time.Hour + if d, err := time.ParseDuration(os.Getenv("MESH_DISK_MEASURE_EVERY")); err == nil && d >= time.Minute { + every = d + } + go k.Loop(context.Background(), every, 3*time.Minute) + } + tools := append(Tools(Here()), ReadingTools(k)...) // An empty name serves as the module the runtime names (MESH_SERVED_MODULE): disk-load. - if err := stdio.Serve("", Tools(Here())); err != nil { + if err := stdio.Serve("", tools); err != nil { fmt.Fprintln(os.Stderr, err) os.Exit(1) } diff --git a/modules/disk-load/cmd/disk-load/measure.go b/modules/disk-load/cmd/disk-load/measure.go new file mode 100644 index 00000000..de9e0369 --- /dev/null +++ b/modules/disk-load/cmd/disk-load/measure.go @@ -0,0 +1,273 @@ +package main + +// How an item is measured (novox/hq ADR 0233 §22 to §25), moved here from the backup holder, which only reads the +// readings now (ADR 0571 clause 3). Never an unbounded walk of something large. + +import ( + "context" + "fmt" + "os" + "regexp" + "strconv" + "strings" + "time" +) + +// The bounds on a walk: a walk is for small items, and one that meets a large one stops and says so. +var ( + WalkFor = 10 * time.Minute + WalkFiles = 2_000_000 + 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: " ", then the status of find. +func walkCommand(path string) string { + return fmt.Sprintf("timeout %d find '%s' -xdev -printf '%%y %%s %%T@\\n' 2>/dev/null | head -n %d | "+ + "awk 'BEGIN{s=0;m=0;n=0} {n++; if ($1==\"f\") s+=$2; if ($3>m) m=$3} END {printf \"%%d %%.0f %%d\\n\", s, m, n}'; "+ + "echo \"${PIPESTATUS[0]:-0}\"", int(WalkFor.Seconds()), path, WalkFiles) +} + +// topCommand is the newest change among a path and its top-level entries, and how many there are. +func topCommand(path string) string { + return "timeout 60 find '" + path + "' -maxdepth 1 -printf '%T@\\n' 2>/dev/null | " + + "awk 'BEGIN{m=0;n=0} {n++; if ($1>m) m=$1} END {printf \"%.0f %d\\n\", m, n-1}'" +} + +func epochTime(s string) *time.Time { + epoch, err := strconv.ParseInt(s, 10, 64) + if err != nil || epoch <= 0 { + return nil + } + t := time.Unix(epoch, 0).UTC() + return &t +} + +// Measurer measures items through a runner that has the root the walks need. +type Measurer struct { + Run Runner + Now func() time.Time +} + +func (b Measurer) exists(ctx context.Context, path string) bool { + _, err := b.Run(ctx, "test", "-e", path) + return err == nil +} + +// Measure is one item, measured now, by its own method. +func (b Measurer) Measure(ctx context.Context, it Item) Measured { + at := b.Now().UTC() + m := Measured{MeasuredAt: &at} + if !b.exists(ctx, it.Path) { + m.Error = it.Path + " does not exist" + zero := int64(0) + m.SizeBytes, m.Precision = &zero, "exact" + return m + } + m.Redundancy = b.redundancyOf(ctx, it.Path) + switch it.Measure { + case MeasureDataset, 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]) + } else if err != nil { + m.Error = err.Error() + } + m.Precision = "no size: the newest change among its top-level entries only" + if it.Measure == MeasureShallow { + return m + } + out, err = b.Run(ctx, "zfs", "list", "-Hp", "-o", "name,used", it.Path) + f := strings.Fields(out) + if err != nil || len(f) != 2 { + m.Precision = "no size: it is on no ZFS dataset to read one from; the newest change among its top-level entries only" + return m + } + used, err := strconv.ParseInt(f[1], 10, 64) + if err != nil { + return m + } + m.SizeBytes = &used + m.Precision = "dataset " + f[0] + ": its whole size, from the filesystem's counters; the newest change among the top-level entries" + return m + } + out, err := b.Run(ctx, "bash", "-c", walkCommand(it.Path)) + if err != nil { + m.Error = err.Error() + return m + } + lines := strings.Split(strings.TrimSpace(out), "\n") + f := strings.Fields(lines[0]) + if len(f) != 3 || len(lines) < 2 { + m.Error = "the measurement said " + strings.TrimSpace(out) + return m + } + size, err1 := strconv.ParseInt(f[0], 10, 64) + files, err2 := strconv.Atoi(f[2]) + if err1 != nil || err2 != nil { + m.Error = "the measurement said " + strings.TrimSpace(out) + return m + } + m.SizeBytes, m.LastWrite = &size, epochTime(f[1]) + m.Precision = "exact" + if status := strings.TrimSpace(lines[len(lines)-1]); status == "124" || files >= WalkFiles { + m.Precision = fmt.Sprintf("partial: measured partially — stopped after %s or %d files; the size is a lower bound, "+ + "and this item wants measure: dataset or shallow", WalkFor, WalkFiles) + } + return m +} + +// 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 { + out, err := b.Run(ctx, "findmnt", "-no", "SOURCE,FSTYPE", "--target", path) + if err != nil { + return nil + } + f := strings.Fields(out) + if len(f) < 2 { + return nil + } + source, fstype := f[0], f[1] + yes, no := true, false + switch { + case fstype == "zfs": + pool, _, _ := strings.Cut(source, "/") + r := &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() + return r + } + status, err := b.Run(ctx, "zpool", "status", "-x", pool) + if err != nil { + r.Said = "zpool status: " + err.Error() + return r + } + health, status = strings.TrimSpace(health), strings.TrimSpace(status) + r.Said = "health " + health + "; " + firstLineOf(status) + if health == "ONLINE" && strings.Contains(status, "is healthy") { + r.Healthy = &yes + } else { + r.Healthy = &no + r.Said = "health " + health + "; " + oneLineOf(status) + } + return r + case strings.HasPrefix(source, "/dev/md"): + device := strings.TrimPrefix(source, "/dev/") + r := &Redundancy{Kind: "md", Where: device} + mdstat, err := b.Run(ctx, "cat", "/proc/mdstat") + if err != nil { + r.Said = err.Error() + return r + } + healthy, said, found := mdHealth(mdstat, device) + if !found { + r.Said = device + " is not in /proc/mdstat" + return r + } + r.Said = said + if healthy { + r.Healthy = &yes + } else { + r.Healthy = &no + } + return r + case fstype == "btrfs": + r := &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()) + return r + } + r.Healthy, r.Said = &yes, firstLineOf(stats) + return r + } + return nil +} + +var mdMembers = regexp.MustCompile(`\[([U_]+)\]`) + +// mdHealth reads one array's block of /proc/mdstat: healthy when every member is up and it is not recovering. +func mdHealth(mdstat, device string) (bool, string, bool) { + lines := strings.Split(mdstat, "\n") + for i, l := range lines { + if !strings.HasPrefix(l, device+" ") { + continue + } + block := l + for j := i + 1; j < len(lines) && strings.HasPrefix(lines[j], " "); j++ { + block += " " + strings.TrimSpace(lines[j]) + } + members := mdMembers.FindStringSubmatch(block) + healthy := members != nil && !strings.Contains(members[1], "_") && !strings.Contains(block, "recovery") + return healthy, oneLineOf(block), true + } + return false, "", false +} + +func firstLineOf(s string) string { + line, _, _ := strings.Cut(strings.TrimSpace(s), "\n") + return line +} + +func oneLineOf(s string) string { return strings.Join(strings.Fields(s), " ") } diff --git a/modules/disk-load/cmd/disk-load/readings.go b/modules/disk-load/cmd/disk-load/readings.go new file mode 100644 index 00000000..baebdb55 --- /dev/null +++ b/modules/disk-load/cmd/disk-load/readings.go @@ -0,0 +1,139 @@ +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 `..`, 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. + +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 + +// StateReadings is the state disk-load keeps. +const StateReadings = "readings" + +// 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 : 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" +} diff --git a/modules/disk-load/cmd/disk-load/readings_test.go b/modules/disk-load/cmd/disk-load/readings_test.go new file mode 100644 index 00000000..237aed35 --- /dev/null +++ b/modules/disk-load/cmd/disk-load/readings_test.go @@ -0,0 +1,115 @@ +package main + +import ( + "context" + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +type memStore struct{ m map[string]json.RawMessage } + +func (s *memStore) Get(k string) (json.RawMessage, error) { return s.m[k], nil } +func (s *memStore) Put(k string, v any) error { + raw, _ := json.Marshal(v) + s.m[k] = raw + return nil +} +func (s *memStore) Keys() ([]string, error) { + var out []string + for k := range s.m { + out = append(out, k) + } + return out, nil +} + +func size(n int64) *int64 { return &n } + +// A reading is kept under .. 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 { + t.Fatal(err) + } + walks := 0 + run := func(_ context.Context, name string, args ...string) (string, error) { + switch { + case name == "test": + return "", nil + case name == "findmnt": + return "", os.ErrNotExist + case name == "bash" && strings.Contains(args[1], "-maxdepth 1"): + return "1760000000 2\n", nil + case name == "bash": + walks++ + return "4096 1760000000 3\n0\n", nil + } + return "", os.ErrNotExist + } + 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) {}, + 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 { + t.Fatalf("first round: %d written %d kept %v, keys %d", w, kept, err, len(store.m)) + } + now = now.Add(time.Hour) + if w, kept, _ := k.Round(context.Background(), false); w != 0 || kept != 2 || walks != 1 { + t.Fatalf("second round: %d written, %d kept, %d walks; wanted none written and the walk not repeated", w, kept, walks) + } + if _, _, err := k.MeasureNow(context.Background(), "pg", "data"); err != nil || walks != 2 { + t.Fatalf("measure now: %v, %d walks", err, walks) + } + 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") + } +} diff --git a/modules/disk-load/cmd/disk-load/readings_tools.go b/modules/disk-load/cmd/disk-load/readings_tools.go new file mode 100644 index 00000000..d47e74c9 --- /dev/null +++ b/modules/disk-load/cmd/disk-load/readings_tools.go @@ -0,0 +1,92 @@ +package main + +import ( + "context" + "fmt" + "sort" + "strings" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// ReadingTools are the tools over the state `readings`: every key, one key, how full the state is, and one item +// measured now (a module's tools are its command line). +func ReadingTools(k *Keeper) []stdio.Tool { + return []stdio.Tool{ + { + Name: "disk_readings", + Description: "The data readings disk-load keeps for this machine's data items (ADR 0233, ADR 0571 clause 3): every key " + + "`..` with its size, newest write, measured-at, precision and seven-day peak, or one key with " + + "all it holds when `key` is given; and how full the state is: keys kept, their bytes, and the cap the bus puts on " + + "a state (none declared). Filter by `module`. (r)", + Input: map[string]any{ + "key": map[string]any{"type": "string", "description": "one key, .., with everything it holds"}, + "module": map[string]any{"type": "string", "description": "only this module's items"}, + }, + Run: func(args map[string]any) (any, error) { + if key := text(args, "key"); key != "" { + raw, err := k.Store.Get(key) + if err != nil { + return nil, err + } + if len(raw) == 0 { + return nil, fmt.Errorf("no reading is kept at %q", key) + } + r, err := DecodeReading(raw) + return r, err + } + keys, err := k.Store.Keys() + if err != nil { + return nil, err + } + 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:"-"` + } + var lines []line + total := 0 + for _, key := range keys { + raw, err := k.Store.Get(key) + if err != nil { + return nil, err + } + total += len(raw) + r, err := DecodeReading(raw) + if err != nil || (module != "" && r.Module != module) { + continue + } + lines = append(lines, line{Key: key, SizeBytes: r.SizeBytes, PeakBytes: r.Peak, Precision: r.Precision, Error: r.Error}) + } + return map[string]any{"readings": lines, "fill": map[string]any{"keys": len(keys), "bytes": total, + "cap": "none declared: the bus's own limit on a state applies"}}, nil + }, + }, + { + Name: "disk_measure", + Description: "Measure one data item of this machine now, whatever its method (a walk too), and write its reading when it " + + "changed. Reads the disks as root: a walk is bounded by time and files and says when it was partial.", + Input: map[string]any{ + "module": map[string]any{"type": "string", "description": "the module that declares the item"}, + "item": map[string]any{"type": "string", "description": "the item's id"}, + }, + Run: func(args map[string]any) (any, error) { + module, item := text(args, "module"), text(args, "item") + if module == "" || item == "" { + return nil, fmt.Errorf("module and item are required") + } + r, written, err := k.MeasureNow(context.Background(), module, item) + if err != nil { + return nil, err + } + return map[string]any{"written": written, "reading": r, "key": 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 + }, + }, + } +} diff --git a/modules/disk-load/cmd/disk-load/tools_test.go b/modules/disk-load/cmd/disk-load/tools_test.go index 88147715..d1de23a8 100644 --- a/modules/disk-load/cmd/disk-load/tools_test.go +++ b/modules/disk-load/cmd/disk-load/tools_test.go @@ -35,9 +35,11 @@ func TestTheManifestNamesExactlyTheToolsTheBundleServesAndNoMachine(t *testing.T t.Fatal(err) } var served []string - for _, tl := range Tools(newFake(t).machine()) { + k := &Keeper{Store: &memStore{m: map[string]json.RawMessage{}}} + for _, tl := range append(Tools(newFake(t).machine()), ReadingTools(k)...) { served = append(served, tl.Name) - if !strings.HasSuffix(tl.Description, "(r)") { + // disk_measure writes the module's own state of readings and nothing else of the machine. + if !strings.HasSuffix(tl.Description, "(r)") && tl.Name != "disk_measure" { t.Errorf("%s only reads and does not say so", tl.Name) } } diff --git a/modules/disk-load/go.mod b/modules/disk-load/go.mod index edf8078e..5e5cae21 100644 --- a/modules/disk-load/go.mod +++ b/modules/disk-load/go.mod @@ -1,5 +1,5 @@ module diskload -go 1.22 +go 1.26.0 -require git.novox.be/novox/mesh-sdk/go v0.1.6 +require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029210043-32f5ce371268 diff --git a/modules/disk-load/go.sum b/modules/disk-load/go.sum index 0dd60617..3eadb468 100644 --- a/modules/disk-load/go.sum +++ b/modules/disk-load/go.sum @@ -1,2 +1,2 @@ -git.novox.be/novox/mesh-sdk/go v0.1.6 h1:9qzdYONYbJdWcu6sxQcq9v1LI0JxcfkiKYkMUzJSkVQ= -git.novox.be/novox/mesh-sdk/go v0.1.6/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= +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= diff --git a/modules/disk-load/module.json b/modules/disk-load/module.json index 9b48041d..a2617cee 100644 --- a/modules/disk-load/module.json +++ b/modules/disk-load/module.json @@ -7,7 +7,9 @@ "disk_filesystems", "disk_folders", "disk_devices", - "cpu_top" + "cpu_top", + "disk_readings", + "disk_measure" ], "build": { "artifacts": [ @@ -23,5 +25,8 @@ ] } ] - } + }, + "state": [ + "readings" + ] }