jschoubben
15a15cbda5
S1 is urgent after half an hour for a control node (the node running mesh-controller). S9 also hears the bus module's slow consumers and closed connections, each closure counted once. S21 is one finding per heard machine from its failed haves and its provisioner clients' reports (PROV_APPLIED). DEAD_LETTERS is counted again at a letter's age and every five minutes while letters are held, on the watchman's own pulse. needs-operator waits for L6: its inputs are not on the bus. Pins mesh-sdk b6ef01d.
92 lines
3.7 KiB
Go
92 lines
3.7 KiB
Go
package watch
|
|
|
|
import (
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
// The dead letters' findings (novox/hq ADR 0564 §5, issue 330): a consumer DEAD_LETTERS holds a message for is said for as
|
|
// long as it holds any, `bus.<stream>.<consumer>.max-deliveries`, and the notices waiting behind the controller's consumer
|
|
// after the lease write are said as `bus.DEAD_LETTER_NOTICES.controller.notices-waiting`.
|
|
|
|
// DeadRecountEvery is how often DEAD_LETTERS is counted again while a consumer's finding stands: a letter delivered again
|
|
// or dropped by the controller's verb, which the watchman does not hear, ends its finding within it. The expiry of the
|
|
// watchman's own pulse, never a loop.
|
|
const DeadRecountEvery = 5 * time.Minute
|
|
|
|
// KindNoticesWaiting is the finding of notices waiting while the controller's consumer stands after the lease write.
|
|
const KindNoticesWaiting = "notices-waiting"
|
|
|
|
// DeadLettersHeld is DEAD_LETTERS' count per consumer, `<stream>.<consumer>`, read whole: each consumer held writes its
|
|
// finding, each one no longer held has it deleted. who words a consumer.
|
|
func (w *Watchman) DeadLettersHeld(held map[string]int, who func(stream, consumer string) string) {
|
|
w.mu.Lock()
|
|
defer w.mu.Unlock()
|
|
want := map[string]bool{}
|
|
keys := make([]string, 0, len(held))
|
|
for k := range held {
|
|
keys = append(keys, k)
|
|
}
|
|
sort.Strings(keys)
|
|
for _, key := range keys {
|
|
n := held[key]
|
|
stream, consumer, ok := strings.Cut(key, ".")
|
|
if !ok || n <= 0 {
|
|
continue
|
|
}
|
|
messages, them := "a message", "it is"
|
|
if n > 1 {
|
|
messages, them = fmt.Sprintf("%d messages", n), "they are"
|
|
}
|
|
f := Finding{Scope: ScopeBus, ID: key, Kind: KindDeadLetters, Severity: Warning, Source: "S9",
|
|
Machine: consumerMachine(stream, consumer),
|
|
Summary: fmt.Sprintf("%s gave up on %s; %s kept in DEAD_LETTERS until delivered again or dropped, with why",
|
|
who(stream, consumer), messages, them),
|
|
Said: fmt.Sprintf("%d held for %s (counted again every %s while any is held, so one delivered again or dropped "+
|
|
"ends this within it)", n, key, DeadRecountEvery),
|
|
Needs: "deliver them again or drop them, from the mesh MCP server."}
|
|
want[f.Key()] = true
|
|
w.find(f)
|
|
}
|
|
for key := range w.findings {
|
|
if strings.HasPrefix(key, ScopeBus+".") && strings.HasSuffix(key, "."+KindDeadLetters) && !want[key] {
|
|
w.unfind(key)
|
|
}
|
|
}
|
|
if len(want) == 0 {
|
|
w.unsignal("dead", "recount")
|
|
return
|
|
}
|
|
w.signal("dead", "recount", DeadRecountEvery, signalDue{Do: dueRecount,
|
|
Finding: Finding{Scope: ScopeBus, ID: "DEAD_LETTERS", Kind: KindDeadLetters, Severity: Warning, Summary: "counted again", Source: "S9"}})
|
|
}
|
|
|
|
// consumerMachine is the machine a module's consumer on EVENTS is on; empty for the others.
|
|
func consumerMachine(stream, consumer string) string {
|
|
if stream == "EVENTS" {
|
|
if node, _, ok := strings.Cut(consumer, "_"); ok {
|
|
return node
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// NoticesWaiting says whether the lease is handed away while the controller's consumer on DEAD_LETTER_NOTICES still
|
|
// stands: every notice then waits in the queue, kept a week, until the lease write's step removes the consumer.
|
|
func (w *Watchman) NoticesWaiting(waiting bool) {
|
|
w.mu.Lock()
|
|
defer w.mu.Unlock()
|
|
f := Finding{Scope: ScopeBus, ID: "DEAD_LETTER_NOTICES.controller", Kind: KindNoticesWaiting, Severity: Warning, Source: "dead-letters",
|
|
Summary: "the lease names no controller but the controller's consumer on DEAD_LETTER_NOTICES still stands, so every " +
|
|
"notice of a message given up on waits in the queue, kept a week, and nothing keeps it in DEAD_LETTERS",
|
|
Said: "the consumer controller on DEAD_LETTER_NOTICES stands after the lease write",
|
|
Needs: "remove the consumer controller on DEAD_LETTER_NOTICES with the bus module's tool, the lease write's step."}
|
|
if waiting {
|
|
w.find(f)
|
|
return
|
|
}
|
|
w.unfind(f.Key())
|
|
}
|