package main // The watcher's watcher (novox/hq to-be 45 §5, signal S10, ADR 0227 rule 6): on a machine that is not // the control node, it listens for the self-check's heartbeat and for the bus itself. When either has // been silent past its bound it sends to Telegram directly over HTTPS — not through the bus — says so // once more an hour later if it still is, and says so again when it returns. // // Two signals: // // - self-check: the controller's `doctor` heartbeat. Bound: twice the interval the heartbeat says // it runs at, else twice five minutes (to-be 45 §3, S10). Heard by the time the heartbeat says it // was made, not by when it arrived: a backlog delivered after a restart is old news, not a sign of // life. // - bus: a round trip through the bus server — this module asking its own tool on its own machine // — every minute. Bound: three minutes. // // Before a signal is first heard it counts from when the watcher started: a heartbeat that never comes // is exactly what this is for. import ( "fmt" "sync" "time" ) const ( SelfCheck = "self-check" Bus = "bus" DefaultInterval = 5 * time.Minute BusBound = 3 * time.Minute RemindAfter = time.Hour // Skew is how far in the future a heartbeat's time may be and still be believed. Skew = 2 * time.Minute ) // Message is one thing said to the operator. type Message struct { Title string Body string } func (m Message) Text() string { if m.Body == "" { return m.Title } return m.Title + "\n" + m.Body } // Sender is the Telegram channel. type Sender interface { Ready() error Send(Message) (string, error) } type signal struct { name string lastHeard time.Time bound time.Duration silentAt time.Time // when it was said silent; zero while it is heard heardBefore time.Time // the last word before it went silent reminded bool owed *Message // a message that could not be sent yet lastRun string // the heartbeat's run id, for the status lastErr string // the bus probe's last error } // Watcher watches. type Watcher struct { Telegram Sender Now func() time.Time Logf func(string, ...any) Node string mu sync.Mutex started time.Time signals map[string]*signal sent []sentNote sendErr string sendOK time.Time misplace string // why this machine is the wrong one to watch from, or "" control string // the machine the controller is on, as last read badBeats int lastBad string } type sentNote struct { At time.Time `json:"at"` Signal string `json:"signal"` What string `json:"what"` Outcome string `json:"outcome"` } func NewWatcher(tg Sender, now func() time.Time, logf func(string, ...any), node string) *Watcher { w := &Watcher{Telegram: tg, Now: now, Logf: logf, Node: node, started: now(), signals: map[string]*signal{ SelfCheck: {name: SelfCheck, bound: 2 * DefaultInterval}, Bus: {name: Bus, bound: BusBound}, }} return w } // Heartbeat takes one self-check heartbeat. func (w *Watcher) Heartbeat(b Beat) { w.mu.Lock() defer w.mu.Unlock() now := w.Now() s := w.signals[SelfCheck] at := b.At if at.IsZero() { at = now } if at.After(now.Add(Skew)) { w.badBeats++ w.lastBad = fmt.Sprintf("a heartbeat from the future (%s), not believed", at.UTC().Format(time.RFC3339)) w.Logf("[mesh-watcher] %s", w.lastBad) return } if at.After(s.lastHeard) { s.lastHeard = at s.lastRun = b.Run } if b.Interval > 0 { s.bound = 2 * b.Interval } } // BadHeartbeat counts a heartbeat that could not be read: said, never read as a sign of life. func (w *Watcher) BadHeartbeat(err error) { w.mu.Lock() defer w.mu.Unlock() w.badBeats++ w.lastBad = err.Error() w.Logf("[mesh-watcher] a heartbeat could not be read (not counted as heard): %v", err) } // BusAnswered takes the outcome of one round trip through the bus. func (w *Watcher) BusAnswered(err error) { w.mu.Lock() defer w.mu.Unlock() s := w.signals[Bus] if err != nil { if s.lastErr != err.Error() { w.Logf("[mesh-watcher] the bus did not answer a round trip: %v", err) } s.lastErr = err.Error() return } s.lastErr = "" s.lastHeard = w.Now() } // Placed records where the controller runs, and says when that is this machine. func (w *Watcher) Placed(controlNodes []string) { w.mu.Lock() defer w.mu.Unlock() w.control = "" was := w.misplace w.misplace = "" for _, n := range controlNodes { if w.control != "" { w.control += ", " } w.control += n if n == w.Node && w.Node != "" { w.misplace = "this machine holds " + n + "'s controller or bus: a watcher here goes down with what it watches; assign mesh-watcher to another machine" } } if w.misplace != "" && was == "" { w.Logf("[mesh-watcher] %s", w.misplace) } } // Tick decides what is silent, what returned, and sends what is owed. func (w *Watcher) Tick() { w.mu.Lock() now := w.Now() var out []struct { s *signal m Message w string } for _, name := range []string{SelfCheck, Bus} { s := w.signals[name] since := s.lastHeard if since.IsZero() { since = w.started } silent := now.Sub(since) switch { case s.silentAt.IsZero() && silent > s.bound: s.silentAt, s.reminded, s.heardBefore = now, false, since m := w.silentMessage(s, silent, false) s.owed = &m case !s.silentAt.IsZero() && silent <= s.bound: m := Message{ Title: "CLEARED: " + w.what(s) + " heard again", Body: "silent for about " + roughly(s.lastHeard.Sub(s.heardBefore)) + "; told by mesh-watcher on " + w.Node + ", directly over HTTPS", } s.silentAt = time.Time{} s.owed = &m case !s.silentAt.IsZero() && !s.reminded && now.Sub(s.silentAt) >= RemindAfter: s.reminded = true m := w.silentMessage(s, silent, true) s.owed = &m } if s.owed != nil { out = append(out, struct { s *signal m Message w string }{s, *s.owed, name}) } } w.mu.Unlock() for _, o := range out { err := w.send(o.w, o.m) w.mu.Lock() if err == nil { o.s.owed = nil } w.mu.Unlock() } } func (w *Watcher) what(s *signal) string { if s.name == SelfCheck { return "the controller's self-check (doctor)" } return "the bus" } func (w *Watcher) silentMessage(s *signal, silent time.Duration, again bool) Message { title := "URGENT: self-check-silent: " + w.what(s) + " has not been heard for " + roughly(silent) if s.name == Bus { title = "URGENT: bus-silent: " + w.what(s) + " has not answered a round trip for " + roughly(silent) } if again { title = "STILL SILENT: " + title[len("URGENT: "):] } body := "bound: " + roughly(s.bound) + "; told by mesh-watcher on " + w.Node + ", directly over HTTPS, not through the mesh" if s.lastHeard.IsZero() { body += "\nnot heard once since the watcher started" } else { body += "\nlast heard: " + s.lastHeard.UTC().Format("2006-01-02 15:04") + " UTC" } if s.name == SelfCheck { body += "\nthe controller, the control node or the bus may be down; the mesh's own messages pass through them" } return Message{Title: title, Body: body} } func (w *Watcher) send(signalName string, m Message) error { _, err := w.Telegram.Send(m) w.mu.Lock() defer w.mu.Unlock() note := sentNote{At: w.Now(), Signal: signalName, What: m.Title, Outcome: "sent"} if err != nil { note.Outcome = "failed: " + err.Error() if w.sendErr != err.Error() { w.Logf("[mesh-watcher] cannot send to Telegram (tried again every minute): %v", err) } w.sendErr = err.Error() } else { if w.sendErr != "" { w.Logf("[mesh-watcher] Telegram sends again") } w.sendErr = "" w.sendOK = w.Now() w.Logf("[mesh-watcher] told the operator: %s", m.Title) } w.sent = append(w.sent, note) if len(w.sent) > 100 { w.sent = w.sent[len(w.sent)-100:] } return err } // SignalStatus is one signal as the status says it. type SignalStatus struct { Signal string `json:"signal"` LastHeard string `json:"last_heard"` Ago string `json:"ago"` Bound string `json:"bound"` Silent bool `json:"silent"` SaidSilent string `json:"said_silent_at,omitempty"` Owed string `json:"not_sent_yet,omitempty"` LastRun string `json:"last_run,omitempty"` LastFailure string `json:"last_failure,omitempty"` } // Status is the watcher's account of itself, its own health first. type Status struct { Verdict string `json:"verdict"` Node string `json:"watching_from"` Control string `json:"controller_and_bus_on,omitempty"` Misplaced string `json:"misplaced,omitempty"` Telegram string `json:"telegram"` LastSent string `json:"last_sent,omitempty"` Signals []SignalStatus `json:"signals"` BadBeats int `json:"unreadable_heartbeats"` LastBad string `json:"last_unreadable,omitempty"` Started string `json:"started"` Listening string `json:"listening"` } func (w *Watcher) Status(listening string) Status { ready := w.Telegram.Ready() w.mu.Lock() defer w.mu.Unlock() now := w.Now() st := Status{Node: w.Node, Control: w.control, Misplaced: w.misplace, Started: w.started.UTC().Format(time.RFC3339), BadBeats: w.badBeats, LastBad: w.lastBad, Listening: listening} switch { case ready != nil: st.Telegram = "CANNOT SEND: " + ready.Error() case w.sendErr != "": st.Telegram = "FAILING: " + w.sendErr default: st.Telegram = "ready" } if !w.sendOK.IsZero() { st.LastSent = w.sendOK.UTC().Format(time.RFC3339) } anySilent := false for _, name := range []string{SelfCheck, Bus} { s := w.signals[name] ss := SignalStatus{Signal: name, Bound: roughly(s.bound), Silent: !s.silentAt.IsZero(), LastRun: s.lastRun, LastFailure: s.lastErr} if s.lastHeard.IsZero() { ss.LastHeard = "never since start" ss.Ago = roughly(now.Sub(w.started)) + " since start" } else { ss.LastHeard = s.lastHeard.UTC().Format(time.RFC3339) ss.Ago = roughly(now.Sub(s.lastHeard)) } if ss.Silent { ss.SaidSilent = s.silentAt.UTC().Format(time.RFC3339) anySilent = true } if s.owed != nil { ss.Owed = s.owed.Title } st.Signals = append(st.Signals, ss) } switch { case st.Telegram != "ready": st.Verdict = "BLIND: the watcher cannot tell the operator anything — " + st.Telegram case w.misplace != "": st.Verdict = "MISPLACED: " + w.misplace case listening != "listening": st.Verdict = "NOT LISTENING: heartbeats cannot reach the watcher — " + listening case anySilent: st.Verdict = "ALARM: a watched signal is silent — see signals" default: st.Verdict = "ok: watching" } return st } // LastHeard is each signal's last word, and the recent sends. func (w *Watcher) LastHeard() map[string]any { w.mu.Lock() defer w.mu.Unlock() now := w.Now() out := map[string]any{} for name, s := range w.signals { if s.lastHeard.IsZero() { out[name] = "never since the watcher started " + roughly(now.Sub(w.started)) + " ago" } else { out[name] = s.lastHeard.UTC().Format(time.RFC3339) + " (" + roughly(now.Sub(s.lastHeard)) + " ago)" } } sent := make([]sentNote, 0, len(w.sent)) for i := len(w.sent) - 1; i >= 0; i-- { sent = append(sent, w.sent[i]) } out["sent"] = sent return out } func roughly(d time.Duration) string { switch { case d < 0: return "a moment" case d < 2*time.Minute: return fmt.Sprintf("%d s", int(d.Seconds())) case d < 2*time.Hour: return fmt.Sprintf("%d min", int(d.Minutes())) case d < 48*time.Hour: return fmt.Sprintf("%.1f h", d.Hours()) } return fmt.Sprintf("%d days", int(d.Hours()/24)) }