mesh-controller/cmd/mesh-controller/reconnects.go

package main

import (
	"context"
	"fmt"
	"sort"
	"strings"
	"time"

	"github.com/nats-io/nats.go"

	"github.com/novox/mesh-controller/internal/conditions"
	"github.com/novox/mesh-controller/internal/link"
)

// D15: no client of the bus reconnects in a loop (novox/hq issue 327).
//
// The server's connection total was the one number that would show a client reconnecting, and every verb
// the controller served opened and closed a connection of its own, hundreds an hour, so a loop was
// invisible in it. The bus's own module reads the server's record of closed connections
// (`nats_closed_connections`); this asks it every run, and says each user whose connections were dropped —
// closed by anything but the client itself: a read or write error, a stale connection, a slow consumer, a
// refused login — more often than reconnectBound in the last hour.

const (
	probeReconnectsID = "D15"
	kindBusReconnects = "bus-reconnects"
	// reconnectBound is how many dropped connections in an hour one user may have before it is said:
	// a client that loses its connection every five minutes. Provisional.
	reconnectBound = 12
	// busModule is the module that is the bus, and closedTool its tool that reads closed connections.
	busModule  = "nats"
	closedTool = "nats_closed_connections"
	closedAsk  = 20 * time.Second
)

// closedConnections is what the bus's module answers.
type closedConnections struct {
	Hours   float64 `json:"hours"`
	Reaches bool    `json:"reaches"`
	Users   []struct {
		User           string  `json:"user"`
		Closed         int     `json:"closed"`
		Dropped        int     `json:"dropped"`
		DroppedPerHour float64 `json:"dropped_per_hour"`
		Names          []struct {
			Name    string         `json:"name"`
			Closed  int            `json:"closed"`
			Dropped int            `json:"dropped"`
			Reasons map[string]int `json:"reasons"`
		} `json:"names"`
	} `json:"users"`
}

// probeReconnects is D15.
func probeReconnects(ctx context.Context, d *doctor) ([]conditions.Observation, error) {
	if d.js == nil {
		return nil, fmt.Errorf("no bus to ask the bus's module over")
	}
	on, err := d.open.inventory.Running(ctx, busModule)
	if err != nil {
		return nil, err
	}
	if len(on) == 0 {
		return nil, nil // no bus module assigned: a mesh whose bus is not the mesh's module
	}
	read, err := askClosed(ctx, d.js.Conn(), on[0])
	if isNothingServes(err) {
		// A bus module older than its tool has nothing it can say, which is not a failure of the probe.
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return reconnecting(read), nil
}

// askClosed asks the bus's module, on its machine, who closed connections in the last hour.
func askClosed(ctx context.Context, conn *nats.Conn, node string) (closedConnections, error) {
	var read closedConnections
	answer, err := link.AskModuleToolOn(ctx, conn, busModule, closedTool, node, map[string]any{"hours": 1}, closedAsk)
	if err != nil {
		return read, err
	}
	if answer.Error != "" {
		return read, fmt.Errorf("%s on %s answered %s with an error: %s", busModule, node, closedTool, answer.Error)
	}
	if err := unmarshalAnswer(answer, &read); err != nil {
		return read, fmt.Errorf("%s on %s answered %s with something unreadable: %w", busModule, node, closedTool, err)
	}
	return read, nil
}

// reconnecting is one observation per user whose connections were dropped more than reconnectBound times
// an hour.
func reconnecting(read closedConnections) []conditions.Observation {
	hours := read.Hours
	if hours <= 0 {
		hours = 1
	}
	var out []conditions.Observation
	for _, u := range read.Users {
		perHour := float64(u.Dropped) / hours
		if perHour <= reconnectBound {
			continue
		}
		reasons := map[string]int{}
		var names []string
		for _, n := range u.Names {
			if n.Dropped == 0 {
				continue
			}
			names = append(names, fmt.Sprintf("%q ×%d", n.Name, n.Dropped))
			for r, c := range n.Reasons {
				if r != "Client Closed" {
					reasons[r] += c
				}
			}
		}
		var why []string
		for r, c := range reasons {
			why = append(why, fmt.Sprintf("%s ×%d", r, c))
		}
		sort.Strings(why)
		partial := ""
		if !read.Reaches {
			partial = " (at least: the server's record of closed connections does not reach back the whole hour)"
		}
		who, machine := busUserWords(u.User)
		out = append(out, conditions.Observation{Scope: conditions.ScopeBus, ID: u.User, Kind: kindBusReconnects,
			Machine: machine, Severity: conditions.Warning,
			Summary: fmt.Sprintf("the bus dropped %s's connection %d times in the last hour%s, more than %d: a "+
				"client reconnecting in a loop", u.User, u.Dropped, partial, reconnectBound),
			Said: fmt.Sprintf("%d of %d closed connections dropped in %.0f h; by name %s; why %s", u.Dropped,
				u.Closed, hours, strings.Join(names, ", "), strings.Join(why, ", ")),
			Headline: clipWords(conditions.Capital(who)+" keeps losing the bus", 60),
			Explanation: conditions.Capital(fmt.Sprintf("%s lost its connection to the bus %d times in the last hour "+
				"and connected again each time. While it reconnects, what it says and what it is asked waits.", who,
				u.Dropped)),
			Resolved: "Resolved: " + who + " stays connected"})
	}
	return out
}

// busUserWords is a bus user as the operator says it, and the machine it is on: `node.<machine>` the
// node-engine, `<machine>.node-tools` the tool runner, `controller` the controller, `<machine>.<module>` a
// module.
func busUserWords(user string) (string, string) {
	switch {
	case user == "controller":
		return "the controller", ""
	case strings.HasPrefix(user, "node."):
		m := strings.TrimPrefix(user, "node.")
		return "the node-engine on " + m, m
	}
	if m, module, ok := strings.Cut(user, "."); ok {
		if module == "node-tools" {
			return "the tool runner on " + m, m
		}
		return module + " on " + m, m
	}
	return "a client of the bus", ""
}

// clipWords is words at most n characters long, cut at a word.
func clipWords(s string, n int) string {
	if len(s) <= n {
		return s
	}
	cut := s[:n]
	if i := strings.LastIndex(cut, " "); i > 0 {
		cut = cut[:i]
	}
	return cut
}