Files
mesh-catalog/modules/nats/cmd/nats-tools/tools.go
T
jschoubben 4fdb047cc8
mesh/merge-gate pass: builds nats, sent nowhere; merging this needs a person's bus upgrade; 1 wait(s) for a person; every machine composes with the change …
mesh/repo-check pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-check-typescript.sh (typescript)
mesh/delivery delivered
nats: say whether a user's grant is in force on the running server, not only written (hq issue 353)
nats_user_can read the user list on the machine. A list written is not a list the bus enforces until
the server reloads it: on 2026-10-09 the controller asked between the write and the reload's effect.
The answer now carries in_force — allowed in the file, and the server's config_load_time after the
file's write — with why when it is not, and anything not known is not in force.
2026-10-12 12:13:03 +02:00

575 lines
20 KiB
Go

package main
// What the bus's tools answer, shaped from the server's own monitoring documents. Pure: each takes the
// raw document and the caller's arguments, so every answer is tested against a captured document.
import (
"encoding/json"
"fmt"
"regexp"
"sort"
"strings"
"time"
)
// ---- the server ----------------------------------------------------------------------------------
// Server is what a person asks first: which server, how loaded, what it allows.
func Server(varz, healthz []byte) (map[string]any, error) {
var v map[string]any
if err := json.Unmarshal(varz, &v); err != nil {
return nil, err
}
out := map[string]any{}
for _, k := range []string{"version", "go", "start", "uptime", "max_payload", "max_connections", "max_pending",
"connections", "total_connections", "subscriptions", "slow_consumers", "in_msgs", "out_msgs", "in_bytes",
"out_bytes", "mem", "cpu", "cores"} {
if val, ok := v[k]; ok {
out[k] = val
}
}
if js, ok := v["jetstream"].(map[string]any); ok {
if stats, ok := js["stats"].(map[string]any); ok {
out["jetstream"] = stats
}
}
var h map[string]any
if json.Unmarshal(healthz, &h) == nil {
out["health"] = h
}
return out, nil
}
// ---- connections ---------------------------------------------------------------------------------
// Connection is one client, named by the mesh user it authenticated as.
type Connection struct {
CID int64 `json:"cid"`
User string `json:"user"`
Name string `json:"name"`
IP string `json:"ip"`
Uptime string `json:"uptime"`
Idle string `json:"idle"`
RTT string `json:"rtt"`
InMsgs int64 `json:"in_msgs"`
OutMsgs int64 `json:"out_msgs"`
InBytes int64 `json:"in_bytes"`
OutBytes int64 `json:"out_bytes"`
PendingBytes int64 `json:"pending_bytes"`
Subscriptions int64 `json:"subscriptions"`
}
func connectionsOf(connz []byte) ([]Connection, error) {
var doc struct {
Connections []struct {
Connection
AuthorizedUser string `json:"authorized_user"`
} `json:"connections"`
}
if err := json.Unmarshal(connz, &doc); err != nil {
return nil, err
}
out := make([]Connection, 0, len(doc.Connections))
for _, c := range doc.Connections {
conn := c.Connection
conn.User = c.AuthorizedUser
out = append(out, conn)
}
return out, nil
}
// Connections is every client, narrowed to users containing `user`, sorted by `by` (out_bytes,
// in_bytes, pending, msgs, subscriptions), at most `limit`.
func Connections(connz []byte, user, by string, limit int) (map[string]any, error) {
all, err := connectionsOf(connz)
if err != nil {
return nil, err
}
var out []Connection
for _, c := range all {
if user == "" || strings.Contains(c.User, user) || strings.Contains(c.Name, user) {
out = append(out, c)
}
}
key := map[string]func(Connection) int64{
"in_bytes": func(c Connection) int64 { return c.InBytes }, "pending": func(c Connection) int64 { return c.PendingBytes },
"msgs": func(c Connection) int64 { return c.InMsgs + c.OutMsgs },
"subscriptions": func(c Connection) int64 { return c.Subscriptions },
}[by]
if key == nil {
key = func(c Connection) int64 { return c.OutBytes }
}
sort.SliceStable(out, func(i, j int) bool { return key(out[i]) > key(out[j]) })
total := len(out)
if limit > 0 && len(out) > limit {
out = out[:limit]
}
return map[string]any{"total": total, "connections": out}, nil
}
// ---- closed connections --------------------------------------------------------------------------
// ClientClosed is the reason the server gives a connection its client closed itself; every other reason
// is the connection dropped: read or write errors, a stale connection, a slow consumer, a refused login,
// the server shutting down.
const ClientClosed = "Client Closed"
// ClosedName is one connection name of one user, as the server's record of closed connections has it.
type ClosedName struct {
Name string `json:"name"`
// Closed is how many connections of this name closed in the window; Dropped how many of them the
// client did not close itself.
Closed int `json:"closed"`
Dropped int `json:"dropped"`
Reasons map[string]int `json:"reasons"`
// Lived is the median time a connection of this name stayed open, in seconds.
Lived float64 `json:"lived_seconds"`
Last time.Time `json:"last"`
}
// ClosedUser is one bus user's closed connections in the window, per hour, and by name.
type ClosedUser struct {
User string `json:"user"`
Closed int `json:"closed"`
Dropped int `json:"dropped"`
ClosedPerHour float64 `json:"closed_per_hour"`
DroppedPerHour float64 `json:"dropped_per_hour"`
Names []ClosedName `json:"names"`
}
// ClosedConnections is who opened and closed connections to the bus in the last `hours` hours (novox/hq
// issue 327): per user and connection name, how many closed, how many were dropped rather than closed by
// their client, and why, from the server's own record of closed connections (`/connz?state=closed`). The
// server keeps the newest ten thousand: `reaches` says whether its record goes back the whole window.
func ClosedConnections(connz []byte, now time.Time, hours float64, user string) (map[string]any, error) {
var doc struct {
Connections []struct {
Name string `json:"name"`
AuthorizedUser string `json:"authorized_user"`
Start time.Time `json:"start"`
Stop *time.Time `json:"stop"`
Reason string `json:"reason"`
} `json:"connections"`
}
if err := json.Unmarshal(connz, &doc); err != nil {
return nil, err
}
if hours <= 0 {
hours = 1
}
since := now.Add(-time.Duration(hours * float64(time.Hour)))
type nameKey struct{ user, name string }
names := map[nameKey]*ClosedName{}
lived := map[nameKey][]float64{}
var oldest time.Time
for _, c := range doc.Connections {
if c.Stop == nil {
continue
}
if oldest.IsZero() || c.Stop.Before(oldest) {
oldest = *c.Stop
}
if c.Stop.Before(since) || user != "" && !strings.Contains(c.AuthorizedUser, user) && !strings.Contains(c.Name, user) {
continue
}
who := c.AuthorizedUser
if who == "" {
who = "(no user: refused before it logged in)"
}
k := nameKey{who, c.Name}
n := names[k]
if n == nil {
n = &ClosedName{Name: c.Name, Reasons: map[string]int{}}
names[k] = n
}
n.Closed++
reason := c.Reason
if reason == "" {
reason = "unsaid"
}
n.Reasons[reason]++
if reason != ClientClosed {
n.Dropped++
}
if c.Stop.After(n.Last) {
n.Last = *c.Stop
}
lived[k] = append(lived[k], c.Stop.Sub(c.Start).Seconds())
}
byUser := map[string]*ClosedUser{}
for k, n := range names {
durations := lived[k]
sort.Float64s(durations)
n.Lived = durations[len(durations)/2]
u := byUser[k.user]
if u == nil {
u = &ClosedUser{User: k.user}
byUser[k.user] = u
}
u.Closed += n.Closed
u.Dropped += n.Dropped
u.Names = append(u.Names, *n)
}
out := make([]ClosedUser, 0, len(byUser))
for _, u := range byUser {
u.ClosedPerHour = float64(u.Closed) / hours
u.DroppedPerHour = float64(u.Dropped) / hours
sort.Slice(u.Names, func(i, j int) bool {
if u.Names[i].Closed != u.Names[j].Closed {
return u.Names[i].Closed > u.Names[j].Closed
}
return u.Names[i].Name < u.Names[j].Name
})
out = append(out, *u)
}
sort.Slice(out, func(i, j int) bool {
if out[i].Closed != out[j].Closed {
return out[i].Closed > out[j].Closed
}
return out[i].User < out[j].User
})
return map[string]any{"since": since.UTC(), "hours": hours, "kept": len(doc.Connections),
"reaches": !oldest.IsZero() && !oldest.After(since) || len(doc.Connections) < maxClosedKept,
"users": out,
"note": "dropped is every connection its client did not close itself; a long-lived client dropped and " +
"back again is a reconnect, a client opening and closing at a steady pace is a client of short " +
"connections"}, nil
}
// maxClosedKept is how many closed connections the server keeps (nats-server's default max_closed_clients).
const maxClosedKept = 10000
// ---- subscriptions -------------------------------------------------------------------------------
// Subscriptions is who listens on what: subjects containing `subject`, with the user listening when
// connz names it, the busiest first.
func Subscriptions(subsz, connz []byte, subject string, limit int) (map[string]any, error) {
var doc struct {
Total int `json:"num_subscriptions"`
List []struct {
Subject string `json:"subject"`
Queue string `json:"qgroup"`
Msgs int64 `json:"msgs"`
CID int64 `json:"cid"`
} `json:"subscriptions_list"`
}
if err := json.Unmarshal(subsz, &doc); err != nil {
return nil, err
}
users := map[int64]string{}
if conns, err := connectionsOf(connz); err == nil {
for _, c := range conns {
users[c.CID] = c.User
}
}
type sub struct {
Subject string `json:"subject"`
Queue string `json:"queue,omitempty"`
Msgs int64 `json:"msgs"`
User string `json:"user,omitempty"`
}
var out []sub
for _, s := range doc.List {
if subject != "" && !strings.Contains(s.Subject, subject) {
continue
}
out = append(out, sub{Subject: s.Subject, Queue: s.Queue, Msgs: s.Msgs, User: users[s.CID]})
}
sort.SliceStable(out, func(i, j int) bool { return out[i].Msgs > out[j].Msgs })
matched := len(out)
if limit > 0 && len(out) > limit {
out = out[:limit]
}
return map[string]any{"subscriptions_on_server": doc.Total, "matched": matched, "subscriptions": out}, nil
}
// ---- JetStream -----------------------------------------------------------------------------------
type jszDoc struct {
Streams int `json:"streams"`
Consumers int `json:"consumers"`
Messages int64 `json:"messages"`
Bytes int64 `json:"bytes"`
Accounts []struct {
Streams []struct {
Name string `json:"name"`
Config struct {
Subjects []string `json:"subjects"`
Retention string `json:"retention"`
MaxAge int64 `json:"max_age"`
MaxMsgsPerSubject int64 `json:"max_msgs_per_subject"`
MaxBytes int64 `json:"max_bytes"`
Description string `json:"description"`
} `json:"config"`
State struct {
Messages int64 `json:"messages"`
Bytes int64 `json:"bytes"`
FirstSeq int64 `json:"first_seq"`
LastSeq int64 `json:"last_seq"`
LastTS string `json:"last_ts"`
NumSubjects int64 `json:"num_subjects"`
ConsumerCount int64 `json:"consumer_count"`
} `json:"state"`
Consumers []struct {
Name string `json:"name"`
Config struct {
FilterSubject string `json:"filter_subject"`
FilterSubjects []string `json:"filter_subjects"`
DeliverSubject string `json:"deliver_subject"`
MaxDeliver int64 `json:"max_deliver"`
} `json:"config"`
NumPending int64 `json:"num_pending"`
NumAckPending int64 `json:"num_ack_pending"`
NumRedelivered int64 `json:"num_redelivered"`
NumWaiting int64 `json:"num_waiting"`
Delivered struct {
LastActive string `json:"last_active"`
} `json:"delivered"`
} `json:"consumer_detail"`
} `json:"stream_detail"`
} `json:"account_details"`
}
// Streams is every stream with its retention and size — or one stream with every consumer's backlog.
func Streams(jsz []byte, stream string) (map[string]any, error) {
var doc jszDoc
if err := json.Unmarshal(jsz, &doc); err != nil {
return nil, err
}
var rows []map[string]any
for _, a := range doc.Accounts {
for _, s := range a.Streams {
if stream != "" && s.Name != stream {
continue
}
row := map[string]any{"name": s.Name, "subjects": s.Config.Subjects, "retention": s.Config.Retention,
"messages": s.State.Messages, "bytes": s.State.Bytes, "subjects_held": s.State.NumSubjects,
"consumers": s.State.ConsumerCount, "last": s.State.LastTS}
if s.Config.MaxAge > 0 {
row["max_age_hours"] = s.Config.MaxAge / 3_600_000_000_000
}
if s.Config.MaxMsgsPerSubject > 0 {
row["max_per_subject"] = s.Config.MaxMsgsPerSubject
}
if stream != "" {
row["description"] = s.Config.Description
var consumers []map[string]any
for _, c := range s.Consumers {
filters := c.Config.FilterSubjects
if c.Config.FilterSubject != "" {
filters = append(filters, c.Config.FilterSubject)
}
kind := "pull"
if c.Config.DeliverSubject != "" {
kind = "push"
}
consumers = append(consumers, map[string]any{"name": c.Name, "kind": kind, "filters": filters,
"pending": c.NumPending, "unacknowledged": c.NumAckPending, "redelivered": c.NumRedelivered,
"waiting_pulls": c.NumWaiting, "last_delivered": c.Delivered.LastActive, "max_deliver": c.Config.MaxDeliver})
}
sort.Slice(consumers, func(i, j int) bool { return consumers[i]["name"].(string) < consumers[j]["name"].(string) })
row["consumer_detail"] = consumers
}
rows = append(rows, row)
}
}
sort.Slice(rows, func(i, j int) bool { return rows[i]["name"].(string) < rows[j]["name"].(string) })
if stream != "" && len(rows) == 0 {
return nil, fmt.Errorf("no stream called %s", stream)
}
return map[string]any{"streams": doc.Streams, "consumers": doc.Consumers, "messages": doc.Messages,
"bytes": doc.Bytes, "detail": rows}, nil
}
// Backlog is every consumer with work it has not finished — pending or unacknowledged — the busiest first.
func Backlog(jsz []byte) (map[string]any, error) {
var doc jszDoc
if err := json.Unmarshal(jsz, &doc); err != nil {
return nil, err
}
var rows []map[string]any
for _, a := range doc.Accounts {
for _, s := range a.Streams {
for _, c := range s.Consumers {
if c.NumPending == 0 && c.NumAckPending == 0 {
continue
}
rows = append(rows, map[string]any{"stream": s.Name, "consumer": c.Name, "pending": c.NumPending,
"unacknowledged": c.NumAckPending, "redelivered": c.NumRedelivered, "last_delivered": c.Delivered.LastActive})
}
}
}
sort.Slice(rows, func(i, j int) bool {
return rows[i]["pending"].(int64)+rows[i]["unacknowledged"].(int64) > rows[j]["pending"].(int64)+rows[j]["unacknowledged"].(int64)
})
return map[string]any{"consumers_behind": len(rows), "backlog": rows}, nil
}
// Buckets is every module's state (novox/hq ADR 0201): the key-value buckets, with their keys and size.
func Buckets(jsz []byte) (map[string]any, error) {
var doc jszDoc
if err := json.Unmarshal(jsz, &doc); err != nil {
return nil, err
}
var rows []map[string]any
for _, a := range doc.Accounts {
for _, s := range a.Streams {
name, ok := strings.CutPrefix(s.Name, "KV_")
if !ok {
continue
}
module, state, _ := strings.Cut(name, "_")
rows = append(rows, map[string]any{"bucket": name, "module": module, "state": state,
"keys_including_deleted": s.State.NumSubjects, "values_kept": s.State.Messages, "bytes": s.State.Bytes,
"history": s.Config.MaxMsgsPerSubject, "last_change": s.State.LastTS, "watchers": s.State.ConsumerCount})
}
}
sort.Slice(rows, func(i, j int) bool { return rows[i]["bucket"].(string) < rows[j]["bucket"].(string) })
return map[string]any{"buckets": rows}, nil
}
// ---- users and grants ----------------------------------------------------------------------------
// User is one bus user's grants, as the controller composed them — never its password hash.
type User struct {
User string `json:"user"`
Publish []string `json:"publish"`
Subscribe []string `json:"subscribe"`
AllowResponses bool `json:"allow_responses"`
}
var (
userBlock = regexp.MustCompile(`\{\s*user:\s*"([^"]+)"`)
allowList = regexp.MustCompile(`(publish|subscribe):\s*\{\s*allow:\s*\[([^\]]*)\]`)
quoted = regexp.MustCompile(`"((?:[^"\\]|\\.)*)"`)
)
// ParseUsers reads the composed user list (the controller's broker.ComposeAccounts shape).
func ParseUsers(conf []byte) []User {
text := string(conf)
idx := userBlock.FindAllStringSubmatchIndex(text, -1)
var out []User
for i, m := range idx {
end := len(text)
if i+1 < len(idx) {
end = idx[i+1][0]
}
block := text[m[0]:end]
u := User{User: text[m[2]:m[3]], Publish: []string{}, Subscribe: []string{}, AllowResponses: strings.Contains(block, "allow_responses")}
for _, a := range allowList.FindAllStringSubmatch(block, -1) {
var subjects []string
for _, q := range quoted.FindAllStringSubmatch(a[2], -1) {
subjects = append(subjects, q[1])
}
if a[1] == "publish" {
u.Publish = subjects
} else {
u.Subscribe = subjects
}
}
out = append(out, u)
}
sort.Slice(out, func(i, j int) bool { return out[i].User < out[j].User })
return out
}
// Users is every user's grants, or those of users containing `user`; with `grants`, the subjects too.
func Users(conf []byte, user string) map[string]any {
all := ParseUsers(conf)
if user == "" {
type summary struct {
User string `json:"user"`
Publish int `json:"publish_grants"`
Subscribe int `json:"subscribe_grants"`
}
var out []summary
for _, u := range all {
out = append(out, summary{u.User, len(u.Publish), len(u.Subscribe)})
}
return map[string]any{"users": out}
}
var out []User
for _, u := range all {
if strings.Contains(u.User, user) {
out = append(out, u)
}
}
return map[string]any{"users": out}
}
// SubjectMatches says whether a concrete subject falls under a grant's pattern: `*` one token, `>` the rest.
func SubjectMatches(pattern, subject string) bool {
p, s := strings.Split(pattern, "."), strings.Split(subject, ".")
for i, tok := range p {
if tok == ">" {
return len(s) > i
}
if i >= len(s) || (tok != "*" && tok != s[i]) {
return false
}
}
return len(p) == len(s)
}
// UserCan answers whether a user may publish or subscribe to a subject, and which grant allows it.
func UserCan(conf []byte, user, action, subject string) (map[string]any, error) {
for _, u := range ParseUsers(conf) {
if u.User != user {
continue
}
grants := u.Publish
if action == "subscribe" {
grants = u.Subscribe
} else if action != "publish" {
return nil, fmt.Errorf("action is publish or subscribe, not %q", action)
}
for _, g := range grants {
if SubjectMatches(g, subject) {
return map[string]any{"user": user, "action": action, "subject": subject, "allowed": true, "by": g}, nil
}
}
answer := map[string]any{"user": user, "action": action, "subject": subject, "allowed": false}
if action == "publish" && u.AllowResponses {
answer["note"] = "a reply to a request this user received is allowed regardless (allow_responses)"
}
return answer, nil
}
return nil, fmt.Errorf("the bus has no user %q; nats_users lists them", user)
}
// InForce adds to a UserCan answer whether the running server holds what the file says (novox/hq issue 353):
// `in_force` only when the grant is allowed in the file and the server's last configuration load
// (`config_load_time` of /varz) came after the file was last written. Anything not known is not in force, said
// in `why`: an asker waiting for its grant must not take a list sent for a list the bus enforces.
func InForce(answer map[string]any, written time.Time, writtenErr error, varz []byte, varzErr error) map[string]any {
answer["in_force"] = false
switch {
case writtenErr != nil:
answer["why"] = writtenErr.Error()
return answer
case varzErr != nil:
answer["why"] = "the server's last configuration load could not be read: " + varzErr.Error()
return answer
}
var v struct {
ConfigLoadTime time.Time `json:"config_load_time"`
}
if err := json.Unmarshal(varz, &v); err != nil || v.ConfigLoadTime.IsZero() {
answer["why"] = "the server does not say when it last loaded its configuration"
return answer
}
answer["file_written"] = written.UTC().Format(time.RFC3339Nano)
answer["server_loaded"] = v.ConfigLoadTime.UTC().Format(time.RFC3339Nano)
loaded := !v.ConfigLoadTime.Before(written)
allowed, _ := answer["allowed"].(bool)
switch {
case !allowed:
answer["why"] = "the user list on the machine does not allow it"
case !loaded:
answer["why"] = "the user list on the machine allows it, and the server has not reloaded since it was written"
default:
answer["in_force"] = true
}
return answer
}