feat(prometheus): add service and service_id to the stream metrics

.requirements

@@ -17,5 +17,5 @@
 
 APISIX_PACKAGE_NAME=apisix
 
-APISIX_RUNTIME=1.3.18
+APISIX_RUNTIME=1.3.19
 APISIX_DASHBOARD_COMMIT=fa2fd0f60f8afffb096476333b9ba63b4c518fa3

apisix/plugins/prometheus/exporter.lua

@@ -238,25 +238,29 @@ local _M = {
 
 
 local function init_stream_metrics()
+    -- service and service_id follow the http metrics: both hold the id unless
+    -- prefer_name puts the name in service, and both are empty for a session
+    -- that never got as far as a route with a service
     metrics.stream_connection_total = prometheus:counter("stream_connection_total",
         "Total number of connections handled per stream route in APISIX",
-        {"route"})
+        {"route", "service", "service_id"})
 
     -- Keyed by listen_addr rather than by route: a session can end before any
     -- stream route is matched, and the byte counters come from nginx, which
-    -- only knows the listening address.
+    -- only knows the listening address. service and service_id split that
+    -- total further.
     metrics.stream_active_connections = prometheus:gauge(
         "stream_active_connections",
-        "Number of stream sessions currently being proxied per listening address",
-        {"listen_addr"})
+        "Number of stream sessions currently being proxied per listening address and service",
+        {"listen_addr", "service", "service_id"})
 
     metrics.stream_status = prometheus:counter("stream_status",
         "Stream sessions per termination status in APISIX",
-        {"code", "listen_addr", "node"})
+        {"code", "listen_addr", "service", "service_id", "node"})
 
     metrics.stream_bandwidth = prometheus:counter("stream_bandwidth",
         "Total bandwidth in bytes proxied by the stream subsystem in APISIX",
-        {"listen_addr", "type", "side"})
+        {"listen_addr", "service", "service_id", "type", "side"})
 
     xrpc.init_metrics(prometheus)
 end
@@ -317,6 +321,48 @@ local STREAM_PUBLISHED_PREFIX = "stream_bytes_published:"
 local STREAM_PUBLISH_LOCK = "stream_bytes_publishing"
 local STREAM_PUBLISH_LOCK_TTL = 10
 
+
+-- The same rule as the http metrics: the id, or the name with prefer_name.
+-- Resolved once per session: preread labels the zone with it and the log
+-- phase reuses it, so all four metrics name a session the same way even if
+-- its service is renamed while it is open.
+local function stream_service_labels(conf, ctx)
+    local labels = ctx.prometheus_stream_service
+    if labels then
+        return labels[1], labels[2]
+    end
+
+    local service, service_id = "", ""
+    local id = ctx.service_id
+    local name = ctx.service_name
+    if not id then
+        -- a route with an upstream_id keeps its service_id without the service
+        -- being merged in, and the http metrics still label it with it
+        id = ctx.matched_route.value.service_id
+        if id then
+            local fetched = conf.prefer_name == true and service_fetch(id)
+            name = fetched and fetched.value.name
+        end
+    end
+
+    if id then
+        service_id = tostring(id)
+        service = conf.prefer_name == true and name or service_id
+    end
+
+    ctx.prometheus_stream_service = {service, service_id}
+    return service, service_id
+end
+
+
+-- The zone is read where neither the session nor the plugin conf is at hand,
+-- so a session carries its label values into the zone itself; "" on a slot
+-- that was never labelled.
+local function stream_zone_labels(entry)
+    local labels = entry.labels
+    return labels[1] or "", labels[2] or ""
+end
+
 local stream_metrics_lib
 local stream_metrics_lib_checked = false
 local stream_zone_unavailable = false
@@ -343,16 +389,19 @@ local function stream_metrics_zone()
 end
 
 
-local function publish_stream_bytes(dict, listen_addr, direction, total)
+local function publish_stream_bytes(dict, listen_addr, service, service_id, direction,
+                                    total)
     local field = direction[1]
 
     if type(total) ~= "number" then
         core.log.error("stream metrics zone reported no ", field, " for ",
-                       listen_addr)
+                       listen_addr, " service ", service_id)
         return
     end
 
-    local key = STREAM_PUBLISHED_PREFIX .. listen_addr .. ":" .. field
+    -- a service name can hold any character but this one
+    local key = STREAM_PUBLISHED_PREFIX .. listen_addr .. "\31" .. service .. "\31"
+                .. service_id .. "\31" .. field
 
     local published = dict:get(key)
     if not published then
