From 3bfe23de7e506961e52f6226447ca1483c3c7424 Mon Sep 17 00:00:00 2001 From: Twinkll Sisodia Date: Mon, 24 Aug 2026 11:47:27 -0400 Subject: [PATCH] feat(tcp): add praxis_tcp_connections_total counter metric Increment a Prometheus counter on every accepted TCP connection, labeled by listener name. Gives operators visibility into connection throughput per listener over time. Closes #1025 Signed-off-by: Twinkll Sisodia --- examples/README.md | 1 + .../observability/tcp-connections-total.yaml | 24 +++++ protocol/src/tcp/metrics.rs | 23 ++++- protocol/src/tcp/mod.rs | 2 +- protocol/src/tcp/proxy.rs | 4 +- tests/integration/tests/suite/examples/mod.rs | 1 + .../suite/examples/tcp_connections_total.rs | 94 +++++++++++++++++++ tests/integration/tests/suite/main.rs | 1 + .../tests/suite/tcp_connections_total.rs | 84 +++++++++++++++++ 9 files changed, 230 insertions(+), 4 deletions(-) create mode 100644 examples/configs/observability/tcp-connections-total.yaml create mode 100644 tests/integration/tests/suite/examples/tcp_connections_total.rs create mode 100644 tests/integration/tests/suite/tcp_connections_total.rs diff --git a/examples/README.md b/examples/README.md index e73e2a56b..ecf84ae8d 100644 --- a/examples/README.md +++ b/examples/README.md @@ -41,6 +41,7 @@ page. | [process-logging.yaml](configs/observability/process-logging.yaml) | Non-blocking process logs written to a file | | [tcp-access-log.yaml](configs/observability/tcp-access-log.yaml) | Structured JSON logging of TCP connection events (connect and disconnect) | | [tcp-connection-metrics.yaml](configs/observability/tcp-connection-metrics.yaml) | Prometheus histogram for TCP connection duration | +| [tcp-connections-total.yaml](configs/observability/tcp-connections-total.yaml) | Prometheus counter for total accepted TCP connections per listener | | [trace-context.yaml](configs/observability/trace-context.yaml) | W3C Trace Context header propagation | | [tracing-otlp.yaml](configs/observability/tracing-otlp.yaml) | Exports distributed tracing spans to an OpenTelemetry Collector via OTLP/gRPC | diff --git a/examples/configs/observability/tcp-connections-total.yaml b/examples/configs/observability/tcp-connections-total.yaml new file mode 100644 index 000000000..ee20a4c6a --- /dev/null +++ b/examples/configs/observability/tcp-connections-total.yaml @@ -0,0 +1,24 @@ +# TCP Connections Total +# +# Prometheus counter for total accepted TCP connections per listener. +# Incremented once per connection, regardless of duration. Scraped +# via the admin /metrics endpoint. +# +# Usage: +# cargo run -p praxis-proxy -- -c examples/configs/observability/tcp-connections-total.yaml +# +# Exercise: +# for i in 1 2 3; do echo "hi" | nc localhost 5432; done +# curl -s http://localhost:9901/metrics | grep tcp_connections_total + +admin: + address: "127.0.0.1:9901" + +insecure_options: + allow_private_upstreams: true + +listeners: + - name: postgres + address: "127.0.0.1:5432" + protocol: tcp + upstream: "127.0.0.1:15432" diff --git a/protocol/src/tcp/metrics.rs b/protocol/src/tcp/metrics.rs index ce765ef01..0d6c8a9ba 100644 --- a/protocol/src/tcp/metrics.rs +++ b/protocol/src/tcp/metrics.rs @@ -3,7 +3,7 @@ //! Prometheus metrics for TCP connection lifecycle. -use metrics::{SharedString, histogram}; +use metrics::{SharedString, counter, histogram}; use crate::http::pingora::metrics::is_recorder_installed; @@ -11,6 +11,9 @@ use crate::http::pingora::metrics::is_recorder_installed; // Constants // ----------------------------------------------------------------------------- +/// Counter for total accepted TCP connections. +const TCP_CONNECTIONS_TOTAL: &str = "praxis_tcp_connections_total"; + /// Histogram for TCP connection duration in seconds. const TCP_CONNECTION_DURATION_SECONDS: &str = "praxis_tcp_connection_duration_seconds"; @@ -18,6 +21,17 @@ const TCP_CONNECTION_DURATION_SECONDS: &str = "praxis_tcp_connection_duration_se // Metric Recording // ----------------------------------------------------------------------------- +/// Increment the total TCP connections counter for the given listener. +/// +/// No-op when the Prometheus recorder has not been installed +/// (i.e. when the admin interface is disabled). +pub(crate) fn record_tcp_connection_accepted(listener: SharedString) { + if !is_recorder_installed() { + return; + } + counter!(TCP_CONNECTIONS_TOTAL, "listener" => listener).increment(1); +} + /// Record TCP connection duration for a closed connection. /// /// The `reason` label captures the disconnect cause (e.g. `completed`, @@ -46,7 +60,12 @@ mod tests { use super::*; #[test] - fn record_without_recorder_does_not_panic() { + fn record_accepted_without_recorder_does_not_panic() { + record_tcp_connection_accepted(SharedString::const_str("test-listener")); + } + + #[test] + fn record_duration_without_recorder_does_not_panic() { record_tcp_connection_duration(SharedString::const_str("test-listener"), "completed", 1.5); } diff --git a/protocol/src/tcp/mod.rs b/protocol/src/tcp/mod.rs index c893aa1de..576c842e3 100644 --- a/protocol/src/tcp/mod.rs +++ b/protocol/src/tcp/mod.rs @@ -13,7 +13,7 @@ use tokio::sync::{Semaphore, watch}; use crate::{ListenerPipelines, Protocol}; -/// TCP connection metrics (Prometheus histograms). +/// TCP connection metrics (Prometheus counters and histograms). pub(crate) mod metrics; /// Bidirectional TCP proxy application. pub(crate) mod proxy; diff --git a/protocol/src/tcp/proxy.rs b/protocol/src/tcp/proxy.rs index 771911967..c5023404e 100644 --- a/protocol/src/tcp/proxy.rs +++ b/protocol/src/tcp/proxy.rs @@ -316,8 +316,10 @@ impl ServerApp for PingoraTcpProxy { None }; + let listener_label = self.listener_label_for(&local_addr); let _active_connection = - crate::http::pingora::metrics::ActiveConnectionGuard::acquire(self.listener_label_for(&local_addr)); + crate::http::pingora::metrics::ActiveConnectionGuard::acquire(listener_label.clone()); + super::metrics::record_tcp_connection_accepted(listener_label); info!("connection_accepted"); diff --git a/tests/integration/tests/suite/examples/mod.rs b/tests/integration/tests/suite/examples/mod.rs index 99b40d607..d6903879e 100644 --- a/tests/integration/tests/suite/examples/mod.rs +++ b/tests/integration/tests/suite/examples/mod.rs @@ -65,6 +65,7 @@ mod sticky_sessions; mod stream_buffer; mod subset_lb; mod tcp_connection_metrics; +mod tcp_connections_total; mod timeout; mod trace_context; #[cfg(feature = "otel")] diff --git a/tests/integration/tests/suite/examples/tcp_connections_total.rs b/tests/integration/tests/suite/examples/tcp_connections_total.rs new file mode 100644 index 000000000..1e672b96a --- /dev/null +++ b/tests/integration/tests/suite/examples/tcp_connections_total.rs @@ -0,0 +1,94 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Praxis Contributors + +//! Functional integration tests for the tcp-connections-total example +//! configuration. + +use std::{ + collections::HashMap, + io::{Read as _, Write as _}, + net::TcpStream, + time::Duration, +}; + +use praxis_test_utils::{free_port, http_get, start_full_proxy, start_tcp_tagged_backend, wait_for_tcp}; + +// ----------------------------------------------------------------------------- +// Tests +// ----------------------------------------------------------------------------- + +#[test] +fn tcp_connections_total_example_emits_counter() { + let backend_port = start_tcp_tagged_backend("pg"); + let proxy_port = free_port(); + let admin_port = free_port(); + let config = super::load_example_config( + "observability/tcp-connections-total.yaml", + proxy_port, + HashMap::from([ + ("127.0.0.1:5432", proxy_port), + ("127.0.0.1:15432", backend_port), + ("127.0.0.1:9901", admin_port), + ]), + ); + + let _proxy = start_full_proxy(&config); + wait_for_tcp(&format!("127.0.0.1:{proxy_port}")); + wait_for_tcp(&format!("127.0.0.1:{admin_port}")); + + let mut stream = TcpStream::connect(format!("127.0.0.1:{proxy_port}")).expect("TCP connect"); + stream.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + stream.set_write_timeout(Some(Duration::from_secs(2))).unwrap(); + stream.write_all(b"SELECT 1").expect("TCP write"); + stream.shutdown(std::net::Shutdown::Write).expect("shutdown write"); + + let mut buf = Vec::new(); + stream.read_to_end(&mut buf).expect("TCP read"); + drop(stream); + + std::thread::sleep(Duration::from_millis(100)); + + let (status, body) = http_get(&format!("127.0.0.1:{admin_port}"), "/metrics", None); + assert_eq!(status, 200, "/metrics should return 200"); + assert!( + body.contains("praxis_tcp_connections_total"), + "metrics should contain praxis_tcp_connections_total counter: {body}" + ); + assert!( + body.contains("listener=\"postgres\""), + "metrics should contain listener=postgres label: {body}" + ); +} + +#[test] +fn tcp_connections_total_example_forwards_traffic() { + let backend_port = start_tcp_tagged_backend("dbdata"); + let proxy_port = free_port(); + let admin_port = free_port(); + let config = super::load_example_config( + "observability/tcp-connections-total.yaml", + proxy_port, + HashMap::from([ + ("127.0.0.1:5432", proxy_port), + ("127.0.0.1:15432", backend_port), + ("127.0.0.1:9901", admin_port), + ]), + ); + + let _proxy = start_full_proxy(&config); + wait_for_tcp(&format!("127.0.0.1:{proxy_port}")); + + let mut stream = TcpStream::connect(format!("127.0.0.1:{proxy_port}")).expect("TCP connect"); + stream.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + stream.set_write_timeout(Some(Duration::from_secs(2))).unwrap(); + stream.write_all(b"hello").expect("TCP write"); + stream.shutdown(std::net::Shutdown::Write).expect("shutdown write"); + + let mut buf = Vec::new(); + stream.read_to_end(&mut buf).expect("TCP read"); + let resp = String::from_utf8_lossy(&buf); + assert!( + resp.contains("dbdata"), + "tcp-connections-total example should forward to tagged backend, got: {resp}" + ); +} diff --git a/tests/integration/tests/suite/main.rs b/tests/integration/tests/suite/main.rs index d3bda0dca..9553bf839 100644 --- a/tests/integration/tests/suite/main.rs +++ b/tests/integration/tests/suite/main.rs @@ -82,6 +82,7 @@ mod stream_buffer_adapter; mod streaming_terminal_response; mod tcp_access_log; mod tcp_connection_metrics; +mod tcp_connections_total; mod tcp_edge_cases; mod tcp_load_balancer; mod tls; diff --git a/tests/integration/tests/suite/tcp_connections_total.rs b/tests/integration/tests/suite/tcp_connections_total.rs new file mode 100644 index 000000000..cd86e7a60 --- /dev/null +++ b/tests/integration/tests/suite/tcp_connections_total.rs @@ -0,0 +1,84 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Praxis Contributors + +//! Integration tests for TCP connections total counter. + +use std::{ + io::{Read as _, Write as _}, + net::TcpStream, + time::Duration, +}; + +use praxis_core::config::Config; +use praxis_test_utils::{free_port, http_get, start_full_proxy, start_tcp_tagged_backend, wait_for_tcp}; + +// ----------------------------------------------------------------------------- +// Tests +// ----------------------------------------------------------------------------- + +#[test] +fn tcp_connections_total_increments_on_each_connection() { + let backend_port = start_tcp_tagged_backend("counter-test"); + let proxy_port = free_port(); + let admin_port = free_port(); + + let yaml = format!( + r#" +admin: + address: "127.0.0.1:{admin_port}" + +insecure_options: + allow_private_upstreams: true + +listeners: + - name: tcp-counter-test + address: "127.0.0.1:{proxy_port}" + protocol: tcp + upstream: "127.0.0.1:{backend_port}" +"# + ); + + let config = Config::from_yaml(&yaml).unwrap(); + let _proxy = start_full_proxy(&config); + wait_for_tcp(&format!("127.0.0.1:{proxy_port}")); + wait_for_tcp(&format!("127.0.0.1:{admin_port}")); + + for i in 0..3 { + let mut stream = + TcpStream::connect(format!("127.0.0.1:{proxy_port}")).unwrap_or_else(|e| panic!("connect {i}: {e}")); + stream.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + stream.set_write_timeout(Some(Duration::from_secs(2))).unwrap(); + stream.write_all(b"ping").expect("write"); + stream.shutdown(std::net::Shutdown::Write).expect("shutdown"); + let mut buf = Vec::new(); + stream.read_to_end(&mut buf).expect("read"); + } + + std::thread::sleep(Duration::from_millis(100)); + + let (status, body) = http_get(&format!("127.0.0.1:{admin_port}"), "/metrics", None); + assert_eq!(status, 200, "/metrics should return 200"); + assert!( + body.contains("praxis_tcp_connections_total"), + "/metrics should contain praxis_tcp_connections_total: {body}" + ); + assert!( + body.contains("listener=\"tcp-counter-test\""), + "/metrics should contain listener=tcp-counter-test label: {body}" + ); + + let count_line = body + .lines() + .find(|l| l.contains("praxis_tcp_connections_total") && l.contains("tcp-counter-test") && !l.starts_with('#')) + .expect("should find counter line"); + let count: f64 = count_line + .split_whitespace() + .last() + .expect("counter should have a value") + .parse() + .expect("counter value should parse as f64"); + assert!( + count >= 3.0, + "counter should be at least 3 after 3 connections, got: {count}" + ); +}