This is an automated email from the ASF dual-hosted git repository.
AlinsRan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/apisix.git
The following commit(s) were added to refs/heads/master by this push:
new abdac4b32c feat(prometheus): stream active connections, termination
status and bandwidth (#13796)
abdac4b32c is described below
commit abdac4b32c6b531af97afe95d79f92adc4b078b1
Author: AlinsRan <[email protected]>
AuthorDate: Mon Aug 17 16:11:52 2026 +0800
feat(prometheus): stream active connections, termination status and
bandwidth (#13796)
---
.requirements | 2 +-
apisix/cli/config.lua | 1 +
apisix/cli/ngx_tpl.lua | 7 +
apisix/plugin.lua | 5 +
apisix/plugins/prometheus/exporter.lua | 272 ++++++++++++++++++++
ci/linux-install-openresty.sh | 6 +-
docs/en/latest/plugins/prometheus.md | 65 +++++
docs/zh/latest/plugins/prometheus.md | 3 +
t/APISIX.pm | 6 +
t/stream-plugin/prometheus-metrics.t | 436 +++++++++++++++++++++++++++++++++
10 files changed, 799 insertions(+), 4 deletions(-)
diff --git a/.requirements b/.requirements
index c4546b2547..8bd4cf8ae8 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 98c3ff5812..ab34d8469a 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 2f9a5c0051..df2641106d 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 04d8ccd13d..6f988fd7d9 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 4b6ead0b71..fbb600e186 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,212 @@ 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 instead of only landing when it ends. The zone holds
+-- process wide totals, so it is read while the exposition is built rather than
+-- on a timer of its own -- which means once per refresh_interval, in the
+-- privileged agent that fills the metrics cache.
+local STREAM_BANDWIDTH_DIRECTIONS = {
+ {"downstream_ingress", "ingress", "downstream"},
+ {"downstream_egress", "egress", "downstream"},
+ {"upstream_egress", "egress", "upstream"},
+ {"upstream_ingress", "ingress", "upstream"},
+}
+
+-- 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
+-- reads 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:"
+
+-- Advancing a baseline is a read-modify-write and shdict has no
+-- compare-and-set, so only one reader converts totals to deltas at a time.
+-- Whoever loses the race leaves the baselines alone and the delta it skipped
+-- is picked up by the next read. The expiry keeps a worker that dies mid
+-- publish from wedging the metric.
+local STREAM_PUBLISH_LOCK = "stream_bytes_publishing"
+local STREAM_PUBLISH_LOCK_TTL = 10
+
+local stream_metrics_lib
+local stream_metrics_lib_checked = false
+local stream_zone_unavailable = false
+
+
+-- Only the runtime check is latched. Whether the zone itself is readable is
+-- decided per read: it comes and goes with the configuration, and caching a
+-- miss would leave the worker silently blind once it came back.
+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
+
+ stream_metrics_lib = lib
+ return lib
+end
+
+
+local function publish_stream_bytes(dict, listen_addr, direction, total)
+ local field = direction[1]
+
+ if type(total) ~= "number" then
+ core.log.error("stream metrics zone reported no ", field, " for ",
+ listen_addr)
+ return
+ end
+
+ local key = STREAM_PUBLISHED_PREFIX .. listen_addr .. ":" .. field
+
+ local published = dict:get(key)
+ if not published then
+ -- 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 read is not counted.
+ local ok, err = dict:set(key, total)
+ if not ok then
+ core.log.error("failed to baseline stream bandwidth for ", key, ":
", err)
+ end
+ return
+ end
+
+ if total == published then
+ return
+ end
+
+ -- Advance the baseline before counting: a write that failed while the
+ -- delta was counted would re-count the same range on every later read.
+ local ok, err = dict:set(key, total)
+ if not ok then
+ core.log.error("failed to advance the stream bandwidth baseline for ",
+ key, ": ", err)
+ return
+ end
+
+ -- a total below what was published means the zone was recreated, which
+ -- 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]))
+ end
+end
+
+
+local function collect_stream_zone_metrics()
+ if not metrics.stream_active_connections then
+ return
+ end
+
+ local lib = stream_metrics_zone()
+ if not lib then
+ return
+ end
+
+ local dict = prometheus.dict
+
+ -- Taken before the zone is sampled, not after: two readers that sampled at
+ -- different instants would otherwise serialize in the opposite order, and
+ -- the older sample would pull the baseline back over a range the fresher
+ -- one had already counted. shdict has no compare-and-set, so this is an
+ -- expiring key rather than a real mutex.
+ local publishing, add_err = dict:add(STREAM_PUBLISH_LOCK, true,
+ STREAM_PUBLISH_LOCK_TTL)
+ if not publishing and add_err ~= "exists" then
+ core.log.error("failed to take the stream bandwidth lock: ", add_err)
+ end
+
+ local entries, err = lib.dump()
+ if not entries then
+ if publishing then
+ dict:delete(STREAM_PUBLISH_LOCK)
+ end
+
+ -- report the transition, not every read: without the zone in the
+ -- configuration this is a steady state, not an incident
+ if not stream_zone_unavailable then
+ stream_zone_unavailable = true
+ core.log.warn("stream bandwidth and active connection metrics are
off: ", err)
+ end
+ return
+ end
+ stream_zone_unavailable = false
+
+ for _, entry in ipairs(entries) do
+ local listen_addr = entry.listen_addr
+
+ -- 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))
+
+ if publishing then
+ for _, direction in ipairs(STREAM_BANDWIDTH_DIRECTIONS) do
+ publish_stream_bytes(dict, listen_addr, direction,
entry[direction[1]])
+ end
+ end
+ end
+
+ if publishing then
+ dict:delete(STREAM_PUBLISH_LOCK)
+ end
+end
+
+
function _M.http_init(prometheus_enabled_in_stream)
-- todo: support hot reload, we may need to update the lua-prometheus
-- library
@@ -675,6 +878,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 +945,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
@@ -832,6 +1100,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/ci/linux-install-openresty.sh b/ci/linux-install-openresty.sh
index 1200b8e367..07f05e82b8 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
diff --git a/docs/en/latest/plugins/prometheus.md
b/docs/en/latest/plugins/prometheus.md
index 45a830209c..4061120905 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 054fed1f4d..f8eeb52c8e 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 05d477cbf2..677812bf8d 100644
--- a/t/APISIX.pm
+++ b/t/APISIX.pm
@@ -445,10 +445,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 0000000000..f7ecb83e4f
--- /dev/null
+++ b/t/stream-plugin/prometheus-metrics.t
@@ -0,0 +1,436 @@
+#
+# 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
+ # The endpoint serves a cache that the privileged agent refills on this
+ # interval, so the zone is sampled there rather than per request. At the
+ # default 15s no block would live long enough to see its own traffic.
+ # The cache expires after twice the interval and every refill makes two
+ # etcd round trips, so this keeps enough headroom that a slow runner does
+ # not serve an expired cache -- which logs an error and trips the
+ # no_error_log assertions.
+ my $extra_yaml_config = <<_EOC_;
+stream_plugins:
+ - prometheus
+plugin_attr:
+ prometheus:
+ refresh_interval: 0.5
+_EOC_
+
+ $block->set_value("extra_yaml_config", $extra_yaml_config);
+
+ if (!defined $block->request) {
+ $block->set_value("request", "GET /t");
+ }
+
+ # Every scrape needs the stream block in its own config. The zone backing
+ # the bandwidth and active connection metrics is declared there and is read
+ # while the exposition is built, so a block generated without a stream
+ # block has nothing to read -- and the reload into it drops the zone, which
+ # takes the running totals with it.
+ if ($block->request =~ m{/apisix/prometheus/metrics}) {
+ $block->set_value("stream_enable", 1);
+ }
+});
+
+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 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
+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 exposition has to report
+ -- it once the cache has been refilled with the session running.
+ -- Raw socket rather than ngx.location.capture: capturing into an
+ -- APISIX route leaves the upstream connect without a usable
+ -- api_ctx.
+ ngx.sleep(1.5)
+
+ 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
+
+ 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
+ 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]