@@ -384,7 +433,7 @@ local function publish_stream_bytes(dict, listen_addr, direction, total)
     -- rebaselines rather than emitting a negative delta
     if total > published then
         metrics.stream_bandwidth:inc(total - published,
-            gen_arr(listen_addr, direction[2], direction[3]))
+            gen_arr(listen_addr, service, service_id, direction[2], direction[3]))
     end
 end
 
@@ -430,14 +479,17 @@ local function collect_stream_zone_metrics()
 
     for _, entry in ipairs(entries) do
         local listen_addr = entry.listen_addr
+        local service, service_id = stream_zone_labels(entry)
 
         -- the gauge carries no baseline and every reader writes the same
         -- value, so it is published whether or not this one took the lock
-        metrics.stream_active_connections:set(entry.active, gen_arr(listen_addr))
+        metrics.stream_active_connections:set(entry.active,
+            gen_arr(listen_addr, service, service_id))
 
         if publishing then
             for _, direction in ipairs(STREAM_BANDWIDTH_DIRECTIONS) do
-                publish_stream_bytes(dict, listen_addr, direction, entry[direction[1]])
+                publish_stream_bytes(dict, listen_addr, service, service_id, direction,
+                                     entry[direction[1]])
             end
         end
     end
@@ -944,7 +996,10 @@ function _M.stream_log(conf, ctx)
         end
     end
 
-    metrics.stream_connection_total:inc(1, gen_arr(route_id))
+    -- empty when the session ended before a route with a service matched
+    local service, service_id = stream_service_labels(conf, ctx)
+
+    metrics.stream_connection_total:inc(1, gen_arr(route_id, service, service_id))
 
     -- empty when the session ended before a node was picked
     local node = ""
@@ -953,7 +1008,41 @@ function _M.stream_log(conf, ctx)
     end
 
     metrics.stream_status:inc(1, gen_arr(stream_status_code(ctx),
-        stream_listen_addr(ctx), node))
+        stream_listen_addr(ctx), service, service_id, node))
+end
+
+
+-- logged once per worker, it would otherwise repeat on every session
+local stream_label_error_logged = false
+
+
+-- Labels the session in the metrics zone as it starts, so that its active
+-- count and every byte it moves from now on are accounted under its service
+-- while it is still open. A session that never gets here -- it ended before
+-- routing, its route has no service, or a plugin running before this one
+-- rejected it -- stays in the unlabelled total of its listen_addr.
+function _M.stream_preread(conf, ctx)
+    local service, service_id = stream_service_labels(conf, ctx)
+    if service_id == "" then
+        return
+    end
+
+    local lib = stream_metrics_zone()
+    if not lib then
+        return
+    end
+
+    -- "not accounted" is a listening address the zone does not count, such as
+    -- a unix socket. Anything else -- a full zone, or labels too long for it --
+    -- leaves the session in the unlabelled total, while apisix_stream_status
+    -- still has its service.
+    local ok, err = lib.set_labels({service, service_id})
+    if not ok and err ~= "not accounted" and not stream_label_error_logged then
+        stream_label_error_logged = true
+        core.log.warn("failed to label stream sessions, first seen on service ",
+                      service_id, ", they are only counted in the listen_addr ",
+                      "total: ", err)
+    end
 end
 
 

apisix/stream/plugins/prometheus.lua

@@ -34,6 +34,7 @@ local _M = {
     version = 0.1,
     priority = 500,
     name = plugin_name,
+    preread = exporter.stream_preread,
     log  = exporter.stream_log,
     destroy = exporter.destroy,
     init = exporter.stream_init,

ci/linux-install-openresty.sh

@@ -61,19 +61,19 @@ else
     sudo apt-get -y update --fix-missing
     sudo apt-get install -y build-essential gcc g++ cpanminus libxml2-dev libxslt-dev
 
-    if [ "$APISIX_RUNTIME" != "1.3.18" ]; then
+    if [ "$APISIX_RUNTIME" != "1.3.19" ]; then
         echo "Please update the apisix-runtime-debug checksum for APISIX_RUNTIME=$APISIX_RUNTIME" >&2
         exit 1
     fi
 
     case "$ARCH" in
         x86_64|amd64)
             DEB_ARCH="amd64"
