Add health_details: server and stream details in the readiness probes (#144)

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

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

README.md

@@ -385,6 +385,7 @@ nuts {
     event_id_format <format>     # Event ids: sequence (default) or sequence_time (sequence and message time)
     event_type <source> [name]   # Event names: message (default), topic, or header <name>
     payload_format <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 <path>           # Legacy readiness endpoint (empty/default: /healthz)
     live_path <path>             # Process liveness endpoint (empty/default: /livez)
     ready_path <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).
 

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()

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",

docs/CONFIGURATION.md

@@ -76,6 +76,7 @@ enforced. For public or browser-facing routes, issue short-lived tokens with
 | `replay_max_messages <count>` | `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 <source> [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 <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 <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, `<sequence>-<unix ns>`. 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 <seconds>` | `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. |

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.

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)

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: <url>; rel="nuts" header
 	// so that clients and upstream APIs can discover the event hub automatically.

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(),
 		}
 	}
 

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")
+	}
+}

website/docs/usage.md

@@ -65,6 +65,7 @@ nuts {
     event_id_format <format>            # Event ids: sequence (default) or sequence_time
     event_type <source> [name]          # Event names: message (default), topic, or header <name>
     payload_format <format>             # Event data: envelope (default) or raw
+    health_details [true|false]         # Readiness probes add NATS server and stream details
 
     # Health probes
     live_path  <path>                   # Liveness probe (default: /livez)