diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/health.go b/modules/mesh-knowledge/cmd/mesh-knowledge/health.go new file mode 100644 index 000000000..92d03c5bc --- /dev/null +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/health.go @@ -0,0 +1,210 @@ +package main + +// The module's health once its records are on the ledger (novox/hq to-be 55 §1.1, ADR 0592 §33 changed by ADR 0597 §12): +// unhealthy, naming the cause, when a kept thing does not answer, when the search index is behind the ledger, when a head lags +// behind its history, and when the stream of the verdicts is at its cap. It names where the records are served from and the +// result of the move's last check, and counts the drafts idle more than a week, the records the form-only check refused and the +// parts waiting more than 30 minutes; such a part is asked of the operator once, and the ask clears when it is judged. A record +// waiting for its review is not the module failing, so those counts leave it healthy. + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/buscalls" +) + +// PartWaits is how long a part may wait on the judge before the operator is asked once. +const PartWaits = 30 * time.Minute + +// ledgerHealth adds what the ledger form owes to the verdict of the watch. +func (k *KB) ledgerHealth(base HealthAnswer) HealthAnswer { + ms, _ := k.Store.(*movableStore) + l, err := k.ledger() + if err != nil { + if ms != nil { + base.Why += "; the records are served from the " + SourceDatabase + } + return base + } + var broken, notes []string + notes = append(notes, "the records are served from the ledger") + // A kept thing that does not answer, and a head behind its history. + var lag buscalls.Lagged + if err := l.pipe.Call(buscalls.LedgerLag, buscalls.LedgerRequest{Ledger: LedgerName}, &lag); err != nil { + broken = append(broken, "the ledger does not answer: "+err.Error()) + } else if len(lag.Instances) > 0 { + names := make([]string, 0, 3) + for i, in := range lag.Instances { + if i == 3 { + break + } + names = append(names, fmt.Sprintf("%s (state at %d of %d)", in.Instance, in.State, in.History)) + } + broken = append(broken, fmt.Sprintf("%d head(s) are behind their history: %s", len(lag.Instances), strings.Join(names, ", "))) + } + // The index is behind the ledger when it holds fewer records than the ledger does. + st := l.IndexState() + l.mu.Lock() + heads, indexed := len(l.heads), 0 + if l.index != nil { + indexed = len(l.index.docsOf) + } + l.mu.Unlock() + switch { + case !st.Built: + notes = append(notes, "the search index is being built") + case heads > 0 && indexed < heads-reservedCount(l): + broken = append(broken, fmt.Sprintf("the search index holds %d of %d records: it is behind the ledger", indexed, heads)) + default: + notes = append(notes, fmt.Sprintf("the search index holds %d documents", st.Docs)) + } + // The stream of the verdicts at its cap. + var fill buscalls.StreamFilled + if err := l.pipe.Call(buscalls.StreamFill, buscalls.StreamRequest{Stream: "reviews"}, &fill); err != nil { + broken = append(broken, "the stream of the verdicts does not answer: "+err.Error()) + } else if fill.MaxBytes > 0 && fill.Bytes >= fill.MaxBytes { + broken = append(broken, fmt.Sprintf("the stream of the verdicts is at its cap of %d bytes", fill.MaxBytes)) + } + // The move's last check. + if kv := k.States; kv != nil { + if e, err := kv.State(findingsState).Get(checkKey); err == nil && e != nil { + var c struct { + Differences []string `json:"differences"` + At time.Time `json:"at"` + } + if json.Unmarshal(e.Value, &c) == nil { + if len(c.Differences) == 0 { + notes = append(notes, "the last check of the database against the ledger was clean ("+c.At.Local().Format("2006-01-02 15:04")+")") + } else { + notes = append(notes, fmt.Sprintf("the last check of the database against the ledger found %d difference(s)", len(c.Differences))) + } + } + } + } + // What waits on a person, counted and left healthy. + idle, waiting := k.waitingCounts() + if idle > 0 { + notes = append(notes, fmt.Sprintf("%d draft(s) untouched for more than a week", idle)) + } + if waiting > 0 { + notes = append(notes, fmt.Sprintf("%d part(s) waiting on the judge for more than 30 minutes", waiting)) + } + if refused := k.formRefused(); refused > 0 { + notes = append(notes, fmt.Sprintf("%d record(s) the form-only check refused", refused)) + } + if len(broken) > 0 { + base.State, base.Healthy = "unhealthy", false + base.Why = strings.Join(broken, "; ") + "; " + strings.Join(notes, "; ") + return base + } + base.Why += "; " + strings.Join(notes, "; ") + return base +} + +func reservedCount(l *ledgerStore) int { + n := 0 + for _, h := range l.heads { + if st := h.Entry.Current.Status; st == StatusReserved || st == StatusWithdrawn || st == StatusRetracted { + n++ + } + } + return n +} + +// waitingCounts are the drafts idle more than DraftIdle and the parts waiting more than PartWaits; a part that waits is +// asked of the operator once, and the ask is cleared when it is judged (the mark leaves `asked`). +func (k *KB) waitingCounts() (idle, waiting int) { + if k.States == nil { + return 0, 0 + } + now := k.now() + if keys, err := k.States.State(draftsState).Keys(); err == nil { + for _, key := range keys { + if e, err := k.States.State(draftsState).Get(key); err == nil && e != nil { + var d Draft + if json.Unmarshal(e.Value, &d) == nil && now.Sub(d.Touched) > DraftIdle { + idle++ + } + } + } + } + live := map[string]bool{} + if keys, err := k.States.State(askedState).Keys(); err == nil { + for _, key := range keys { + e, err := k.States.State(askedState).Get(key) + if err != nil || e == nil { + continue + } + var a struct { + Part string `json:"part"` + Record string `json:"record"` + At time.Time `json:"at"` + } + if json.Unmarshal(e.Value, &a) != nil || now.Sub(a.At) <= PartWaits { + continue + } + waiting++ + live[key] = true + k.askPartWaits(key, a.Record, a.Part) + } + } + k.clearPartWaits(live) + return idle, waiting +} + +const partWaitPrefix = "part-wait." + +// askPartWaits asks the operator once, through the operator channel, about a part waiting more than 30 minutes. +func (k *KB) askPartWaits(key, record, part string) { + if k.Notify == nil || k.Store == nil { + return + } + ctx := context.Background() + if first, err := k.Store.Told(ctx, partWaitPrefix+key); err != nil || !first { + return + } + if err := k.Notify("knowledge."+partWaitPrefix+key, fmt.Sprintf("The review of %s waits on the judge: its part %q has waited more than 30 minutes.", record, part), + "A review is waiting on the judge: "+record, "The knowledge base asked the judge for a part of this record and no answer has come. It asks again by itself; the tool knowledge_asked lists what is in flight.", + record+" is judged again"); err != nil { + _ = k.Store.Untold(ctx, partWaitPrefix+key) + } +} + +// clearPartWaits clears the ask of every part that is no longer waiting. +func (k *KB) clearPartWaits(live map[string]bool) { + if k.Clear == nil || k.Store == nil { + return + } + ctx := context.Background() + told, err := k.Store.ToldOf(ctx, partWaitPrefix) + if err != nil { + return + } + for _, t := range told { + if live[strings.TrimPrefix(t, partWaitPrefix)] { + continue + } + if err := k.Clear("knowledge."+t, "its part was judged"); err == nil { + _ = k.Store.Untold(ctx, t) + } + } +} + +// formRefused counts the records a form-only version would be refused for. +func (k *KB) formRefused() int { + done, err := k.FormOnly("", nil, false) + if err != nil { + return 0 + } + n := 0 + for _, d := range done { + if strings.HasPrefix(d.Skipped, "the check refused") { + n++ + } + } + return n +} diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/health_test.go b/modules/mesh-knowledge/cmd/mesh-knowledge/health_test.go new file mode 100644 index 000000000..546e4388d --- /dev/null +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/health_test.go @@ -0,0 +1,83 @@ +package main + +import ( + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/buscalls" +) + +// The health on the ledger (to-be 55 §1.1, AC56): it names where the records are served from and the move's last check, is +// unhealthy for a ledger or a stream that does not answer, a head behind its history, an index behind the ledger and a stream +// at its cap, and counts what waits on a person (an idle draft, a part waiting on the judge, a form-only check refused) without +// failing; a part waiting more than 30 minutes is asked of the operator once, and the ask clears when it is judged. +func TestTheHealthOnTheLedgerNamesTheSourceAndFailsForWhatTheLedgerOwes(t *testing.T) { + w, f := ledgerWorld(t) + ls := w.k.Store.(*ledgerStore) + if err := ls.BuildIndex(t.Context()); err != nil { + t.Fatal(err) + } + f.fill = buscalls.StreamFilled{Fill: buscalls.Fill{Count: 3, Bytes: 100, MaxBytes: 512 << 20}} + base := HealthAnswer{State: "healthy", Healthy: true, Why: "the database serves"} + ans := w.k.ledgerHealth(base) + if !ans.Healthy || !strings.Contains(ans.Why, "served from the ledger") || !strings.Contains(ans.Why, "search index holds") { + t.Fatalf("%+v", ans) + } + // Each thing the ledger owes, one at a time. + for name, c := range map[string]struct { + set func() + want string + }{ + "a head behind": {func() { + f.lag = buscalls.Lagged{Instances: []buscalls.InstanceLag{{Instance: "decision/0290", History: 9, State: 7}}} + }, "behind their history"}, + "a stream at a cap": {func() { f.fill.Bytes = f.fill.MaxBytes }, "at its cap"}, + "a stream that does not answer": {func() { f.noFill = true }, "does not answer"}, + } { + f.lag, f.noFill = buscalls.Lagged{}, false + f.fill = buscalls.StreamFilled{Fill: buscalls.Fill{Bytes: 100, MaxBytes: 512 << 20}} + c.set() + got := w.k.ledgerHealth(base) + if got.Healthy || !strings.Contains(got.Why, c.want) { + t.Errorf("%s: %+v", name, got) + } + } + f.lag, f.noFill, f.fill = buscalls.Lagged{}, false, buscalls.StreamFilled{} + // An index behind the ledger: a record the ledger holds and the index does not. + ls.index.mu.Lock() + ls.index.docsOf = map[Key][]string{} + ls.index.mu.Unlock() + if got := w.k.ledgerHealth(base); got.Healthy || !strings.Contains(got.Why, "behind the ledger") { + t.Errorf("an index behind: %+v", got) + } + _ = ls.BuildIndex(t.Context()) + // What waits on a person is counted and leaves it healthy; a part waiting more than 30 minutes is asked once. + w.k.States.State(askedState).Put("decision.0300.abc", map[string]any{"part": "Context", "record": "ADR 0300", "at": w.clock.Add(-time.Hour)}) + w.k.States.State(draftsState).Put("decision.0301", Draft{Kind: KindDecision, Number: 301, Author: "a@n", Touched: w.clock.AddDate(0, 0, -8)}) + got := w.k.ledgerHealth(base) + if !got.Healthy || !strings.Contains(got.Why, "1 part(s) waiting") || !strings.Contains(got.Why, "1 draft(s) untouched") { + t.Fatalf("%+v", got) + } + w.k.ledgerHealth(base) + if len(w.told) != 1 { + t.Fatalf("the operator was asked %d times", len(w.told)) + } + w.k.States.State(askedState).Delete("decision.0300.abc") + w.k.ledgerHealth(base) + if len(w.cleared) != 1 { + t.Fatalf("the ask was not cleared: %v", w.cleared) + } +} + +// The records served from the database say so, and fail nothing. +func TestTheHealthOnTheDatabaseSaysSo(t *testing.T) { + db := newMemStore() + states := newFakeStates() + m := newMovableStore(db, newLedgerStore(newFakeLedger(t), states), states) + k := &KB{Store: m, States: states} + ans := k.ledgerHealth(HealthAnswer{State: "healthy", Healthy: true, Why: "the database serves"}) + if !ans.Healthy || !strings.Contains(ans.Why, "served from the database") { + t.Fatalf("%+v", ans) + } +} diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/ledger_fake_test.go b/modules/mesh-knowledge/cmd/mesh-knowledge/ledger_fake_test.go index 88c275b6d..377eb6023 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/ledger_fake_test.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/ledger_fake_test.go @@ -30,6 +30,10 @@ type fakeLedger struct { // failApply, when set, refuses the apply of that operation. failApply string account string + // lag and fill are what lag and a stream's fill answer; noFill makes the stream not answer. + lag buscalls.Lagged + fill buscalls.StreamFilled + noFill bool } func newFakeLedger(t *testing.T) *fakeLedger { @@ -97,6 +101,13 @@ func (f *fakeLedger) Call(method string, req, answer any) error { page.Operations = append(page.Operations, o) } out = page + case buscalls.LedgerLag: + out = f.lag + case buscalls.StreamFill: + if f.noFill { + return fmt.Errorf("the stream does not answer") + } + out = f.fill default: return fmt.Errorf("the fake ledger answers no %s", method) } diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/main.go b/modules/mesh-knowledge/cmd/mesh-knowledge/main.go index e10b33243..818e9b0cb 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/main.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/main.go @@ -106,7 +106,7 @@ func serve(k *KB, status string) { k.kick = make(chan struct{}, 1) go hear(k) go background(k) - if err := stdio.Serve("", append(append(Tools(k, callerOf), healthTool(status, k.now), highestTool(k), marksTool(k)), moduleTools(k, callerOf)...)); err != nil { + if err := stdio.Serve("", append(append(Tools(k, callerOf), healthTool(status, k.now, k), highestTool(k), marksTool(k)), moduleTools(k, callerOf)...)); err != nil { logf("%v", err) os.Exit(1) } @@ -617,14 +617,20 @@ func (k *KB) Health(ready Healther) HealthAnswer { return HealthAnswer{State: "healthy", Healthy: true, Why: why} } -func healthTool(status string, now func() time.Time) stdio.Tool { +func healthTool(status string, now func() time.Time, k *KB) stdio.Tool { return stdio.Tool{Name: HealthTool, Description: "How the knowledge base is, as its watch last saw it: healthy when its database serves this version, unhealthy " + "otherwise. A rule that judges the whole record set and fails — a decision accepted that nothing cites, a supersession " + "or a change said on one end only, a record extending a superseded one — is named but leaves it healthy: it is told " + "to the operator as a warning for the records' authors (novox/hq issue 504). So is a proposed record that ran out of " + "review tries: it waits on its author (novox/hq issue 508).", - Run: func(map[string]any) (any, error) { return ReadStatus(status, now()), nil }} + Run: func(map[string]any) (any, error) { + ans := ReadStatus(status, now()) + if k != nil { + ans = k.ledgerHealth(ans) // once the records are on the ledger, what the ledger form owes (health.go) + } + return ans, nil + }} } // ---- the verbs --------------------------------------------------------------------------------------------------