-            EXPECTED_SHA256="0d7cbe27cd0303c6f6b3cad27338a697cf2e2e809ceeac9a57b2d1fb78a3a20e"
+            EXPECTED_SHA256="0f350a142c915a70d4501d68259a1506ffdd8efd6a1464802c8935b1f36fdf3b"
             ;;
         arm64|aarch64)
             DEB_ARCH="arm64"
-            EXPECTED_SHA256="a7cf5040837e4d34f456e97ff9b858ce64bf95e3b2f647f5d3021de5a9770741"
+            EXPECTED_SHA256="67fc887818f2899874e26fc539b9f49c78c7f8e73aed495683df9b694250db54"
             ;;
         *)
             echo "Unsupported architecture: $ARCH" >&2

docs/en/latest/plugins/prometheus.md

@@ -202,6 +202,20 @@ The following labels are used to differentiate `apisix_http_status` metrics.
 | llm_model          | Effective target model for the AI request. Uses the model configured on the AI instance when present; otherwise uses the model requested by the client. Empty for traditional HTTP traffic. |
 | response_source    | Response origin: `apisix` for responses generated by APISIX, `nginx` for NGINX proxy errors, or `upstream` for responses received from the Upstream. |
 
+### Service labels on the Stream metrics
+
+All four Stream metrics carry `service` and `service_id`, following the same rule as the HTTP metrics: `service_id` is the ID of the Service the session's Stream Route belongs to, and `service` is that ID too, or the Service's name when `prefer_name` is `true` on the Route's `prometheus` Plugin. Both are empty for a session that never reached a Stream Route with a Service, for example one that failed its TLS handshake or matched no Route. Such sessions are only counted in the per `listen_addr` total, so summing over `service` gives the same totals as before.
+
+`apisix_stream_active_connections` and `apisix_stream_bandwidth` are split per Service while sessions are still open: the Stream `prometheus` Plugin labels each session with its Service when it starts, and from then on its active count and the bytes it moves are accounted under that Service.
+
+### Labels for `apisix_stream_connection_total`
+
+| Name        | Description                                                                             |
+| ----------- | --------------------------------------------------------------------------------------- |
+| route       | ID of the matched Stream Route, or its name when `prefer_name` is `true`.                |
+| service     | ID of the Service of the matched Stream Route when `prefer_name` is `false` (default), and name of the Service when `prefer_name` is `true`. Empty when the session did not reach a Stream Route that belongs to a Service. |
+| service_id  | ID of the Service of the matched Stream Route. Empty when the session did not reach a Stream Route that belongs to a Service. |
+
 ### Labels for `apisix_stream_active_connections`
 
 The gauge is incremented when a session is accepted and decremented when it
@@ -210,6 +224,8 @@ ends, so it reflects live concurrency without waiting for sessions to finish.
 | Name        | Description                                                                             |
 | ----------- | --------------------------------------------------------------------------------------- |
 | listen_addr | Listening address the client connected to, for example `0.0.0.0:9100`.                   |
+| service     | ID of the Service of the matched Stream Route when `prefer_name` is `false` (default), and name of the Service when `prefer_name` is `true`. Empty when the session did not reach a Stream Route that belongs to a Service. |
+| service_id  | ID of the Service of the matched Stream Route. Empty when the session did not reach a Stream Route that belongs to a Service. |
 
 ### Labels for `apisix_stream_status`
 
@@ -225,6 +241,8 @@ clean close. No synthetic code is introduced.
 | ----------- | --------------------------------------------------------------------------------------- |
 | code        | How the session ended: `200` for a normal close, worker shutdown, or a missing or unrecognized termination reason; `400` for a client-side problem such as a reset or invalid preread data; `403` when rejected by an access rule; `500` for an internal error; `502` for an upstream or transport problem such as a connect failure, reset, or idle timeout; `503` when rejected by a connection limit. |
 | listen_addr | Listening address the client connected to, for example `0.0.0.0:9100`.                   |
+| service     | ID of the Service of the matched Stream Route when `prefer_name` is `false` (default), and name of the Service when `prefer_name` is `true`. Empty when the session did not reach a Stream Route that belongs to a Service. |
+| service_id  | ID of the Service of the matched Stream Route. Empty when the session did not reach a Stream Route that belongs to a Service. |
 | node        | Address of the upstream node used, empty when no node was selected.                      |
 
 For UDP only a subset of the codes occurs, since UDP has no close, FIN or
