package main import ( "fmt" "sort" "strings" "time" "github.com/novox/mesh-controller/internal/broker" "github.com/novox/mesh-controller/internal/conditions" "github.com/novox/mesh-controller/internal/inventory" "github.com/novox/mesh-controller/internal/link" ) // The signals table (novox/hq to-be 45 §3, ADR 0227 rule 5), compiled in. // // **Every signal the core expects has a watchdog; its absence is a condition.** A row says what is // expected, from whom, after what, within what bound, and what is raised when it does not come. One // table, read three ways: by the watchdog loop (watchdogs.go), which runs every row's watch over the // facts it gathered; by `doctor signals`, which says the age of every row's newest signal; and by the // test generated from it (signals_test.go), which suppresses each signal in turn and asserts its // condition — so a row added without a watch, or a watch that does not fire, fails the build. // // A row not watched yet says why, and in which phase it will be: a watchdog of a signal nothing emits // would be a condition that can only cry wolf. Bounds marked provisional are set from what Phase 0 // measured (`durations`) and corrected in Phase 1's first live week; a corrected bound is a change to // this table, reviewed like code. // signalRow is one row of the table. type signalRow struct { Row string Signal string Emitter string Trigger string Bound string Kind string Severity conditions.Severity // Phase is the phase of to-be 45 its watchdog is built in. Phase int // Deferred says why it is not watched yet; empty for a row that is. Deferred string // needs is the part of the facts the row reads: an error there is the row blind, which is // itself said (probe-failed) rather than read as nothing wrong. needs func(f *signalFacts) error // watch is what is wrong now, from the facts. watch func(f *signalFacts) []conditions.Observation // newest is the time of the newest signal of this row, for `doctor signals`; zero when none. newest func(f *signalFacts) time.Time } // The bounds, provisional where to-be 45 says so. const ( // heartbeatEvery is the interval a node-engine or node tools that say none have always used. heartbeatEvery = 60 * time.Second // heartbeatsMissed is how many intervals may pass in silence (S1, S11). heartbeatsMissed = 3 // controlNodeUrgentAfter is how long the control node may be silent before it is urgent (S1). controlNodeUrgentAfter = 30 * time.Minute // reportAtLeast is the least a machine is given to report a send (S2). reportAtLeast = 2 * time.Minute // tierAtLeast is the least a plan's tier is given (S3), the bound `status` calls a plan late at. tierAtLeast = planWaitBound // loopDeafAfter is how long the event loop may take nothing while its consumers hold some (S4). loopDeafAfter = 2 * time.Minute // askAtLeast is the least a build ask is given, and askDefault the bound while nothing is // measured (S6): the build seat declares no timeout of its own. askAtLeast = 20 * time.Minute askDefault = time.Hour // callDefault is the bound of a verb that declares none (S7); push's and build's are longer. callDefault = 10 * time.Minute // advisoryQuiet is how long the bus must be quiet about a thing before its advisory clears (S9). advisoryQuiet = time.Hour // staleRefusalsAllowed in staleRefusalsWithin are what S13 lets pass from one writer. staleRefusalsAllowed = 5 staleRefusalsWithin = 5 * time.Minute // leaseBound is how long the lease may go unrenewed (S12): the key's age. leaseBound = broker.LeaseTTL // waitBound is how long a walk may wait for its delivery's word before it is said, and waitUrgentAfter // before it is urgent (S16, novox/hq ADR 0239): a group member may wait behind another for a while; four // hours without a word is the delivery's owner broken, whatever it says of itself. waitBound = 30 * time.Minute waitUrgentAfter = 4 * time.Hour ) // callBounds are the verbs that may run longer than callDefault, and how long (S7). var callBounds = map[string]time.Duration{ "push": 30 * time.Minute, "rotate": 30 * time.Minute, "give": 5 * time.Minute, "assign": 15 * time.Minute, "unassign": 15 * time.Minute, "command": 30 * time.Minute, "doctor": 3 * time.Minute, } // callBound is a verb's bound. func callBound(verb string) time.Duration { if b, ok := callBounds[verb]; ok { return b } return callDefault } // signalsTable is the table, in to-be 45's order. var signalsTable = []signalRow{ {Row: "S1", Signal: "machine heartbeat", Emitter: "node-engine", Trigger: "its interval", Bound: "3 × the interval it says (60 s when it says none), provisional; not raised while the machine " + "said it is asleep or shutting down (ADR 0211); urgent after 30 min for a control node", Kind: "silent", Severity: conditions.Warning, Phase: 1, needs: func(f *signalFacts) error { return f.machinesErr }, watch: watchHeartbeats, newest: func(f *signalFacts) time.Time { return newestOf(f.machines, func(m machineFacts) time.Time { return m.lastHeard }) }}, {Row: "S2", Signal: "report after a send", Emitter: "node-engine", Trigger: "each declaration sent", Bound: "max(2 min, 3 × that machine's last apply duration), provisional; not while the machine is silent or asleep", Kind: "sent-not-reported", Severity: conditions.Warning, Phase: 1, needs: func(f *signalFacts) error { return f.machinesErr }, watch: watchReports, newest: func(f *signalFacts) time.Time { return newestOf(f.machines, func(m machineFacts) time.Time { return m.reportedAt }) }}, {Row: "S3", Signal: "plan tier progress", Emitter: "controller's plan", Trigger: "each tier entered", Bound: "max(30 min, 3 × the p90 of that repository's measured tiers), provisional; not while the " + "build seat is paused under it", Kind: "stalled", Severity: conditions.Warning, Phase: 1, needs: func(f *signalFacts) error { return f.plansErr }, watch: watchPlans, newest: func(f *signalFacts) time.Time { return newestOf(f.plans, func(p planFacts) time.Time { return p.entered }) }}, {Row: "S4", Signal: "the controller's event loop takes a message", Emitter: "controller", Trigger: "while its consumers have pending messages", Bound: "2 min", Kind: "controller-deaf", Severity: conditions.Urgent, Phase: 1, needs: func(f *signalFacts) error { return f.loopErr }, watch: watchLoop, newest: func(f *signalFacts) time.Time { return f.loop.took }}, {Row: "S5", Signal: "a merge announced becomes a plan, or nothing reads it", Emitter: "announcer → controller", Trigger: "each merge", Bound: "10 min (the catch-up pass of issue 266)", Kind: "merge-not-acted", Severity: conditions.Urgent, Phase: 1, needs: func(f *signalFacts) error { return f.mergesErr }, watch: watchMerges, newest: func(f *signalFacts) time.Time { return f.mergesPassed }}, {Row: "S6", Signal: "a build asked → its outcome", Emitter: "build seat", Trigger: "each ask", Bound: "max(20 min, 3 × the p90 of measured builds), 1 h while nothing is measured, provisional — the " + "build seat declares no timeout; and an ask the queue gave up on (dead) at once", Kind: "ask-lost", Severity: conditions.Warning, Phase: 1, needs: func(f *signalFacts) error { return f.asksErr }, watch: watchAsks, newest: func(f *signalFacts) time.Time { return newestOf(f.asks, func(a askFacts) time.Time { return a.since }) }}, {Row: "S7", Signal: "a call running → finished", Emitter: "controller", Trigger: "each call", Bound: "the verb's bound: push, rotate and command 30 min, assign and unassign 15 min, doctor 3 min, " + "any other 10 min", Kind: "call-hung", Severity: conditions.Warning, Phase: 1, needs: func(*signalFacts) error { return nil }, watch: watchCalls, newest: func(f *signalFacts) time.Time { return newestOf(f.calls, func(c link.Call) time.Time { return c.Started }) }}, {Row: "S8", Signal: "a provider's failing word repeated", Emitter: "provider", Trigger: "every 15 min while failing (ADR 0224)", Bound: "30 min", Kind: kindProviderSilent, Severity: conditions.Warning, Phase: 1, needs: func(f *signalFacts) error { return f.standingsErr }, watch: watchProviders, newest: func(f *signalFacts) time.Time { return newestOf(f.standings, func(c conditions.Condition) time.Time { return c.LastObserved }) }}, {Row: "S9", Signal: "bus advisories: maximum deliveries, consumer deleted; the controller's own slow " + "consumer and refused subjects", Emitter: "bus server's advisory subjects; the controller's connection", Trigger: "any", Bound: "any occurrence; clears after an hour without another, a deleted consumer " + "once it exists again or the mesh no longer expects it, and a message given up on once DEAD_LETTERS " + "no longer holds it (novox/hq issue 330)", Kind: "slow-consumer, max-deliveries, refused, consumer-lost", Severity: conditions.Warning, Phase: 1, needs: func(f *signalFacts) error { return f.advisoriesErr }, watch: watchAdvisories, newest: func(f *signalFacts) time.Time { return newestOf(f.advisories, func(a link.Advisory) time.Time { return a.Last }) }}, {Row: "S10", Signal: "the self-check's heartbeat", Emitter: "controller's doctor", Trigger: "every run", Bound: "2 × its interval; watched from a second machine by mesh-watcher, and here as well", Kind: "self-check-silent", Severity: conditions.Urgent, Phase: 1, needs: func(*signalFacts) error { return nil }, watch: watchSelfCheck, newest: func(f *signalFacts) time.Time { return f.selfCheck.last }}, {Row: "S11", Signal: "node tools heartbeat", Emitter: "node tools", Trigger: "its interval", Bound: "3 × the interval it says (60 s when it says none), provisional; only where node-tools is assigned", Kind: "tools-silent", Severity: conditions.Warning, Phase: 1, needs: func(f *signalFacts) error { return f.machinesErr }, watch: watchTools, newest: func(f *signalFacts) time.Time { return newestOf(f.machines, func(m machineFacts) time.Time { return m.toolsHeard }) }}, {Row: "S12", Signal: "the controller lease renewed", Emitter: "controller", Trigger: "every 5 s", Bound: "15 s (the key's age); a holder that lost the lease, or stopped renewing and was taken over, and a " + "lease bucket found raised again from nothing, are said for an hour after; a controller serving without " + "the lease, for as long as it does", Kind: "lease-lost", Severity: conditions.Urgent, Phase: 2, needs: func(f *signalFacts) error { return f.leaseErr }, watch: watchLease, newest: func(f *signalFacts) time.Time { return f.lease.renewed }}, {Row: "S13", Signal: "stale refusals", Emitter: "every receiver (rule 2)", Trigger: "each refusal", Bound: "more than 5 from one writer in 5 min: a controller epoch, a controller that claimed none, or a " + "machine's node-engine whose accounts the controller refused", Kind: "stale-writer", Severity: conditions.Warning, Phase: 2, needs: func(*signalFacts) error { return nil }, watch: watchStaleRefusals, newest: func(f *signalFacts) time.Time { return newestOf(f.staleRefusals, func(w link.WriterRefusals) time.Time { return w.Last }) }}, {Row: "S14", Signal: "facts snapshot exported", Emitter: "controller", Trigger: "when it moved, and daily", Bound: "2 days: a snapshot older than that, or none kept since this controller began two days ago", Kind: "facts-stale", Severity: conditions.Warning, Phase: 5, needs: func(*signalFacts) error { return nil }, watch: watchFacts, newest: func(f *signalFacts) time.Time { return f.facts.taken }}, {Row: "S15", Signal: "a hand act with a cause already recorded", Emitter: "hand-act log", Trigger: "each act", Bound: "the second within 14 days; clears when fewer than two remain within 14 days", Kind: "healer-wanted", Severity: conditions.Warning, Phase: 3, needs: func(f *signalFacts) error { return f.handActsErr }, watch: watchHandActs, newest: func(f *signalFacts) time.Time { return newestOf(f.handActs, func(a link.HandAct) time.Time { return a.At }) }}, {Row: "S16", Signal: "a walk waiting for its delivery's word is let go", Emitter: "mesh-delivery, through deliver", Trigger: "each merge whose walk waits (novox/hq ADR 0239)", Bound: "30 min, then urgent after 4 h — raised by the controller whatever mesh-delivery says of itself, " + "naming `plans go` as the way on", Kind: kindWalkWaiting, Severity: conditions.Warning, Phase: 3, needs: func(f *signalFacts) error { return f.plansErr }, watch: watchWaits, newest: func(f *signalFacts) time.Time { return newestOf(f.waits, func(w waitFacts) time.Time { return w.since }) }}, {Row: "S18", Signal: "a batch of merges is cut into its walk", Emitter: "controller's merge window", Trigger: "each batch (novox/hq ADR 0276)", Bound: "a minute past merge-window-at-most, while no walk is open: the controller failed to cut it", Kind: kindBatchNotCut, Severity: conditions.Warning, Phase: 3, needs: func(f *signalFacts) error { return f.plansErr }, watch: watchBatchesNotCut, newest: func(f *signalFacts) time.Time { return newestOf(f.batches, func(b batchFacts) time.Time { return b.atMost }) }}, {Row: "S19", Signal: "a batch waiting behind an open walk is cut when it ends", Emitter: "controller's merge window", Trigger: "each batch closed while a walk is open (novox/hq ADR 0276)", Bound: "the walk's bound, a tier's bound for each of its tiers, from when the batch closed: naming the walk", Kind: kindBatchBehindWalk, Severity: conditions.Warning, Phase: 3, needs: func(f *signalFacts) error { return f.plansErr }, watch: watchBatchesBehind, newest: func(f *signalFacts) time.Time { return newestOf(f.batches, func(b batchFacts) time.Time { return b.closed }) }}, {Row: "S20", Signal: "a send refused for the assignment generation it was composed from", Emitter: "node-engine", Trigger: "each refusal (novox/hq issue 234)", Bound: "none: raised at the first refusal, naming its sender, the generation it came from and the one the " + "machine applied; cleared an hour after the last", Kind: kindOlderGeneration, Severity: conditions.Warning, Phase: 2, needs: func(f *signalFacts) error { return f.refusedSendsErr }, watch: watchGenerationRefusals, newest: func(f *signalFacts) time.Time { return newestOf(f.refusedSends, func(s inventory.Send) time.Time { if s.RefusedAt == nil { return time.Time{} } return *s.RefusedAt }) }}, {Row: "S17", Signal: "a send held for the bus's planned step is told to a person", Emitter: "controller's plan", Trigger: "each send refused because it would replace the bus outside its step (novox/hq issue 336)", Bound: "none: raised at the first refusal, for the operator, naming what waits, the bus build from and to, " + "since when and the mesh-controller.bus call; cleared once the bus's machine has been sent the build the mesh holds", Kind: kindBusStepWaiting, Severity: conditions.Warning, Phase: 3, needs: func(f *signalFacts) error { if f.plansErr != nil { return f.plansErr } return f.busErr }, watch: watchBusWaits, newest: func(f *signalFacts) time.Time { return newestOf(f.bus.waits, func(w busWaitFacts) time.Time { return w.since }) }}, {Row: "S21", Signal: "a clean apply", Emitter: "node-engine", Trigger: "each report of a declaration refused or applied only in part (novox/hq issue 233)", Bound: "one apply interval (5 min) past the first report of the same failure, judged by time while the machine " + "is heard — the node-engine says a re-apply only when how it went changed; not while it is silent or " + "asleep; naming the refusal, or the package manager that does not answer with its words; cleared by the " + "clean apply it then says", Kind: kindApplyFailing, Severity: conditions.Warning, Phase: 2, needs: func(f *signalFacts) error { return f.doingErr }, watch: watchApplyFailing, newest: func(f *signalFacts) time.Time { return newestOf(f.doing, func(d inventory.Doing) time.Time { return d.At }) }}, } // watchFacts is S14: the snapshot a merge check is fed is older than its bound, or none was kept since // this controller began that long ago (novox/hq to-be 45 §9). A check fed a stale snapshot judges a change // against a mesh that no longer is, which is the fault the snapshot exists to end. func watchFacts(f *signalFacts) []conditions.Observation { since := f.facts.taken if since.IsZero() { since = f.facts.began } if since.IsZero() || f.now.Sub(since) <= factsStaleAfter { return nil } said := "none kept since this controller began " + ago(f.now.Sub(since)) + " ago" if !f.facts.taken.IsZero() { said = "the newest kept was taken " + ago(f.now.Sub(f.facts.taken)) + " ago" } if f.facts.err != nil { said += "; the last attempt: " + oneLine(f.facts.err.Error()) } return []conditions.Observation{{Scope: conditions.ScopeCore, ID: "facts", Kind: "facts-stale", Token: "stale", Severity: conditions.Warning, Summary: fmt.Sprintf("the facts snapshot merge checks are fed is stale (bound %s): a change is judged against a "+ "mesh that no longer is", ago(factsStaleAfter)), Said: said}} } // newestOf is the newest time among things. func newestOf[T any](list []T, at func(T) time.Time) time.Time { var newest time.Time for _, x := range list { if t := at(x); t.After(newest) { newest = t } } return newest } // ago is a duration as the summaries say it. func ago(d time.Duration) string { if d < time.Minute { return d.Round(time.Second).String() } return d.Round(time.Minute).String() } // heartbeatBound is a machine's S1 or S11 bound from the interval it says. func heartbeatBound(every time.Duration) time.Duration { if every <= 0 { every = heartbeatEvery } return heartbeatsMissed * every } func watchHeartbeats(f *signalFacts) []conditions.Observation { var out []conditions.Observation for _, m := range f.machines { if m.lastHeard.IsZero() || m.asleep() { // Never heard is a machine that has not joined, which status says; asleep is not lost. continue } bound := heartbeatBound(m.every) silent := f.now.Sub(m.lastHeard) if silent <= bound { continue } severity := conditions.Warning if m.control && silent > controlNodeUrgentAfter { severity = conditions.Urgent } out = append(out, conditions.Observation{Scope: conditions.ScopeMachine, ID: m.name, Kind: "silent", Machine: m.name, Severity: severity, Summary: fmt.Sprintf("%s has not been heard from since %s (bound %s)", m.name, m.lastHeard.UTC().Format("2006-01-02 15:04 MST"), bound), Said: fmt.Sprintf("silent for %s", ago(silent))}) } return out } func watchReports(f *signalFacts) []conditions.Observation { var out []conditions.Observation for _, m := range f.machines { if m.sentAt.IsZero() || m.reportedCurrent || m.asleep() { continue } if !m.lastHeard.IsZero() && f.now.Sub(m.lastHeard) > heartbeatBound(m.every) { continue // silent: S1 says it, and a silent machine reports nothing } bound := max(reportAtLeast, 3*m.lastApply) waited := f.now.Sub(m.sentAt) if waited <= bound { continue } out = append(out, conditions.Observation{Scope: conditions.ScopeMachine, ID: m.name, Kind: "sent-not-reported", Machine: m.name, Severity: conditions.Warning, Summary: fmt.Sprintf("%s was sent a declaration at %s and has not reported applying it (bound %s)", m.name, m.sentAt.UTC().Format("2006-01-02 15:04 MST"), ago(bound)), Said: fmt.Sprintf("waiting %s for the report of the declaration sent", ago(waited))}) } return out } func watchPlans(f *signalFacts) []conditions.Observation { var out []conditions.Observation for _, p := range f.plans { // Under a paused build seat, or held only by the bus's planned step (S17, novox/hq issue 336): a wait // for a person, said as itself, not a walk running late. if p.paused || p.bus { continue } in := f.now.Sub(p.entered) if in <= p.bound { continue } out = append(out, conditions.Observation{Scope: conditions.ScopePlan, ID: p.id, Kind: "stalled", Headline: "Delivery of " + repoName(p.repository) + " is stuck halfway", Explanation: fmt.Sprintf("Its walk across the machines has been at the same step for %s, longer than "+ "usual. Nothing is lost, and the mesh keeps it where it is.", humanDuration(f.now.Sub(p.entered))), Resolved: "Delivery of " + repoName(p.repository) + " moves again", Severity: conditions.Warning, Summary: fmt.Sprintf("the plan for %s %s has been at tier %d of %d since %s (bound %s): %s", p.repository, short(p.commit), p.tier+1, p.tiers, p.entered.UTC().Format("2006-01-02 15:04 MST"), ago(p.bound), p.waiting), Said: fmt.Sprintf("at tier %d for %s: %s", p.tier+1, ago(in), p.waiting)}) } return out } // kindWalkWaiting is S16's kind: plan..waiting. const kindWalkWaiting = "waiting" // watchWaits is S16: a walk that has waited for its delivery's word past its bound, said by the controller // itself — a delivery's owner that is up and never says go (a bug, a delivery stuck in its own table) would // otherwise leave a merge waiting for ever with nothing open (novox/hq ADR 0239). func watchWaits(f *signalFacts) []conditions.Observation { var out []conditions.Observation for _, w := range f.waits { in := f.now.Sub(w.since) if in <= waitBound { continue } severity := conditions.Warning if in > waitUrgentAfter { severity = conditions.Urgent } out = append(out, conditions.Observation{Scope: conditions.ScopePlan, ID: w.id, Kind: kindWalkWaiting, Severity: severity, Summary: fmt.Sprintf("the walk of %s %s has waited %s for %s's word to start: `mesh-delivery.show` for the "+ "deliveries that landed as %s says why; `plans go %s --why …` starts it by hand", w.repository, short(w.commit), ago(in), w.awaits, mergesWords(w), w.id), Said: fmt.Sprintf("waiting since %s for %s", w.since.UTC().Format(time.RFC3339), w.awaits), Headline: deliveryName(w.modules, w.repository) + " waiting to start", Explanation: walkWaitingWords(w, in, severity), Needs: waitingNeeds(severity), Actions: waitingActions(w.id, severity), Resolved: deliveryName(w.modules, w.repository) + " no longer waiting"}) } return out } // mergesWords is the merges a waiting walk answers (novox/hq ADR 0276): every delivery it carries, not one // commit that a supersession may already have ended (issue 362). Its own commit for a walk naming none. func mergesWords(w waitFacts) string { if len(w.merges) == 0 { return short(w.commit) } var out []string for _, m := range w.merges { out = append(out, m.Repository+"@"+short(m.Commit)) } return readableList(out) } func watchLoop(f *signalFacts) []conditions.Observation { if f.loop.pending == 0 { return nil } since := f.loop.took if since.IsZero() { since = f.started } if f.now.Sub(since) <= loopDeafAfter { return nil } return []conditions.Observation{{Scope: conditions.ScopeCore, ID: "controller", Kind: "controller-deaf", Token: "deaf", Machine: f.host, Severity: conditions.Urgent, Summary: fmt.Sprintf("the controller's event loop has taken nothing for %s while its consumers hold %d "+ "message(s): reports, builds and merges are not being acted on", ago(f.now.Sub(since)), f.loop.pending), Said: fmt.Sprintf("%d pending (%s), last taken %s", f.loop.pending, f.loop.where, since.UTC().Format(time.RFC3339))}} } func watchMerges(f *signalFacts) []conditions.Observation { var out []conditions.Observation for _, m := range f.merges { out = append(out, conditions.Observation{Scope: conditions.ScopeMerge, ID: m.Repo + "." + short(m.Commit), Kind: "merge-not-acted", Token: "not-acted", Severity: conditions.Urgent, Summary: fmt.Sprintf("%s/%s merged into %s (%s) was never handed to the controller by the bus; "+ "acted on late by the catch-up", m.Owner, m.Repo, m.Base, short(m.Commit)), Said: fmt.Sprintf("announced %s, %s behind it", m.At.UTC().Format(time.RFC3339), readableList(m.Modules))}) } return out } func watchAsks(f *signalFacts) []conditions.Observation { var out []conditions.Observation for _, a := range f.asks { var said string switch { case a.state == link.AskDead: said = fmt.Sprintf("the %s queue handed it out as often as it may and it was never settled", a.seat) case a.state == link.AskInFlight && f.now.Sub(a.since) > a.bound: said = fmt.Sprintf("in flight on %s for %s (bound %s)", orSomewhere(a.on), ago(f.now.Sub(a.since)), ago(a.bound)) default: continue } out = append(out, conditions.Observation{Scope: conditions.ScopeBuild, ID: a.id, Kind: "ask-lost", Token: "lost", Machine: a.on, Severity: conditions.Warning, Summary: fmt.Sprintf("the build %s of %s has no outcome: %s", a.id, a.what, said), Said: said}) } return out } func orSomewhere(node string) string { if node == "" { return "a machine that did not say which" } return node } func watchCalls(f *signalFacts) []conditions.Observation { var out []conditions.Observation for _, c := range f.calls { bound := callBound(c.Verb) running := f.now.Sub(c.Started) if running <= bound { continue } out = append(out, conditions.Observation{Scope: conditions.ScopeCall, ID: c.ID, Kind: "call-hung", Token: "hung", Machine: f.host, Severity: conditions.Warning, Summary: fmt.Sprintf("%s.%s (call %s) has been running since %s, past its bound of %s", c.Seat, c.Verb, c.ID, c.Started.UTC().Format("2006-01-02 15:04 MST"), ago(bound)), Said: fmt.Sprintf("running %s, asked by %s", ago(running), orSomebody(c.Caller))}) } return out } func orSomebody(caller string) string { if caller == "" { return "a caller the bus did not name" } return caller } func watchProviders(f *signalFacts) []conditions.Observation { var out []conditions.Observation for _, c := range f.standings { quiet := f.now.Sub(c.LastObserved) if quiet <= providerSaysAgainWithin { continue } module, node, consumer, ok := providerOf(c) if !ok { continue } out = append(out, conditions.Observation{Scope: conditions.ScopeProvider, ID: module + "." + node + "." + consumer, Token: "silent", Kind: kindProviderSilent, Machine: node, Severity: conditions.Warning, Summary: fmt.Sprintf("%s on %s said it keeps failing %s and has said nothing since %s: it stopped "+ "saying anything, so its last word is all the mesh has", module, node, consumer, c.LastObserved.UTC().Format("2006-01-02 15:04 MST")), Said: fmt.Sprintf("not said again for %s", ago(quiet))}) } return out } func watchAdvisories(f *signalFacts) []conditions.Observation { var out []conditions.Observation for _, a := range f.advisories { if f.now.Sub(a.Last) > advisoryQuiet { continue } if a.Kind == link.AdvisoryConsumerLost && !f.lostConsumers[a.Stream+"."+a.Consumer] { continue // it exists again, or the mesh no longer expects it: a removal, not a loss } if a.Kind == link.AdvisoryMaxDeliveries && a.Token == "" { // A consumer's max-deliveries key is the dead-letter row's (novox/hq issue 330): said while // DEAD_LETTERS holds what it gave up on, cleared when that is delivered again or dropped. The // listener records no such advisory today; one read here as well added a look to the count and // overwrote the row's words on every look (novox/hq issue 440). Only one that could not be kept // (AdvisoryNotKept), which DEAD_LETTERS cannot say, is said from here. continue } severity := conditions.Warning times := "" if a.Count > 1 { times = fmt.Sprintf(" (%d times since %s)", a.Count, a.First.UTC().Format("15:04 MST")) } machine := "" if a.ID == "controller" { machine = f.host } o := conditions.Observation{Scope: conditions.ScopeBus, ID: a.ID, Kind: a.Kind, Token: a.Token, Machine: machine, Severity: severity, Summary: a.Said + times, Said: a.Said, // The advisory is remembered and read again on every look for an hour: the condition counts // what the bus said and when, not the looks (novox/hq issue 402). Happened: a.Last, Times: a.Count} if a.Kind == link.AdvisoryMaxDeliveries && a.Token == link.AdvisoryNotKept { o.Machine = consumerMachine(a.Stream, a.Consumer) o.Headline = clip(conditions.Capital(fmt.Sprintf("%s gave up on a message, not kept", consumerWho(a.Stream, a.Consumer))), 60) o.Explanation = "A listener on the bus could not handle a message, and the mesh could not keep it " + "for you yet. It tries again every minute." o.Resolved = "Resolved: the message is kept" } out = append(out, o) } return append(out, watchDeadLetters(f)...) } // watchDeadLetters says each consumer that DEAD_LETTERS holds a message for (novox/hq issue 330): open // while it holds any, so it clears when they are delivered again or dropped, never because the server // stopped saying it. func watchDeadLetters(f *signalFacts) []conditions.Observation { keys := make([]string, 0, len(f.deadLetters)) for k := range f.deadLetters { keys = append(keys, k) } sort.Strings(keys) var out []conditions.Observation for _, key := range keys { n := f.deadLetters[key] stream, consumer, _ := strings.Cut(key, ".") messages, them := "a message", "it" if n > 1 { messages, them = fmt.Sprintf("%d messages", n), "them" } who := consumerWho(stream, consumer) // DEAD_LETTERS is a record of what was given up on: the condition says when the newest held message // was kept, and its count is a running total of the messages given up on while it is open, never // how often the controller looked (novox/hq issues 402 and 440). It is at least the number held now, // and does not go down when one is delivered again or dropped. Without that time it counts its // looks, as a source of the present does. happened, times := f.deadLettersNewest[key], 0 if !happened.IsZero() { times = n } out = append(out, conditions.Observation{Scope: conditions.ScopeBus, ID: key, Kind: link.AdvisoryMaxDeliveries, Happened: happened, Times: times, Machine: consumerMachine(stream, consumer), Severity: conditions.Warning, Summary: fmt.Sprintf("%s gave up on %s; %s kept in %s until delivered again or dropped, with why, "+ "through the controller's dead-letters verb", link.ConsumerInWords(stream, consumer), messages, map[bool]string{true: "they are", false: "it is"}[n > 1], broker.DeadLettersStream), Said: fmt.Sprintf("%d held for %s", n, key), Headline: clip(conditions.Capital(fmt.Sprintf("%s could not handle %s", who, messages)), 60), Needs: fmt.Sprintf("deliver %s again or drop %s, from the mesh MCP server.", them, them), Explanation: conditions.Capital(fmt.Sprintf("%s was handed %s several times and gave up, so what %s "+ "asked for was not done. The mesh keeps %s until you deliver %s again or drop %s.", who, messages, them, them, them, them)), Resolved: "Resolved: the messages it gave up on were delivered again or dropped"}) } return out } // consumerWho is a durable consumer's holder as the operator says it: a module on its machine, the // controller, or a seat's holders. func consumerWho(stream, consumer string) string { switch { case consumer == broker.ControllerName: return "the controller" case strings.HasPrefix(stream, "SEAT_") && strings.HasSuffix(consumer, "_worker"): seat := strings.ToLower(strings.ReplaceAll(strings.TrimSuffix(strings.TrimPrefix(consumer, "SEAT_"), "_worker"), "_", "-")) return "the holder of " + seat case stream == broker.EventsStream: if node, module, ok := strings.Cut(consumer, "_"); ok { return module + " on " + node } } return "a listener on the bus" } // consumerMachine is the machine a module's consumer is on; empty for the others. func consumerMachine(stream, consumer string) string { if stream == broker.EventsStream { if node, _, ok := strings.Cut(consumer, "_"); ok { return node } } return "" } // clip is words at most n characters long, cut at a word. func clip(s string, n int) string { if len(s) <= n { return s } cut := s[:n] if i := strings.LastIndex(cut, " "); i > 0 { cut = cut[:i] } return cut } func watchSelfCheck(f *signalFacts) []conditions.Observation { every := f.selfCheck.every if every <= 0 { every = doctorEvery } since := f.selfCheck.last if since.IsZero() { since = f.started } if f.now.Sub(since) <= 2*every { return nil } return []conditions.Observation{{Scope: conditions.ScopeCore, ID: "doctor", Kind: "self-check-silent", Machine: f.host, Severity: conditions.Urgent, Summary: fmt.Sprintf("the self-check has not finished a run since %s (it runs every %s): the mesh's "+ "invariants are not being checked", since.UTC().Format("2006-01-02 15:04 MST"), every), Said: fmt.Sprintf("no run for %s", ago(f.now.Sub(since)))}} } func watchTools(f *signalFacts) []conditions.Observation { var out []conditions.Observation for _, m := range f.machines { if !m.tools || m.asleep() { continue } bound := heartbeatBound(m.toolsEvery) since := m.toolsHeard if since.IsZero() { // Not heard since this controller started: silent since then at the most. since = f.toolsHeardFrom } silent := f.now.Sub(since) if silent <= bound { continue } heard := "not since this controller started at " + since.UTC().Format("2006-01-02 15:04 MST") if !m.toolsHeard.IsZero() { heard = "since " + m.toolsHeard.UTC().Format("2006-01-02 15:04 MST") } out = append(out, conditions.Observation{Scope: conditions.ScopeMachine, ID: m.name, Kind: "tools-silent", Machine: m.name, Severity: conditions.Warning, Summary: fmt.Sprintf("the node tools on %s have not said they are there %s (bound %s): nothing can "+ "ask that machine anything", m.name, heard, bound), Said: fmt.Sprintf("silent for %s", ago(silent))}) } return out } func watchStaleRefusals(f *signalFacts) []conditions.Observation { var out []conditions.Observation for _, w := range f.staleRefusals { if w.Count <= staleRefusalsAllowed { continue } scope, id, machine := conditions.ScopeCore, "controller.unnamed", "" named := w.Writer switch { case w.Epoch > 0: id = fmt.Sprintf("controller.epoch-%d", w.Epoch) if e, ok := f.epochs[w.Epoch]; ok { named = fmt.Sprintf("the controller of epoch %d (%s%s)", w.Epoch, e.Instance, endedWords(e)) } case w.Writer == link.WriterNodeEngine(firstOf(w.Receivers)) || strings.HasPrefix(w.Writer, "the node-engine on "): node := strings.TrimPrefix(w.Writer, "the node-engine on ") scope, id, machine = conditions.ScopeMachine, node, node } if machine == "" && len(w.Receivers) > 0 { machine = w.Receivers[0] } var also []string for _, r := range w.Receivers { if r != machine && r != "controller" { also = append(also, r) } } out = append(out, conditions.Observation{Scope: scope, ID: id, Kind: "stale-writer", Token: "stale-writer", Machine: machine, Also: also, Severity: conditions.Warning, Summary: fmt.Sprintf("%s was refused %d time(s) in %s as older than what its receivers hold (%s): a "+ "writer is sending what the mesh has moved past", named, w.Count, staleRefusalsWithin, strings.Join(w.Receivers, ", ")), Said: fmt.Sprintf("%d stale refusals in %s, the last at %s", w.Count, staleRefusalsWithin, w.Last.UTC().Format(time.RFC3339))}) } return out } // firstOf is a list's first, empty for none. func firstOf(list []string) string { if len(list) == 0 { return "" } return list[0] } // endedWords is how an epoch ended, for a sentence naming it; nothing while it is held. func endedWords(e inventory.Epoch) string { if e.Ended == nil { return ", still holding the lease" } return fmt.Sprintf(", %s at %s", e.How, e.Ended.UTC().Format("15:04:05 MST")) } // watchLease is S12: the lease held and renewed by this controller, and every holder that lost it. func watchLease(f *signalFacts) []conditions.Observation { var out []conditions.Observation l := f.lease if l.unleased != "" { out = append(out, conditions.Observation{Scope: conditions.ScopeCore, ID: "controller.lease", Token: "unleased", Kind: "lease-lost", Machine: f.host, Severity: conditions.Urgent, Summary: "the controller serves WITHOUT the lease: nothing keeps a second controller from acting " + "beside it, and its declarations carry no epoch. A bus whose user list is older than this " + "controller does not grant it the lease's bucket: a push of the machine holding the bus sends " + "the list that does, and the controller takes the lease within five seconds", Said: l.unleased}) } else if l.held && !l.renewed.IsZero() && f.now.Sub(l.renewed) > leaseBound { out = append(out, conditions.Observation{Scope: conditions.ScopeCore, ID: "controller.lease", Token: "late", Kind: "lease-lost", Machine: f.host, Severity: conditions.Urgent, Summary: fmt.Sprintf("the controller has not renewed its lease (epoch %d) since %s, past the key's age "+ "of %s: another controller may take it", l.epoch, l.renewed.UTC().Format(time.RFC3339), leaseBound), Said: fmt.Sprintf("not renewed for %s", ago(f.now.Sub(l.renewed)))}) } if !l.reset.IsZero() && f.now.Sub(l.reset) <= advisoryQuiet { out = append(out, conditions.Observation{Scope: conditions.ScopeCore, ID: "controller.lease", Token: "reset", Kind: "lease-lost", Machine: f.host, Severity: conditions.Urgent, Summary: "the controller lease's bucket was raised again from nothing: whatever held the lease before " + "does not know it lost it", Said: l.resetSaid}) } var lost []string newest := inventory.Epoch{} for _, e := range l.ended { if e.Ended == nil || e.How == inventory.EpochReleased || f.now.Sub(*e.Ended) > advisoryQuiet { continue } lost = append(lost, fmt.Sprintf("epoch %d (%s) %s at %s", e.Epoch, e.Instance, e.How, e.Ended.UTC().Format("15:04:05 MST"))) if newest.Ended == nil || e.Ended.After(*newest.Ended) { newest = e } } if len(lost) > 0 { how := "lost it: its renewal was refused or could not be made" if newest.How == inventory.EpochExpired { how = "stopped renewing it without giving it back, and was taken over" } out = append(out, conditions.Observation{Scope: conditions.ScopeCore, ID: "controller.lease", Token: "lost", Kind: "lease-lost", Machine: newest.Host, Severity: conditions.Urgent, Summary: fmt.Sprintf("the controller of epoch %d (%s) %s; the controller of epoch %d acts now", newest.Epoch, newest.Instance, how, l.epoch), Said: strings.Join(lost, "; ")}) } return out } // handActsWithin is how far back a repeated cause counts (S15). const handActsWithin = 14 * 24 * time.Hour // watchHandActs is S15: a cause recorded by hand twice within a fortnight is a healer wanted, named by // the cause. **A heal is never a hand act** (healers.go), so a cause a healer exists for and a person // still repaired twice says the healer is not enough — its reach or its budget — and is said so. **An act // that is a person's decision by design is no repair** (handActVerbs), so it never counts; one counted // before this was so clears on the next tick, as any condition the row no longer sees. func watchHandActs(f *signalFacts) []conditions.Observation { recent := make([]link.HandAct, 0, len(f.handActs)) for _, a := range repairs(f.handActs) { if f.now.Sub(a.At) <= handActsWithin { recent = append(recent, a) } } repeated := link.RepeatedCauses(recent, f.now) causes := make([]string, 0, len(repeated)) for c := range repeated { causes = append(causes, c) } sort.Strings(causes) var out []conditions.Observation for _, cause := range causes { var acts []string var newest link.HandAct for _, a := range recent { if a.Cause != cause { continue } acts = append(acts, fmt.Sprintf("%s %s by %s: %s", a.At.UTC().Format("2006-01-02 15:04"), strings.TrimSpace(a.Verb+" "+strings.Join(a.Args, " ")), a.By, a.Why)) if a.At.After(newest.At) { newest = a } } wanted := "a healer is wanted for it" if h := healerNamedFor(cause); h != "" { wanted = fmt.Sprintf("healer %s answers this cause and a person still repaired it: its reach or its "+ "budget is not enough", h) } out = append(out, conditions.Observation{Scope: conditions.ScopeMesh, ID: "hand-acts." + cause, Token: "healer-wanted", Kind: "healer-wanted", Severity: conditions.Warning, Summary: fmt.Sprintf("%q was repaired by hand %d times in %d days, the last by %s: %s", cause, repeated[cause], int(handActsWithin.Hours()/24), newest.By, wanted), Said: strings.Join(acts, "; "), Headline: conditions.Capital(causeWords(cause)) + " keeps being fixed by hand", Explanation: fmt.Sprintf("A person repaired %s by hand %d times in %d days, so an automatic repair is "+ "wanted for it. Nothing is broken now.", causeWords(cause), repeated[cause], int(handActsWithin.Hours()/24)), Resolved: "Resolved: no more hand repairs of " + causeWords(cause)}) } return out } // watchedRows are the rows a watchdog runs for. func watchedRows() []signalRow { var out []signalRow for _, r := range signalsTable { if r.watch != nil { out = append(out, r) } } return out } // kindsOf is a row's condition kinds, one or several. func kindsOf(r signalRow) []string { var out []string for _, k := range strings.Split(r.Kind, ",") { out = append(out, strings.TrimSpace(k)) } return out }