mesh-catalog/modules/nats/cmd/nats-tools/watch.go

package main

// The watch: every user's slow consumer and every connection the bus closed for breaking its rules,
// said on the bus (novox/hq ADR 0357, issue 270).
//
// **The server says these of other users only in its record of closed connections.** It closes every
// client it judges a slow consumer, and keeps the last 10,000 closed connections with the user and the
// reason (`/connz?state=closed&auth=1`) — about two days of the mesh's traffic. It publishes neither in
// the mesh's account; a system account would hear them, and the bus has none. So the bundle that already
// reads the record for `nats_closed_connections` looks at it every WatchEvery and publishes each closure
// that matters as an event the controller raises as S9.
//
// **Said at least once, counted once.** A closure is its user, its connection id and when it opened —
// the id alone is reused after the server restarts. Each one said is remembered while this process runs;
// one whose publish failed is tried again at the next look. On start the last hour (S9's window) is said
// again, so a restart repeats closures rather than losing them; the controller counts each by its
// identity, so a repeat changes no count.

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"sort"
	"strings"
	"sync"
	"time"
)

const (
	// WatchEvery is how often the watch reads the record of closed connections. With the controller's
	// watchdogs ticking every 30 s too, a closure is a condition about a minute after the server closed it.
	WatchEvery = 30 * time.Second
	// WatchSaysAgain is how far back the first look says closures again: S9's window, an hour.
	WatchSaysAgain = time.Hour
	// EventSlowConsumer and EventConnectionClosed are the events the watch publishes (module.json `emits`).
	EventSlowConsumer     = "bus.slow-consumer"
	EventConnectionClosed = "bus.connection-closed"
	// EventWatch is the watch's heartbeat, after every look: whether it is healthy and why not. The controller
	// raises a watch that stops saying it, or says it is not healthy (novox/hq ADR 0357, review 274): a dead
	// watch must never read as no slow consumers.
	EventWatch = "bus.watch"
)

// notSaid are the reasons a closure is routine: the client's own choice, the network or a machine asleep
// (D15 says a user dropped in a loop), the server's own shutdown, and a refused login, which names no user.
// Every other reason is said — one the module does not know included — because a new way for the bus to
// throw a client out is exactly what must not pass in silence.
var notSaid = map[string]bool{
	ClientClosed:             true,
	"Stale Connection":       true,
	"Read Error":             true,
	"Write Error":            true,
	"TLS Handshake Failure":  true,
	"Authentication Timeout": true,
	"Authentication Failure": true,
	"Server Shutdown":        true,
}

// isSlowConsumer says a closure's reason is the server judging the client a slow consumer: "Slow Consumer
// (Pending Bytes)" or "Slow Consumer (Write Deadline)".
func isSlowConsumer(reason string) bool { return strings.HasPrefix(reason, "Slow Consumer") }

// Closure is one closed connection as the watch says it. The user and the connection's own name, never a
// credential: the record the server keeps names who logged in, and nothing they logged in with.
type Closure struct {
	User   string    `json:"user"`
	Name   string    `json:"name,omitempty"`
	CID    uint64    `json:"cid"`
	Reason string    `json:"reason"`
	Opened time.Time `json:"opened"`
	Closed time.Time `json:"closed"`
	// Lived is how long the connection was open, in seconds.
	Lived float64 `json:"lived_seconds"`
	// Pending is what the server still had to write to it when it closed, in bytes, where it says.
	Pending int64 `json:"pending_bytes,omitempty"`
}

// identity is what tells one closure from another across a server restart.
func (c Closure) identity() string {
	return fmt.Sprintf("%s/%d/%s", c.User, c.CID, c.Opened.UTC().Format(time.RFC3339Nano))
}

// event is the event a closure is said as.
func (c Closure) event() string {
	if isSlowConsumer(c.Reason) {
		return EventSlowConsumer
	}
	return EventConnectionClosed
}

