diff --git a/checks/health/go.mod b/checks/health/go.mod index f6437708..9a268a24 100644 --- a/checks/health/go.mod +++ b/checks/health/go.mod @@ -2,4 +2,4 @@ module checks/health go 1.26.0 -require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a +require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029221656-196c5b6af277 diff --git a/checks/health/go.sum b/checks/health/go.sum index ea2abb0e..fc894705 100644 --- a/checks/health/go.sum +++ b/checks/health/go.sum @@ -1,2 +1,2 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a h1:elC2IjB+qbeXG+pWItLz71DenegbrdsqbcH5kyGo56Q= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029214732-6b8c6943811a/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= diff --git a/modules/disk-load/cmd/disk-load/busread.go b/modules/disk-load/cmd/disk-load/busread.go new file mode 100644 index 00000000..5542fca0 --- /dev/null +++ b/modules/disk-load/cmd/disk-load/busread.go @@ -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 +} diff --git a/modules/disk-load/cmd/disk-load/keeper.go b/modules/disk-load/cmd/disk-load/keeper.go index 3c8cd7bb..ba74a622 100644 --- a/modules/disk-load/cmd/disk-load/keeper.go +++ b/modules/disk-load/cmd/disk-load/keeper.go @@ -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 diff --git a/modules/disk-load/cmd/disk-load/main.go b/modules/disk-load/cmd/disk-load/main.go index 9e4a19ff..6946af88 100644 --- a/modules/disk-load/cmd/disk-load/main.go +++ b/modules/disk-load/cmd/disk-load/main.go @@ -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 diff --git a/modules/disk-load/cmd/disk-load/measure.go b/modules/disk-load/cmd/disk-load/measure.go index de9e0369..2973080b 100644 --- a/modules/disk-load/cmd/disk-load/measure.go +++ b/modules/disk-load/cmd/disk-load/measure.go @@ -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: " ", 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()) diff --git a/modules/disk-load/cmd/disk-load/readings.go b/modules/disk-load/cmd/disk-load/readings.go index baebdb55..3cad1734 100644 --- a/modules/disk-load/cmd/disk-load/readings.go +++ b/modules/disk-load/cmd/disk-load/readings.go @@ -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 `..`, 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 : 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" diff --git a/modules/disk-load/cmd/disk-load/readings_test.go b/modules/disk-load/cmd/disk-load/readings_test.go index 237aed35..e4970ca1 100644 --- a/modules/disk-load/cmd/disk-load/readings_test.go +++ b/modules/disk-load/cmd/disk-load/readings_test.go @@ -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 .. 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") - } } diff --git a/modules/disk-load/cmd/disk-load/readings_tools.go b/modules/disk-load/cmd/disk-load/readings_tools.go index d47e74c9..322fe275 100644 --- a/modules/disk-load/cmd/disk-load/readings_tools.go +++ b/modules/disk-load/cmd/disk-load/readings_tools.go @@ -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 }, }, diff --git a/modules/disk-load/cmd/disk-load/tools_test.go b/modules/disk-load/cmd/disk-load/tools_test.go index d1de23a8..672e6e94 100644 --- a/modules/disk-load/cmd/disk-load/tools_test.go +++ b/modules/disk-load/cmd/disk-load/tools_test.go @@ -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) } } diff --git a/modules/disk-load/go.mod b/modules/disk-load/go.mod index 5e5cae21..936c7c3e 100644 --- a/modules/disk-load/go.mod +++ b/modules/disk-load/go.mod @@ -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 diff --git a/modules/disk-load/go.sum b/modules/disk-load/go.sum index 3eadb468..fc894705 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.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= diff --git a/modules/disk-load/module.json b/modules/disk-load/module.json index a2617cee..edc915d0 100644 --- a/modules/disk-load/module.json +++ b/modules/disk-load/module.json @@ -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" ] }