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)