// Closures are the closed connections in the record that are to be said, closed at or after since, oldest
// first.
func Closures(connz []byte, since time.Time) ([]Closure, error) {
	var doc struct {
		Connections []struct {
			CID            uint64     `json:"cid"`
			Name           string     `json:"name"`
			AuthorizedUser string     `json:"authorized_user"`
			Start          time.Time  `json:"start"`
			Stop           *time.Time `json:"stop"`
			Reason         string     `json:"reason"`
			Pending        int64      `json:"pending_bytes"`
		} `json:"connections"`
	}
	if err := json.Unmarshal(connz, &doc); err != nil {
		return nil, fmt.Errorf("the bus's record of closed connections cannot be read: %w", err)
	}
	var out []Closure
	for _, c := range doc.Connections {
		if c.Stop == nil || c.Stop.Before(since) || c.AuthorizedUser == "" || notSaid[c.Reason] {
			continue
		}
		reason := c.Reason
		if reason == "" {
			reason = "unsaid"
		}
		out = append(out, Closure{User: c.AuthorizedUser, Name: c.Name, CID: c.CID, Reason: reason,
			Opened: c.Start.UTC(), Closed: c.Stop.UTC(), Lived: c.Stop.Sub(c.Start).Seconds(), Pending: c.Pending})
	}
	sort.SliceStable(out, func(i, j int) bool { return out[i].Closed.Before(out[j].Closed) })
	return out, nil
}

// Watch is the watch's state, shared with the tools that report it.
type Watch struct {
	monitor Monitor
	emit    func(string, any) error
	now     func() time.Time
	started time.Time

	mu sync.Mutex
	// said is every closure said, by identity, with when it closed: forgotten once older than the hour, since
	// the record is never read that far back again.
	said map[string]time.Time
	// failed are the closures whose publish failed, by identity, tried again at every look until said or
	// older than the hour.
	failed map[string]failedClosure
	// lastLook is when the watch last looked; lastRead when a look last read the record whole.
	lastLook, lastRead time.Time
	// failingSince is when the reads began to fail, zero while they succeed.
	failingSince time.Time
	lastErr      string
	saidN        map[string]int
	newest       []Closure
	// faults are the looks that panicked and were recovered: the loop goes on, and each is said.
	faults    int
	lastFault string
}

type failedClosure struct {
	closure Closure
	err     string
	since   time.Time
}

// NewWatch is a watch that has not looked yet.
func NewWatch(monitor Monitor, emit func(string, any) error) *Watch {
	return &Watch{monitor: monitor, emit: emit, now: time.Now, started: time.Now(), said: map[string]time.Time{},
		saidN: map[string]int{}, failed: map[string]failedClosure{}}
}

// readPage is how many closed connections one request asks for, newest first.
const readPage = 256

// readSince reads the record newest first (`sort=stop`), a page at a time, until a page reaches back past since
// or the record ends: a look reads only what closed since the last one, not all 10,000 kept.
func (w *Watch) readSince(ctx context.Context, since time.Time) ([]Closure, error) {
	var out []Closure
	for offset := 0; offset < maxClosedKept; offset += readPage {
		page, err := w.monitor(ctx, fmt.Sprintf("/connz?state=closed&auth=1&sort=stop&limit=%d&offset=%d", readPage, offset))
		if err != nil {
			return nil, err
		}
		closures, oldest, n, err := closuresOfPage(page, since)
		if err != nil {
			return nil, err
		}
		out = append(out, closures...)
		if n < readPage || (!oldest.IsZero() && oldest.Before(since)) {
			break
		}
	}
	sort.SliceStable(out, func(i, j int) bool { return out[i].Closed.Before(out[j].Closed) })
	return out, nil
}

// closuresOfPage is one page's closures to say, its oldest stop and how many connections it held.
func closuresOfPage(page []byte, since time.Time) ([]Closure, time.Time, int, error) {
	var doc struct {
		Connections []struct {
			Stop *time.Time `json:"stop"`
		} `json:"connections"`
	}
	if err := json.Unmarshal(page, &doc); err != nil {
		return nil, time.Time{}, 0, fmt.Errorf("the bus's record of closed connections cannot be read: %w", err)
	}
	var oldest time.Time
	for _, c := range doc.Connections {
		if c.Stop != nil && (oldest.IsZero() || c.Stop.Before(oldest)) {
			oldest = *c.Stop
		}
	}
	closures, err := Closures(page, since)
	return closures, oldest, len(doc.Connections), err
}