@@ -239,6 +257,8 @@ counted; the HTTP subsystem cannot contribute to it.
 | Name        | Description                                                                             |
 | ----------- | --------------------------------------------------------------------------------------- |
 | listen_addr | Listening address the client connected to, for example `0.0.0.0:9100`.                   |
+| service     | ID of the Service of the matched Stream Route when `prefer_name` is `false` (default), and name of the Service when `prefer_name` is `true`. Empty when the session did not reach a Stream Route that belongs to a Service. |
+| service_id  | ID of the Service of the matched Stream Route. Empty when the session did not reach a Stream Route that belongs to a Service. |
 | side        | Which connection the bytes crossed: `downstream` between APISIX and the client, `upstream` between APISIX and the upstream. |
 | type        | Direction relative to APISIX, matching `apisix_bandwidth`: `ingress` for bytes APISIX received, `egress` for bytes APISIX sent. |
 
@@ -763,28 +783,31 @@ You should see an output similar to the following:
 ```text
 # HELP apisix_stream_connection_total Total number of connections handled per Stream Route in APISIX
 # TYPE apisix_stream_connection_total counter
-apisix_stream_connection_total{route="prometheus-route"} 1
-# HELP apisix_stream_active_connections Number of stream sessions currently being proxied per listening address
+apisix_stream_connection_total{route="prometheus-route",service="",service_id=""} 1
+# HELP apisix_stream_active_connections Number of stream sessions currently being proxied per listening address and service
 # TYPE apisix_stream_active_connections gauge
-apisix_stream_active_connections{listen_addr="0.0.0.0:9100"} 0
+apisix_stream_active_connections{listen_addr="0.0.0.0:9100",service="",service_id=""} 0
 # HELP apisix_stream_status Stream sessions per termination status in APISIX
 # TYPE apisix_stream_status counter
-apisix_stream_status{code="200",listen_addr="0.0.0.0:9100",node="54.237.103.220:80"} 1
+apisix_stream_status{code="200",listen_addr="0.0.0.0:9100",service="",service_id="",node="54.237.103.220:80"} 1
 # HELP apisix_stream_bandwidth Total bandwidth in bytes proxied by the stream subsystem in APISIX
 # TYPE apisix_stream_bandwidth counter
-apisix_stream_bandwidth{listen_addr="0.0.0.0:9100",type="ingress",side="downstream"} 78
-apisix_stream_bandwidth{listen_addr="0.0.0.0:9100",type="egress",side="downstream"} 219
-apisix_stream_bandwidth{listen_addr="0.0.0.0:9100",type="egress",side="upstream"} 78
-apisix_stream_bandwidth{listen_addr="0.0.0.0:9100",type="ingress",side="upstream"} 219
+apisix_stream_bandwidth{listen_addr="0.0.0.0:9100",service="",service_id="",type="ingress",side="downstream"} 78
+apisix_stream_bandwidth{listen_addr="0.0.0.0:9100",service="",service_id="",type="egress",side="downstream"} 219
+apisix_stream_bandwidth{listen_addr="0.0.0.0:9100",service="",service_id="",type="egress",side="upstream"} 78
+apisix_stream_bandwidth{listen_addr="0.0.0.0:9100",service="",service_id="",type="ingress",side="upstream"} 219
 ```
 
-The exact Upstream address and byte counts depend on the request. The active-connections gauge is `0` above because the request completed before the scrape; scrape while a connection remains open to observe a positive value.
+The Stream Route in this example does not belong to a Service, so `service` and `service_id` are empty. The exact Upstream address and byte counts depend on the request. The active-connections gauge is `0` above because the request completed before the scrape; scrape while a connection remains open to observe a positive value.
 
 :::note
 
 `apisix_stream_active_connections` and `apisix_stream_bandwidth` are backed by
 an NGINX shared memory zone, sized by `nginx_config.stream.metrics_zone_size`
 (default `1m`). They require APISIX-Runtime; on a runtime without it the two
-metrics are simply not published.
+metrics are simply not published. The zone holds one slot per listening address
+plus one per Service seen on it, about 760 slots for `1m`; at most three
+quarters of them go to Services. When they run out, sessions of a new Service
+are only counted in their `listen_addr` total and a warning is logged.
 
 :::