diff --git a/modules/nats/cmd/nats-tools/main.go b/modules/nats/cmd/nats-tools/main.go index 80e99d55..bafecedd 100644 --- a/modules/nats/cmd/nats-tools/main.go +++ b/modules/nats/cmd/nats-tools/main.go @@ -62,6 +62,17 @@ func tools(monitor Monitor, users func(context.Context) ([]byte, error)) []stdio } return Connections(connz, arg(a, "user"), arg(a, "sort"), limitOf(a, 50)) }}, + {Name: "nats_closed_connections", + Description: "Who opened and closed connections to the bus lately (novox/hq issue 327): per user and connection name, how many closed in the last hours, how many were dropped rather than closed by their client and why, how long each lived, and the rate per hour — the question behind a connection total that keeps rising while the open count stays flat. Narrowed by part of a user or connection name.", + Input: map[string]any{"hours": num("how many hours back (default 1)"), "user": str("part of a user or connection name, e.g. controller")}, + Run: func(a map[string]any) (any, error) { + connz, err := get(fmt.Sprintf("/connz?state=closed&auth=1&limit=%d", maxClosedKept)) + if err != nil { + return nil, err + } + hours, _ := a["hours"].(float64) + return ClosedConnections(connz, time.Now(), hours, arg(a, "user")) + }}, {Name: "nats_subscriptions", Description: "Who listens on what: every subscription whose subject contains the text given, with its queue group, the messages it received and the user holding it, the busiest first.", Input: map[string]any{"subject": str("part of a subject, e.g. mesh.seat.anthropic-licence-manager or $SRV"), "limit": num("at most this many (default 100)")}, diff --git a/modules/nats/cmd/nats-tools/server_test.go b/modules/nats/cmd/nats-tools/server_test.go index 773469e2..57ab02ee 100644 --- a/modules/nats/cmd/nats-tools/server_test.go +++ b/modules/nats/cmd/nats-tools/server_test.go @@ -2,10 +2,11 @@ package main import ( "fmt" + "io" "math/rand" + "net/http" "os" "regexp" - "runtime/debug" "sync" "testing" "time" @@ -26,19 +27,11 @@ func TestTheImageIsTheServerTestedHere(t *testing.T) { if said == nil { t.Fatal("the Dockerfile does not say which nats release its digest is (`# upstream: nats X.Y.Z-alpine`)") } - info, ok := debug.ReadBuildInfo() - if !ok { - t.Fatal("no build information to read the tested server's version from") + // The server's own word for its release, compiled into it: a test binary's build information may list + // no dependencies (the build seat's did not, so this read "these tests run no nats-server"). + if server.VERSION != string(said[1]) { + t.Fatalf("the image runs nats %s and these tests run %s: move them together", said[1], server.VERSION) } - for _, dep := range info.Deps { - if dep.Path == "github.com/nats-io/nats-server/v2" { - if dep.Version != "v"+string(said[1]) { - t.Fatalf("the image runs nats %s and these tests run %s: move them together", said[1], dep.Version) - } - return - } - } - t.Fatal("these tests run no nats-server") } // **A consumer with several filters is handed every message** (novox/hq issue 266). @@ -167,3 +160,51 @@ func TestAConsumerWithSeveralFiltersIsHandedEveryMessage(t *testing.T) { time.Sleep(200 * time.Millisecond) } } + +// The record of closed connections is the server's own, read through its monitoring endpoint as the tool +// reads it (novox/hq issue 327): two connections a client opened and closed are both there, by user and +// name, closed by their client. +func TestTheServersRecordOfClosedConnectionsIsRead(t *testing.T) { + s, err := server.NewServer(&server.Options{Port: -1, HTTPPort: -1, NoLog: true, NoSigs: true, + Users: []*server.User{{Username: "controller", Password: "x"}}}) + if err != nil { + t.Fatal(err) + } + go s.Start() + if !s.ReadyForConnections(10 * time.Second) { + t.Fatal("no server") + } + defer s.Shutdown() + for i := 0; i < 2; i++ { + nc, err := nats.Connect(s.ClientURL(), nats.UserInfo("controller", "x"), nats.Name("mesh-controller verb conditions")) + if err != nil { + t.Fatal(err) + } + nc.Close() + } + monitor := func(path string) []byte { + resp, err := http.Get("http://" + s.MonitorAddr().String() + path) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + return body + } + var users []ClosedUser + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + out, err := ClosedConnections(monitor("/connz?state=closed&auth=1&limit=10000"), time.Now(), 1, "") + if err != nil { + t.Fatal(err) + } + if users = out["users"].([]ClosedUser); len(users) == 1 && users[0].Closed == 2 { + break + } + time.Sleep(50 * time.Millisecond) + } + if len(users) != 1 || users[0].User != "controller" || users[0].Closed != 2 || users[0].Dropped != 0 || + users[0].Names[0].Name != "mesh-controller verb conditions" { + t.Fatalf("%+v", users) + } +} diff --git a/modules/nats/cmd/nats-tools/tools.go b/modules/nats/cmd/nats-tools/tools.go index 15f06486..af0c725a 100644 --- a/modules/nats/cmd/nats-tools/tools.go +++ b/modules/nats/cmd/nats-tools/tools.go @@ -9,6 +9,7 @@ import ( "regexp" "sort" "strings" + "time" ) // ---- the server ---------------------------------------------------------------------------------- @@ -106,6 +107,138 @@ func Connections(connz []byte, user, by string, limit int) (map[string]any, erro 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 diff --git a/modules/nats/cmd/nats-tools/tools_test.go b/modules/nats/cmd/nats-tools/tools_test.go index 677510b0..786487cc 100644 --- a/modules/nats/cmd/nats-tools/tools_test.go +++ b/modules/nats/cmd/nats-tools/tools_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "strings" "testing" + "time" ) const connz = `{"connections":[ @@ -125,3 +126,47 @@ func TestWhetherAUserMayDoSomething(t *testing.T) { } } } + +// novox/hq issue 327: who opens and closes connections, per user and name, and which were dropped rather +// than closed by their client — read from the server's record of closed connections. +func TestClosedConnectionsArePerUserAndNameAndSayWhichWereDropped(t *testing.T) { + const closed = `{"connections":[ + {"cid":10,"name":"mesh-controller verb conditions","authorized_user":"controller","start":"2026-10-08T14:12:00Z","stop":"2026-10-08T14:12:01Z","reason":"Client Closed"}, + {"cid":11,"name":"mesh-controller verb conditions","authorized_user":"controller","start":"2026-10-08T14:13:00Z","stop":"2026-10-08T14:13:03Z","reason":"Client Closed"}, + {"cid":12,"name":"ace.node-tools","authorized_user":"ace.node-tools","start":"2026-10-08T13:00:00Z","stop":"2026-10-08T14:14:00Z","reason":"Stale Connection"}, + {"cid":13,"name":"ace.node-tools","authorized_user":"ace.node-tools","start":"2026-10-08T14:14:05Z","stop":"2026-10-08T14:15:00Z","reason":"Read Error"}, + {"cid":5,"name":"old","authorized_user":"controller","start":"2026-10-08T10:00:00Z","stop":"2026-10-08T10:00:01Z","reason":"Client Closed"}, + {"cid":14,"name":"","start":"2026-10-08T14:16:00Z","stop":"2026-10-08T14:16:00Z","reason":"Authentication Failure"}]}` + now, _ := time.Parse(time.RFC3339, "2026-10-08T14:20:00Z") + out, err := ClosedConnections([]byte(closed), now, 1, "") + if err != nil { + t.Fatal(err) + } + users := out["users"].([]ClosedUser) + if len(users) != 3 { + t.Fatalf("%+v", users) + } + byUser := map[string]ClosedUser{} + for _, u := range users { + byUser[u.User] = u + } + c := byUser["controller"] + if c.Closed != 2 || c.Dropped != 0 || c.ClosedPerHour != 2 || len(c.Names) != 1 || + c.Names[0].Name != "mesh-controller verb conditions" || c.Names[0].Lived != 3 { + t.Errorf("the controller's: %+v", c) + } + tools := byUser["ace.node-tools"] + if tools.Closed != 2 || tools.Dropped != 2 || tools.Names[0].Reasons["Stale Connection"] != 1 { + t.Errorf("the runtime's: %+v", tools) + } + if byUser["(no user: refused before it logged in)"].Dropped != 1 { + t.Errorf("a refused login is not said: %+v", users) + } + if out["reaches"] != true { + t.Error("a record older than the window does not reach back the whole window") + } + narrowed, _ := ClosedConnections([]byte(closed), now, 1, "node-tools") + if got := narrowed["users"].([]ClosedUser); len(got) != 1 || got[0].User != "ace.node-tools" { + t.Errorf("narrowed: %+v", got) + } +} diff --git a/modules/nats/module.json b/modules/nats/module.json index 7ecb8b32..d8647b3e 100644 --- a/modules/nats/module.json +++ b/modules/nats/module.json @@ -38,6 +38,7 @@ "tools": [ "nats_server", "nats_connections", + "nats_closed_connections", "nats_subscriptions", "nats_streams", "nats_backlog",