// Look reads what closed since the last look and says every closure not said yet, and tries again every
// closure whose publish failed. A fault in a look is recovered and said: the watch goes on.
func (w *Watch) Look(ctx context.Context) {
	defer func() {
		if r := recover(); r != nil {
			w.mu.Lock()
			w.faults++
			w.lastFault = fmt.Sprintf("a look stopped on a fault at %s: %v", w.now().UTC().Format(time.RFC3339), r)
			w.mu.Unlock()
			fmt.Fprintln(os.Stderr, w.lastFault)
		}
	}()
	now := w.now()
	hour := now.Add(-WatchSaysAgain)
	w.mu.Lock()
	// The first look says the last hour again; a later one reads from one interval before the last whole
	// read, so a closure the server recorded late is not missed — the said set tells what was said.
	since := hour
	if !w.lastRead.IsZero() && w.lastRead.Add(-WatchEvery).After(hour) {
		since = w.lastRead.Add(-WatchEvery)
	}
	w.mu.Unlock()
	read, cancel := context.WithTimeout(ctx, 20*time.Second)
	defer cancel()
	closures, err := w.readSince(read, since)

	w.mu.Lock()
	w.lastLook = now
	for id, closed := range w.said {
		if closed.Before(hour) {
			delete(w.said, id)
		}
	}
	for id, f := range w.failed {
		if f.closure.Closed.Before(hour) {
			delete(w.failed, id) // S9 would no longer say it
		}
	}
	if err != nil {
		w.lastErr = err.Error()
		if w.failingSince.IsZero() {
			w.failingSince = now
		}
		fmt.Fprintf(os.Stderr, "the watch could not read the bus's closed connections: %v\n", err)
	} else {
		w.lastErr, w.failingSince, w.lastRead = "", time.Time{}, now
	}
	todo := map[string]Closure{}
	for _, c := range closures {
		if _, done := w.said[c.identity()]; !done {
			todo[c.identity()] = c
		}
	}
	for id, f := range w.failed {
		if _, done := w.said[id]; !done {
			todo[id] = f.closure
		}
	}
	w.mu.Unlock()

	order := make([]Closure, 0, len(todo))
	for _, c := range todo {
		order = append(order, c)
	}
	sort.SliceStable(order, func(i, j int) bool { return order[i].Closed.Before(order[j].Closed) })
	for _, c := range order {
		if w.emit == nil {
			continue
		}
		err := w.emit(c.event(), c)
		w.mu.Lock()
		if err != nil {
			// Tried again at the next look: it stays out of the said set.
			f := w.failed[c.identity()]
			if f.since.IsZero() {
				f.since = now
			}
			f.closure, f.err = c, err.Error()
			w.failed[c.identity()] = f
			w.mu.Unlock()
			fmt.Fprintf(os.Stderr, "%s for %s not published, tried again at the next look: %v\n", c.event(), c.User, err)
			continue
		}
		delete(w.failed, c.identity())
		w.said[c.identity()] = c.Closed
		w.saidN[c.event()]++
		w.newest = append(w.newest, c)
		if len(w.newest) > 10 {
			w.newest = w.newest[len(w.newest)-10:]
		}
		w.mu.Unlock()
	}
	w.beat()
}

// Beat is the watch's heartbeat as it publishes it.
type Beat struct {
	At       time.Time  `json:"at"`
	Healthy  bool       `json:"healthy"`
	Why      string     `json:"why,omitempty"`
	LastRead *time.Time `json:"last_read,omitempty"`
	Every    string     `json:"every"`
}

