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"
)

// 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 tool that reports it.
type Watch struct {
	monitor Monitor
	emit    func(string, any) error
	now     func() time.Time

	mu sync.Mutex
	// said is every closure said, by identity, with when it closed: forgotten once older than the look's
	// window, since the record will not be read that far back again.
	said     map[string]time.Time
	lastLook time.Time
	lastErr  string
	saidN    map[string]int
	failed   map[string]string
	newest   []Closure
	stopped  string
}

// 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, said: map[string]time.Time{},
		saidN: map[string]int{}, failed: map[string]string{}}
}

// Look reads the record once and says every closure not said yet.
func (w *Watch) Look(ctx context.Context) {
	now := w.now()
	// The first look says the last hour again; every later one reads that far back too, so a closure whose
	// publish failed is still in reach, and what is said is told apart by the said set, not by time.
	since := now.Add(-WatchSaysAgain)
	read, cancel := context.WithTimeout(ctx, 20*time.Second)
	defer cancel()
	connz, err := w.monitor(read, fmt.Sprintf("/connz?state=closed&auth=1&limit=%d", maxClosedKept))
	var closures []Closure
	if err == nil {
		closures, err = Closures(connz, since)
	}
	w.mu.Lock()
	w.lastLook = now
	if err != nil {
		w.lastErr = err.Error()
		w.mu.Unlock()
		fmt.Fprintf(os.Stderr, "the watch could not read the bus's closed connections: %v\n", err)
		return
	}
	w.lastErr = ""
	for id, closed := range w.said {
		if closed.Before(since) {
			delete(w.said, id)
		}
	}
	var todo []Closure
	for _, c := range closures {
		if _, done := w.said[c.identity()]; !done {
			todo = append(todo, c)
		}
	}
	w.mu.Unlock()

	for _, c := range todo {
		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.
			w.failed[c.identity()] = err.Error()
			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()
	}
}

// Run is the watch's life: a look every WatchEvery until the context ends.
func (w *Watch) Run(ctx context.Context) {
	defer func() {
		if r := recover(); r != nil {
			w.mu.Lock()
			w.stopped = fmt.Sprintf("the watch stopped on a fault: %v", r)
			w.mu.Unlock()
			fmt.Fprintln(os.Stderr, w.stopped)
		}
	}()
	// 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)
	}
}

// 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()
	out := map[string]any{
		"running":     w.stopped == "",
		"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 w.stopped != "" {
		out["stopped"] = w.stopped
	}
	if !w.lastLook.IsZero() {
		out["last_look"] = w.lastLook.UTC()
	}
	if w.lastErr != "" {
		out["last_error"] = w.lastErr
	}
	if len(w.failed) > 0 {
		out["not_published"] = w.failed
	}
	return out
}