diff --git a/modules/disk-load/.gitignore b/modules/disk-load/.gitignore index 96984eae..76f82dbe 100644 --- a/modules/disk-load/.gitignore +++ b/modules/disk-load/.gitignore @@ -1 +1,2 @@ -disk-load +/disk-load +/cmd/disk-load/disk-load diff --git a/modules/disk-load/cmd/disk-load/data.go b/modules/disk-load/cmd/disk-load/data.go new file mode 100644 index 00000000..7882c804 --- /dev/null +++ b/modules/disk-load/cmd/disk-load/data.go @@ -0,0 +1,123 @@ +package main + +// Where disk-load learns the data items it measures (novox/hq ADR 0577 clause 2): the node-engine's state data +// (delivery.EngineDataBucket), one key `.` per module placed on this node that declares data items, as the +// engine composed them. disk-load watches its own node's keys and composes nothing: it reads no resolved values, no +// manifests and no placement, so a manifest its mesh-sdk cannot judge never silences it, and a round reads one watched +// map instead of several hundred keys over the tunnel (issue 1955). + +import ( + "context" + "errors" + "fmt" + "sync" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// Watched is this node's data items as the engine's state data holds them, by module, kept from a watch of its keys. +type Watched struct { + Node string + + mu sync.Mutex + byKey map[string]string // key to its module, for a delete that carries no value + byModule map[string][]catalogue.PlacedDataItem + ready bool +} + +// ErrNotWatchedYet is a round asked before the watch handed over the engine's current keys: no item is known yet, and +// none is taken for gone. +var ErrNotWatchedYet = errors.New("the node-engine's state " + delivery.EngineDataBucket + " is not watched yet: " + + "no data item of this machine is known") + +// Take keeps one change of a key and answers the module it is of and whether its items changed. +func (w *Watched) Take(ch stdio.StateChange) (string, bool, error) { + w.mu.Lock() + defer w.mu.Unlock() + if w.byKey == nil { + w.byKey, w.byModule = map[string]string{}, map[string][]catalogue.PlacedDataItem{} + } + if ch.Op != stdio.OpPut { + module, known := w.byKey[ch.Key] + if !known { + return "", false, nil + } + delete(w.byKey, ch.Key) + delete(w.byModule, module) + return module, true, nil + } + d, err := delivery.DecodeEngineData(ch.Value) + if err != nil { + return "", false, fmt.Errorf("%s: %w", ch.Key, err) + } + if d.Node != w.Node { + return "", false, fmt.Errorf("%s names the machine %q, and this is %s", ch.Key, d.Node, w.Node) + } + for _, it := range d.Items { + if it.Module != d.Module { + return "", false, fmt.Errorf("%s holds an item of %q, and is %s's", ch.Key, it.Module, d.Module) + } + } + changed := !sameItems(w.byModule[d.Module], d.Items) + w.byKey[ch.Key] = d.Module + w.byModule[d.Module] = d.Items + return d.Module, changed, nil +} + +// Ready says the watch handed over every current key: from now on an item missing is an item gone. +func (w *Watched) Ready() { + w.mu.Lock() + w.ready = true + w.mu.Unlock() +} + +// All is every module's items, as the Keeper reads them; refused until the watch is ready. +func (w *Watched) All(context.Context) (map[string][]catalogue.PlacedDataItem, error) { + w.mu.Lock() + defer w.mu.Unlock() + if !w.ready { + return nil, ErrNotWatchedYet + } + out := make(map[string][]catalogue.PlacedDataItem, len(w.byModule)) + for m, its := range w.byModule { + out[m] = append([]catalogue.PlacedDataItem(nil), its...) + } + return out, nil +} + +func sameItems(a, b []catalogue.PlacedDataItem) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} + +// Follow watches this node's keys of the engine's state data, and asks the keeper to measure a module whose items changed +// after the watch began; it returns once the current keys were handed over (stdio.KeptState.Follow). +func (w *Watched) Follow(k *Keeper) error { + pattern, err := delivery.EngineDataNodeKeys(w.Node) + if err != nil { + return err + } + stdio.State(delivery.EngineDataRead).Follow(pattern, func(ch stdio.StateChange) error { + module, changed, err := w.Take(ch) + if err != nil || !changed || ch.Current { + return err + } + k.Due(module) + return nil + }) + w.Ready() + // The handed-over set is complete now: a module whose key went while the watch was down is retired once. + k.RetireUnkeyed(context.Background()) + // A round refused while the watch caught up runs now, not at the next hour (the delivery's readings within minutes). + go k.RoundIfRefused(context.Background()) + return nil +} diff --git a/modules/disk-load/cmd/disk-load/data_test.go b/modules/disk-load/cmd/disk-load/data_test.go new file mode 100644 index 00000000..c8a86f46 --- /dev/null +++ b/modules/disk-load/cmd/disk-load/data_test.go @@ -0,0 +1,190 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "os" + "strings" + "testing" + "time" + + 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 +// this machine as the watch hands them over, and none is known (none taken for gone) before the watch caught up; a key +// that changes after the watch began has its module measured again, and one deleted is measured no more; an item whose +// measurement fails is a failed item of the round, which measures the rest, and the round's key says so. +func TestTheItemsComeFromTheEnginesKeysAndARoundSaysHowFarItGot(t *testing.T) { + key := func(module string) string { k, _ := delivery.EngineDataKey("ace", module); return k } + value := func(module string, items ...catalogue.PlacedDataItem) json.RawMessage { + raw, err := delivery.EncodeEngineData(delivery.EngineData{Node: "ace", Module: module, Digest: "sha256:" + module, Items: items}) + if err != nil { + t.Fatal(err) + } + return raw + } + pg := catalogue.PlacedDataItem{Module: "pg", ID: "data", Class: "valuable", Path: "/srv/pg", Measure: catalogue.MeasureShallow} + gone := catalogue.PlacedDataItem{Module: "pg", ID: "wal", Class: "valuable", Path: "/srv/missing", Measure: catalogue.MeasureShallow} + web := catalogue.PlacedDataItem{Module: "web", ID: "uploads", Class: "valuable", Path: "/srv/web", Measure: catalogue.MeasureShallow} + measured := map[string]int{} + run := func(_ context.Context, name string, args ...string) (string, error) { + switch { + case name == "test" && args[1] == "/srv/missing": + return "", errors.New("no such file") + case name == "test": + return "", nil + case name == "bash" && strings.Contains(args[1], "-maxdepth 1"): + measured[strings.Split(args[1], "'")[1]]++ + return "1760000000 2\n", nil + } + return "", os.ErrNotExist + } + now := time.Date(2026, 11, 1, 0, 30, 0, 0, time.UTC) + readings, rounds := &memStore{m: map[string]json.RawMessage{}}, &memStore{m: map[string]json.RawMessage{}} + w := &Watched{Node: "ace"} + var said []string + k := &Keeper{Node: "ace", Items: w.All, Store: readings, Rounds: rounds, + Say: func(f string, a ...any) { said = append(said, f) }, + Measurer: Measurer{Run: run, Now: func() time.Time { return now }}} + round := func() Round { + var r Round + raw, _ := rounds.Get("ace") + if err := json.Unmarshal(raw, &r); err != nil { + t.Fatalf("the round's key: %v", err) + } + return r + } + + // Before the watch caught up nothing is known, and a round says so in its key rather than retiring every reading. + if _, _, err := k.Round(context.Background(), false); !errors.Is(err, ErrNotWatchedYet) { + t.Fatalf("a round before the watch: %v", err) + } + if r := round(); r.Ended == nil || !strings.Contains(r.Error, "not watched yet") { + t.Fatalf("the round before the watch is kept as %+v", r) + } + refusedAt := round().Started + for _, step := range []struct { + name string + change stdio.StateChange + module string + changed bool + err bool + }{ + {"a current key", stdio.StateChange{Key: key("pg"), Op: stdio.OpPut, Value: value("pg", pg, gone), Current: true}, "pg", true, false}, + {"another current key", stdio.StateChange{Key: key("web"), Op: stdio.OpPut, Value: value("web", web), Current: true}, "web", true, false}, + {"the same items again", stdio.StateChange{Key: key("web"), Op: stdio.OpPut, Value: value("web", web)}, "web", false, false}, + {"another machine's items", stdio.StateChange{Key: key("x"), Op: stdio.OpPut, Value: json.RawMessage( + `{"v":1,"node":"g14","module":"x","digest":"d","items":[]}`)}, "", false, true}, + {"a delete of a key never seen", stdio.StateChange{Key: key("y"), Op: stdio.OpDelete}, "", false, false}, + } { + module, changed, err := w.Take(step.change) + if module != step.module || changed != step.changed || (err != nil) != step.err { + t.Errorf("%s: taken as %q, changed %v, %v", step.name, module, changed, err) + } + } + w.Ready() + + // The round refused before the watch caught up runs once it has (Watched.Follow), not at the next hour: it measures + // every item it was handed; the missing one fails, and the round goes on and says it. + now = now.Add(time.Minute) + if !k.RoundIfRefused(context.Background()) || k.RoundIfRefused(context.Background()) { + t.Fatal("the round refused before the watch was not run once when the watch caught up") + } + r := round() + if !r.Started.After(refusedAt) { + t.Fatalf("the round's key still says the refused round: %+v", r) + } + if r.Ended == nil || r.Items != 3 || r.Measured != 3 || len(r.Failed) != 1 || r.Failed[0].Item != "pg.wal" || + !strings.Contains(r.Failed[0].Why, "does not exist") || measured["/srv/pg"] != 1 || measured["/srv/web"] != 1 { + t.Fatalf("the round's key says %+v; measured %v", r, measured) + } + if !strings.Contains(strings.Join(said, "\n"), "the round ended") || !strings.Contains(strings.Join(said, "\n"), "starts") { + t.Errorf("the journal said %v", said) + } + + // A key that changes after the watch began has its module measured again, at once, by Work. + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go k.Work(ctx) + moved := web + moved.Path = "/srv/web2" + if module, changed, err := w.Take(stdio.StateChange{Key: key("web"), Op: stdio.OpPut, Value: value("web", moved)}); err != nil || !changed { + t.Fatalf("a changed key: %s %v %v", module, changed, err) + } + k.Due("web") + waitFor(t, func() bool { + raw := readingOf(readings, "ace.web.uploads") + return raw != nil && raw.Path == "/srv/web2" + }) + // A key deleted: its module's items are measured no more, and its reading is retired. + if module, changed, _ := w.Take(stdio.StateChange{Key: key("web"), Op: stdio.OpDelete}); module != "web" || !changed { + t.Fatalf("a deleted key was taken as %q, %v", module, changed) + } + k.Due("web") + waitFor(t, func() bool { r := readingOf(readings, "ace.web.uploads"); return r != nil && !r.RetiredAt.IsZero() }) + before := measured["/srv/web2"] + if _, _, err := k.Round(context.Background(), false); err != nil { + t.Fatal(err) + } + if measured["/srv/web2"] != before || round().Items != 2 { + t.Errorf("an item whose key was deleted was measured again: %d times, round %+v", measured["/srv/web2"]-before, round()) + } + + // 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", + PlacedDataItem: catalogue.PlacedDataItem{Module: "oldie", ID: "x", Path: "/srv/oldie"}}) + for _, restart := range []struct { + name string + keys []stdio.StateChange + retired bool + }{ + {"no key yet", nil, false}, + {"pg's key alone", []stdio.StateChange{{Key: key("pg"), Op: stdio.OpPut, Value: value("pg", pg, gone), Current: true}}, true}, + } { + w2 := &Watched{Node: "ace"} + said = nil + k2 := &Keeper{Node: "ace", Items: w2.All, Store: readings, Say: k.Say, Measurer: k.Measurer} + for _, ch := range restart.keys { + if _, _, err := w2.Take(ch); err != nil { + t.Fatal(err) + } + } + w2.Ready() + k2.RetireUnkeyed(context.Background()) + oldie, pgData := readingOf(readings, "ace.oldie.x"), readingOf(readings, "ace.pg.data") + if oldie == nil || pgData == nil || oldie.RetiredAt.IsZero() == restart.retired || !pgData.RetiredAt.IsZero() { + t.Errorf("%s: oldie %+v, pg %+v; oldie retired: want %v, pg never", restart.name, oldie, pgData, restart.retired) + } + if !restart.retired && !strings.Contains(strings.Join(said, "\n"), "holds no key") { + t.Errorf("%s: the skip was not said: %v", restart.name, said) + } + } +} + +func readingOf(s *memStore, key string) *health.Reading { + raw, _ := s.Get(key) + if raw == nil { + return nil + } + r, err := health.DecodeReading(raw) + if err != nil { + return nil + } + return &r +} + +func waitFor(t *testing.T, done func() bool) { + t.Helper() + for end := time.Now().Add(5 * time.Second); time.Now().Before(end); time.Sleep(5 * time.Millisecond) { + if done() { + return + } + } + t.Fatal("not done within 5 s") +} diff --git a/modules/disk-load/cmd/disk-load/keeper.go b/modules/disk-load/cmd/disk-load/keeper.go index 388ef136..95998f3f 100644 --- a/modules/disk-load/cmd/disk-load/keeper.go +++ b/modules/disk-load/cmd/disk-load/keeper.go @@ -7,12 +7,14 @@ import ( "errors" "fmt" "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" "strings" "sync" + "sync/atomic" "time" stdio "git.novox.be/novox/mesh-sdk/go" @@ -25,47 +27,130 @@ type Store interface { Keys() ([]string, error) } -// busStore is the state through the runtime. -type busStore struct{} +// busStore is one of disk-load's states through the runtime: readings, or rounds. +type busStore struct{ state string } -func (busStore) Get(key string) (json.RawMessage, error) { - e, err := stdio.State(StateReadings).Get(key) +func (b busStore) Get(key string) (json.RawMessage, error) { + e, err := stdio.State(b.state).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) +func (b busStore) Put(key string, v any) error { + _, err := stdio.State(b.state).Put(key, v) return err } -func (busStore) Keys() ([]string, error) { return stdio.State(StateReadings).Keys() } +func (b busStore) Keys() ([]string, error) { return stdio.State(b.state).Keys() } // Keeper measures the items of its node and keeps the readings. type Keeper struct { Node string - // Items is every data item the modules placed on this node declare, composed from the bus (frombus.DataItems). + // Items is every data item the modules placed on this node declare, as the node-engine composed them and its state + // data holds them (Watched.All, novox/hq ADR 0577 clause 2). Items func(ctx context.Context) (map[string][]catalogue.PlacedDataItem, error) Measurer Measurer Store Store - Say func(string, ...any) + // Rounds is the state rounds, where each round says how far it got (ADR 0577 clause 3); nil keeps none. + Rounds Store + Say func(string, ...any) mu sync.Mutex // one round at a time + + // pending are the modules whose items changed since they were last measured (Due), measured by Work. + pmu sync.Mutex + pending map[string]bool + wake chan struct{} + + // unwatched says a round was refused because the watch had not caught up (ErrNotWatchedYet): one runs as soon as + // it has (Watched.Follow), not an hour later. + unwatched atomic.Bool +} + +// RoundIfRefused runs one round now when the last was refused for a watch not caught up, and says whether it did. +func (k *Keeper) RoundIfRefused(ctx context.Context) bool { + if !k.unwatched.CompareAndSwap(true, false) { + return false + } + _, _, _ = k.Round(ctx, false) + return true +} + +// SayEvery is how many items a round measures between two lines saying how far it got. +const SayEvery = 50 + +// RoundVersion is the version of a round's layout. +const RoundVersion = 1 + +// Round is the value of a node's key of the state rounds: when its last round started and ended, how many items it +// had to measure, how many it measured, wrote and kept, and each item that failed and why. Ended is nil while it runs; +// Error says why a round measured nothing at all. +type Round struct { + V int `json:"v"` + Node string `json:"node"` + Walks bool `json:"walks,omitempty"` + Started time.Time `json:"started"` + Ended *time.Time `json:"ended,omitempty"` + Items int `json:"items"` + Measured int `json:"measured"` + Written int `json:"written"` + Kept int `json:"kept"` + Failed []FailedItem `json:"failed,omitempty"` + Error string `json:"error,omitempty"` +} + +// FailedItem is one item a round could not measure or keep, `.`, and why. +type FailedItem struct { + Item string `json:"item"` + Why string `json:"why"` +} + +// putRound writes the round to this node's key; a round that cannot be written is said, and measures on. +func (k *Keeper) putRound(r Round) { + if k.Rounds == nil { + return + } + if err := k.Rounds.Put(k.Node, r); err != nil { + k.Say("the round's outcome cannot be kept in the state %s: %v", StateRounds, err) + } } // 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). +// +// A round says how far it got (novox/hq ADR 0577 clause 3): it writes its outcome to this node's key of the state rounds +// when it starts, every SayEvery items and when it ends, and says the same in the journal. An item that fails is a +// failed item of the round, never its end. func (k *Keeper) Round(ctx context.Context, walks bool) (written, kept int, err error) { k.mu.Lock() defer k.mu.Unlock() + r := Round{V: RoundVersion, Node: k.Node, Walks: walks, Started: k.Measurer.Now().UTC()} + end := func() { + at := k.Measurer.Now().UTC() + r.Ended = &at + k.putRound(r) + } + k.putRound(r) + k.Say("a round of measuring the data declared here starts (walks: %v)", walks) items, err := k.Items(ctx) if err != nil { + if errors.Is(err, ErrNotWatchedYet) { + k.unwatched.Store(true) + } + r.Error = err.Error() + end() + k.Say("the round measured nothing: %v", err) return 0, 0, err } modules := make([]string, 0, len(items)) for m := range items { modules = append(modules, m) + for _, it := range items[m] { + if it.Class != ClassCache { + r.Items++ + } + } } sort.Strings(modules) var failed []string @@ -74,42 +159,175 @@ func (k *Keeper) Round(ctx context.Context, walks bool) (written, kept int, err if it.Class == ClassCache { continue } - w, err := k.measureOne(ctx, it, walks) - if err != nil { + w, said, err := k.measureOne(ctx, it, walks) + r.Measured++ + switch { + case err != nil: failed = append(failed, fmt.Sprintf("%s.%s: %v", module, it.ID, err)) - continue + r.Failed = append(r.Failed, FailedItem{Item: module + "." + it.ID, Why: err.Error()}) + case said != "": + // Measured, and the measurement failed: its reading says so, and the round counts it failed. + r.Failed = append(r.Failed, FailedItem{Item: module + "." + it.ID, Why: said}) } - if w { + if err == nil && w { written++ - } else { + } else if err == nil { kept++ } + r.Written, r.Kept = written, kept + if r.Measured%SayEvery == 0 { + k.Say("the round has measured %d of %d items: %d written, %d unchanged, %d failed", r.Measured, r.Items, + written, kept, len(r.Failed)) + k.putRound(r) + } } } - retired, err := k.retireGone(items) - written += retired - if err != nil { - failed = append(failed, err.Error()) - } + // A round retires nothing: a reading is retired only when its module's key says so (MeasureModule), never because + // an item is missing from what the watch holds now, which may be empty or partial. + end() + k.Say("the round ended: %d of %d items measured, %d readings written, %d unchanged, %d failed", r.Measured, r.Items, + written, kept, len(r.Failed)) if len(failed) > 0 { return written, kept, errors.New("readings not kept: " + strings.Join(failed, "; ")) } return written, kept, nil } -// retireGone marks retired every reading of this node whose item the node no longer declares — its module was unassigned -// there, or is no longer pulled in — with when and why, and keeps it: the record of what the module left behind, until a -// person deletes it (ADR 0233). It is cleared when the item is declared again (measureOne). -func (k *Keeper) retireGone(declared map[string][]catalogue.PlacedDataItem) (int, error) { +// Due asks Work to measure a module's items: its key of the engine's state data changed (novox/hq ADR 0577 clause 2). +func (k *Keeper) Due(module string) { + k.pmu.Lock() + if k.pending == nil { + k.pending, k.wake = map[string]bool{}, make(chan struct{}, 1) + } + k.pending[module] = true + wake := k.wake + k.pmu.Unlock() + select { + case wake <- struct{}{}: + default: + } +} + +// Work measures each module Due names, one at a time, until ctx ends. +func (k *Keeper) Work(ctx context.Context) { + k.pmu.Lock() + if k.pending == nil { + k.pending, k.wake = map[string]bool{}, make(chan struct{}, 1) + } + wake := k.wake + k.pmu.Unlock() + for { + select { + case <-ctx.Done(): + return + case <-wake: + } + for { + k.pmu.Lock() + var module string + for m := range k.pending { + module = m + break + } + delete(k.pending, module) + k.pmu.Unlock() + if module == "" { + break + } + k.MeasureModule(ctx, module) + } + } +} + +// MeasureModule measures one module's items now, as its key of the engine's state data changed, and retires the +// readings of a module that no longer declares any here. A walk is not repeated within a day for an item that did not +// change (measureOne). +func (k *Keeper) MeasureModule(ctx context.Context, module string) { + k.mu.Lock() + defer k.mu.Unlock() + items, err := k.Items(ctx) + if err != nil { + k.Say("%s's data items changed and are not measured: %v", module, err) + return + } + n, err := k.retireGone(module, items[module]) + if err != nil { + k.Say("%s's readings of items it no longer declares here are not retired: %v", module, err) + } else if n > 0 { + k.Say("%s no longer declares %d item(s) here: their readings are retired, measured no more", module, n) + } + if len(items[module]) == 0 { + return + } + written, kept := 0, 0 + var failed []string + for _, it := range items[module] { + if it.Class == ClassCache { + continue + } + w, said, err := k.measureOne(ctx, it, false) + switch { + case err != nil: + failed = append(failed, it.ID+": "+err.Error()) + case said != "": + failed = append(failed, it.ID+": "+said) + case w: + written++ + default: + kept++ + } + } + k.Say("%s's data items changed and were measured: %d readings written, %d unchanged, %d failed%s", module, written, + kept, len(failed), map[bool]string{true: " (" + strings.Join(failed, "; ") + ")", false: ""}[len(failed) > 0]) +} + +// retireGone marks retired every reading of this node of module whose item its key of the engine's state data no longer +// names — the key was deleted (the module was unassigned there, or is no longer pulled in) or names fewer items — with +// when and why, and keeps it: the record of what the module left behind, until a person deletes it (ADR 0233). It is +// 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) +} + +// RetireUnkeyed retires, once the watch handed over every current key, the readings of this node whose module has no +// key in that set: a key deleted while the watch was not up (disk-load restarted, the machine rebooted or slept) never +// reaches disk-load as a delete. An empty set is skipped and said: the engine has not written its keys yet, and an +// empty set would retire every reading. +func (k *Keeper) RetireUnkeyed(ctx context.Context) { + k.mu.Lock() + defer k.mu.Unlock() + items, err := k.Items(ctx) + if err != nil { + k.Say("the readings of modules with no key are not looked for: %v", err) + return + } + if len(items) == 0 { + k.Say("the node-engine's state %s holds no key of this machine yet: no reading is retired for it", delivery.EngineDataBucket) + return + } + var declared []catalogue.PlacedDataItem + for _, its := range items { + declared = append(declared, its...) + } + n, err := k.retire(func(r health.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) + case n > 0: + k.Say("%d reading(s) of modules with no key of this machine retired, measured no more", n) + } +} + +// 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) { keys, err := k.Store.Keys() if err != nil { return 0, err } declaredNow := map[string]bool{} - for module, its := range declared { - for _, it := range its { - declaredNow[health.ReadingKey(k.Node, module, it.ID)] = true - } + for _, it := range declared { + declaredNow[health.ReadingKey(k.Node, it.Module, it.ID)] = true } now := k.Measurer.Now().UTC() n := 0 @@ -122,7 +340,7 @@ func (k *Keeper) retireGone(declared map[string][]catalogue.PlacedDataItem) (int continue } r, err := health.DecodeReading(raw) - if err != nil || r.Node != k.Node || !r.RetiredAt.IsZero() { + if err != nil || r.Node != k.Node || !of(r) || !r.RetiredAt.IsZero() { continue } r.RetiredAt = now @@ -135,12 +353,15 @@ func (k *Keeper) retireGone(declared map[string][]catalogue.PlacedDataItem) (int return n, nil } -func (k *Keeper) measureOne(ctx context.Context, it catalogue.PlacedDataItem, walks bool) (bool, error) { +// measureOne measures one item, unless it is a walk done within WalkEvery of an unchanged item, and writes its reading +// 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 raw, err := k.Store.Get(key) if err != nil { - return false, err + return false, "", err } if len(raw) > 0 { if p, err := health.DecodeReading(raw); err == nil { @@ -148,8 +369,9 @@ func (k *Keeper) measureOne(ctx context.Context, it catalogue.PlacedDataItem, wa } } now := k.Measurer.Now() - if prev != nil && prev.RetiredAt.IsZero() && it.Measure == catalogue.MeasureWalk && !walks && prev.MeasuredAt != nil && now.Sub(*prev.MeasuredAt) < WalkEvery { - return false, nil + if prev != nil && prev.RetiredAt.IsZero() && prev.PlacedDataItem == it && it.Measure == catalogue.MeasureWalk && !walks && + prev.MeasuredAt != nil && now.Sub(*prev.MeasuredAt) < WalkEvery { + return false, "", nil } m := k.Measurer.Measure(ctx, it) next, changed := health.NextReading(prev, k.Node, it, m, now) @@ -158,9 +380,9 @@ func (k *Keeper) measureOne(ctx context.Context, it catalogue.PlacedDataItem, wa next.RetiredAt, next.RetiredWhy, changed = time.Time{}, "", true } if !changed { - return false, nil + return false, m.Error, nil } - return true, k.Store.Put(key, next) + return true, m.Error, k.Store.Put(key, next) } // MeasureNow measures one item now, walk or not, and writes its reading when it changed. @@ -178,7 +400,7 @@ func (k *Keeper) MeasureNow(ctx context.Context, module, item string) (health.Re if it.Class == ClassCache { 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) + w, _, err := k.measureOne(ctx, it, true) if err != nil { return health.Reading{}, false, err } @@ -201,12 +423,7 @@ func (k *Keeper) Loop(ctx context.Context, every time.Duration, first time.Durat 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) - } + _, _, _ = k.Round(ctx, false) // it says its outcome itself, in the journal and its key of the state rounds select { case <-ctx.Done(): return diff --git a/modules/disk-load/cmd/disk-load/main.go b/modules/disk-load/cmd/disk-load/main.go index 578539fc..32fd45b6 100644 --- a/modules/disk-load/cmd/disk-load/main.go +++ b/modules/disk-load/cmd/disk-load/main.go @@ -4,42 +4,52 @@ // 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), composed from the bus (no file). +// seven-day peak as its own state `readings` (ADR 0571 clause 3). It learns the items from the node-engine's state data, +// watching its own node's keys, and composes nothing (ADR 0577 clause 2); each round says how far it got in its state +// `rounds` (clause 3). package main import ( "context" "fmt" "os" - "strings" "time" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/announce" - "git.novox.be/novox/mesh-sdk/go/catalogue" - "git.novox.be/novox/mesh-sdk/go/compose/frombus" - "git.novox.be/novox/mesh-sdk/go/compose/fromruntime" ) func logf(format string, args ...any) { fmt.Fprintf(os.Stderr, "[disk-load] "+format+"\n", args...) } +// Kept is every state disk-load keeps, as it announces them: the readings, readable by restic, nfs-server, mounts and +// the self-check (their manifests' `reads`; checks/readable holds it), and the rounds, its own tools' alone. +var Kept = []announce.State{{Name: StateReadings, Readable: true}, {Name: StateRounds}} + func main() { - // Readable: restic, nfs-server, mounts and the self-check read it (their manifests' `reads`); checks/readable holds it. - if err := stdio.Keeps(announce.State{Name: StateReadings, Readable: true}); err != nil { + if err := stdio.Keeps(Kept...); err != nil { logf("%v", err) os.Exit(1) } 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, node) - }} + watched := &Watched{Node: node} + k := &Keeper{Node: node, Store: busStore{StateReadings}, Rounds: busStore{StateRounds}, Say: logf, + Measurer: Measurer{Run: RootRunner, Now: time.Now}, Items: watched.All} if node != "" { 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) + ctx := context.Background() + go k.Work(ctx) + go func() { + <-stdio.Served() // the runtime's channel first: a watch is asked through it + if err := watched.Follow(k); err != nil { + logf("this machine's keys of the node-engine's data items cannot be watched, and nothing is measured: %v", err) + return + } + logf("watching this machine's keys of the node-engine's data items") + }() + go k.Loop(ctx, 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. @@ -48,15 +58,3 @@ func main() { os.Exit(1) } } - -// 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, node string) (map[string][]catalogue.PlacedDataItem, error) { - items, unplaced, err := frombus.DataItems(ctx, fromruntime.Reader(), 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/readings.go b/modules/disk-load/cmd/disk-load/readings.go index 3cad1734..c8528d30 100644 --- a/modules/disk-load/cmd/disk-load/readings.go +++ b/modules/disk-load/cmd/disk-load/readings.go @@ -8,5 +8,9 @@ import "git.novox.be/novox/mesh-sdk/go/health" // StateReadings is the state disk-load keeps. const StateReadings = health.ReadingsState +// StateRounds is its state of each node's last round, key `` (novox/hq ADR 0577 clause 3): apart from the readings, +// so a reader of the readings never meets a round. +const StateRounds = "rounds" + // 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 d934ad27..eb1d5785 100644 --- a/modules/disk-load/cmd/disk-load/readings_test.go +++ b/modules/disk-load/cmd/disk-load/readings_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "os" "strings" + "sync" "testing" "time" @@ -12,15 +13,26 @@ import ( "git.novox.be/novox/mesh-sdk/go/health" ) -type memStore struct{ m map[string]json.RawMessage } +type memStore struct { + mu sync.Mutex + m map[string]json.RawMessage +} -func (s *memStore) Get(k string) (json.RawMessage, error) { return s.m[k], nil } +func (s *memStore) Get(k string) (json.RawMessage, error) { + s.mu.Lock() + defer s.mu.Unlock() + return s.m[k], nil +} func (s *memStore) Put(k string, v any) error { raw, _ := json.Marshal(v) + s.mu.Lock() + defer s.mu.Unlock() s.m[k] = raw return nil } func (s *memStore) Keys() ([]string, error) { + s.mu.Lock() + defer s.mu.Unlock() var out []string for k := range s.m { out = append(out, k) @@ -66,11 +78,16 @@ func TestARoundMeasuresTheDeclaredItemsAndWritesWhatChanged(t *testing.T) { if _, _, err := k.MeasureNow(context.Background(), "pg", "data"); err != nil || walks != 2 { t.Fatalf("measure now: %v, %d walks", err, walks) } - // An item no longer declared is kept, retired with when and why; declared again, it is not retired any more. + // A round retires nothing, even when no item is declared (the watched set may be empty or partial); an item its + // module's key no longer names is kept, retired with when and why; declared again, it is not retired any more. delete(placed, "pg") if _, _, err := k.Round(context.Background(), false); err != nil { t.Fatal(err) } + if kept, _ := health.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"]) 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) diff --git a/modules/disk-load/cmd/disk-load/readings_tools.go b/modules/disk-load/cmd/disk-load/readings_tools.go index 322fe275..044e3d46 100644 --- a/modules/disk-load/cmd/disk-load/readings_tools.go +++ b/modules/disk-load/cmd/disk-load/readings_tools.go @@ -2,9 +2,11 @@ package main import ( "context" + "encoding/json" "fmt" "sort" "strings" + "time" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/health" @@ -68,6 +70,37 @@ func ReadingTools(k *Keeper) []stdio.Tool { "cap": "none declared: the bus's own limit on a state applies"}}, nil }, }, + { + Name: "disk_round", + Description: "How far this machine's last round of measuring its data items got (novox/hq ADR 0577 clause 3), from " + + "disk-load's state `rounds`, this machine's key ``: when it started and ended (no end while it runs), how " + + "many items it had, measured, wrote and kept unchanged, each item that failed and why, and how old the round is. " + + "A round that never ends, or fails every item, shows here. (r)", + Input: map[string]any{}, + Run: func(args map[string]any) (any, error) { + node := k.Node + if k.Rounds == nil { + return nil, fmt.Errorf("this disk-load keeps no rounds") + } + raw, err := k.Rounds.Get(node) + if err != nil { + return nil, err + } + if len(raw) == 0 { + return nil, fmt.Errorf("no round of %s is kept yet: its first round starts a few minutes after disk-load starts", node) + } + var r Round + if err := json.Unmarshal(raw, &r); err != nil { + return nil, fmt.Errorf("the round of %s is not readable: %w", node, err) + } + state := "ended" + if r.Ended == nil { + state = "running" + } + return map[string]any{"round": r, "state": state, + "age": k.Measurer.Now().UTC().Sub(r.Started).Round(time.Second).String()}, 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 " + diff --git a/modules/disk-load/cmd/disk-load/whole_integration_test.go b/modules/disk-load/cmd/disk-load/whole_integration_test.go new file mode 100644 index 00000000..2cff6f38 --- /dev/null +++ b/modules/disk-load/cmd/disk-load/whole_integration_test.go @@ -0,0 +1,267 @@ +//go:build integration + +package main + +// disk-load whole against a real bus (novox/hq ADR 0577 clause 5, the group's integration test): the built binary, played +// over its real stdin and stdout as the node's tool runtime plays it (mesh/state.* answered from a real nats-server's +// key-value buckets, a watch handed over from the bus's own watch, mesh/publish onto its subjects). A key of the +// node-engine's state data written after disk-load started has its module's item read within a minute, and disk-load +// composes nothing to learn it. + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "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" +) + +type wire struct { + ID json.RawMessage `json:"id,omitempty"` + Method string `json:"method,omitempty"` + Params json.RawMessage `json:"params,omitempty"` + Result json.RawMessage `json:"result,omitempty"` +} + +// player is the node's tool runtime, as far as disk-load needs it, over a real bus. +type player struct { + ctx context.Context + nc *nats.Conn + js jetstream.JetStream + mu sync.Mutex + in io.Writer + next int + asked []string // every state call, ` ` +} + +func (p *player) send(m map[string]any) { + raw, _ := json.Marshal(m) + p.mu.Lock() + defer p.mu.Unlock() + _, _ = p.in.Write(append(raw, '\n')) +} + +func (p *player) answer(id json.RawMessage, result any, err error) { + if err != nil { + p.send(map[string]any{"jsonrpc": "2.0", "id": id, "error": map[string]any{"code": -32000, "message": err.Error()}}) + return + } + p.send(map[string]any{"jsonrpc": "2.0", "id": id, "result": result}) +} + +// bucket is a state's bucket: another module's `.`, or disk-load's own by its local name. +func bucket(state string) string { + if owner, name, other := strings.Cut(state, "."); other { + return owner + "_" + name + } + return "disk-load_" + state +} + +func (p *player) serve(out io.Reader) { + scanner := bufio.NewScanner(out) + scanner.Buffer(make([]byte, 1<<20), 1<<24) + for scanner.Scan() { + var m wire + if json.Unmarshal(scanner.Bytes(), &m) != nil || m.Method == "" { + continue // an answer to a change handed over + } + go p.handle(m) + } +} + +func (p *player) handle(m wire) { + var a struct { + State string `json:"state"` + Key string `json:"key"` + Value json.RawMessage `json:"value"` + Body json.RawMessage `json:"body"` + } + _ = json.Unmarshal(m.Params, &a) + p.mu.Lock() + p.asked = append(p.asked, m.Method+" "+a.State+" "+a.Key) + p.mu.Unlock() + if m.Method == "mesh/publish" { + var e struct { + Key string `json:"key"` + Body json.RawMessage `json:"body"` + } + _ = json.Unmarshal(m.Params, &e) + p.answer(m.ID, map[string]any{}, p.nc.Publish("mesh.mod.disk-load.event."+e.Key, e.Body)) + return + } + if !strings.HasPrefix(m.Method, "mesh/state.") { + p.answer(m.ID, nil, errors.New("nats: no responders available for request")) + return + } + kv, err := p.js.KeyValue(p.ctx, bucket(a.State)) + if err != nil { + p.answer(m.ID, nil, err) + return + } + switch m.Method { + case "mesh/state.get": + e, err := kv.Get(p.ctx, a.Key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + p.answer(m.ID, nil, nil) + return + } + if err != nil { + p.answer(m.ID, nil, err) + return + } + p.answer(m.ID, map[string]any{"key": e.Key(), "value": json.RawMessage(e.Value()), "revision": e.Revision()}, nil) + case "mesh/state.put": + rev, err := kv.Put(p.ctx, a.Key, a.Value) + p.answer(m.ID, map[string]any{"revision": rev}, err) + case "mesh/state.keys": + keys, err := kv.Keys(p.ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + keys, err = []string{}, nil + } + p.answer(m.ID, keys, err) + case "mesh/state.watch": + p.watch(m.ID, a.State, kv, a.Key) + default: + p.answer(m.ID, nil, errors.New("not played here")) + } +} + +// watch hands over the bus's watch of the pattern: every current value, then the answer to the watch, then each change. +func (p *player) watch(id json.RawMessage, state string, kv jetstream.KeyValue, pattern string) { + var watchID string + _ = json.Unmarshal(id, &watchID) + if pattern == "" { + pattern = ">" + } + w, err := kv.Watch(p.ctx, strings.ReplaceAll(pattern, "**", ">")) + if err != nil { + p.answer(id, nil, err) + return + } + current := true + for e := range w.Updates() { + if e == nil { + current = false + p.answer(id, map[string]any{}, nil) + continue + } + op := "put" + if e.Operation() != jetstream.KeyValuePut { + op = "delete" + } + p.mu.Lock() + p.next++ + n := p.next + p.mu.Unlock() + p.send(map[string]any{"jsonrpc": "2.0", "id": fmt.Sprintf("rt-%d", n), "method": "mesh/state", "params": map[string]any{ + "watch": watchID, "change": map[string]any{"state": state, "key": e.Key(), "op": op, + "value": json.RawMessage(e.Value()), "revision": e.Revision(), "current": current}}}) + } +} + +// 0577/AC5 (the group's integration test): a key of the node-engine's state data written after disk-load started is +// read within a minute, its item's reading is on the bus, and disk-load read no composition input to learn it. +func TestDiskLoadWholeMeasuresAnItemWithinAMinuteOfItsKey(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + bin := filepath.Join(t.TempDir(), "disk-load") + build := exec.Command("go", "build", "-o", bin, ".") + build.Stderr = os.Stderr + if err := build.Run(); err != nil { + t.Fatalf("disk-load does not build: %v", err) + } + nc, err := nats.Connect(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + for _, b := range []string{delivery.EngineDataBucket, health.ReadingsBucket, "disk-load_" + StateRounds} { + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b}); err != nil { + t.Fatal(err) + } + } + cmd := exec.CommandContext(ctx, bin) + cmd.Env = append(os.Environ(), "MESH_NODE=ace", "MESH_MODULE=disk-load") + stdin, _ := cmd.StdinPipe() + stdout, _ := cmd.StdoutPipe() + cmd.Stderr = os.Stderr + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = stdin.Close(); cancel(); _ = cmd.Wait() }) + p := &player{ctx: ctx, nc: nc, js: js, in: stdin} + go p.serve(stdout) + + // Once disk-load watches its machine's keys, the engine writes one. + watching := func() bool { + p.mu.Lock() + defer p.mu.Unlock() + for _, a := range p.asked { + if a == "mesh/state.watch "+delivery.EngineDataRead+" ace.*" { + return true + } + } + return false + } + for end := time.Now().Add(20 * time.Second); !watching(); time.Sleep(50 * time.Millisecond) { + if time.Now().After(end) { + t.Fatalf("disk-load did not watch its machine's keys of %s: asked %v", delivery.EngineDataBucket, p.asked) + } + } + data, _ := js.KeyValue(ctx, delivery.EngineDataBucket) + dir := t.TempDir() + item := catalogue.PlacedDataItem{Module: "pg", ID: "data", Class: "valuable", Path: dir, Measure: catalogue.MeasureShallow} + raw, err := delivery.EncodeEngineData(delivery.EngineData{Node: "ace", Module: "pg", Digest: "sha256:pg", Items: []catalogue.PlacedDataItem{item}}) + if err != nil { + t.Fatal(err) + } + key, _ := delivery.EngineDataKey("ace", "pg") + written := time.Now() + if _, err := data.Put(ctx, key, raw); err != nil { + t.Fatal(err) + } + readings, _ := js.KeyValue(ctx, health.ReadingsBucket) + for { + e, err := readings.Get(ctx, health.ReadingKey("ace", "pg", "data")) + if err == nil { + r, err := health.DecodeReading(e.Value()) + if err != nil || r.Path != dir || r.MeasuredAt == nil { + t.Fatalf("the reading is %s (%v)", e.Value(), err) + } + break + } + if time.Since(written) > time.Minute { + t.Fatalf("no reading within a minute of the key's write: %v", err) + } + time.Sleep(200 * time.Millisecond) + } + // It composed nothing: every state it asked is the engine's data or its own. + p.mu.Lock() + defer p.mu.Unlock() + for _, a := range p.asked { + f := strings.Fields(a) + if len(f) > 1 && strings.HasPrefix(f[0], "mesh/state.") && f[1] != delivery.EngineDataRead && f[1] != StateReadings && f[1] != StateRounds { + t.Errorf("disk-load read %s, which is no state of its own nor the engine's data", a) + } + } +} diff --git a/modules/disk-load/go.mod b/modules/disk-load/go.mod index 939c2017..f6636513 100644 --- a/modules/disk-load/go.mod +++ b/modules/disk-load/go.mod @@ -2,13 +2,21 @@ module diskload go 1.26.0 -require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031213356-5ae69df8ccde +require ( + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031230538-c28c645ca406 + github.com/nats-io/nats.go v1.54.0 +) require ( + github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect + github.com/google/go-tpm v0.9.8 // indirect github.com/klauspost/compress v1.20.0 // indirect - github.com/nats-io/nats.go v1.54.0 // indirect + github.com/minio/highwayhash v1.0.4 // indirect + github.com/nats-io/jwt/v2 v2.8.1 // indirect + github.com/nats-io/nats-server/v2 v2.11.17 // indirect github.com/nats-io/nkeys v0.4.16 // indirect github.com/nats-io/nuid v1.0.1 // indirect golang.org/x/crypto v0.57.0 // indirect golang.org/x/sys v0.48.0 // indirect + golang.org/x/time v0.15.0 // indirect ) diff --git a/modules/disk-load/go.sum b/modules/disk-load/go.sum index 0e43aead..d7e576d2 100644 --- a/modules/disk-load/go.sum +++ b/modules/disk-load/go.sum @@ -1,7 +1,17 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031213356-5ae69df8ccde h1:RyISctWEfdaykq69I3CYKOZ4N1wUx6PpiPbql4kpXkg= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031213356-5ae69df8ccde/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031230538-c28c645ca406 h1:Z4RcH5Ft4OAAG3F538NXIHP9F3pkyk3j0LtP9mqu1qQ= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031230538-c28c645ca406/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= +github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= +github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY= github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA= github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= +github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4= +github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ= +github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU= +github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg= +github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I= +github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24= github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA= github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k= github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= @@ -10,5 +20,8 @@ github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= diff --git a/modules/disk-load/module.json b/modules/disk-load/module.json index 4b4313ca..7b162bf9 100644 --- a/modules/disk-load/module.json +++ b/modules/disk-load/module.json @@ -9,6 +9,7 @@ "disk_devices", "cpu_top", "disk_readings", + "disk_round", "disk_measure" ], "build": { @@ -27,14 +28,13 @@ ] }, "state": [ - "readings" + "readings", + { + "name": "rounds", + "per-machine": true + } ], "reads": [ - "mesh-config.resolved", - "mesh-delivery-assigner.placement", - "mesh-delivery-builder.builds", - "mesh-delivery-indexer.manifests", - "mesh-delivery-indexer.modules", - "node-engine.have" + "mesh-host.data" ] }