// beat says the watch is there, and whether it is healthy. One that cannot be published is the controller's to
// notice by its absence.
func (w *Watch) beat() {
	if w.emit == nil {
		return
	}
	w.mu.Lock()
	healthy, why := w.healthLocked()
	b := Beat{At: w.now().UTC(), Healthy: healthy, Why: why, Every: WatchEvery.String()}
	if !w.lastRead.IsZero() {
		at := w.lastRead.UTC()
		b.LastRead = &at
	}
	w.mu.Unlock()
	if err := w.emit(EventWatch, b); err != nil {
		fmt.Fprintf(os.Stderr, "the watch's heartbeat not published: %v\n", err)
	}
}

// Run is the watch's life: a look every WatchEvery until the context ends. A look that faults is recovered in
// Look, so the loop never ends but with the context.
func (w *Watch) Run(ctx context.Context) {
	// The first look one interval in, not at once: the runtime that publishes for this bundle answers only
	// once the bundle serves, and the first look reads the last hour back anyway.
	t := time.NewTicker(WatchEvery)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}
		w.Look(ctx)
	}
}

// WatchStale is how long the watch may go without reading the record whole, or keep a closure unpublished,
// before it is not healthy: three looks.
const WatchStale = 3 * WatchEvery

// Health is the watch's own verdict: healthy while it has read the record whole within WatchStale (after the
// same grace from its start) and holds no closure unpublished for longer. Said in every heartbeat, so the
// controller raises a watch that cannot read (novox/hq ADR 0357, review 274).
func (w *Watch) Health() (bool, string) {
	w.mu.Lock()
	defer w.mu.Unlock()
	return w.healthLocked()
}

func (w *Watch) healthLocked() (bool, string) {
	now := w.now()
	switch {
	case w.lastRead.IsZero() && now.Sub(w.started) > WatchStale:
		why := fmt.Sprintf("the watch of slow consumers has not read the bus's closed connections since it started %s ago",
			now.Sub(w.started).Round(time.Second))
		if w.lastErr != "" {
			why += ": " + w.lastErr
		}
		return false, why
	case !w.lastRead.IsZero() && now.Sub(w.lastRead) > WatchStale:
		why := fmt.Sprintf("the watch of slow consumers last read the bus's closed connections %s ago",
			now.Sub(w.lastRead).Round(time.Second))
		if w.lastErr != "" {
			why += ": " + w.lastErr
		}
		return false, why
	}
	for _, f := range w.failed {
		if now.Sub(f.since) > WatchStale {
			return false, fmt.Sprintf("the watch could not publish %s for %s for %s: %s", f.closure.event(),
				f.closure.User, now.Sub(f.since).Round(time.Second), f.err)
		}
	}
	return true, ""
}

// Report is the watch as the tool `nats_bus_watch` shows it.
func (w *Watch) Report() map[string]any {
	w.mu.Lock()
	defer w.mu.Unlock()
	healthy, why := w.healthLocked()
	out := map[string]any{
		"healthy":     healthy,
		"every":       WatchEvery.String(),
		"reads_back":  WatchSaysAgain.String(),
		"events":      []string{EventSlowConsumer, EventConnectionClosed},
		"said":        w.saidN,
		"remembered":  len(w.said),
		"newest_said": w.newest,
		"not_said_reasons": func() []string {
			var r []string
			for k := range notSaid {
				r = append(r, k)
			}
			sort.Strings(r)
			return r
		}(),
	}
	if why != "" {
		out["why"] = why
	}
	if w.faults > 0 {
		out["faults_recovered"], out["last_fault"] = w.faults, w.lastFault
	}
	if !w.lastRead.IsZero() {
		out["last_read"] = w.lastRead.UTC()
	}
	if !w.lastLook.IsZero() {
		out["last_look"] = w.lastLook.UTC()
	}
	if w.lastErr != "" {
		out["last_error"] = w.lastErr
	}
	if len(w.failed) > 0 {
		failed := map[string]string{}
		for id, f := range w.failed {
			failed[id] = f.err
		}
		out["not_published"] = failed
	}
	return out
}