From 4ed2002d3054d9eed0d5447213da1abd2da2b9b1 Mon Sep 17 00:00:00 2001 From: AlinsRan Date: Mon, 10 Aug 2026 14:27:44 +0800 Subject: [PATCH 1/4] feat(prometheus): stream active connections, termination status and bandwidth A TCP or UDP proxy could not be monitored from the stream metrics we had. The only one was a per route connection counter, nginx reports a stream $status of 200 for every failure that happens after the upstream connection is established, and no byte counter of a live session is reachable from Lua, so a long-lived connection produced nothing until it ended. Add three metrics, keyed by listen_addr because a session can end before any stream route is matched and because the byte counters come from nginx, which only knows the listening address: - apisix_stream_active_connections, live concurrency per listening address - apisix_stream_status, one count per finished session. $stream_session_reason narrows 200 down to genuine closes and folds timeouts and resets onto the codes nginx itself uses; plugin rejections are carried on ctx because a stream plugin rejects by closing the session, which never sets $status - apisix_stream_bandwidth, bytes per direction and side, read once a second from the nginx zone so it grows while a connection is still open apisix_stream_metrics_zone, $stream_session_reason and $stream_listen_addr come from apisix-nginx-module 1.19.9, so .requirements moves to the first runtime carrying it. Everything degrades on an older runtime: the collector logs once and stays off, and the status metric falls back to $status. --- .requirements | 2 +- apisix/cli/config.lua | 1 + apisix/cli/ngx_tpl.lua | 7 + apisix/plugin.lua | 5 + apisix/plugins/prometheus/exporter.lua | 200 ++++++++++++ docs/en/latest/plugins/prometheus.md | 65 ++++ docs/zh/latest/plugins/prometheus.md | 3 + t/APISIX.pm | 6 + t/stream-plugin/prometheus-metrics.t | 406 +++++++++++++++++++++++++ 9 files changed, 694 insertions(+), 1 deletion(-) create mode 100644 t/stream-plugin/prometheus-metrics.t diff --git a/.requirements b/.requirements index c4546b254777..8bd4cf8ae875 100644 --- a/.requirements +++ b/.requirements @@ -17,5 +17,5 @@ APISIX_PACKAGE_NAME=apisix -APISIX_RUNTIME=1.3.11 +APISIX_RUNTIME=1.3.14 APISIX_DASHBOARD_COMMIT=c8d3466d3c36386d3888efbc8250cd8183c77298 diff --git a/apisix/cli/config.lua b/apisix/cli/config.lua index 98c3ff581218..ab34d8469a2b 100644 --- a/apisix/cli/config.lua +++ b/apisix/cli/config.lua @@ -117,6 +117,7 @@ local _M = { access_log_format = "$remote_addr [$time_local] $protocol $status $bytes_sent $bytes_received $session_time", -- luacheck: pop access_log_format_escape = "default", + metrics_zone_size = "1m", lua_shared_dict = { ["etcd-cluster-health-check-stream"] = "10m", ["lrucache-lock-stream"] = "10m", diff --git a/apisix/cli/ngx_tpl.lua b/apisix/cli/ngx_tpl.lua index 4fa56d57a4cc..64ba66fe2a49 100644 --- a/apisix/cli/ngx_tpl.lua +++ b/apisix/cli/ngx_tpl.lua @@ -146,6 +146,13 @@ stream { lua_max_running_timers {* max_running_timers *}; {% end %} + # backs apisix_stream_active_connections and apisix_stream_bandwidth; the + # counters live in nginx so that they keep moving during a long-lived + # session instead of only being known once it ends + {% if use_apisix_base and enabled_stream_plugins["prometheus"] and stream.metrics_zone_size then %} + apisix_stream_metrics_zone {* stream.metrics_zone_size *}; + {% end %} + lua_shared_dict lrucache-lock-stream {* stream.lua_shared_dict["lrucache-lock-stream"] *}; lua_shared_dict etcd-cluster-health-check-stream {* stream.lua_shared_dict["etcd-cluster-health-check-stream"] *}; lua_shared_dict worker-events-stream {* stream.lua_shared_dict["worker-events-stream"] *}; diff --git a/apisix/plugin.lua b/apisix/plugin.lua index 04d8ccd13def..6f988fd7d9b1 100644 --- a/apisix/plugin.lua +++ b/apisix/plugin.lua @@ -1410,6 +1410,11 @@ function _M.run_plugin(phase, plugins, api_ctx) core.log.warn(plugins[i].name, " exits with status code ", code) end + -- a stream session is rejected by closing it, so the + -- code never reaches $status; keep it for the log + -- phase, which still runs after ngx_exit + api_ctx.stream_rejected_code = code + ngx_exit(1) end end diff --git a/apisix/plugins/prometheus/exporter.lua b/apisix/plugins/prometheus/exporter.lua index 4b6ead0b719c..a813c4506134 100644 --- a/apisix/plugins/prometheus/exporter.lua +++ b/apisix/plugins/prometheus/exporter.lua @@ -14,6 +14,7 @@ -- See the License for the specific language governing permissions and -- limitations under the License. -- +local require = require local base_prometheus = require("prometheus") local tonumber = tonumber local core = require("apisix.core") @@ -241,10 +242,142 @@ local function init_stream_metrics() "Total number of connections handled per stream route in APISIX", {"route"}) + -- 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. + metrics.stream_active_connections = prometheus:gauge( + "stream_active_connections", + "Number of stream sessions currently being proxied per listening address", + {"listen_addr"}) + + metrics.stream_status = prometheus:counter("stream_status", + "Stream sessions per termination status in APISIX", + {"code", "listen_addr", "node"}) + + metrics.stream_bandwidth = prometheus:counter("stream_bandwidth", + "Total bandwidth in bytes proxied by the stream subsystem in APISIX", + {"listen_addr", "type", "side"}) + xrpc.init_metrics(prometheus) end +-- src/stream/ngx_stream.h; 403 is reachable through ngx_stream_access_module +-- and the stream ip-restriction plugin +local STREAM_NGINX_CODES = { + [200] = true, [400] = true, [403] = true, + [500] = true, [502] = true, [503] = true, +} + +-- $stream_session_reason carries the fine grained termination reason; the +-- metric aggregates it onto the status codes nginx itself uses for stream +-- sessions, so no synthetic code ever shows up in apisix_stream_status. +local STREAM_REASON_TO_CODE = { + closed = "200", + -- a worker going away is a gateway side action, not a failed session + shutdown = "200", + client_rst = "400", + client_read_error = "400", + client_error = "400", + upstream_rst = "502", + upstream_read_error = "502", + upstream_error = "502", + connect_timeout = "502", + connect_failed = "502", + recv_timeout = "502", + send_timeout = "502", + upstream_timeout = "502", +} + + +-- Active session counts and byte counters are maintained by nginx in a shared +-- memory zone (apisix_stream_metrics_zone) so that they keep moving during a +-- long-lived session. The zone is global, so a single worker publishes it. +local STREAM_BANDWIDTH_DIRECTIONS = { + {"downstream_ingress", "ingress", "downstream"}, + {"downstream_egress", "egress", "downstream"}, + {"upstream_egress", "egress", "upstream"}, + {"upstream_ingress", "ingress", "upstream"}, +} + +local stream_zone_totals = {} + + +local function collect_stream_zone_metrics(premature, stream_metrics) + -- metrics is cleared while stream_init re-registers them, the timer may + -- well fire in that window + if premature or not metrics.stream_active_connections then + return + end + + local entries, err = stream_metrics.dump() + if not entries then + core.log.error("failed to read the stream metrics zone: ", err) + return + end + + for _, entry in ipairs(entries) do + local listen_addr = entry.listen_addr + + metrics.stream_active_connections:set(entry.active, gen_arr(listen_addr)) + + local seen = stream_zone_totals[listen_addr] + if not seen then + -- The zone holds totals while the metric holds a counter that + -- outlives this worker, so the first read only takes a baseline; + -- otherwise a worker restart would replay everything. + seen = {} + stream_zone_totals[listen_addr] = seen + for _, direction in ipairs(STREAM_BANDWIDTH_DIRECTIONS) do + seen[direction[1]] = entry[direction[1]] + end + + else + for _, direction in ipairs(STREAM_BANDWIDTH_DIRECTIONS) do + local field = direction[1] + local total = entry[field] + local delta = total - seen[field] + + -- a negative delta means the zone was recreated, rebaseline + if delta > 0 then + metrics.stream_bandwidth:inc(delta, + gen_arr(listen_addr, direction[2], direction[3])) + end + seen[field] = total + end + end + end +end + + +local function init_stream_zone_collector() + if ngx.get_phase() ~= "init_worker" or ngx.worker.id() ~= 0 then + return + end + + local ok, stream_metrics = pcall(require, "resty.apisix.stream.metrics") + if not ok then + core.log.warn("stream bandwidth and active connection metrics need a ", + "runtime providing resty.apisix.stream.metrics") + return + end + + local _, err = stream_metrics.dump() + if err then + core.log.warn("stream bandwidth and active connection metrics are off: ", err) + return + end + + core.table.clear(stream_zone_totals) + + local timer_ok, timer_err = ngx.timer.every(1, collect_stream_zone_metrics, + stream_metrics) + if not timer_ok then + core.log.error("failed to start the stream metrics timer: ", timer_err) + end +end + + function _M.http_init(prometheus_enabled_in_stream) -- todo: support hot reload, we may need to update the lua-prometheus -- library @@ -467,6 +600,8 @@ function _M.stream_init() prometheus = base_prometheus.init("prometheus-metrics", metric_prefix) init_stream_metrics() + + init_stream_zone_collector() end @@ -675,6 +810,62 @@ function _M.http_log(conf, ctx) end +-- Keeps the label inside the set of codes nginx itself uses for stream +-- sessions. A rejecting plugin can return anything -- limit-conn's +-- rejected_code is operator supplied -- and letting that through would put an +-- unbounded, user controlled value on the metric. +local function stream_reject_code(code) + -- a rejection is never a success, whatever the plugin was configured to + -- return; 200 has to keep meaning "the peer closed" + if type(code) ~= "number" or code < 400 then + return "500" + end + + if STREAM_NGINX_CODES[code] then + return tostring(code) + end + + if code < 500 then + return "400" + end + + return "500" +end + + +local function stream_status_code(ctx) + -- stream plugins reject by closing the session (plugin.lua run_plugin + -- calls ngx_exit(1)), so the code they returned never reaches $status + if ctx.stream_rejected_code then + return stream_reject_code(ctx.stream_rejected_code) + end + + local status = ctx.var.status + + -- nginx reports every post-connect failure as 200, so only a 200 needs + -- the reason to tell a normal close from a timeout or a reset + if status ~= "200" then + return status or "200" + end + + return STREAM_REASON_TO_CODE[ctx.var.stream_session_reason] or "200" +end + + +-- The metrics zone keys its slots by the configured listening address, so the +-- status metric has to use the same one. $server_addr is the address the +-- connection was accepted on, which differs on a wildcard listen; it is only +-- a fallback for a runtime without the apisix-nginx-module variable. +local function stream_listen_addr(ctx) + local listen_addr = ctx.var.stream_listen_addr + if listen_addr then + return listen_addr + end + + return ctx.var.server_addr .. ":" .. ctx.var.server_port +end + + function _M.stream_log(conf, ctx) local route_id = "" local matched_route = ctx.matched_route and ctx.matched_route.value @@ -686,6 +877,15 @@ function _M.stream_log(conf, ctx) end metrics.stream_connection_total:inc(1, gen_arr(route_id)) + + -- empty when the session ended before a node was picked + local node = "" + if ctx.balancer_ip and ctx.balancer_port then + node = ctx.balancer_ip .. ":" .. ctx.balancer_port + end + + metrics.stream_status:inc(1, gen_arr(stream_status_code(ctx), + stream_listen_addr(ctx), node)) end diff --git a/docs/en/latest/plugins/prometheus.md b/docs/en/latest/plugins/prometheus.md index 45a830209c8b..4061120905d1 100644 --- a/docs/en/latest/plugins/prometheus.md +++ b/docs/en/latest/plugins/prometheus.md @@ -130,6 +130,9 @@ The following metrics are exported by the `prometheus` Plugin by default. See [g | apisix_shared_dict_free_space_bytes | gauge | The remaining space in an [NGINX shared dictionary](https://github.com/openresty/lua-nginx-module#ngxshareddict). | | apisix_upstream_status | gauge | Health check status of upstream nodes, available if health checks are configured on the upstream. A value of `1` represents healthy and `0` represents unhealthy. | | apisix_stream_connection_total | counter | Total number of connections handled per Stream Route. | +| apisix_stream_active_connections | gauge | Number of Stream sessions currently being proxied, per listening address. Covers TCP connections and UDP sessions alike. | +| apisix_stream_status | counter | Counted once per Stream session when it ends, classified by how it ended. | +| apisix_stream_bandwidth | counter | Total bandwidth in bytes proxied by the Stream subsystem, per listening address and direction. | ## Labels @@ -155,6 +158,49 @@ The following labels are used to differentiate `apisix_http_status` metrics. | request_type | traditional_http / ai_chat / ai_stream | | llm_model | For non-traditional_http requests, name of the llm_model | +### Labels for `apisix_stream_active_connections` + +The gauge is incremented when a session is accepted and decremented when it +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`. | + +### Labels for `apisix_stream_status` + +NGINX reports a Stream `$status` of 200 for every failure that happens after +the upstream connection is established, so a timeout or a reset is +indistinguishable from a clean close on its own. This metric narrows 200 down +to sessions that really ended on a close signal, using the termination reason +the runtime records, and folds the rest onto the status codes NGINX itself +uses for Stream sessions. No synthetic code is introduced. + +| Name | Description | +| ----------- | --------------------------------------------------------------------------------------- | +| code | How the session ended: `200` normal close, `400` client-side problem such as a reset or invalid preread data, `403` rejected by an access rule, `500` internal error, `502` upstream or transport problem such as a connect failure, a reset, or an idle timeout, `503` rejected by a connection limit. | +| listen_addr | Listening address the client connected to, for example `0.0.0.0:9100`. | +| 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 +reset to observe. + +### Labels for `apisix_stream_bandwidth` + +Bytes keep accumulating while a connection is open rather than only at session +end, so a long-lived connection is visible as it runs. Only Stream traffic is +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`. | +| 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. | + +Under plain forwarding `downstream`/`ingress` tracks `upstream`/`egress`, and +`upstream`/`ingress` tracks `downstream`/`egress`; a lasting mismatch is itself +a signal that one side stopped reading. + ### Labels for `apisix_bandwidth` The following labels are used to differentiate `apisix_bandwidth` metrics. @@ -665,4 +711,23 @@ You should see an output similar to the following: # 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="1"} 1 +# HELP apisix_stream_active_connections Number of stream sessions currently being proxied per listening address +# TYPE apisix_stream_active_connections gauge +apisix_stream_active_connections{listen_addr="0.0.0.0:9100"} 1 +# 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="127.0.0.1:1995"} 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"} 5 +apisix_stream_bandwidth{listen_addr="0.0.0.0:9100",type="egress",side="upstream"} 5 ``` + +:::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. + +::: diff --git a/docs/zh/latest/plugins/prometheus.md b/docs/zh/latest/plugins/prometheus.md index 054fed1f4dda..f8eeb52c8e90 100644 --- a/docs/zh/latest/plugins/prometheus.md +++ b/docs/zh/latest/plugins/prometheus.md @@ -130,6 +130,9 @@ Prometheus 中有不同类型的指标。要了解它们之间的区别，请参 | apisix_shared_dict_free_space_bytes | gauge | [NGINX 共享字典](https://github.com/openresty/lua-nginx-module#ngxshareddict) 中剩余的空间。 | | apisix_upstream_status | gauge | 上游节点的健康检查状态，如果在上游配置了健康检查，则可用。值为 `1` 表示健康，`0` 表示不健康。 | | apisix_stream_connection_total | counter | 每个 Stream Route 处理的总连接数。 | +| apisix_stream_active_connections | gauge | 当前正在代理的 Stream 会话数，按监听地址区分。同时覆盖 TCP 连接与 UDP 会话。 | +| apisix_stream_status | counter | 每条 Stream 会话结束时计数一次，按结束方式分类。 | +| apisix_stream_bandwidth | counter | Stream 子系统代理的总流量（字节），按监听地址与方向区分。 | ## 标签 diff --git a/t/APISIX.pm b/t/APISIX.pm index e0b86560b040..95bbcce66ace 100644 --- a/t/APISIX.pm +++ b/t/APISIX.pm @@ -448,10 +448,16 @@ _EOC_ ngx.say("hello world") _EOC_ + # backs apisix_stream_active_connections and apisix_stream_bandwidth + my $stream_metrics_zone = $version =~ m/\/apisix-nginx-module/ + ? "apisix_stream_metrics_zone 1m;" : ""; + my $stream_config = $block->stream_config // <<_EOC_; $lua_deps_path lua_socket_log_errors off; + $stream_metrics_zone + lua_shared_dict lrucache-lock-stream 10m; lua_shared_dict plugin-limit-conn-stream 10m; lua_shared_dict etcd-cluster-health-check-stream 10m; diff --git a/t/stream-plugin/prometheus-metrics.t b/t/stream-plugin/prometheus-metrics.t new file mode 100644 index 000000000000..a1f274e707bd --- /dev/null +++ b/t/stream-plugin/prometheus-metrics.t @@ -0,0 +1,406 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +BEGIN { + if ($ENV{TEST_NGINX_CHECK_LEAK}) { + $SkipReason = "unavailable for the hup tests"; + + } else { + $ENV{TEST_NGINX_USE_HUP} = 1; + undef $ENV{TEST_NGINX_USE_STAP}; + } +} + +use t::APISIX; + +my $nginx_binary = $ENV{'TEST_NGINX_BINARY'} || 'nginx'; +my $version = eval { `$nginx_binary -V 2>&1` }; + +if ($version !~ m/\/apisix-nginx-module/) { + plan(skip_all => "apisix-nginx-module not installed"); +} else { + plan('no_plan'); +} + +repeat_each(1); +no_long_string(); +no_shuffle(); +no_root_location(); + +add_block_preprocessor(sub { + my ($block) = @_; + + # stream_plugins replaces the default list rather than extending it, so + # every plugin these tests touch has to be named here + my $extra_yaml_config = <<_EOC_; +stream_plugins: + - prometheus +_EOC_ + + $block->set_value("extra_yaml_config", $extra_yaml_config); + + if (!defined $block->request) { + $block->set_value("request", "GET /t"); + } +}); + +run_tests; + +__DATA__ + +=== TEST 1: pre-create the metrics endpoint and a stream route +--- config + location /t { + content_by_lua_block { + local data = { + { + url = "/apisix/admin/routes/metrics", + data = [[{ + "plugins": { + "public-api": {} + }, + "uri": "/apisix/prometheus/metrics" + }]] + }, + { + url = "/apisix/admin/stream_routes/1", + data = [[{ + "plugins": { + "prometheus": {} + }, + "upstream": { + "type": "roundrobin", + "nodes": [{ + "host": "127.0.0.1", + "port": 1995, + "weight": 1 + }] + } + }]] + } + } + + local t = require("lib.test_admin").test + + for _, data in ipairs(data) do + local code, body = t(data.url, ngx.HTTP_PUT, data.data) + if code > 300 then + ngx.say(body) + return + end + end + } + } +--- response_body + + + +=== TEST 2: proxy a session +--- stream_request +hello +--- stream_response +hello world + + + +=== TEST 3: the session is counted as a normal close, not as an error +--- request +GET /apisix/prometheus/metrics +--- response_body eval +qr/apisix_stream_status\{code="200",listen_addr="[^"]+",node="127.0.0.1:1995"\} 1$/m + + + +=== TEST 4: bandwidth and active connections come from the nginx zone +The collector publishes the zone once a second, so drive the traffic between +two ticks instead of relying on whatever happened in the previous block. + +The upstream has to stay open after answering: the shared fake upstreams close +as soon as they have written their line, and nginx finalizes the session on the +upstream's EOF no matter that the client is still connected -- which is why the +gauge would read 0 while the probe believes its session is live. + +The probe cannot live at /t: with the stream subsystem enabled Test::Nginx +installs its own `location = /t`, and an exact match wins. +--- extra_stream_config +server { + listen 1993; + content_by_lua_block { + local sock = ngx.req.socket() + sock:receive("1") + ngx.say("hello world") + ngx.flush(true) + -- answer, then hold the session open for the probe to observe + ngx.sleep(10) + } +} +--- config + location /probe { + content_by_lua_block { + local t = require("lib.test_admin").test + local code = t("/apisix/admin/stream_routes/1", ngx.HTTP_PUT, [[{ + "plugins": { + "prometheus": {} + }, + "upstream": { + "type": "roundrobin", + "nodes": [{ + "host": "127.0.0.1", + "port": 1993, + "weight": 1 + }] + } + }]]) + if code > 300 then + ngx.say("route: ", code) + return + end + + ngx.sleep(1.5) + + local sock = ngx.socket.tcp() + local ok, err = sock:connect("127.0.0.1", 1985) + if not ok then + ngx.say("connect: ", err) + return + end + + local bytes + bytes, err = sock:send("hello") + if not bytes then + ngx.say("send: ", err) + return + end + + -- the fake upstream answers with ngx.say, so the line is terminated + local line + line, err = sock:receive("*l") + if not line then + ngx.say("receive: ", err) + return + end + + -- The session is still open here, so the gauge must already be + -- counting it once the collector has ticked. Scrape over a real + -- socket: a subrequest would re-enter APISIX's routing and die in + -- the balancer with "invalid api_ctx". + ngx.sleep(1.5) + + local scr = ngx.socket.tcp() + scr:settimeout(5000) + ok, err = scr:connect("127.0.0.1", 1984) + if not ok then + ngx.say("scrape connect: ", err) + return + end + scr:send("GET /apisix/prometheus/metrics HTTP/1.0\r\n" + .. "Host: localhost\r\n\r\n") + local body = scr:receive("*a") + scr:close() + + local live = body and ngx.re.match(body, + [[apisix_stream_active_connections\{listen_addr="0\.0\.0\.0:1985"[^}]*\} (\d+)]]) + ngx.say("live=", live and live[1] or "no-series") + + ok, err = sock:close() + if not ok then + ngx.say("close: ", err) + return + end + + ngx.sleep(1.5) + } + } +--- request +GET /probe +--- stream_enable +--- timeout: 20 +--- response_body +live=1 + + + +=== TEST 5: bandwidth for client to gateway +nginx-lua-prometheus sorts the exposition, so each series is asserted on its +own rather than in one order-dependent pattern. +--- request +GET /apisix/prometheus/metrics +--- response_body_like eval +qr/apisix_stream_bandwidth\{listen_addr="0\.0\.0\.0:1985",type="ingress",side="downstream"\} [1-9]\d*/ +--- no_error_log +[error] + + + +=== TEST 6: bandwidth for gateway to client +nginx-lua-prometheus sorts the exposition, so each series is asserted on its +own rather than in one order-dependent pattern. +--- request +GET /apisix/prometheus/metrics +--- response_body_like eval +qr/apisix_stream_bandwidth\{listen_addr="0\.0\.0\.0:1985",type="egress",side="downstream"\} [1-9]\d*/ +--- no_error_log +[error] + + + +=== TEST 7: bandwidth for gateway to upstream +nginx-lua-prometheus sorts the exposition, so each series is asserted on its +own rather than in one order-dependent pattern. +--- request +GET /apisix/prometheus/metrics +--- response_body_like eval +qr/apisix_stream_bandwidth\{listen_addr="0\.0\.0\.0:1985",type="egress",side="upstream"\} [1-9]\d*/ +--- no_error_log +[error] + + + +=== TEST 8: bandwidth for upstream to gateway +nginx-lua-prometheus sorts the exposition, so each series is asserted on its +own rather than in one order-dependent pattern. +--- request +GET /apisix/prometheus/metrics +--- response_body_like eval +qr/apisix_stream_bandwidth\{listen_addr="0\.0\.0\.0:1985",type="ingress",side="upstream"\} [1-9]\d*/ +--- no_error_log +[error] + + + +=== TEST 9: the gauge drops back to zero once the session is gone +TEST 4 asserted it reads 1 while its session is open; the probe then closed it +and outlived another tick, so the published value has to be 0 by now. +--- request +GET /apisix/prometheus/metrics +--- response_body_like eval +qr/apisix_stream_active_connections\{listen_addr="0\.0\.0\.0:1985"\} 0$/m +--- no_error_log +[error] + + + +=== TEST 10: an unreachable upstream is counted as 502 +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + local code, body = t("/apisix/admin/stream_routes/1", ngx.HTTP_PUT, [[{ + "plugins": { + "prometheus": {} + }, + "upstream": { + "type": "roundrobin", + "nodes": [{ + "host": "127.0.0.1", + "port": 1979, + "weight": 1 + }] + } + }]]) + if code > 300 then + ngx.say(body) + return + end + } + } +--- response_body + + + +=== TEST 11: hit the unreachable upstream +--- stream_request +hello +--- error_log +connect() failed + + + +=== TEST 12: the failing node is still reported on a 502 +--- request +GET /apisix/prometheus/metrics +--- response_body eval +qr/apisix_stream_status\{code="502",listen_addr="[^"]+",node="127.0.0.1:1979"\} 1$/m + + + +=== TEST 13: point a route at an upstream that accepts and then says nothing +--- extra_stream_config +server { + listen 1993; + content_by_lua_block { + -- accept and stay silent, so the session dies on proxy_timeout + ngx.sleep(5) + } +} +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + local code, body = t("/apisix/admin/stream_routes/1", ngx.HTTP_PUT, [[{ + "plugins": { + "prometheus": {} + }, + "upstream": { + "type": "roundrobin", + "nodes": [{ + "host": "127.0.0.1", + "port": 1993, + "weight": 1 + }] + } + }]]) + if code > 300 then + ngx.say(body) + return + end + } + } +--- response_body + + + +=== TEST 14: let a session die on the idle timeout +--- extra_stream_config +server { + listen 1993; + content_by_lua_block { + ngx.sleep(5) + } +} +--- stream_server_config + proxy_timeout 500ms; + preread_by_lua_block { + apisix.stream_preread_phase() + } + proxy_pass apisix_backend; +--- stream_request +hello +--- timeout: 10 + + + +=== TEST 15: nginx calls that session a 200, the metric must not +This is the point of the feature. nginx reports $status 200 for every failure +after the upstream connection is up -- apisix-nginx-module's t/stream/metrics.t +pins that for the same case -- so an idle timeout has to reach the metric as +502 through $stream_session_reason, not as a success. +--- request +GET /apisix/prometheus/metrics +--- response_body_like eval +qr/apisix_stream_status\{code="502",listen_addr="0\.0\.0\.0:1985",node="127\.0\.0\.1:1993"\}/ +--- no_error_log +[error] From 0caa09aacf719bf9c78269fc2527556e331e3d87 Mon Sep 17 00:00:00 2001 From: AlinsRan Date: Mon, 10 Aug 2026 15:34:54 +0800 Subject: [PATCH 2/4] ci: pin the apisix-runtime-debug checksums for 1.3.14 ci/linux-install-openresty.sh guards the debug deb with a hardcoded version and sha256 per architecture, and bails out when .requirements moves without them. Every job that installs the runtime failed on that guard. Digests taken from the published 1.3.14 release assets. --- ci/linux-install-openresty.sh | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/ci/linux-install-openresty.sh b/ci/linux-install-openresty.sh index 1200b8e3677f..07f05e82b84b 100755 --- a/ci/linux-install-openresty.sh +++ b/ci/linux-install-openresty.sh @@ -61,7 +61,7 @@ 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.11" ]; then + if [ "$APISIX_RUNTIME" != "1.3.14" ]; then echo "Please update the apisix-runtime-debug checksum for APISIX_RUNTIME=$APISIX_RUNTIME" >&2 exit 1 fi @@ -69,11 +69,11 @@ else case "$ARCH" in x86_64|amd64) DEB_ARCH="amd64" - EXPECTED_SHA256="6c03f0a47a80e84c595c7e067f7d05fc69890237f9191af55108a284b356c4ee" + EXPECTED_SHA256="2d2350347c982e4467ff9326b5b93fcb9af2089b33b02bcd426e84a5adacf6f2" ;; arm64|aarch64) DEB_ARCH="arm64" - EXPECTED_SHA256="cdc124262a1acb2de170f12a2180cdc357ba867d6447cd08a9ba1639994d4e50" + EXPECTED_SHA256="495320e6377b96ab8d8a80a980a845e346c88160e2798ba46113a4da9042af4d" ;; *) echo "Unsupported architecture: $ARCH" >&2 From 7054e74d25c368b58a58fe1f5de81548467b6d6d Mon Sep 17 00:00:00 2001 From: AlinsRan Date: Mon, 10 Aug 2026 16:32:02 +0800 Subject: [PATCH 3/4] test: read the zone for the live check instead of the exposition The live-gauge case scraped the metrics endpoint while its session was open and found no series, although the later case asserting 0 on the same series passed -- so the collector had published, just not before this block scraped. Whether it has ticked by then depends on when the stream plugins finish loading, which is not what this case is about. Read the zone directly, which is where the count lives while the session is open; the published gauge stays covered by the case after it. --- t/stream-plugin/prometheus-metrics.t | 32 +++++++++++----------------- 1 file changed, 12 insertions(+), 20 deletions(-) diff --git a/t/stream-plugin/prometheus-metrics.t b/t/stream-plugin/prometheus-metrics.t index a1f274e707bd..8d7fe6621052 100644 --- a/t/stream-plugin/prometheus-metrics.t +++ b/t/stream-plugin/prometheus-metrics.t @@ -193,27 +193,19 @@ server { return end - -- The session is still open here, so the gauge must already be - -- counting it once the collector has ticked. Scrape over a real - -- socket: a subrequest would re-enter APISIX's routing and die in - -- the balancer with "invalid api_ctx". - ngx.sleep(1.5) - - local scr = ngx.socket.tcp() - scr:settimeout(5000) - ok, err = scr:connect("127.0.0.1", 1984) - if not ok then - ngx.say("scrape connect: ", err) - return + -- The session is still open here, so the zone must be counting + -- it. Read the zone rather than the exposition: whether the + -- collector has published yet depends on when the stream plugins + -- finished loading, which is not what this case is about. TEST 9 + -- covers the published gauge. + local zone = require("resty.apisix.stream.metrics") + local live = "no-slot" + for _, e in ipairs(zone.dump() or {}) do + if e.listen_addr == "0.0.0.0:1985" then + live = e.active + end end - scr:send("GET /apisix/prometheus/metrics HTTP/1.0\r\n" - .. "Host: localhost\r\n\r\n") - local body = scr:receive("*a") - scr:close() - - local live = body and ngx.re.match(body, - [[apisix_stream_active_connections\{listen_addr="0\.0\.0\.0:1985"[^}]*\} (\d+)]]) - ngx.say("live=", live and live[1] or "no-series") + ngx.say("live=", live) ok, err = sock:close() if not ok then From fcb114561dbaab0471dbbfb852703b69a91cdd72 Mon Sep 17 00:00:00 2001 From: AlinsRan Date: Wed, 12 Aug 2026 14:36:45 +0800 Subject: [PATCH 4/4] refactor(prometheus): read the stream metrics zone at scrape time The zone was published by a per-second timer pinned to worker 0, because the bandwidth counter needs a total-to-delta conversion and the baseline sat in worker memory, which only one worker could own. Moving the baseline into the metric dict removes that constraint: every worker claims against the same value, so the conversion can run wherever the scrape lands. The range is claimed with an atomic incr so two concurrent scrapes cannot both count it -- the loser hands the excess back. Drops up to two seconds of staleness (one timer tick plus one counter sync), the gap where a reload lost the bytes between the old worker's last tick and the new worker's first, and the timer itself. --- apisix/plugins/prometheus/exporter.lua | 169 +++++++++++++++++-------- t/stream-plugin/prometheus-metrics.t | 44 +++++-- 2 files changed, 147 insertions(+), 66 deletions(-) diff --git a/apisix/plugins/prometheus/exporter.lua b/apisix/plugins/prometheus/exporter.lua index a813c4506134..4ce90ec28815 100644 --- a/apisix/plugins/prometheus/exporter.lua +++ b/apisix/plugins/prometheus/exporter.lua @@ -292,7 +292,9 @@ local STREAM_REASON_TO_CODE = { -- Active session counts and byte counters are maintained by nginx in a shared -- memory zone (apisix_stream_metrics_zone) so that they keep moving during a --- long-lived session. The zone is global, so a single worker publishes it. +-- long-lived session instead of only landing when it ends. The zone holds +-- process wide totals, so it is read once per scrape, by whichever worker +-- happens to serve the metrics endpoint. local STREAM_BANDWIDTH_DIRECTIONS = { {"downstream_ingress", "ingress", "downstream"}, {"downstream_egress", "egress", "downstream"}, @@ -300,80 +302,141 @@ local STREAM_BANDWIDTH_DIRECTIONS = { {"upstream_ingress", "ingress", "upstream"}, } -local stream_zone_totals = {} +-- How much of each zone total has already been added to the counter. This +-- lives in the metric dict rather than in worker memory so that every worker +-- claims against the same value, and so that it is dropped together with the +-- counters it describes whenever that dict is flushed. +local STREAM_PUBLISHED_PREFIX = "stream_bytes_published:" +local stream_metrics_lib +local stream_metrics_lib_checked = false -local function collect_stream_zone_metrics(premature, stream_metrics) - -- metrics is cleared while stream_init re-registers them, the timer may - -- well fire in that window - if premature or not metrics.stream_active_connections then + +local function stream_metrics_zone() + if stream_metrics_lib_checked then + return stream_metrics_lib + end + stream_metrics_lib_checked = true + + local ok, lib = pcall(require, "resty.apisix.stream.metrics") + if not ok then + core.log.warn("stream bandwidth and active connection metrics need a ", + "runtime providing resty.apisix.stream.metrics") + return nil + end + + local _, err = lib.dump() + if err then + core.log.warn("stream bandwidth and active connection metrics are off: ", err) + return nil + end + + stream_metrics_lib = lib + return lib +end + + +-- Move the published total forward and return how much of the move is ours to +-- count. Two workers scraping at the same instant read the same starting +-- point, so the range is claimed with an atomic incr rather than a +-- read-modify-write: the loser sees the key run past the zone total, hands the +-- excess back and counts only what is left. +local function claim_stream_bytes(dict, key, total, delta) + -- the init argument covers the key being flushed between the read and the + -- claim: it rebaselines instead of failing + local claimed, err = dict:incr(key, delta, total - delta) + if not claimed then + core.log.error("failed to claim stream bandwidth for ", key, ": ", err) + return 0 + end + + if claimed <= total then + return delta + end + + local excess = claimed - total + local _, incr_err = dict:incr(key, -excess) + if incr_err then + core.log.error("failed to return an over claim for ", key, ": ", incr_err) + end + + return delta - excess +end + + +local function publish_stream_bytes(dict, listen_addr, direction, total) + local field = direction[1] + local key = STREAM_PUBLISHED_PREFIX .. listen_addr .. ":" .. field + + -- The zone counts from when nginx started while the counter outlives a + -- reload, so the first sight of a slot only takes a baseline; replaying + -- the whole total into a counter that survived would double it. The cost + -- is that traffic before the first scrape is not counted. + if dict:add(key, total) then return end - local entries, err = stream_metrics.dump() - if not entries then - core.log.error("failed to read the stream metrics zone: ", err) + local published = dict:get(key) + if not published then + -- evicted between the add and the get, rebaseline on the next scrape + dict:set(key, total) return end - for _, entry in ipairs(entries) do - local listen_addr = entry.listen_addr + local delta = total - published + if delta == 0 then + return + end - metrics.stream_active_connections:set(entry.active, gen_arr(listen_addr)) + if delta < 0 then + -- Either the zone was recreated under us, or another worker is midway + -- through a claim it has not handed back yet. Read once more before + -- rewinding: a claim in flight lasts microseconds, a recreated zone + -- stays low. + published = dict:get(key) or total + delta = total - published + if delta >= 0 then + return + end - local seen = stream_zone_totals[listen_addr] - if not seen then - -- The zone holds totals while the metric holds a counter that - -- outlives this worker, so the first read only takes a baseline; - -- otherwise a worker restart would replay everything. - seen = {} - stream_zone_totals[listen_addr] = seen - for _, direction in ipairs(STREAM_BANDWIDTH_DIRECTIONS) do - seen[direction[1]] = entry[direction[1]] - end + dict:incr(key, delta) + return + end - else - for _, direction in ipairs(STREAM_BANDWIDTH_DIRECTIONS) do - local field = direction[1] - local total = entry[field] - local delta = total - seen[field] - - -- a negative delta means the zone was recreated, rebaseline - if delta > 0 then - metrics.stream_bandwidth:inc(delta, - gen_arr(listen_addr, direction[2], direction[3])) - end - seen[field] = total - end - end + local counted = claim_stream_bytes(dict, key, total, delta) + if counted > 0 then + metrics.stream_bandwidth:inc(counted, + gen_arr(listen_addr, direction[2], direction[3])) end end -local function init_stream_zone_collector() - if ngx.get_phase() ~= "init_worker" or ngx.worker.id() ~= 0 then +local function collect_stream_zone_metrics() + if not metrics.stream_active_connections then return end - local ok, stream_metrics = pcall(require, "resty.apisix.stream.metrics") - if not ok then - core.log.warn("stream bandwidth and active connection metrics need a ", - "runtime providing resty.apisix.stream.metrics") + local lib = stream_metrics_zone() + if not lib then return end - local _, err = stream_metrics.dump() - if err then - core.log.warn("stream bandwidth and active connection metrics are off: ", err) + local entries, err = lib.dump() + if not entries then + core.log.error("failed to read the stream metrics zone: ", err) return end - core.table.clear(stream_zone_totals) + local dict = prometheus.dict + + for _, entry in ipairs(entries) do + local listen_addr = entry.listen_addr + + metrics.stream_active_connections:set(entry.active, gen_arr(listen_addr)) - local timer_ok, timer_err = ngx.timer.every(1, collect_stream_zone_metrics, - stream_metrics) - if not timer_ok then - core.log.error("failed to start the stream metrics timer: ", timer_err) + for _, direction in ipairs(STREAM_BANDWIDTH_DIRECTIONS) do + publish_stream_bytes(dict, listen_addr, direction, entry[direction[1]]) + end end end @@ -600,8 +663,6 @@ function _M.stream_init() prometheus = base_prometheus.init("prometheus-metrics", metric_prefix) init_stream_metrics() - - init_stream_zone_collector() end @@ -1032,6 +1093,10 @@ local function collect(yieldable) -- collect ngx.shared.DICT status shared_dict_status() + -- the stream zone is process wide, reading it here keeps the exposition + -- exact at the moment of the scrape + collect_stream_zone_metrics() + -- across all services nginx_status() diff --git a/t/stream-plugin/prometheus-metrics.t b/t/stream-plugin/prometheus-metrics.t index 8d7fe6621052..61f956edc9ca 100644 --- a/t/stream-plugin/prometheus-metrics.t +++ b/t/stream-plugin/prometheus-metrics.t @@ -125,8 +125,8 @@ qr/apisix_stream_status\{code="200",listen_addr="[^"]+",node="127.0.0.1:1995"\} === TEST 4: bandwidth and active connections come from the nginx zone -The collector publishes the zone once a second, so drive the traffic between -two ticks instead of relying on whatever happened in the previous block. +The zone is read while the metrics endpoint is served, so a session that is +still open has to show up in that same scrape. The upstream has to stay open after answering: the shared fake upstreams close as soon as they have written their line, and nginx finalizes the session on the @@ -193,19 +193,35 @@ server { return end - -- The session is still open here, so the zone must be counting - -- it. Read the zone rather than the exposition: whether the - -- collector has published yet depends on when the stream plugins - -- finished loading, which is not what this case is about. TEST 9 - -- covers the published gauge. - local zone = require("resty.apisix.stream.metrics") - local live = "no-slot" - for _, e in ipairs(zone.dump() or {}) do - if e.listen_addr == "0.0.0.0:1985" then - live = e.active - end + -- The session is still open here, so this scrape has to report it. + -- Raw socket rather than ngx.location.capture: capturing into an + -- APISIX route leaves the upstream connect without a usable + -- api_ctx. + local scrape = ngx.socket.tcp() + ok, err = scrape:connect("127.0.0.1", 1984) + if not ok then + ngx.say("scrape connect: ", err) + return end - ngx.say("live=", live) + + ok, err = scrape:send("GET /apisix/prometheus/metrics HTTP/1.0\r\n" + .. "Host: 127.0.0.1\r\n\r\n") + if not ok then + ngx.say("scrape send: ", err) + return + end + + local body, rerr, partial = scrape:receive("*a") + scrape:close() + body = body or partial + if not body then + ngx.say("scrape: ", rerr) + return + end + + local live = body:match('apisix_stream_active_connections' + .. '{listen_addr="0%.0%.0%.0:1985"[^}]*} (%d+)') + ngx.say("live=", live or "no-series") ok, err = sock:close() if not ok then