package main // Reaching the bus server's own monitoring API (the HTTP endpoints nats-server serves: varz, connz, // subsz, jsz, healthz). Read-only by the server's own design: nothing here can change the bus. // // **Two ways in, the first that answers.** The module's configuration binds the endpoint inside its // container; published on the machine's loopback, a host-side client reaches it directly — and where the // binding is the container's own loopback (as the module shipped until this bundle), only a process inside // the container can. So the endpoint is asked directly first, and through the container second; the day // the binding is moved, the second way simply stops being used. import ( "context" "fmt" "io" "net/http" "os" "os/exec" "strings" "time" ) // Monitor fetches a monitoring path, e.g. "/varz". type Monitor func(ctx context.Context, path string) ([]byte, error) func env(key, fallback string) string { if v := strings.TrimSpace(os.Getenv(key)); v != "" { return v } return fallback } // LiveMonitor is the server as this machine reaches it. func LiveMonitor() Monitor { base := env("MESH_NATS_MONITOR", "http://127.0.0.1:8222") container := env("MESH_NATS_CONTAINER", "mesh-broker-nats") client := &http.Client{Timeout: 5 * time.Second} return func(ctx context.Context, path string) ([]byte, error) { req, err := http.NewRequestWithContext(ctx, http.MethodGet, base+path, nil) if err == nil { if resp, err := client.Do(req); err == nil { defer resp.Body.Close() if resp.StatusCode == http.StatusOK { return io.ReadAll(io.LimitReader(resp.Body, 64<<20)) } } } // Through the container: its own loopback is where the endpoint listens. out, err := exec.CommandContext(ctx, "docker", "exec", container, "wget", "-qO-", "http://127.0.0.1:8222"+path).Output() if err != nil { return nil, fmt.Errorf("the bus's monitoring endpoint answered neither at %s nor inside %s: %w", base, container, err) } return out, nil } } // ReadUsers reads the composed user list the server includes: in the container, where it is mounted // read-only; the machine's copy is readable by root alone. func ReadUsers(ctx context.Context) ([]byte, error) { container := env("MESH_NATS_CONTAINER", "mesh-broker-nats") path := env("MESH_NATS_USERS_FILE", "/etc/nats/accounts.conf") out, err := exec.CommandContext(ctx, "docker", "exec", container, "cat", path).Output() if err != nil { return nil, fmt.Errorf("the bus's user list could not be read inside %s: %w", container, err) } return out, nil } // UsersWritten is when the composed user list the server includes was last written, read in the container // to the nanosecond (`stat -c %y`): what the server's last reload must come after for the list to be in force. func UsersWritten(ctx context.Context) (time.Time, error) { container := env("MESH_NATS_CONTAINER", "mesh-broker-nats") path := env("MESH_NATS_USERS_FILE", "/etc/nats/accounts.conf") out, err := exec.CommandContext(ctx, "docker", "exec", container, "stat", "-c", "%y", path).Output() if err != nil { return time.Time{}, fmt.Errorf("when the bus's user list was written could not be read inside %s: %w", container, err) } return ParseWritten(string(out)) } // ParseWritten reads `stat -c %y`: "2026-10-09 17:38:13.422211788 +0000". func ParseWritten(s string) (time.Time, error) { t, err := time.Parse("2006-01-02 15:04:05.999999999 -0700", strings.TrimSpace(s)) if err != nil { return time.Time{}, fmt.Errorf("when the bus's user list was written is not a time: %q", strings.TrimSpace(s)) } return t, nil }