mesh-catalog/modules/mesh-watcher/cmd/mesh-watcher/watcher.go

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 self-check's heartbeat (mesh-self-check.checked). 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 self-check"
	}
	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))
}