From f6cc5da8de0f5f5d9615a0065662936470910d6e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bartosz=20Pacho=C5=82ek?= Date: Mon, 28 Sep 2026 00:10:36 +0200 Subject: [PATCH] Add health_details: server and stream details in the readiness probes (#144) To learn which nats-server a NUTS instance is connected to, or what its stream's limits and state are, an operator had to query NATS separately. Some behaviour now depends on the server version (a restore keeps or replaces the stream's creation time depending on the server), so seeing it from NUTS helps triage. health_details adds nats_server (version, name, cluster, domain) and stream_info (subjects, storage, replicas, message and byte counts, first and last sequence, consumers, limits, creation time) to the readiness probes' JSON. The stream details come from the read the probe already makes, so there is no extra request. nats_server is left out while disconnected, and stream_info when the stream cannot be read. Off by default: probe paths are often public, and the details name servers, clusters and limits. Closes #144 Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 6 +++ Caddyfile.test | 16 +++++++ README.md | 20 ++++++++ caddyfile.go | 11 +++++ caddyfile_test.go | 14 ++++++ docs/CONFIGURATION.md | 1 + features/sse_streaming.feature | 13 +++++ functional_test/steps_test.go | 8 ++++ handler.go | 7 +++ serve.go | 66 ++++++++++++++++++++++++- serve_integration_test.go | 88 ++++++++++++++++++++++++++++++++++ website/docs/usage.md | 1 + 12 files changed, 250 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ee0549e..d1681e5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- `health_details` adds the NATS server the handler is connected to + (version, name, cluster, domain) and the stream as the probe read it + (subjects, storage, replicas, message and byte counts, first and last + sequence, consumers, limits, creation time) to the readiness probes' JSON + (#144). Off by default: probe paths are often public, and the details name + servers, clusters and limits. - `payload_format raw` sends the NATS payload itself as each event's data, without the JSON envelope: a `data:` line per line of it, which EventSource joins back (#143). A payload SSE cannot carry, not UTF-8 text diff --git a/Caddyfile.test b/Caddyfile.test index 6371eda..5cd6f38 100644 --- a/Caddyfile.test +++ b/Caddyfile.test @@ -93,6 +93,22 @@ } } + # The same stream whose readiness probes add the NATS server and the + # stream's details, for features/sse_streaming.feature. + handle /details* { + uri strip_prefix /details + nuts { + nats_url {$NATS_URL:nats://localhost:4222} + stream_name {$STREAM_NAME:EVENTS} + topic_prefix {$TOPIC_PREFIX:events.} + heartbeat_interval 1 + nats_idle_heartbeat 1 + nats_ping_interval 1 + allowed_origins * + health_details + } + } + handle /example* { root * /srv/example uri strip_prefix /example diff --git a/README.md b/README.md index 4e81d1e..a6c6af5 100644 --- a/README.md +++ b/README.md @@ -385,6 +385,7 @@ nuts { event_id_format # Event ids: sequence (default) or sequence_time (sequence and message time) event_type [name] # Event names: message (default), topic, or header payload_format # Event data: envelope (default) or raw (the NATS payload itself) + health_details [true|false] # Readiness probes add the NATS server and stream details (default: off) health_path # Legacy readiness endpoint (empty/default: /healthz) live_path # Process liveness endpoint (empty/default: /livez) ready_path # NATS/stream readiness endpoint (empty/default: /readyz) @@ -937,6 +938,25 @@ availability: {"status":"degraded","nats":"disconnected","stream":"unavailable"} ``` +With `health_details`, the readiness and legacy health endpoints also report +the NATS server the handler is connected to and the stream as the probe read +it, which helps triage without querying NATS separately (#144): + +```json +{"status":"ok","nats":"connected","stream":"available", + "nats_server":{"version":"2.15.0","name":"nats-1","cluster":"east"}, + "stream_info":{"name":"EVENTS","subjects":["events.>"],"storage":"file", + "replicas":3,"messages":1523,"bytes":210432,"first_seq":1,"last_seq":1523, + "consumers":12,"max_msgs":10000,"max_bytes":-1,"max_age_seconds":3600, + "max_consumers":-1,"created":"2026-09-27T10:00:00Z"}} +``` + +`nats_server` is left out while NATS is disconnected, and `stream_info` when +the stream cannot be read. The details are off by default: probe paths are +often reachable by load balancers or the public, and they name servers, +clusters and limits. Turn them on for routes that only trusted callers reach, +or restrict the probe paths with Caddy matchers. + Operational runbooks and Kubernetes probe examples are in [docs/OPERATIONS.md](docs/OPERATIONS.md). diff --git a/caddyfile.go b/caddyfile.go index 5e0d431..dcda788 100644 --- a/caddyfile.go +++ b/caddyfile.go @@ -302,6 +302,17 @@ func (h *Handler) UnmarshalCaddyfile(d *caddyfile.Dispenser) error { } h.EventIDFormat = d.Val() + case "health_details": + if d.NextArg() { + b, err := strconv.ParseBool(d.Val()) + if err != nil { + return d.Errf("invalid health_details: %v", err) + } + h.HealthDetails = b + } else { + h.HealthDetails = true + } + case "payload_format": if !d.NextArg() { return d.ArgErr() diff --git a/caddyfile_test.go b/caddyfile_test.go index 449c384..6460bc8 100644 --- a/caddyfile_test.go +++ b/caddyfile_test.go @@ -16,6 +16,13 @@ func TestHandler_UnmarshalCaddyfile(t *testing.T) { expected *Handler expectError bool }{ + { + name: "health_details with an explicit value", + caddyfile: `nuts { + health_details true + }`, + expected: &Handler{HealthDetails: true}, + }, { name: "full configuration", caddyfile: `nuts { @@ -47,6 +54,7 @@ func TestHandler_UnmarshalCaddyfile(t *testing.T) { event_id_format sequence_time event_type header Event-Type payload_format raw + health_details subscriber_jwt_key secret-key subscriber_jwt_cookie nuts_session allowed_origins https://example.com https://other.com @@ -83,6 +91,7 @@ func TestHandler_UnmarshalCaddyfile(t *testing.T) { EventType: "header", EventTypeHeader: "Event-Type", PayloadFormat: "raw", + HealthDetails: true, SubscriberJWTKey: "secret-key", SubscriberJWTCookie: "nuts_session", AllowedOrigins: []string{"https://example.com", "https://other.com"}, @@ -599,6 +608,11 @@ func TestHandler_UnmarshalCaddyfile_RejectsInvalidOptionalConfig(t *testing.T) { line: "event_type topic extra", wantErr: "argument", }, + { + name: "health_details that is not a boolean", + line: "health_details sometimes", + wantErr: "invalid health_details", + }, { name: "unknown payload_format", line: "payload_format text", diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 2d6e321..0a76823 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -76,6 +76,7 @@ enforced. For public or browser-facing routes, issue short-lived tokens with | `replay_max_messages ` | `replay_max_messages` | `0` | Integer `>= 0` | `0` is unlimited. When reached during replay, NUTS closes the stream cleanly and the client reconnects from its last event ID. Only history counts: messages published after the request arrived never trip the cap. | | `event_type [header name]` | `event_type` | `message` | `message`, `topic`, or `header` followed by a header name | The SSE event name of each message: `message`, its topic without `topic_prefix`, or the value of a NATS message header. A name outside `[A-Za-z0-9._:-]{1,100}`, a reserved name (`connected`, `reset`, `open`, `error`) or a missing header falls back to `message`. | | (the name after `event_type header`) | `event_type_header` | Empty | Printable ASCII without spaces or colons | The header `event_type header` reads; case-sensitive, as NATS headers are. Rejected unless `event_type` is `header`. | +| `health_details [bool]` | `health_details` | `false` | No argument means `true`; optional boolean accepted | The readiness probes (`ready_path`, `health_path`) add `nats_server` (version, name, cluster, domain) and `stream_info` (subjects, storage, replicas, message and byte counts, first and last sequence, consumers, limits, creation time). Off by default because probe paths are often public and the details name servers, clusters and limits. | | `payload_format ` | `payload_format` | `envelope` | `envelope` or `raw` | What an event's data holds: the JSON envelope `{"topic","payload","time"}`, or the NATS payload itself, a `data:` line per line of it. In raw mode a payload that is not UTF-8 text, or holds a carriage return, is dropped and counted as `nuts_messages_dropped_total{reason="raw_not_text"}`. | | `event_id_format ` | `event_id_format` | `sequence` | `sequence` or `sequence_time` | What an event's id holds: the message's stream sequence, or the sequence and the stored time of the message, `-`. With `sequence_time`, a client that resumes with a cursor from a stream that was since recreated or restored replays the stream from its start instead of skipping its first messages; both forms are accepted as cursors either way. Enable it only for clients that treat ids as opaque strings. | | `replay_window ` | `replay_window` | `0` | Integer `>= 0` | `0` preserves retained replay. Positive values bound old replay cursors to `StartTime(now - replay_window)` while preserving exact sequence replay inside the window. Also applies when stream info cannot be read or the cursor is ahead of the stream. Live messages are never filtered. | diff --git a/features/sse_streaming.feature b/features/sse_streaming.feature index 1471a38..1669fc9 100644 --- a/features/sse_streaming.feature +++ b/features/sse_streaming.feature @@ -69,6 +69,19 @@ Feature: SSE Streaming with JetStream | /events/readyz/ | available | | /events/healthz/ | connected | + # health_details adds the NATS server and the stream's details to the + # readiness probes (#144); without it they stay terse. + Scenario: Readiness details are opt-in + When I request SSE endpoint "/details/readyz" + Then I should receive HTTP status 200 + And the response should contain "nats_server" + And the response should contain "version" + And the response should contain "stream_info" + And the response should contain "EVENTS" + When I request SSE endpoint "/events/readyz" + Then I should receive HTTP status 200 + And the response should not contain "nats_server" + Scenario: A backlog larger than the prefetch replays on one connection # client_buffer_size is 64 by default; the stream stops pulling while the # client catches up instead of disconnecting it. diff --git a/functional_test/steps_test.go b/functional_test/steps_test.go index a5bf0fe..b3a8057 100644 --- a/functional_test/steps_test.go +++ b/functional_test/steps_test.go @@ -706,6 +706,13 @@ func theResponseShouldContain(text string) error { return nil } +func theResponseShouldNotContain(text string) error { + if strings.Contains(tc.httpBody, text) { + return fmt.Errorf("response contains %q: %s", text, tc.httpBody) + } + return nil +} + func iSendOPTIONSRequestToWithOrigin(endpoint, origin string) error { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() @@ -1194,6 +1201,7 @@ func InitializeScenario(ctx *godog.ScenarioContext) { ctx.Step(`^the connected event should list topic "([^"]*)"$`, theConnectedEventShouldListTopic) ctx.Step(`^I should receive HTTP status (\d+)$`, iShouldReceiveHTTPStatus) ctx.Step(`^the response should contain "([^"]*)"$`, theResponseShouldContain) + ctx.Step(`^the response should not contain "([^"]*)"$`, theResponseShouldNotContain) ctx.Step(`^the response header "([^"]*)" should be "([^"]*)"$`, theResponseHeaderShouldBe) ctx.Step(`^the SSE response header "([^"]*)" should be "([^"]*)"$`, theSSEResponseHeaderShouldBe) ctx.Step(`^I should receive a heartbeat comment$`, iShouldReceiveAHeartbeatComment) diff --git a/handler.go b/handler.go index 3cb96b1..6519c81 100644 --- a/handler.go +++ b/handler.go @@ -114,6 +114,13 @@ type Handler struct { // dropped (#143). PayloadFormat string `json:"payload_format,omitempty"` + // HealthDetails adds the NATS server's version and names and the + // stream's configuration and state to the readiness probes' JSON + // (health_path, ready_path). Off by default: probe paths are often + // reachable by load balancers or the public, and the details name + // servers, clusters and limits (#144). + HealthDetails bool `json:"health_details,omitempty"` + // HubURL is the URL advertised in the Link header for hub discovery. // When set, SSE responses include a Link: ; rel="nuts" header // so that clients and upstream APIs can discover the event hub automatically. diff --git a/serve.go b/serve.go index 63f3604..287cd06 100644 --- a/serve.go +++ b/serve.go @@ -1691,11 +1691,65 @@ func (h *Handler) serveLiveCheck(w http.ResponseWriter) error { } // serveReadinessCheck responds with NATS / stream readiness status. +// natsServerDetails is the NATS server a handler is connected to, reported +// by the readiness probes with health_details (#144). +type natsServerDetails struct { + Version string `json:"version"` + Name string `json:"name,omitempty"` + Cluster string `json:"cluster,omitempty"` + Domain string `json:"domain,omitempty"` +} + +// streamDetails is the configured stream as the readiness probe read it, +// reported with health_details (#144). Zero limits mean none. +type streamDetails struct { + Name string `json:"name"` + Subjects []string `json:"subjects"` + Storage jetstream.StorageType `json:"storage"` + Replicas int `json:"replicas"` + Messages uint64 `json:"messages"` + Bytes uint64 `json:"bytes"` + FirstSeq uint64 `json:"first_seq"` + LastSeq uint64 `json:"last_seq"` + Consumers int `json:"consumers"` + MaxMsgs int64 `json:"max_msgs"` + MaxBytes int64 `json:"max_bytes"` + MaxAgeSeconds int64 `json:"max_age_seconds"` + MaxConsumers int `json:"max_consumers"` + Created time.Time `json:"created"` +} + +func newStreamDetails(info *jetstream.StreamInfo) *streamDetails { + if info == nil { + return nil + } + return &streamDetails{ + Name: info.Config.Name, + Subjects: info.Config.Subjects, + Storage: info.Config.Storage, + Replicas: info.Config.Replicas, + Messages: info.State.Msgs, + Bytes: info.State.Bytes, + FirstSeq: info.State.FirstSeq, + LastSeq: info.State.LastSeq, + Consumers: info.State.Consumers, + MaxMsgs: info.Config.MaxMsgs, + MaxBytes: info.Config.MaxBytes, + MaxAgeSeconds: int64(info.Config.MaxAge / time.Second), + MaxConsumers: info.Config.MaxConsumers, + Created: info.Created, + } +} + func (h *Handler) serveReadinessCheck(w http.ResponseWriter) error { type healthResponse struct { Status string `json:"status"` NATS string `json:"nats"` Stream string `json:"stream"` + // With health_details (#144): the server NATS is connected to, and + // the stream as the probe read it. + NATSServer *natsServerDetails `json:"nats_server,omitempty"` + StreamInfo *streamDetails `json:"stream_info,omitempty"` } resp := healthResponse{ @@ -1743,13 +1797,23 @@ func (h *Handler) serveReadinessCheck(w http.ResponseWriter) error { // the probe past the orchestrator's readiness budget. See // defaultReadinessProbeTimeout for rationale. ctx, cancel := context.WithTimeout(context.Background(), defaultReadinessProbeTimeout) - _, err := js.Stream(ctx, h.StreamName) + stream, err := js.Stream(ctx, h.StreamName) cancel() if err != nil { resp.Status = "degraded" resp.Stream = "unavailable" statusCode = http.StatusServiceUnavailable recordFailure("stream_info_error", zap.Error(err)) + } else if h.HealthDetails { + resp.StreamInfo = newStreamDetails(stream.CachedInfo()) + } + } + if h.HealthDetails && conn != nil && conn.IsConnected() { + resp.NATSServer = &natsServerDetails{ + Version: conn.ConnectedServerVersion(), + Name: conn.ConnectedServerName(), + Cluster: conn.ConnectedClusterName(), + Domain: conn.ConnectedDomain(), } } diff --git a/serve_integration_test.go b/serve_integration_test.go index f3b8e79..6f7065c 100644 --- a/serve_integration_test.go +++ b/serve_integration_test.go @@ -11,12 +11,15 @@ import ( "net/http" "net/http/httptest" "reflect" + "slices" "strconv" "strings" "sync/atomic" "testing" "time" + natsserver "github.com/nats-io/nats-server/v2/server" + "github.com/caddyserver/caddy/v2/modules/caddyhttp" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -1624,3 +1627,88 @@ func TestHandler_RawPayloadFormat(t *testing.T) { t.Fatalf("the event after the unsendable payload = %q", got) } } + +// TestHandler_HealthDetails covers #144: with health_details the readiness +// probes add the NATS server and the stream as the probe read it. Without +// it they stay as they were, a stream that cannot be read has no details, +// and a disconnected handler names no server. +func TestHandler_HealthDetails(t *testing.T) { + ns := startJetStreamServer(t, func(o *natsserver.Options) { + o.JetStreamDomain = "hub" + o.Cluster.Name = "east" // named, without a cluster listener + }) + nc, err := nats.Connect(ns.ClientURL()) + if err != nil { + t.Fatalf("connect: %v", err) + } + defer nc.Close() + if _, err := mustJetStream(t, nc).CreateStream(context.Background(), jetstream.StreamConfig{ + Name: "DETAILS", Subjects: []string{"events.>"}, Storage: jetstream.MemoryStorage, + MaxMsgs: 1000, MaxAge: time.Hour, MaxConsumers: 50, + }); err != nil { + t.Fatalf("CreateStream: %v", err) + } + js, _ := nc.JetStream() + publishRange(t, js, "events.a", 1, 3) + if _, err := mustJetStream(t, nc).CreateOrUpdateConsumer(context.Background(), "DETAILS", jetstream.ConsumerConfig{Durable: "reader"}); err != nil { + t.Fatalf("CreateOrUpdateConsumer: %v", err) + } + + probe := func(t *testing.T, h *Handler) (int, map[string]json.RawMessage) { + t.Helper() + rr := httptest.NewRecorder() + if err := h.ServeHTTP(rr, httptest.NewRequest(http.MethodGet, "/events/readyz", nil), nil); err != nil { + t.Fatalf("ServeHTTP: %v", err) + } + var body map[string]json.RawMessage + if err := json.Unmarshal(rr.Body.Bytes(), &body); err != nil { + t.Fatalf("body %q: %v", rr.Body.String(), err) + } + return rr.Code, body + } + handler := func(t *testing.T, stream string, details bool) *Handler { + h := &Handler{NatsURL: ns.ClientURL(), StreamName: stream, HealthDetails: details, logger: zap.NewNop()} + connectHandler(t, h) + return h + } + + code, body := probe(t, handler(t, "DETAILS", true)) + var server natsServerDetails + var stream streamDetails + if err := json.Unmarshal(body["nats_server"], &server); err != nil || code != http.StatusOK { + t.Fatalf("status %d, nats_server %s: %v", code, body["nats_server"], err) + } + if server.Version != natsserver.VERSION || server.Name != ns.Name() || server.Cluster != "east" || server.Domain != "hub" { + t.Fatalf("nats_server = %+v, want version %s, name %s, cluster east, domain hub", server, natsserver.VERSION, ns.Name()) + } + if err := json.Unmarshal(body["stream_info"], &stream); err != nil { + t.Fatalf("stream_info %s: %v", body["stream_info"], err) + } + if stream.Name != "DETAILS" || !slices.Equal(stream.Subjects, []string{"events.>"}) || stream.Storage != jetstream.MemoryStorage || + stream.Messages != 3 || stream.Bytes == 0 || stream.Consumers != 1 || stream.FirstSeq != 1 || stream.LastSeq != 3 || stream.MaxMsgs != 1000 || + stream.MaxBytes != -1 || stream.MaxAgeSeconds != 3600 || stream.MaxConsumers != 50 || stream.Replicas != 1 || stream.Created.IsZero() { + t.Fatalf("stream_info = %+v", stream) + } + if !strings.Contains(string(body["stream_info"]), `"storage":"memory"`) { + t.Fatalf("storage is not written as a name: %s", body["stream_info"]) + } + + if _, body := probe(t, handler(t, "DETAILS", false)); body["nats_server"] != nil || body["stream_info"] != nil { + t.Fatalf("details without health_details: %v", body) + } + if code, body := probe(t, handler(t, "MISSING", true)); code != http.StatusServiceUnavailable || body["stream_info"] != nil || body["nats_server"] == nil { + t.Fatalf("a stream that cannot be read: status %d, body %v; want 503 with the server and no stream", code, body) + } + + h := handler(t, "DETAILS", true) + ns.Shutdown() + for h.conn.IsConnected() { + time.Sleep(10 * time.Millisecond) + } + if code, body := probe(t, h); code != http.StatusServiceUnavailable || body["nats_server"] != nil { + t.Fatalf("disconnected: status %d, body %v; want 503 without a server", code, body) + } + if newStreamDetails(nil) != nil { + t.Fatal("details of no stream info") + } +} diff --git a/website/docs/usage.md b/website/docs/usage.md index 5c7a534..e5c15de 100644 --- a/website/docs/usage.md +++ b/website/docs/usage.md @@ -65,6 +65,7 @@ nuts { event_id_format # Event ids: sequence (default) or sequence_time event_type [name] # Event names: message (default), topic, or header payload_format # Event data: envelope (default) or raw + health_details [true|false] # Readiness probes add NATS server and stream details # Health probes live_path # Liveness probe (default: /livez)