Readings and last good nights are the owners' published packages: disk-load's diskreadings and restic's backupgood, taken by series (B5b Left 3; ADR 0634 §4; issue 1966)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
@@ -9,10 +9,10 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-catalog/modules/disk-load/diskreadings"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
)
|
||||
|
||||
// ADR 0577 clauses 2, 3 (and the disk-load check of clause 5): the items measured come from the node-engine's keys of
|
||||
@@ -137,7 +137,7 @@ func TestTheItemsComeFromTheEnginesKeysAndARoundSaysHowFarItGot(t *testing.T) {
|
||||
|
||||
// A restart: oldie's key was deleted while the watch was down, so no delete arrives. Once the watch handed over its
|
||||
// keys, oldie's reading is retired and pg's kept; an empty set retires nothing (the engine has not written yet).
|
||||
_ = readings.Put("ace.oldie.x", health.Reading{V: health.ReadingVersion, Node: "ace",
|
||||
_ = readings.Put("ace.oldie.x", diskreadings.Reading{V: diskreadings.ReadingVersion, Node: "ace",
|
||||
PlacedDataItem: catalogue.PlacedDataItem{Module: "oldie", ID: "x", Path: "/srv/oldie"}})
|
||||
for _, restart := range []struct {
|
||||
name string
|
||||
@@ -167,12 +167,12 @@ func TestTheItemsComeFromTheEnginesKeysAndARoundSaysHowFarItGot(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func readingOf(s *memStore, key string) *health.Reading {
|
||||
func readingOf(s *memStore, key string) *diskreadings.Reading {
|
||||
raw, _ := s.Get(key)
|
||||
if raw == nil {
|
||||
return nil
|
||||
}
|
||||
r, err := health.DecodeReading(raw)
|
||||
r, err := diskreadings.DecodeReading(raw)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -6,9 +6,9 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"git.novox.be/novox/mesh-catalog/modules/disk-load/diskreadings"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
"os"
|
||||
"os/exec"
|
||||
"sort"
|
||||
@@ -287,7 +287,7 @@ func (k *Keeper) MeasureModule(ctx context.Context, module string) {
|
||||
// asked only on a change of that module's key (novox/hq ADR 0577 clause 2), never from a round. It is cleared when the
|
||||
// item is declared again (measureOne).
|
||||
func (k *Keeper) retireGone(module string, declared []catalogue.PlacedDataItem) (int, error) {
|
||||
return k.retire(func(r health.Reading) bool { return r.Module == module }, declared)
|
||||
return k.retire(func(r diskreadings.Reading) bool { return r.Module == module }, declared)
|
||||
}
|
||||
|
||||
// RetireUnkeyed retires, once the watch handed over every current key, the readings of this node whose module has no
|
||||
@@ -310,7 +310,7 @@ func (k *Keeper) RetireUnkeyed(ctx context.Context) {
|
||||
for _, its := range items {
|
||||
declared = append(declared, its...)
|
||||
}
|
||||
n, err := k.retire(func(r health.Reading) bool { _, keyed := items[r.Module]; return !keyed }, declared)
|
||||
n, err := k.retire(func(r diskreadings.Reading) bool { _, keyed := items[r.Module]; return !keyed }, declared)
|
||||
switch {
|
||||
case err != nil:
|
||||
k.Say("the readings of modules with no key are not all retired: %v", err)
|
||||
@@ -320,14 +320,14 @@ func (k *Keeper) RetireUnkeyed(ctx context.Context) {
|
||||
}
|
||||
|
||||
// retire marks retired every reading of this node that of picks and declared does not name.
|
||||
func (k *Keeper) retire(of func(health.Reading) bool, declared []catalogue.PlacedDataItem) (int, error) {
|
||||
func (k *Keeper) retire(of func(diskreadings.Reading) bool, declared []catalogue.PlacedDataItem) (int, error) {
|
||||
keys, err := k.Store.Keys()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
declaredNow := map[string]bool{}
|
||||
for _, it := range declared {
|
||||
declaredNow[health.ReadingKey(k.Node, it.Module, it.ID)] = true
|
||||
declaredNow[diskreadings.ReadingKey(k.Node, it.Module, it.ID)] = true
|
||||
}
|
||||
now := k.Measurer.Now().UTC()
|
||||
n := 0
|
||||
@@ -339,7 +339,7 @@ func (k *Keeper) retire(of func(health.Reading) bool, declared []catalogue.Place
|
||||
if err != nil || len(raw) == 0 {
|
||||
continue
|
||||
}
|
||||
r, err := health.DecodeReading(raw)
|
||||
r, err := diskreadings.DecodeReading(raw)
|
||||
if err != nil || r.Node != k.Node || !of(r) || !r.RetiredAt.IsZero() {
|
||||
continue
|
||||
}
|
||||
@@ -357,14 +357,14 @@ func (k *Keeper) retire(of func(health.Reading) bool, declared []catalogue.Place
|
||||
// when it changed. It answers whether it wrote the reading, what the measurement said failed ("" when it did not, or
|
||||
// when the item was not measured now), and an error when the reading could not be read or kept.
|
||||
func (k *Keeper) measureOne(ctx context.Context, it catalogue.PlacedDataItem, walks bool) (bool, string, error) {
|
||||
key := health.ReadingKey(k.Node, it.Module, it.ID)
|
||||
var prev *health.Reading
|
||||
key := diskreadings.ReadingKey(k.Node, it.Module, it.ID)
|
||||
var prev *diskreadings.Reading
|
||||
raw, err := k.Store.Get(key)
|
||||
if err != nil {
|
||||
return false, "", err
|
||||
}
|
||||
if len(raw) > 0 {
|
||||
if p, err := health.DecodeReading(raw); err == nil {
|
||||
if p, err := diskreadings.DecodeReading(raw); err == nil {
|
||||
prev = &p
|
||||
}
|
||||
}
|
||||
@@ -374,7 +374,7 @@ func (k *Keeper) measureOne(ctx context.Context, it catalogue.PlacedDataItem, wa
|
||||
return false, "", nil
|
||||
}
|
||||
m := k.Measurer.Measure(ctx, it)
|
||||
next, changed := health.NextReading(prev, k.Node, it, m, now)
|
||||
next, changed := diskreadings.NextReading(prev, k.Node, it, m, now)
|
||||
if prev != nil && !prev.RetiredAt.IsZero() {
|
||||
// Declared again: it is no longer retired.
|
||||
next.RetiredAt, next.RetiredWhy, changed = time.Time{}, "", true
|
||||
@@ -386,32 +386,32 @@ func (k *Keeper) measureOne(ctx context.Context, it catalogue.PlacedDataItem, wa
|
||||
}
|
||||
|
||||
// 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) (health.Reading, bool, error) {
|
||||
func (k *Keeper) MeasureNow(ctx context.Context, module, item string) (diskreadings.Reading, bool, error) {
|
||||
k.mu.Lock()
|
||||
defer k.mu.Unlock()
|
||||
items, err := k.Items(ctx)
|
||||
if err != nil {
|
||||
return health.Reading{}, false, err
|
||||
return diskreadings.Reading{}, false, err
|
||||
}
|
||||
for _, it := range items[module] {
|
||||
if it.ID != item {
|
||||
continue
|
||||
}
|
||||
if it.Class == ClassCache {
|
||||
return health.Reading{}, false, fmt.Errorf("%s.%s is a cache, which is listed and never measured", module, item)
|
||||
return diskreadings.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 health.Reading{}, false, err
|
||||
return diskreadings.Reading{}, false, err
|
||||
}
|
||||
raw, err := k.Store.Get(health.ReadingKey(k.Node, module, item))
|
||||
raw, err := k.Store.Get(diskreadings.ReadingKey(k.Node, module, item))
|
||||
if err != nil {
|
||||
return health.Reading{}, w, err
|
||||
return diskreadings.Reading{}, w, err
|
||||
}
|
||||
r, err := health.DecodeReading(raw)
|
||||
r, err := diskreadings.DecodeReading(raw)
|
||||
return r, w, err
|
||||
}
|
||||
return health.Reading{}, false, fmt.Errorf("%s declares no data item %q on this machine", module, item)
|
||||
return diskreadings.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
|
||||
|
||||
@@ -6,8 +6,8 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"git.novox.be/novox/mesh-catalog/modules/disk-load/diskreadings"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
@@ -56,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 catalogue.PlacedDataItem) health.Measured {
|
||||
func (b Measurer) Measure(ctx context.Context, it catalogue.PlacedDataItem) diskreadings.Measured {
|
||||
at := b.Now().UTC()
|
||||
m := health.Measured{MeasuredAt: &at}
|
||||
m := diskreadings.Measured{MeasuredAt: &at}
|
||||
if !b.exists(ctx, it.Path) {
|
||||
m.Error = it.Path + " does not exist"
|
||||
zero := int64(0)
|
||||
@@ -120,7 +120,7 @@ func (b Measurer) Measure(ctx context.Context, it catalogue.PlacedDataItem) heal
|
||||
|
||||
// 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) *health.Redundancy {
|
||||
func (b Measurer) redundancyOf(ctx context.Context, path string) *diskreadings.Redundancy {
|
||||
out, err := b.Run(ctx, "findmnt", "-no", "SOURCE,FSTYPE", "--target", path)
|
||||
if err != nil {
|
||||
return nil
|
||||
@@ -134,7 +134,7 @@ func (b Measurer) redundancyOf(ctx context.Context, path string) *health.Redunda
|
||||
switch {
|
||||
case fstype == "zfs":
|
||||
pool, _, _ := strings.Cut(source, "/")
|
||||
r := &health.Redundancy{Kind: "zfs", Where: pool}
|
||||
r := &diskreadings.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()
|
||||
@@ -156,7 +156,7 @@ func (b Measurer) redundancyOf(ctx context.Context, path string) *health.Redunda
|
||||
return r
|
||||
case strings.HasPrefix(source, "/dev/md"):
|
||||
device := strings.TrimPrefix(source, "/dev/")
|
||||
r := &health.Redundancy{Kind: "md", Where: device}
|
||||
r := &diskreadings.Redundancy{Kind: "md", Where: device}
|
||||
mdstat, err := b.Run(ctx, "cat", "/proc/mdstat")
|
||||
if err != nil {
|
||||
r.Said = err.Error()
|
||||
@@ -175,7 +175,7 @@ func (b Measurer) redundancyOf(ctx context.Context, path string) *health.Redunda
|
||||
}
|
||||
return r
|
||||
case fstype == "btrfs":
|
||||
r := &health.Redundancy{Kind: "btrfs", Where: source}
|
||||
r := &diskreadings.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())
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"os"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// The readings' layout is disk-load's own package, published by its series for the readers to take (ADR 0634 §4).
|
||||
func TestTheReadingsLayoutIsAPublishedPackageOfThisModule(t *testing.T) {
|
||||
raw, err := os.ReadFile("../../module.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var m struct {
|
||||
Build struct {
|
||||
Artifacts []struct {
|
||||
Name, Kind, Series string
|
||||
Paths []string
|
||||
} `json:"artifacts"`
|
||||
} `json:"build"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &m); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, a := range m.Build.Artifacts {
|
||||
if a.Name == "diskreadings" && a.Kind == "package" && a.Series == "0.1" && len(a.Paths) == 1 && a.Paths[0] == "diskreadings" {
|
||||
return
|
||||
}
|
||||
}
|
||||
t.Errorf("no package artifact diskreadings of series 0.1 in %s", raw)
|
||||
}
|
||||
@@ -1,12 +1,12 @@
|
||||
package main
|
||||
|
||||
// The readings layout, their key and the peak are mesh-sdk's (health.Reading, health.NextReading; novox/hq ADR 0571
|
||||
// The readings layout, their key and the peak are the package diskreadings (Reading, 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 "git.novox.be/novox/mesh-sdk/go/health"
|
||||
import "git.novox.be/novox/mesh-catalog/modules/disk-load/diskreadings"
|
||||
|
||||
// StateReadings is the state disk-load keeps.
|
||||
const StateReadings = health.ReadingsState
|
||||
const StateReadings = diskreadings.ReadingsState
|
||||
|
||||
// StateRounds is its state of each node's last round, key `<node>` (novox/hq ADR 0577 clause 3): apart from the readings,
|
||||
// so a reader of the readings never meets a round.
|
||||
|
||||
@@ -9,8 +9,8 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-catalog/modules/disk-load/diskreadings"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
)
|
||||
|
||||
type memStore struct {
|
||||
@@ -84,11 +84,11 @@ func TestARoundMeasuresTheDeclaredItemsAndWritesWhatChanged(t *testing.T) {
|
||||
if _, _, err := k.Round(context.Background(), false); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if kept, _ := health.DecodeReading(store.m["ace.pg.data"]); !kept.RetiredAt.IsZero() {
|
||||
if kept, _ := diskreadings.DecodeReading(store.m["ace.pg.data"]); !kept.RetiredAt.IsZero() {
|
||||
t.Fatalf("a round retired a reading: %+v", kept)
|
||||
}
|
||||
k.MeasureModule(context.Background(), "pg")
|
||||
gone, _ := health.DecodeReading(store.m["ace.pg.data"])
|
||||
gone, _ := diskreadings.DecodeReading(store.m["ace.pg.data"])
|
||||
if gone.RetiredAt.IsZero() || !strings.Contains(gone.RetiredWhy, "no longer declared on ace") || gone.FirstSeen.IsZero() {
|
||||
t.Fatalf("a reading of an item no longer declared: %+v", gone)
|
||||
}
|
||||
@@ -96,7 +96,7 @@ func TestARoundMeasuresTheDeclaredItemsAndWritesWhatChanged(t *testing.T) {
|
||||
if _, _, err := k.Round(context.Background(), false); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if back, _ := health.DecodeReading(store.m["ace.pg.data"]); !back.RetiredAt.IsZero() {
|
||||
if back, _ := diskreadings.DecodeReading(store.m["ace.pg.data"]); !back.RetiredAt.IsZero() {
|
||||
t.Fatalf("a reading declared again is still retired: %+v", back)
|
||||
}
|
||||
if _, _, err := k.MeasureNow(context.Background(), "pg", "tmp"); err == nil {
|
||||
|
||||
@@ -8,8 +8,8 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-catalog/modules/disk-load/diskreadings"
|
||||
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
|
||||
@@ -35,7 +35,7 @@ func ReadingTools(k *Keeper) []stdio.Tool {
|
||||
if len(raw) == 0 {
|
||||
return nil, fmt.Errorf("no reading is kept at %q", key)
|
||||
}
|
||||
r, err := health.DecodeReading(raw)
|
||||
r, err := diskreadings.DecodeReading(raw)
|
||||
return r, err
|
||||
}
|
||||
keys, err := k.Store.Keys()
|
||||
@@ -45,12 +45,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 health.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 diskreadings.Reading `json:"-"`
|
||||
}
|
||||
var lines []line
|
||||
total := 0
|
||||
@@ -60,7 +60,7 @@ func ReadingTools(k *Keeper) []stdio.Tool {
|
||||
return nil, err
|
||||
}
|
||||
total += len(raw)
|
||||
r, err := health.DecodeReading(raw)
|
||||
r, err := diskreadings.DecodeReading(raw)
|
||||
if err != nil || (module != "" && r.Module != module) {
|
||||
continue
|
||||
}
|
||||
@@ -118,7 +118,7 @@ func ReadingTools(k *Keeper) []stdio.Tool {
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]any{"written": written, "reading": r, "key": health.ReadingKey(k.Node, module, item),
|
||||
return map[string]any{"written": written, "reading": r, "key": diskreadings.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
|
||||
},
|
||||
},
|
||||
|
||||
@@ -26,9 +26,9 @@ import (
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-catalog/modules/disk-load/diskreadings"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
"git.novox.be/novox/mesh-sdk/go/testbus"
|
||||
)
|
||||
|
||||
@@ -195,7 +195,7 @@ func TestDiskLoadWholeMeasuresAnItemWithinAMinuteOfItsKey(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, b := range []string{delivery.EngineDataBucket, health.ReadingsBucket, "disk-load_" + StateRounds} {
|
||||
for _, b := range []string{delivery.EngineDataBucket, diskreadings.ReadingsBucket, "disk-load_" + StateRounds} {
|
||||
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -240,11 +240,11 @@ func TestDiskLoadWholeMeasuresAnItemWithinAMinuteOfItsKey(t *testing.T) {
|
||||
if _, err := data.Put(ctx, key, raw); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
readings, _ := js.KeyValue(ctx, health.ReadingsBucket)
|
||||
readings, _ := js.KeyValue(ctx, diskreadings.ReadingsBucket)
|
||||
for {
|
||||
e, err := readings.Get(ctx, health.ReadingKey("ace", "pg", "data"))
|
||||
e, err := readings.Get(ctx, diskreadings.ReadingKey("ace", "pg", "data"))
|
||||
if err == nil {
|
||||
r, err := health.DecodeReading(e.Value())
|
||||
r, err := diskreadings.DecodeReading(e.Value())
|
||||
if err != nil || r.Path != dir || r.MeasuredAt == nil {
|
||||
t.Fatalf("the reading is %s (%v)", e.Value(), err)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,5 @@
|
||||
module git.novox.be/novox/mesh-catalog/modules/disk-load/diskreadings
|
||||
|
||||
go 1.26.0
|
||||
|
||||
require git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000
|
||||
@@ -0,0 +1,137 @@
|
||||
package diskreadings
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
// **Disk usage readings and peaks per data item** (novox/hq ADR 0571 clause 3, ADR 0233 §21, §27, §34): disk-load, on every
|
||||
// node, measures each data item the modules of its node declare (catalogue.PlacedDataItem, from compose/frombus.DataItems)
|
||||
// and keeps the newest reading and the item's largest size in seven days as its own state `readings`, one key
|
||||
// `<node>.<module>.<item>`, written only when it changes. The node-backup holder, mounts, nfs-server and the self-check's
|
||||
// probes read it there; the controller keeps none of it. The package is disk-load's own, published by its series (ADR 0634 §4); the readers import it.
|
||||
const (
|
||||
DiskLoadModule = "disk-load"
|
||||
ReadingsState = "readings"
|
||||
ReadingsBucket = DiskLoadModule + "_" + ReadingsState
|
||||
ReadingsRead = DiskLoadModule + "." + ReadingsState
|
||||
)
|
||||
|
||||
// ReadingVersion is the version of a reading's layout.
|
||||
const ReadingVersion = 1
|
||||
|
||||
// PeakWindow is how far back the peak looks (ADR 0233 §34).
|
||||
const PeakWindow = 7 * 24 * time.Hour
|
||||
|
||||
// Redundancy is the redundant storage an item is on, as read: zfs, md or btrfs, its pool, array or filesystem, and its
|
||||
// health (nil when it could not be read).
|
||||
type Redundancy struct {
|
||||
Kind string `json:"kind"`
|
||||
Where string `json:"where"`
|
||||
Healthy *bool `json:"healthy"`
|
||||
Said string `json:"said"`
|
||||
}
|
||||
|
||||
// Measured is what one measurement found. Precision says what the size is: "exact", "dataset <name>: its whole size",
|
||||
// "partial: ..." (a lower bound, never compared), or "no size: ...".
|
||||
type Measured struct {
|
||||
SizeBytes *int64 `json:"size_bytes,omitempty"`
|
||||
LastWrite *time.Time `json:"last_write,omitempty"`
|
||||
MeasuredAt *time.Time `json:"measured_at,omitempty"`
|
||||
Precision string `json:"precision,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
Redundancy *Redundancy `json:"redundancy,omitempty"`
|
||||
}
|
||||
|
||||
// Partial says the size is only a lower bound, which a peak is never made of.
|
||||
func (m Measured) Partial() bool { return strings.HasPrefix(m.Precision, "partial") }
|
||||
|
||||
// DaySize is the largest size measured on one UTC day.
|
||||
type DaySize struct {
|
||||
Day string `json:"day"`
|
||||
Bytes int64 `json:"bytes"`
|
||||
}
|
||||
|
||||
// Reading is the value of one key: the item as placed, what was measured, and the largest size in the seven days up to
|
||||
// it (Peak, measured on PeakAt), read from the day maxima Days.
|
||||
type Reading struct {
|
||||
V int `json:"v"`
|
||||
Node string `json:"node"`
|
||||
catalogue.PlacedDataItem
|
||||
Measured
|
||||
Peak *int64 `json:"peak_bytes,omitempty"`
|
||||
PeakAt string `json:"peak_day,omitempty"`
|
||||
Days []DaySize `json:"days,omitempty"`
|
||||
// FirstSeen is when the item was first measured on this node, kept by every reading after; RetiredAt and RetiredWhy
|
||||
// say its module left the node, written by whoever retires it and carried on unchanged. Each is left out while zero,
|
||||
// so a reading written before reads as it did.
|
||||
FirstSeen time.Time `json:"first_seen,omitzero"`
|
||||
RetiredAt time.Time `json:"retired_at,omitzero"`
|
||||
RetiredWhy string `json:"retired_why,omitempty"`
|
||||
}
|
||||
|
||||
// ReadingKey is the key of an item on a node: `<node>.<module>.<item>`.
|
||||
func ReadingKey(node, module, item string) string { return node + "." + module + "." + item }
|
||||
|
||||
// DecodeReading reads a key's value; it refuses another version.
|
||||
func DecodeReading(raw []byte) (Reading, error) {
|
||||
var r Reading
|
||||
if err := json.Unmarshal(raw, &r); err != nil {
|
||||
return Reading{}, fmt.Errorf("diskreadings: a reading is not readable: %w", err)
|
||||
}
|
||||
if r.V != ReadingVersion {
|
||||
return Reading{}, fmt.Errorf("diskreadings: 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, FirstSeen: now.UTC()}
|
||||
var days []DaySize
|
||||
if prev != nil {
|
||||
days = prev.Days
|
||||
r.RetiredAt, r.RetiredWhy = prev.RetiredAt, prev.RetiredWhy
|
||||
if !prev.FirstSeen.IsZero() {
|
||||
r.FirstSeen = prev.FirstSeen
|
||||
}
|
||||
}
|
||||
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)
|
||||
}
|
||||
@@ -0,0 +1,55 @@
|
||||
package diskreadings
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
// ADR 0571 clause 3: a reading keeps its seven-day peak and is written only when it changes; a partial size is no peak.
|
||||
func TestAReadingKeepsItsPeakAndIsWrittenOnlyWhenItChanges(t *testing.T) {
|
||||
at := time.Date(2026, 10, 29, 12, 0, 0, 0, time.UTC)
|
||||
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
|
||||
}
|
||||
// The first reading sets FirstSeen and every one after keeps it, as it keeps a retirement; an older reading reads as before.
|
||||
if first := time.Date(2026, 10, 20, 1, 0, 0, 0, time.UTC); !prev.FirstSeen.Equal(first) {
|
||||
t.Fatalf("first seen %s, not %s", prev.FirstSeen, first)
|
||||
}
|
||||
retired := *prev
|
||||
retired.RetiredAt, retired.RetiredWhy = at, "its module left ace"
|
||||
if r, _ := NextReading(&retired, "ace", retired.PlacedDataItem, retired.Measured, at.Add(time.Hour)); !r.RetiredAt.Equal(at) ||
|
||||
r.RetiredWhy != "its module left ace" || !r.FirstSeen.Equal(prev.FirstSeen) {
|
||||
t.Fatalf("a retired reading moved on as %+v", r)
|
||||
}
|
||||
if old, err := DecodeReading([]byte(`{"v":1,"node":"ace","module":"pg","item":"data"}`)); err != nil || !old.FirstSeen.IsZero() {
|
||||
t.Fatalf("a reading of before read as %+v: %v", old, err)
|
||||
}
|
||||
if raw, _ := json.Marshal(Reading{V: ReadingVersion}); strings.Contains(string(raw), "first_seen") || strings.Contains(string(raw), "retired") {
|
||||
t.Fatalf("zero fields were written: %s", raw)
|
||||
}
|
||||
if ReadingKey("ace", "pg", "data") != "ace.pg.data" || ReadingsBucket != "disk-load_readings" {
|
||||
t.Fatal("the readings' key or bucket moved")
|
||||
}
|
||||
}
|
||||
@@ -21,3 +21,7 @@ require (
|
||||
golang.org/x/sys v0.48.0 // indirect
|
||||
golang.org/x/time v0.16.0 // indirect
|
||||
)
|
||||
|
||||
require git.novox.be/novox/mesh-catalog/modules/disk-load/diskreadings v0.0.0-00010101000000-000000000000
|
||||
|
||||
replace git.novox.be/novox/mesh-catalog/modules/disk-load/diskreadings => ./diskreadings
|
||||
|
||||
@@ -24,6 +24,15 @@
|
||||
"loads": [
|
||||
"disk-load"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "diskreadings",
|
||||
"kind": "package",
|
||||
"language": "go",
|
||||
"paths": [
|
||||
"diskreadings"
|
||||
],
|
||||
"series": "0.1"
|
||||
}
|
||||
]
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user