From c432b9b5f60ffdc40f075e4bd11dad2c6c23ef98 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 11 Oct 2026 19:30:59 +0200 Subject: [PATCH] Name the fill condition for buckets and logs, and judge each one on its own MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of #231: "store" is the glossary's word for the postgres server, so the condition is bucket-or-log-filling. And one stream that cannot be read no longer fails the whole check: it is said as unread with its reason, an open fill condition for it is kept, and every other bucket and log is still judged (ADR 0297 §6). --- cmd/mesh-controller/bucket_log_fill.go | 212 ++++++++++++++++++++ cmd/mesh-controller/bucket_log_fill_test.go | 130 ++++++++++++ cmd/mesh-controller/doctor.go | 7 +- cmd/mesh-controller/plain_words.go | 12 +- cmd/mesh-controller/store_fill.go | 165 --------------- cmd/mesh-controller/store_fill_test.go | 76 ------- 6 files changed, 355 insertions(+), 247 deletions(-) create mode 100644 cmd/mesh-controller/bucket_log_fill.go create mode 100644 cmd/mesh-controller/bucket_log_fill_test.go delete mode 100644 cmd/mesh-controller/store_fill.go delete mode 100644 cmd/mesh-controller/store_fill_test.go diff --git a/cmd/mesh-controller/bucket_log_fill.go b/cmd/mesh-controller/bucket_log_fill.go new file mode 100644 index 00000000..4fd8fa88 --- /dev/null +++ b/cmd/mesh-controller/bucket_log_fill.go @@ -0,0 +1,212 @@ +package main + +import ( + "context" + "errors" + "fmt" + "sort" + + "github.com/nats-io/nats.go" + + "github.com/novox/mesh-controller/internal/conditions" +) + +// **A filling bucket or log is said before it is full** (novox/hq ADR 0297 §6, issue 501). +// +// A module's bucket and its log each have a cap on the bus, and when either is full the bus refuses +// the next write: the module stops doing what it writes for, and nothing said so before. On +// 2026-10-11 the issue tracker's bucket held a sixth of its cap after two days, growing toward a stop +// nobody would have heard coming. So the self-check reads every module's bucket and log, and one at +// fillRaiseAt of its cap or more raises a condition naming the module, the bucket or log, and how full +// it is. +// +// **Cleared by observation below fillClearBelow, not below fillRaiseAt**: a bucket or log hovering at +// its threshold would otherwise be raised and cleared on every run. Between the two, an open condition +// is kept and none is raised. +// +// **One that cannot be read is said, and the rest are still judged.** An unanswered question about one +// stream is no reason to know nothing of the others: it is a finding of its own, naming the bucket or +// log and why, and a fill condition already open for it is kept, because not knowing is not a pass. + +// The fill thresholds, in percent of a bucket's or log's cap. +const ( + fillRaiseAt = 75 + fillClearBelow = 70 +) + +// The conditions the fill probe raises: one filling, and one that could not be read. +const ( + kindBucketOrLogFilling = "bucket-or-log-filling" + kindBucketOrLogUnread = "bucket-or-log-unread" +) + +// probeFillID is the probe's id in the registry. +const probeFillID = "D-fill" + +// bucketOrLog is one module's bucket or log as the fill probe reads it. +type bucketOrLog struct { + Module string + // Kind is `bucket` or `log`; Name its local name; Stream the stream it is on the bus. + Kind, Name, Stream string +} + +// filled is how full one bucket or log is, read from the bus. +type filled struct { + Bytes, Max uint64 +} + +// percent is how full, in whole percent, rounded down. +func (f filled) percent() uint64 { + if f.Max == 0 { + return 0 + } + return f.Bytes * 100 / f.Max +} + +// fillKey is where a bucket's or log's fill condition is kept. +func fillKey(s bucketOrLog) string { return conditions.Key(conditions.ScopeBus, s.Stream, "filling") } + +// fillObservation is what one bucket's or log's fill says: a finding at fillRaiseAt or above, and, +// while its condition is open, at fillClearBelow or above too; nothing otherwise, which is what clears +// it. The operator's words are the kind's (plain_words.go); the summary and the evidence name the +// module, the bucket or log, and the fill. +func fillObservation(s bucketOrLog, f filled, open bool) (conditions.Observation, bool) { + if f.Max == 0 { + return conditions.Observation{}, false // one with no cap cannot fill + } + // Compared in bytes, not rounded percent: 74.9% is not 75%. + atRaise := f.Bytes*100 >= f.Max*fillRaiseAt + aboveClear := f.Bytes*100 >= f.Max*fillClearBelow + if !atRaise && !(open && aboveClear) { + return conditions.Observation{}, false + } + pct := f.percent() + return conditions.Observation{ + Scope: conditions.ScopeBus, ID: s.Stream, Token: "filling", Kind: kindBucketOrLogFilling, + Severity: conditions.Warning, + Summary: fmt.Sprintf("%s's %s %s holds %s of %s, %d%%: at its cap the bus refuses the module's next write", + s.Module, s.Kind, s.Name, mibWords(f.Bytes), mibWords(f.Max), pct), + Said: fmt.Sprintf("module %s, %s %s (stream %s): %d of %d bytes, %d%%", s.Module, s.Kind, s.Name, + s.Stream, f.Bytes, f.Max, pct), + }, true +} + +// unreadObservation says one bucket or log could not be read, and why. Asked over the network, so one +// look can be wrong: raised on the second look in a row (confirm.go). +func unreadObservation(s bucketOrLog, why error) conditions.Observation { + return conditions.Observation{ + Scope: conditions.ScopeBus, ID: s.Stream, Token: "unread", Kind: kindBucketOrLogUnread, + Severity: conditions.Warning, Confirm: true, + Summary: fmt.Sprintf("%s's %s %s could not be read, so how full it is is not known", s.Module, s.Kind, s.Name), + Said: fmt.Sprintf("module %s, %s %s (stream %s): %v", s.Module, s.Kind, s.Name, s.Stream, why), + } +} + +// keptFilling is an open fill condition said again for a bucket or log that could not be read this +// time: not knowing how full it is now is no reason to say it has room. +func keptFilling(s bucketOrLog, open conditions.Condition, why error) conditions.Observation { + severity := open.Severity + if severity == "" { + severity = conditions.Warning + } + summary := open.Summary + if summary == "" { + summary = fmt.Sprintf("%s's %s %s was filling when it was last read", s.Module, s.Kind, s.Name) + } + return conditions.Observation{ + Scope: conditions.ScopeBus, ID: s.Stream, Token: "filling", Kind: kindBucketOrLogFilling, + Severity: severity, Summary: summary, + Said: fmt.Sprintf("module %s, %s %s (stream %s): not read this time (%v); kept open as it was", + s.Module, s.Kind, s.Name, s.Stream, why), + } +} + +// mibWords is a size as a person reads it, in MiB with one decimal. +func mibWords(b uint64) string { + return fmt.Sprintf("%.1f MiB", float64(b)/(1024*1024)) +} + +// readFill is how full one bucket or log is on the bus; found false when it is not on the bus. +type readFill func(ctx context.Context, s bucketOrLog) (f filled, found bool, err error) + +// openFill is the fill condition open under a key, if one is. +type openFill func(ctx context.Context, key string) (conditions.Condition, bool) + +// judgeFills judges every bucket and log on its own: one that cannot be read is said as unread, with +// its open fill condition kept, and every other is judged by its fill. +func judgeFills(ctx context.Context, all []bucketOrLog, read readFill, open openFill) []conditions.Observation { + var out []conditions.Observation + for _, s := range all { + f, found, err := read(ctx, s) + if err != nil { + out = append(out, unreadObservation(s, err)) + if c, isOpen := open(ctx, fillKey(s)); isOpen { + out = append(out, keptFilling(s, c, err)) + } + continue + } + if !found { + continue // not on the bus: created on the controller's next raise, and nothing there can fill + } + _, isOpen := open(ctx, fillKey(s)) + if o, said := fillObservation(s, f, isOpen); said { + out = append(out, o) + } + } + sort.SliceStable(out, func(i, j int) bool { return out[i].Key() < out[j].Key() }) + return out +} + +// declaredBucketsAndLogs is every module's bucket and log the catalogue declares, by its stream. +func declaredBucketsAndLogs(ctx context.Context, d *doctor) ([]bucketOrLog, error) { + buckets, err := d.open.inventory.DeclaredBuckets(ctx) + if err != nil { + return nil, err + } + logs, err := d.open.inventory.DeclaredLogs(ctx) + if err != nil { + return nil, err + } + var out []bucketOrLog + for _, b := range buckets { + out = append(out, bucketOrLog{Module: b.Module, Kind: "bucket", Name: b.Name, Stream: "KV_" + b.Bucket()}) + } + for _, l := range logs { + out = append(out, bucketOrLog{Module: l.Module, Kind: "log", Name: l.Name, Stream: l.Stream()}) + } + return out, nil +} + +// probeFill reads every module's bucket and log on the bus and says each one filling toward its cap, +// and each one that could not be read. +func probeFill(ctx context.Context, d *doctor) ([]conditions.Observation, error) { + all, err := declaredBucketsAndLogs(ctx, d) + if err != nil { + return nil, err + } + js := d.js.Context() + read := func(ctx context.Context, s bucketOrLog) (filled, bool, error) { + if err := ctx.Err(); err != nil { + return filled{}, false, err + } + info, err := js.StreamInfo(s.Stream, nats.Context(ctx)) + switch { + case errors.Is(err, nats.ErrStreamNotFound): + return filled{}, false, nil + case err != nil: + return filled{}, false, err + case info.Config.MaxBytes <= 0: + return filled{}, true, nil + } + return filled{Bytes: info.State.Bytes, Max: uint64(info.Config.MaxBytes)}, true, nil + } + open := func(ctx context.Context, key string) (conditions.Condition, bool) { + if d.keeper == nil { + return conditions.Condition{}, false + } + // One that cannot be read is taken as open, as confirm.go takes it: unknown is no reason to clear. + c, isOpen, err := d.keeper.Get(ctx, key) + return c, isOpen || err != nil + } + return judgeFills(ctx, all, read, open), nil +} diff --git a/cmd/mesh-controller/bucket_log_fill_test.go b/cmd/mesh-controller/bucket_log_fill_test.go new file mode 100644 index 00000000..e9acd8f2 --- /dev/null +++ b/cmd/mesh-controller/bucket_log_fill_test.go @@ -0,0 +1,130 @@ +package main + +import ( + "context" + "errors" + "strings" + "testing" + + "github.com/novox/mesh-controller/internal/conditions" +) + +// **A filling log or bucket is said at three quarters of its cap and cleared below seven tenths** +// (novox/hq ADR 0297 §6): raised at 75% naming the module, the bucket or log and its fill; not at 74.9%; +// kept between 70% and 75% while it is open, and not raised there when it is not; and cleared — said no +// more — once it reads below 70%, open or not. +func TestAFillingBucketOrLogIsRaisedAtThreeQuartersAndClearedBelowSevenTenths(t *testing.T) { + const mib = 1024 * 1024 + log := bucketOrLog{Module: "mesh-issues", Kind: "log", Name: "changes", Stream: "LOG_mesh-issues_changes"} + bucket := bucketOrLog{Module: "mesh-issues", Kind: "bucket", Name: "issues", Stream: "KV_mesh-issues_issues"} + for _, c := range []struct { + name string + of bucketOrLog + bytes uint64 + max uint64 + open bool + said bool + }{ + {"a log at exactly 75%", log, 768 * mib, 1024 * mib, false, true}, + {"a bucket at exactly 75%", bucket, 48 * mib, 64 * mib, false, true}, + {"a bucket full", bucket, 64 * mib, 64 * mib, false, true}, + {"a log just under 75%", log, 768*mib - 1, 1024 * mib, false, false}, + {"a bucket at 72%, not open", bucket, 64 * mib * 72 / 100, 64 * mib, false, false}, + {"a bucket at 72%, open: kept", bucket, 64 * mib * 72 / 100, 64 * mib, true, true}, + {"a log at exactly 70%, open: kept", log, 700 * mib, 1000 * mib, true, true}, + {"a log just under 70%, open: cleared", log, 700*mib - 1, 1000 * mib, true, false}, + {"a bucket at 10%, open: cleared", bucket, 64 * mib / 10, 64 * mib, true, false}, + {"a bucket with no cap", bucket, 100, 0, true, false}, + } { + o, said := fillObservation(c.of, filled{Bytes: c.bytes, Max: c.max}, c.open) + if said != c.said { + t.Errorf("%s: said %v, want %v", c.name, said, c.said) + continue + } + if !said { + continue + } + if o.Key() != fillKey(c.of) || o.Kind != kindBucketOrLogFilling || o.Severity != conditions.Warning { + t.Errorf("%s: raised as %s (%s, %s)", c.name, o.Key(), o.Kind, o.Severity) + } + for _, part := range []string{c.of.Module, c.of.Kind + " " + c.of.Name, "%", " of "} { + if !strings.Contains(o.Summary, part) { + t.Errorf("%s: does not name %q: %q", c.name, part, o.Summary) + } + } + if !strings.Contains(o.Said, "bytes") { + t.Errorf("%s: the evidence does not say the fill in bytes: %q", c.name, o.Said) + } + } + o, _ := fillObservation(log, filled{Bytes: 800 * mib, Max: 1024 * mib}, false) + if !strings.Contains(o.Summary, "800.0 MiB of 1024.0 MiB, 78%") { + t.Errorf("the fill is said as %q", o.Summary) + } +} + +// **One bucket or log that cannot be read does not stop the others being judged** (review of #231): it is +// said as unread with its reason, a fill condition open for it is kept, and every other bucket and log +// is judged by its own fill. +func TestAnUnreadableBucketOrLogIsSaidAndTheRestAreStillJudged(t *testing.T) { + const mib = 1024 * 1024 + broken := bucketOrLog{Module: "a", Kind: "bucket", Name: "broken", Stream: "KV_a_broken"} + brokenOpen := bucketOrLog{Module: "a", Kind: "log", Name: "was-filling", Stream: "LOG_a_was-filling"} + full := bucketOrLog{Module: "b", Kind: "log", Name: "changes", Stream: "LOG_b_changes"} + empty := bucketOrLog{Module: "c", Kind: "bucket", Name: "quiet", Stream: "KV_c_quiet"} + absent := bucketOrLog{Module: "d", Kind: "bucket", Name: "not-yet", Stream: "KV_d_not-yet"} + refusal := errors.New("the bus did not answer") + read := func(_ context.Context, s bucketOrLog) (filled, bool, error) { + switch s { + case broken, brokenOpen: + return filled{}, false, refusal + case full: + return filled{Bytes: 900 * mib, Max: 1000 * mib}, true, nil + case empty: + return filled{Bytes: 1, Max: 64 * mib}, true, nil + } + return filled{}, false, nil + } + open := func(_ context.Context, key string) (conditions.Condition, bool) { + if key == fillKey(brokenOpen) { + return conditions.Condition{Key: key, Severity: conditions.Warning, + Summary: "a's log was-filling holds 800.0 MiB of 1024.0 MiB, 78%"}, true + } + return conditions.Condition{}, false + } + got := map[string]conditions.Observation{} + for _, o := range judgeFills(context.Background(), []bucketOrLog{broken, brokenOpen, full, empty, absent}, read, open) { + got[o.Key()] = o + } + for _, s := range []bucketOrLog{broken, brokenOpen} { + o, said := got[conditions.Key(conditions.ScopeBus, s.Stream, "unread")] + if !said || o.Kind != kindBucketOrLogUnread || !o.Confirm || !strings.Contains(o.Said, refusal.Error()) { + t.Errorf("%s could not be read and was said as %+v", s.Stream, o) + } + } + if o, kept := got[fillKey(brokenOpen)]; !kept || !strings.Contains(o.Said, "kept open") { + t.Errorf("an open fill condition of a log not read this time was not kept: %+v", o) + } + if _, said := got[fillKey(broken)]; said { + t.Error("a bucket not read and not filling was said filling") + } + if o, said := got[fillKey(full)]; !said || o.Kind != kindBucketOrLogFilling { + t.Errorf("a log at 90%% beside an unreadable one was not judged: %+v", got) + } + if len(got) != 4 { + t.Errorf("said %d findings, want 4 (two unread, one kept, one filling): %v", len(got), got) + } +} + +// Both kinds have plain words of their own, which hold to the plain rule (novox/hq ADR 0253). +func TestAFillingOrUnreadBucketOrLogIsSaidInPlainWords(t *testing.T) { + for _, kind := range []string{kindBucketOrLogFilling, kindBucketOrLogUnread} { + wording, has := plainWordings[kind] + if !has { + t.Errorf("%s has no plain words", kind) + continue + } + if why, ok := conditions.PlainWords(wording(conditions.Observation{Kind: kind})); !ok { + t.Errorf("%s: %s", kind, why) + } + } +} diff --git a/cmd/mesh-controller/doctor.go b/cmd/mesh-controller/doctor.go index 3604a647..dfb74f06 100644 --- a/cmd/mesh-controller/doctor.go +++ b/cmd/mesh-controller/doctor.go @@ -137,10 +137,11 @@ var probeRegistry = []probe{ "proving its sender runs: not by its own account, and not through a tool that runs its command as an account " + "that can", From: "ADR 0259 §8", Kind: kindRootNotFree, Phase: 2, run: probeAgentRoot}, // A module's bucket or log filling toward its cap (novox/hq ADR 0297 §6): said at three quarters, cleared - // below seven tenths, so a store at its threshold is not raised and cleared on every run. - {ID: probeStoreFillID, Asserts: "every module's bucket and log holds less than three quarters of its cap; one " + + // below seven tenths, so one at its threshold is not raised and cleared on every run. One that cannot be + // read is said on its own, and every other is still judged. + {ID: probeFillID, Asserts: "every module's bucket and log holds less than three quarters of its cap; one " + "said filling is cleared once it holds less than seven tenths", From: "ADR 0297, issue 501", - Kind: kindStoreFilling, Phase: 1, run: probeStoreFill}, + Kind: kindBucketOrLogFilling, Raises: []string{kindBucketOrLogUnread}, Phase: 1, run: probeFill}, {ID: "DW", Asserts: "the watchdogs of the signals table ran within three of their intervals", From: "ADR 0227 rule 6: the watchers are watched", Kind: "watchdogs-silent", Phase: 1, run: probeWatchdogs}, // The core's health definitions (novox/hq to-be 45 §8, ADR 0236): what a core component's new build is diff --git a/cmd/mesh-controller/plain_words.go b/cmd/mesh-controller/plain_words.go index a41ec619..3a493a15 100644 --- a/cmd/mesh-controller/plain_words.go +++ b/cmd/mesh-controller/plain_words.go @@ -606,12 +606,18 @@ var plainWordings = map[string]func(conditions.Observation) words{ Explanation: "The bus refused messages from a part of the mesh, so what they carried did not happen.", Resolved: "Resolved: the bus takes the messages again"} }), - kindStoreFilling: worded(func(o conditions.Observation) words { - return words{Headline: "A module's store on the bus is filling up", + kindBucketOrLogFilling: worded(func(o conditions.Observation) words { + return words{Headline: "A module's bucket or log on the bus is filling up", Needs: "decide whether to raise its cap or have the module keep less.", Explanation: "A bucket or log a module keeps on the bus holds three quarters of its cap or more. When it " + "is full, the bus refuses what the module writes there.", - Resolved: "The store has room again"} + Resolved: "It has room again"} + }), + kindBucketOrLogUnread: worded(func(o conditions.Observation) words { + return words{Headline: "A module's bucket or log could not be read", + Explanation: "The controller could not read how full a bucket or log a module keeps on the bus is, so it " + + "cannot say whether it is filling up. It asks again at its next check.", + Resolved: "It can be read again"} }), "stream-wrong": worded(func(o conditions.Observation) words { return words{Headline: "Part of the bus's storage is wrong", diff --git a/cmd/mesh-controller/store_fill.go b/cmd/mesh-controller/store_fill.go deleted file mode 100644 index 39dcaa63..00000000 --- a/cmd/mesh-controller/store_fill.go +++ /dev/null @@ -1,165 +0,0 @@ -package main - -import ( - "context" - "errors" - "fmt" - - "github.com/nats-io/nats.go" - - "github.com/novox/mesh-controller/internal/conditions" -) - -// **A filling bucket or log is said before it is full** (novox/hq ADR 0297 §6, issue 501). -// -// A module's bucket and its log each have a cap on the bus, and when either is full the bus refuses -// the next write: the module stops doing what it writes for, and nothing said so before. On -// 2026-10-11 the issue tracker's bucket held a sixth of its cap after two days, growing toward a stop -// nobody would have heard coming. So the self-check reads every module's bucket and log, and one at -// fillRaiseAt of its cap or more raises a condition naming the module, the store and how full it is. -// -// **Cleared by observation below fillClearBelow, not below fillRaiseAt**: a store hovering at its -// threshold would otherwise be raised and cleared on every run. Between the two, an open condition is -// kept and none is raised. - -// The fill thresholds, in percent of a store's cap. -const ( - fillRaiseAt = 75 - fillClearBelow = 70 -) - -// kindStoreFilling is the condition a filling bucket or log raises. -const kindStoreFilling = "store-filling" - -// probeStoreFillID is the probe's id in the registry. -const probeStoreFillID = "D-fill" - -// moduleStore is one module's bucket or log as the fill probe reads it. -type moduleStore struct { - Module string - // Kind is `bucket` or `log`; Name its local name; Stream the stream it is on the bus. - Kind, Name, Stream string -} - -// filled is how full one store is, read from the bus. -type filled struct { - Bytes, Max uint64 -} - -// percent is how full, in whole percent, rounded down. -func (f filled) percent() uint64 { - if f.Max == 0 { - return 0 - } - return f.Bytes * 100 / f.Max -} - -// fillKey is where a store's fill condition is kept. -func fillKey(s moduleStore) string { return conditions.Key(conditions.ScopeBus, s.Stream, "filling") } - -// fillObservation is what one store's fill says: a finding at fillRaiseAt or above, and, while its -// condition is open, at fillClearBelow or above too; nothing otherwise, which is what clears it. -func fillObservation(s moduleStore, f filled, open bool) (conditions.Observation, bool) { - if f.Max == 0 { - return conditions.Observation{}, false // a store with no cap cannot fill - } - // Compared in bytes, not rounded percent: 74.9% is not 75%. - atRaise := f.Bytes*100 >= f.Max*fillRaiseAt - aboveClear := f.Bytes*100 >= f.Max*fillClearBelow - if !atRaise && !(open && aboveClear) { - return conditions.Observation{}, false - } - pct := f.percent() - fill := fmt.Sprintf("%s of %s, %d%%", mibWords(f.Bytes), mibWords(f.Max), pct) - return conditions.Observation{ - Scope: conditions.ScopeBus, ID: s.Stream, Token: "filling", Kind: kindStoreFilling, - Severity: conditions.Warning, - Summary: fmt.Sprintf("%s's %s %s holds %s: at its cap the bus refuses the module's next write", - s.Module, s.Kind, s.Name, fill), - Headline: fillHeadline(s, pct), - Explanation: fmt.Sprintf("The %s %s of %s holds %s. When it is full, the bus refuses what %s writes "+ - "there, and %s stops doing what it writes for.", s.Kind, s.Name, s.Module, fill, s.Module, s.Module), - Needs: fmt.Sprintf("decide whether to raise the cap of this %s or have %s keep less.", s.Kind, s.Module), - Resolved: fillResolved(s), - Said: fmt.Sprintf("module %s, %s %s (stream %s): %d of %d bytes, %d%%", s.Module, s.Kind, s.Name, - s.Stream, f.Bytes, f.Max, pct), - }, true -} - -// fillHeadline names the store and how full it is, in at most a headline's length. -func fillHeadline(s moduleStore, pct uint64) string { - for _, h := range []string{ - fmt.Sprintf("%s's %s %s is %d%% full", s.Module, s.Kind, s.Name, pct), - fmt.Sprintf("A %s of %s is %d%% full", s.Kind, s.Module, pct), - } { - if len(h) <= conditions.HeadlineMax { - return h - } - } - return fmt.Sprintf("A module's %s is %d%% full", s.Kind, pct) -} - -// fillResolved is the line said when the store has room again, in at most a resolved line's length. -func fillResolved(s moduleStore) string { - if h := fmt.Sprintf("%s's %s %s has room again", s.Module, s.Kind, s.Name); len(h) <= conditions.HeadlineMax { - return h - } - return fmt.Sprintf("A %s of %s has room again", s.Kind, s.Module) -} - -// mibWords is a size as a person reads it, in MiB with one decimal. -func mibWords(b uint64) string { - return fmt.Sprintf("%.1f MiB", float64(b)/(1024*1024)) -} - -// declaredStores is every module's bucket and log the catalogue declares, by the stream each is. -func declaredStores(ctx context.Context, d *doctor) ([]moduleStore, error) { - buckets, err := d.open.inventory.DeclaredBuckets(ctx) - if err != nil { - return nil, err - } - logs, err := d.open.inventory.DeclaredLogs(ctx) - if err != nil { - return nil, err - } - var out []moduleStore - for _, b := range buckets { - out = append(out, moduleStore{Module: b.Module, Kind: "bucket", Name: b.Name, Stream: "KV_" + b.Bucket()}) - } - for _, l := range logs { - out = append(out, moduleStore{Module: l.Module, Kind: "log", Name: l.Name, Stream: l.Stream()}) - } - return out, nil -} - -// probeStoreFill reads every module's bucket and log on the bus and says each one filling toward its -// cap. One not on the bus says nothing here: the controller creates it on its next raise, and what is -// not there cannot fill. -func probeStoreFill(ctx context.Context, d *doctor) ([]conditions.Observation, error) { - stores, err := declaredStores(ctx, d) - if err != nil { - return nil, err - } - js := d.js.Context() - var out []conditions.Observation - for _, s := range stores { - if ctx.Err() != nil { - return nil, ctx.Err() - } - info, err := js.StreamInfo(s.Stream, nats.Context(ctx)) - switch { - case errors.Is(err, nats.ErrStreamNotFound): - continue - case err != nil: - return nil, fmt.Errorf("%s's %s %s cannot be read: %w", s.Module, s.Kind, s.Name, err) - } - if info.Config.MaxBytes <= 0 { - continue - } - f := filled{Bytes: info.State.Bytes, Max: uint64(info.Config.MaxBytes)} - if o, found := fillObservation(s, f, isOpen(ctx, d.keeper, fillKey(s))); found { - out = append(out, o) - } - } - return sortedFound(out), nil -} diff --git a/cmd/mesh-controller/store_fill_test.go b/cmd/mesh-controller/store_fill_test.go deleted file mode 100644 index 7ede9fe7..00000000 --- a/cmd/mesh-controller/store_fill_test.go +++ /dev/null @@ -1,76 +0,0 @@ -package main - -import ( - "strings" - "testing" - - "github.com/novox/mesh-controller/internal/conditions" -) - -// **A filling log or bucket is said at three quarters of its cap and cleared below seven tenths** -// (novox/hq ADR 0297 §6): raised at 75% naming the module, the store and its fill; not at 74.9%; kept -// between 70% and 75% while it is open, and not raised there when it is not; and cleared — said no -// more — once it reads below 70%, open or not. -func TestAFillingStoreIsRaisedAtThreeQuartersAndClearedBelowSevenTenths(t *testing.T) { - const mib = 1024 * 1024 - log := moduleStore{Module: "mesh-issues", Kind: "log", Name: "changes", Stream: "LOG_mesh-issues_changes"} - bucket := moduleStore{Module: "mesh-issues", Kind: "bucket", Name: "issues", Stream: "KV_mesh-issues_issues"} - for _, c := range []struct { - name string - store moduleStore - bytes uint64 - max uint64 - open bool - said bool - }{ - {"a log at exactly 75%", log, 768 * mib, 1024 * mib, false, true}, - {"a bucket at exactly 75%", bucket, 48 * mib, 64 * mib, false, true}, - {"a bucket full", bucket, 64 * mib, 64 * mib, false, true}, - {"a log just under 75%", log, 768*mib - 1, 1024 * mib, false, false}, - {"a bucket at 72%, not open", bucket, 64 * mib * 72 / 100, 64 * mib, false, false}, - {"a bucket at 72%, open: kept", bucket, 64 * mib * 72 / 100, 64 * mib, true, true}, - {"a log at exactly 70%, open: kept", log, 700 * mib, 1000 * mib, true, true}, - {"a log just under 70%, open: cleared", log, 700*mib - 1, 1000 * mib, true, false}, - {"a bucket at 10%, open: cleared", bucket, 64 * mib / 10, 64 * mib, true, false}, - {"a store with no cap", bucket, 100, 0, true, false}, - } { - o, said := fillObservation(c.store, filled{Bytes: c.bytes, Max: c.max}, c.open) - if said != c.said { - t.Errorf("%s: said %v, want %v", c.name, said, c.said) - continue - } - if !said { - continue - } - if o.Key() != fillKey(c.store) || o.Kind != kindStoreFilling || o.Severity != conditions.Warning { - t.Errorf("%s: raised as %s (%s, %s)", c.name, o.Key(), o.Kind, o.Severity) - } - for _, part := range []string{c.store.Module, c.store.Kind + " " + c.store.Name, "%"} { - if !strings.Contains(o.Summary, part) || !strings.Contains(o.Headline, part) { - t.Errorf("%s: does not name %q: %q / %q", c.name, part, o.Summary, o.Headline) - } - } - if !strings.Contains(o.Summary, " of ") || !strings.Contains(o.Said, "bytes") { - t.Errorf("%s: does not say the fill as bytes of the cap: %q / %q", c.name, o.Summary, o.Said) - } - } - o, _ := fillObservation(log, filled{Bytes: 800 * mib, Max: 1024 * mib}, false) - if !strings.Contains(o.Summary, "800.0 MiB of 1024.0 MiB, 78%") { - t.Errorf("the fill is said as %q", o.Summary) - } -} - -// The words the operator reads are plain, for every store a module may name. -func TestAFillingStoreIsSaidInPlainWords(t *testing.T) { - for _, s := range []moduleStore{ - {Module: "mesh-issues", Kind: "log", Name: "changes", Stream: "LOG_mesh-issues_changes"}, - {Module: "a-module-with-a-very-long-name-indeed", Kind: "bucket", Name: "a-bucket-with-a-long-name", - Stream: "KV_a-module-with-a-very-long-name-indeed_a-bucket-with-a-long-name"}, - } { - o, _ := fillObservation(s, filled{Bytes: 90, Max: 100}, false) - if why, ok := conditions.PlainWords(conditions.Words{Headline: o.Headline, Explanation: o.Explanation, - Resolved: o.Resolved, Needs: o.Needs}); !ok { - t.Errorf("%s: %s", s.Stream, why) - } - } -}