feat(nats): add JetStream data stream

packages/nats/_dev/build/docs/README.md

@@ -6,7 +6,7 @@ The integration collects metrics from [NATS monitoring server APIs](https://docs
 
 ## Compatibility
 
-The Nats package is tested with Nats 1.3.0, 2.0.4 and 2.1.4
+The NATS package is tested with NATS 2.10.27 and requires Elastic Agent 9.1+. The `jetstream` dataset requires NATS with JetStream enabled (NATS 2.2+), and consumer metrics require NATS 2.9+.
 
 ## Logs
 
@@ -24,8 +24,8 @@ Please refer to the following [document](https://www.elastic.co/guide/en/ecs/cur
 
 ## Metrics
 
-The default datasets are `stats`, `connections`, `routes` and `subscriptions` while `connection` and `route`
-datasets can be enabled to collect detailed metrics per connection/route.
+The default datasets are `stats`, `connections`, `routes` and `subscriptions` while `connection`, `route`
+and `jetstream` datasets can be enabled to collect detailed metrics per connection/route and JetStream monitoring.
 
 ### stats
 
@@ -93,6 +93,30 @@ Please refer to the following [document](https://www.elastic.co/guide/en/ecs/cur
 
 {{fields "connection"}}
 
+### jetstream
+
+This is the `jetstream` dataset of the Nats package, in charge of retrieving
+JetStream metrics from a NATS server. It collects data from the [/jsz](https://docs.nats.io/learn/monitoring/monitoring-endpoints#jsz-reports-jetstream-state) monitoring endpoint.
+
+The `jetstream` dataset supports four categories of metrics that can be enabled independently:
+
+* `stats` — General JetStream server stats (streams, consumers, messages, memory, storage).
+* `account` — Per-account JetStream metrics (memory, storage, API stats).
+* `stream` — Per-stream metrics (state, config, cluster info).
+* `consumer` — Per-consumer metrics (delivered, ack floor, pending, config). Requires NATS 2.9+.
+
+Account, stream, and consumer metrics can be filtered by name. Filters are cumulative and apply even if a category is not enabled but name filters are configured. When no names are configured, all entities are reported.
+
+This dataset requires Elastic Agent 9.1+ and a NATS server with JetStream enabled.
+
+{{event "jetstream"}}
+
+**ECS Field Reference**
+
+Please refer to the following [document](https://www.elastic.co/guide/en/ecs/current/ecs-field-reference.html) for detailed information on ECS fields.
+
+{{fields "jetstream"}}
+
 ### route
 
 This is the `route` dataset of the Nats package, in charge of retrieving detailed

packages/nats/_dev/deploy/docker/Dockerfile

@@ -1,19 +1,23 @@
-ARG NATS_VERSION=2.0.4
+ARG NATS_VERSION=2.10.27
 FROM nats:$NATS_VERSION
 
 # build stage
-FROM golang:1.20.2-alpine3.17 AS build-env
-RUN apk --no-cache add build-base git mercurial gcc
-RUN go install github.com/nats-io/nats.go/examples/nats-bench@v1.10.0
+FROM golang:1.25-alpine AS build-env
+RUN apk --no-cache add build-base git gcc
+RUN go install github.com/nats-io/nats.go/examples/nats-bench@v1.36.0
+WORKDIR /src
+COPY jetstream-traffic.go .
+RUN go mod init jetstream-traffic && go get github.com/nats-io/nats.go@v1.36.0 && go mod tidy && CGO_ENABLED=0 go build -o /jetstream-traffic jetstream-traffic.go
 
 # create an enhanced container with nc command available since nats is based
 # on scratch image making healthcheck impossible
-FROM alpine:3.17
+FROM alpine:3.20
 COPY --from=0 / /opt/nats
 COPY --from=build-env /go/bin/nats-bench /nats-bench
+COPY --from=build-env /jetstream-traffic /jetstream-traffic
 COPY run.sh /run.sh
 # Expose client, management, and cluster ports
 EXPOSE 4222 8222 6222
-HEALTHCHECK --interval=1s --retries=10 CMD nc -w 1 0.0.0.0 8222 </dev/null
+HEALTHCHECK --interval=1s --retries=30 CMD nc -w 1 0.0.0.0 8222 </dev/null
 # Run via the configuration file
 CMD ["/run.sh"]

packages/nats/_dev/deploy/docker/jetstream-traffic.go

@@ -0,0 +1,127 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+func main() {
+	var nc *nats.Conn
+	var err error
+
+	// Wait for NATS to become ready
+	for i := 0; i < 30; i++ {
+		nc, err = nats.Connect("nats://127.0.0.1:4222")
+		if err == nil {
+			break
+		}
+		time.Sleep(1 * time.Second)
+	}
+	if err != nil {
+		log.Fatalf("Failed to connect to NATS: %v", err)
+	}
+	defer nc.Close()
+
+	var js jetstream.JetStream
+	js, err = jetstream.New(nc)
+	if err != nil {
+		log.Fatalf("Failed to create JetStream context: %v", err)
+	}
+
+	cfg := jetstream.StreamConfig{
+		Name:        "ORDERS",
+		Description: "Orders stream for system testing",
+		Subjects:    []string{"orders.*"},
+		Storage:     jetstream.FileStorage,
+		Retention:   jetstream.LimitsPolicy,
+		MaxMsgs:     10000,
+		MaxBytes:    10485760,
+		MaxAge:      24 * time.Hour,
+		MaxMsgSize:  1048576,
+		Replicas:    1,
+	}
+
+	var stream jetstream.Stream
+	for i := 0; i < 30; i++ {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		stream, err = js.CreateOrUpdateStream(ctx, cfg)
+		cancel()
+		if err == nil {
+			fmt.Printf("Stream created: %s\n", stream.CachedInfo().Config.Name)
+			break
+		}
+		log.Printf("CreateOrUpdateStream attempt %d failed: %v, retrying...", i+1, err)
+		time.Sleep(1 * time.Second)
+	}
+	if err != nil {
+		log.Fatalf("Failed to create stream after retries: %v", err)
+	}
+
+	consumerCfg := jetstream.ConsumerConfig{
+		Name:          "PROCESSOR",
+		Durable:       "PROCESSOR",
+		Description:   "Order processor consumer",
+		DeliverPolicy: jetstream.DeliverAllPolicy,
+		AckPolicy:     jetstream.AckExplicitPolicy,
+		AckWait:       30 * time.Second,
+		MaxDeliver:    5,
+		FilterSubject: "orders.*",
+		ReplayPolicy:  jetstream.ReplayInstantPolicy,
+		MaxAckPending: 1000,
+	}
+
+	var consumer jetstream.Consumer
+	for i := 0; i < 30; i++ {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		consumer, err = js.CreateOrUpdateConsumer(ctx, "ORDERS", consumerCfg)
+		cancel()
+		if err == nil {
+			fmt.Printf("Consumer created: %s\n", consumer.CachedInfo().Config.Name)
+			break
+		}
+		log.Printf("CreateOrUpdateConsumer attempt %d failed: %v, retrying...", i+1, err)
+		time.Sleep(1 * time.Second)
+	}
+	if err != nil {
+		log.Fatalf("Failed to create consumer after retries: %v", err)
+	}
+
+	// Initial batch of messages
+	for i := 0; i < 20; i++ {
+		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+		msg := fmt.Sprintf("order data %d", i)
+		if _, err := js.Publish(ctx, "orders.created", []byte(msg)); err != nil {
+			log.Printf("Initial publish error for message %d: %v", i, err)
+		}
+		cancel()
+	}
+
+	// Continuous loop
+	ticker := time.NewTicker(200 * time.Millisecond)
+	count := 20
+	for range ticker.C {
+		count++
+		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+		msg := fmt.Sprintf("order data %d", count)
+		_, err := js.Publish(ctx, "orders.created", []byte(msg))
+		cancel()
+		if err != nil {
+			log.Printf("Publish error: %v", err)
+			continue
+		}
+
+		if count%2 == 0 && consumer != nil {
+			msgs, err := consumer.Fetch(1, jetstream.FetchMaxWait(50*time.Millisecond))
+			if err == nil {
+				for m := range msgs.Messages() {
+					_ = m.Ack()
+				}
+			}
+		}
+	}
+}

packages/nats/_dev/deploy/docker/run.sh

@@ -1,25 +1,28 @@
 #!/bin/sh
 
 # create logs' directory
-mkdir /var/log/nats
+mkdir -p /var/log/nats
 
 # This script is used with old and new versions of nats,
 # and they use different names for their binaries, detect
 # which one is used and run it.
 
 # NATS 2.X
 if [ -x /opt/nats/nats-server ]; then
-    if [[ -z "${ROUTES}" ]]; then
-        (/opt/nats/nats-server -DV -l /var/log/nats/nats.log --cluster nats://0.0.0.0:6222 --http_port 8222 --port 4222) &
+    if [ -z "${ROUTES}" ]; then
+        (/opt/nats/nats-server -DV -js --server_name nats --cluster_name nats-cluster -l /var/log/nats/nats.log --cluster nats://0.0.0.0:6222 --http_port 8222 --port 4222 --routes nats://nats-routes:6222) &
     else
-        (/opt/nats/nats-server -DV -l /var/log/nats/nats.log --cluster nats://0.0.0.0:6222 --http_port 8222 --port 4222 --routes nats://nats:6222) &
+        (/opt/nats/nats-server -DV -js --server_name nats-routes --cluster_name nats-cluster -l /var/log/nats/nats.log --cluster nats://0.0.0.0:6222 --http_port 8222 --port 4222 --routes nats://nats:6222) &
+    fi
+    if [ -x /jetstream-traffic ]; then
+        /jetstream-traffic &
     fi
     while true; do /nats-bench -np 1 -n 100000000 -ms 16 foo; done
 fi
 
 # NATS 1.X
 if [ -x /opt/nats/gnatsd ]; then
-    if [[ -z "${ROUTES}" ]]; then
+    if [ -z "${ROUTES}" ]; then
         (/opt/nats/gnatsd -DV -l /var/log/nats/nats.log --cluster nats://0.0.0.0:6222 --http_port 8222 --port 4222) &
     else
         (/opt/nats/gnatsd -DV -l /var/log/nats/nats.log --cluster nats://0.0.0.0:6222 --http_port 8222 --port 4222 --routes nats://nats:6222) &

packages/nats/_dev/deploy/variants.yml

@@ -1,6 +1,4 @@
 variants:
   v1:
-    SERVICE_VERSION: 1.3.0
-  v2:
-    SERVICE_VERSION: 2.1.4
-default: v2
+    SERVICE_VERSION: 2.10.27
+default: v1

packages/nats/changelog.yml

@@ -1,4 +1,9 @@
 # newer versions go on top
+- version: "1.13.0"
+  changes:
+    - description: Add JetStream data stream for monitoring JetStream stats, accounts, streams, and consumers. Requires Elastic Agent 9.1+.
+      type: enhancement
+      link: https://github.com/elastic/integrations/pull/20980
 - version: "1.12.0"
   changes:
     - description: Add missing fields to `connection` and `stats` data datastreams.

packages/nats/data_stream/connection/fields/fields.yml

@@ -83,3 +83,9 @@
     - name: id
       type: keyword
       description: A unique numeric identifier assigned by the NATS server to a connection
+    - name: kind
+      type: keyword
+      description: The kind of the connection (Client, Leafnode, etc.)
+    - name: type
+      type: keyword
+      description: The type of the connection

packages/nats/data_stream/jetstream/_dev/test/system/test-default-config.yml

@@ -0,0 +1,9 @@
+vars:
+  hosts:
+    - http://{{Hostname}}:{{Port}}
+data_stream:
+  vars:
+    jetstream_stats_enabled: true
+    jetstream_account_enabled: true
+    jetstream_stream_enabled: true
+    jetstream_consumer_enabled: true

packages/nats/data_stream/jetstream/agent/stream/stream.yml.hbs

@@ -0,0 +1,33 @@
+metricsets: ["jetstream"]
+hosts:
+{{#each hosts}}
+  - {{this}}
+{{/each}}
+period: {{period}}
+jetstream:
+  stats:
+    enabled: {{jetstream_stats_enabled}}
+  account:
+    enabled: {{jetstream_account_enabled}}
+{{#if jetstream_account_names}}
+    names:
+{{#each jetstream_account_names}}
+      - {{this}}
+{{/each}}
+{{/if}}
+  stream:
+    enabled: {{jetstream_stream_enabled}}
+{{#if jetstream_stream_names}}
+    names:
+{{#each jetstream_stream_names}}
+      - {{this}}
+{{/each}}
+{{/if}}
+  consumer:
+    enabled: {{jetstream_consumer_enabled}}
+{{#if jetstream_consumer_names}}
+    names:
+{{#each jetstream_consumer_names}}
+      - {{this}}
+{{/each}}
+{{/if}}

packages/nats/data_stream/jetstream/fields/base-fields.yml

@@ -0,0 +1,20 @@
+- name: data_stream.type
+  type: constant_keyword
+  description: Data stream type.
+- name: data_stream.dataset
+  type: constant_keyword
+  description: Data stream dataset.
+- name: data_stream.namespace
+  type: constant_keyword
+  description: Data stream namespace.
+- name: '@timestamp'
+  type: date
+  description: Event timestamp.
+- name: event.module
+  type: constant_keyword
+  description: Event module
+  value: nats
+- name: event.dataset
+  type: constant_keyword
+  description: Event dataset
+  value: nats.jetstream

packages/nats/data_stream/jetstream/fields/ecs.yml

@@ -0,0 +1,27 @@
+- external: ecs
+  name: service.address
+  dimension: true
+- external: ecs
+  name: agent.id
+  dimension: true
+- external: ecs
+  name: cloud.account.id
+  dimension: true
+- external: ecs
+  name: cloud.region
+  dimension: true
+- external: ecs
+  name: cloud.availability_zone
+  dimension: true
+- external: ecs
+  name: cloud.instance.id
+  dimension: true
+- external: ecs
+  name: cloud.provider
+  dimension: true
+- external: ecs
+  name: container.id
+  dimension: true
+- external: ecs
+  name: host.name
+  dimension: true