use std::sync::OnceLock;
use metrics::{Label, SharedString, counter, gauge, histogram};
use metrics_exporter_prometheus::{Matcher, PrometheusBuilder, PrometheusHandle};
use praxis_core::config::{MetricLabel, MetricLabelsConfig};
const HTTP_REQUESTS_TOTAL: &str = "praxis_http_requests_total";
const HTTP_REQUEST_DURATION_SECONDS: &str = "praxis_http_request_duration_seconds";
const HTTP_REQUEST_BODY_BYTES: &str = "praxis_http_request_body_bytes";
const HTTP_RESPONSE_BODY_BYTES: &str = "praxis_http_response_body_bytes";
const HTTP_ACTIVE_REQUESTS: &str = "praxis_http_active_requests";
const TCP_ACTIVE_CONNECTIONS: &str = "praxis_tcp_active_connections";
const OVERLOAD_REJECTS_TOTAL: &str = "praxis_overload_rejects_total";
const PROCESS_OPEN_FDS: &str = "praxis_process_open_fds";
const PROCESS_MAX_FDS: &str = "praxis_process_max_fds";
const UPSTREAM_REQUESTS_TOTAL: &str = "praxis_upstream_requests_total";
const UPSTREAM_CONNECT_DURATION_SECONDS: &str = "praxis_upstream_connect_duration_seconds";
const UPSTREAM_CONNECT_FAILURES_TOTAL: &str = "praxis_upstream_connect_failures_total";
const UPSTREAM_RETRIES_TOTAL: &str = "praxis_upstream_retries_total";
const UPSTREAM_HEALTHY_ENDPOINTS: &str = "praxis_upstream_healthy_endpoints";
const UPSTREAM_TOTAL_ENDPOINTS: &str = "praxis_upstream_total_endpoints";
const UPSTREAM_HEALTH_TRANSITIONS_TOTAL: &str = "praxis_upstream_health_transitions_total";
const CONFIG_RELOAD_TOTAL: &str = "praxis_config_reload_total";
const CONFIG_RELOAD_LAST_SUCCESS_TIMESTAMP: &str = "praxis_config_reload_last_success_timestamp";
const ERRORS_TOTAL: &str = "praxis_errors_total";
pub(crate) const ERROR_TYPE_FILTER_REJECT: &str = "filter_reject";
pub(crate) const ERROR_TYPE_TIMEOUT: &str = "timeout";
pub(crate) const ERROR_TYPE_UPSTREAM_UNAVAILABLE: &str = "upstream_unavailable";
pub(crate) const ERROR_TYPE_UPSTREAM_PROTOCOL: &str = "upstream_protocol";
pub(crate) const ERROR_TYPE_DOWNSTREAM: &str = "downstream";
pub(crate) const ERROR_TYPE_INTERNAL: &str = "internal";
pub(crate) const OVERLOAD_REASON_MEMORY: &str = "memory";
pub(crate) const OVERLOAD_REASON_GLOBAL_CONNECTIONS: &str = "global_connections";
pub(crate) const OVERLOAD_REASON_LISTENER_CONNECTIONS: &str = "listener_connections";
pub(crate) const OVERLOAD_REASON_FILE_DESCRIPTORS: &str = "file_descriptors";
pub(crate) const RETRY_RESULT_SUCCESS: &str = "success";
pub(crate) const RETRY_RESULT_EXHAUSTED: &str = "exhausted";
pub(crate) const HEALTH_RESULT_HEALTHY: &str = "healthy";
pub(crate) const HEALTH_RESULT_UNHEALTHY: &str = "unhealthy";
pub(crate) const RELOAD_RESULT_SUCCESS: &str = "success";
pub(crate) const RELOAD_RESULT_FAILURE: &str = "failure";
const BODY_SIZE_BUCKETS_BYTES: &[f64] = &[
64.0,
256.0,
1_024.0,
4_096.0,
16_384.0,
65_536.0,
262_144.0,
1_048_576.0,
10_485_760.0,
];
static PROMETHEUS_HANDLE: OnceLock<PrometheusHandle> = OnceLock::new();
static LABEL_CONFIG: OnceLock<MetricLabelsConfig> = OnceLock::new();
static ALL_LABELS: OnceLock<MetricLabelsConfig> = OnceLock::new();
pub fn install_metric_labels(labels: MetricLabelsConfig) {
let _existing = LABEL_CONFIG.set(labels);
}
pub(crate) fn metric_labels() -> &'static MetricLabelsConfig {
LABEL_CONFIG
.get()
.unwrap_or_else(|| ALL_LABELS.get_or_init(MetricLabelsConfig::default))
}
fn selected_labels(pairs: &[(&'static str, Option<SharedString>)]) -> Vec<Label> {
pairs
.iter()
.filter_map(|(name, value)| value.clone().map(|value| Label::new(*name, value)))
.collect()
}
fn label_if(enabled: bool, value: SharedString) -> Option<SharedString> {
enabled.then_some(value)
}
pub fn install_prometheus_recorder() -> &'static PrometheusHandle {
#[expect(
clippy::expect_used,
reason = "recorder installation is a one-time startup operation"
)]
PROMETHEUS_HANDLE.get_or_init(|| {
PrometheusBuilder::new()
.set_buckets_for_metric(
Matcher::Full(HTTP_REQUEST_BODY_BYTES.to_owned()),
BODY_SIZE_BUCKETS_BYTES,
)
.expect("body request histogram buckets must be non-empty")
.set_buckets_for_metric(
Matcher::Full(HTTP_RESPONSE_BODY_BYTES.to_owned()),
BODY_SIZE_BUCKETS_BYTES,
)
.expect("body response histogram buckets must be non-empty")
.install_recorder()
.expect("failed to install Prometheus recorder")
})
}
pub fn render_prometheus() -> Option<String> {
PROMETHEUS_HANDLE.get().map(PrometheusHandle::render)
}
pub(crate) fn is_recorder_installed() -> bool {
PROMETHEUS_HANDLE.get().is_some()
}
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct StatsMetricsSnapshot {
pub http_active_by_listener: std::collections::HashMap<String, u64>,
pub http_active_aggregate: Option<u64>,
pub tcp_active_by_listener: std::collections::HashMap<String, u64>,
pub tcp_active_aggregate: Option<u64>,
pub upstream_requests_by_cluster: std::collections::HashMap<String, u64>,
pub upstream_requests_aggregate: Option<u64>,
pub connect_failures_by_cluster: std::collections::HashMap<String, u64>,
pub connect_failures_aggregate: Option<u64>,
}
pub fn collect_stats_metrics(prometheus_text: &str) -> StatsMetricsSnapshot {
let mut snapshot = StatsMetricsSnapshot::default();
for line in prometheus_text.lines() {
let line = line.trim();
if line.is_empty() || line.starts_with('#') {
continue;
}
if let Some((name, labels, value)) = parse_prometheus_sample(line) {
match name {
HTTP_ACTIVE_REQUESTS => snapshot.record_http_active(&labels, value),
TCP_ACTIVE_CONNECTIONS => snapshot.record_tcp_active(&labels, value),
UPSTREAM_REQUESTS_TOTAL => snapshot.add_upstream_requests(&labels, value),
UPSTREAM_CONNECT_FAILURES_TOTAL => snapshot.add_connect_failures(&labels, value),
_ => {},
}
}
}
snapshot
}
impl StatsMetricsSnapshot {
fn record_http_active(&mut self, labels: &std::collections::HashMap<String, String>, value: u64) {
record_by_listener(
&mut self.http_active_by_listener,
&mut self.http_active_aggregate,
labels,
value,
);
}
fn record_tcp_active(&mut self, labels: &std::collections::HashMap<String, String>, value: u64) {
record_by_listener(
&mut self.tcp_active_by_listener,
&mut self.tcp_active_aggregate,
labels,
value,
);
}
fn add_upstream_requests(&mut self, labels: &std::collections::HashMap<String, String>, value: u64) {
accumulate_by_cluster(
&mut self.upstream_requests_by_cluster,
&mut self.upstream_requests_aggregate,
labels,
value,
);
}
fn add_connect_failures(&mut self, labels: &std::collections::HashMap<String, String>, value: u64) {
accumulate_by_cluster(
&mut self.connect_failures_by_cluster,
&mut self.connect_failures_aggregate,
labels,
value,
);
}
}
fn record_by_listener(
by_listener: &mut std::collections::HashMap<String, u64>,
aggregate: &mut Option<u64>,
labels: &std::collections::HashMap<String, String>,
value: u64,
) {
match labels.get("listener") {
Some(listener) => {
by_listener.insert(listener.clone(), value);
},
None => *aggregate = Some(value),
}
}
fn accumulate_by_cluster(
by_cluster: &mut std::collections::HashMap<String, u64>,
aggregate: &mut Option<u64>,
labels: &std::collections::HashMap<String, String>,
value: u64,
) {
match labels.get("cluster") {
Some(cluster) => {
*by_cluster.entry(cluster.clone()).or_insert(0) += value;
},
None => *aggregate = Some(aggregate.unwrap_or(0) + value),
}
}
type PrometheusSample<'a> = (&'a str, std::collections::HashMap<String, String>, u64);
fn parse_prometheus_sample(line: &str) -> Option<PrometheusSample<'_>> {
let (name_and_labels, value_str) = line.rsplit_once(' ')?;
#[expect(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
reason = "Prometheus counter/gauge values are non-negative integers"
)]
let value = {
let parsed = value_str.parse::<f64>().ok()?;
parsed.round() as u64
};
let (name, labels) = if let Some((name, label_blob)) = name_and_labels.split_once('{') {
let label_blob = label_blob.strip_suffix('}')?;
(name, parse_prometheus_labels(label_blob))
} else {
(name_and_labels, std::collections::HashMap::new())
};
Some((name, labels, value))
}
fn parse_prometheus_labels(input: &str) -> std::collections::HashMap<String, String> {
let mut labels = std::collections::HashMap::new();
let mut rest = input.trim();
while !rest.is_empty() {
let (pair, tail) = split_prometheus_label_pair(rest);
if let Some((key, value)) = parse_prometheus_label_pair(pair) {
labels.insert(key, value);
}
rest = tail;
}
labels
}
fn split_prometheus_label_pair(input: &str) -> (&str, &str) {
let bytes = input.as_bytes();
let mut in_quotes = false;
for (index, byte) in bytes.iter().enumerate() {
match *byte {
b'"' => in_quotes = !in_quotes,
b',' if !in_quotes => {
let head = input.get(..index).unwrap_or(input);
let tail = input.get(index + 1..).unwrap_or("").trim_start();
return (head, tail);
},
_ => {},
}
}
(input, "")
}
fn parse_prometheus_label_pair(pair: &str) -> Option<(String, String)> {
let (key, value) = pair.split_once('=')?;
let value = value.strip_prefix('"')?.strip_suffix('"')?;
Some((key.to_owned(), value.to_owned()))
}
pub fn status_class(code: u16) -> &'static str {
match code {
100..=199 => "1xx",
200..=299 => "2xx",
300..=399 => "3xx",
400..=499 => "4xx",
500..=599 => "5xx",
_ => "unknown",
}
}
pub fn method_label(method: &str) -> &'static str {
match method {
"GET" => "GET",
"POST" => "POST",
"PUT" => "PUT",
"DELETE" => "DELETE",
"PATCH" => "PATCH",
"HEAD" => "HEAD",
"OPTIONS" => "OPTIONS",
"TRACE" => "TRACE",
"CONNECT" => "CONNECT",
_ => "OTHER",
}
}
pub(crate) struct RequestMetricLabels {
pub cluster: SharedString,
pub method: &'static str,
pub route: SharedString,
pub status_class: &'static str,
}
fn selected_request_labels(labels: RequestMetricLabels) -> Vec<Label> {
let selected = metric_labels();
let pairs = [
(
"method",
label_if(
selected.is_enabled(MetricLabel::Method),
SharedString::const_str(labels.method),
),
),
(
"status_class",
label_if(
selected.is_enabled(MetricLabel::StatusClass),
SharedString::const_str(labels.status_class),
),
),
("route", label_if(selected.is_enabled(MetricLabel::Route), labels.route)),
(
"cluster",
label_if(selected.is_enabled(MetricLabel::Cluster), labels.cluster),
),
];
selected_labels(&pairs)
}
pub(crate) fn record_request_metrics(labels: RequestMetricLabels, duration_secs: f64) {
if !is_recorder_installed() {
return;
}
if !metric_labels().all_enabled() {
let emitted = selected_request_labels(labels);
counter!(HTTP_REQUESTS_TOTAL, emitted.clone()).increment(1);
histogram!(HTTP_REQUEST_DURATION_SECONDS, emitted).record(duration_secs);
return;
}
record_request_metrics_all_labels(labels, duration_secs);
}
fn record_request_metrics_all_labels(labels: RequestMetricLabels, duration_secs: f64) {
let cluster = labels.cluster;
let route = labels.route;
counter!(
HTTP_REQUESTS_TOTAL,
"method" => labels.method,
"status_class" => labels.status_class,
"route" => route.clone(),
"cluster" => cluster.clone()
)
.increment(1);
histogram!(
HTTP_REQUEST_DURATION_SECONDS,
"method" => labels.method,
"status_class" => labels.status_class,
"route" => route,
"cluster" => cluster
)
.record(duration_secs);
}
fn selected_body_labels(method: &'static str, status_class: &'static str, cluster: SharedString) -> Vec<Label> {
let selected = metric_labels();
let pairs = [
(
"method",
label_if(
selected.is_enabled(MetricLabel::Method),
SharedString::const_str(method),
),
),
(
"status_class",
label_if(
selected.is_enabled(MetricLabel::StatusClass),
SharedString::const_str(status_class),
),
),
("cluster", label_if(selected.is_enabled(MetricLabel::Cluster), cluster)),
];
selected_labels(&pairs)
}
fn record_body_size_all_labels(
method: &'static str,
status_class: &'static str,
cluster: SharedString,
request_bytes: f64,
response_bytes: f64,
) {
histogram!(
HTTP_REQUEST_BODY_BYTES,
"method" => method,
"status_class" => status_class,
"cluster" => cluster.clone()
)
.record(request_bytes);
histogram!(
HTTP_RESPONSE_BODY_BYTES,
"method" => method,
"status_class" => status_class,
"cluster" => cluster
)
.record(response_bytes);
}
pub(crate) fn record_body_size_metrics(
method: &'static str,
status_class: &'static str,
cluster: SharedString,
request_body_bytes: u64,
response_body_bytes: u64,
) {
if !is_recorder_installed() {
return;
}
#[expect(
clippy::cast_precision_loss,
reason = "body byte counts as histogram observations; exact integer precision not required"
)]
let (request_bytes, response_bytes) = (request_body_bytes as f64, response_body_bytes as f64);
if !metric_labels().all_enabled() {
let emitted = selected_body_labels(method, status_class, cluster);
histogram!(HTTP_REQUEST_BODY_BYTES, emitted.clone()).record(request_bytes);
histogram!(HTTP_RESPONSE_BODY_BYTES, emitted).record(response_bytes);
return;
}
record_body_size_all_labels(method, status_class, cluster, request_bytes, response_bytes);
}
pub struct ActiveRequestGuard {
listener: SharedString,
}
impl ActiveRequestGuard {
pub(crate) fn acquire(listener: SharedString) -> Self {
if is_recorder_installed() {
if metric_labels().is_enabled(MetricLabel::Listener) {
gauge!(HTTP_ACTIVE_REQUESTS, "listener" => listener.clone()).increment(1.0);
} else {
gauge!(HTTP_ACTIVE_REQUESTS).increment(1.0);
}
}
Self { listener }
}
}
impl Drop for ActiveRequestGuard {
fn drop(&mut self) {
if is_recorder_installed() {
if metric_labels().is_enabled(MetricLabel::Listener) {
gauge!(HTTP_ACTIVE_REQUESTS, "listener" => self.listener.clone()).decrement(1.0);
} else {
gauge!(HTTP_ACTIVE_REQUESTS).decrement(1.0);
}
}
}
}
pub(crate) fn record_error(error_type: &'static str) {
if !is_recorder_installed() {
return;
}
counter!(ERRORS_TOTAL, "type" => error_type).increment(1);
}
pub(crate) fn error_type_for(etype: &::pingora_core::ErrorType, source: &::pingora_core::ErrorSource) -> &'static str {
use ::pingora_core::ErrorSource::{Downstream, Internal, Unset};
if matches!(source, Downstream) {
return ERROR_TYPE_DOWNSTREAM;
}
if is_timeout(etype) {
return ERROR_TYPE_TIMEOUT;
}
if is_unreachable(etype) {
return ERROR_TYPE_UPSTREAM_UNAVAILABLE;
}
if matches!(source, Internal | Unset) {
return ERROR_TYPE_INTERNAL;
}
ERROR_TYPE_UPSTREAM_PROTOCOL
}
fn is_timeout(etype: &::pingora_core::ErrorType) -> bool {
use ::pingora_core::ErrorType::{ConnectTimedout, ReadTimedout, TLSHandshakeTimedout, WriteTimedout};
matches!(
etype,
ConnectTimedout | TLSHandshakeTimedout | ReadTimedout | WriteTimedout
)
}
fn is_unreachable(etype: &::pingora_core::ErrorType) -> bool {
use ::pingora_core::ErrorType::{BindError, ConnectError, ConnectNoRoute, ConnectRefused, SocketError};
matches!(
etype,
ConnectRefused | ConnectNoRoute | ConnectError | BindError | SocketError
)
}
pub(crate) fn record_overload_reject(reason: &'static str) {
if !is_recorder_installed() {
return;
}
counter!(OVERLOAD_REJECTS_TOTAL, "reason" => reason).increment(1);
}
pub fn set_process_fd_gauges(usage: praxis_core::fd::FdUsage) {
if !is_recorder_installed() {
return;
}
#[expect(clippy::cast_precision_loss, reason = "descriptor counts fit f64 exactly below 2^53")]
{
gauge!(PROCESS_OPEN_FDS).set(usage.open as f64);
gauge!(PROCESS_MAX_FDS).set(usage.limit as f64);
}
}
pub(crate) fn record_upstream_connect_duration(cluster: SharedString, duration_secs: f64) {
if !is_recorder_installed() {
return;
}
if metric_labels().is_enabled(MetricLabel::Cluster) {
histogram!(UPSTREAM_CONNECT_DURATION_SECONDS, "cluster" => cluster).record(duration_secs);
} else {
histogram!(UPSTREAM_CONNECT_DURATION_SECONDS).record(duration_secs);
}
}
pub(crate) fn record_upstream_request(cluster: SharedString, endpoint: SharedString, status_class: &'static str) {
if !is_recorder_installed() {
return;
}
let selected = metric_labels();
if !selected.all_enabled() {
let pairs = [
("cluster", label_if(selected.is_enabled(MetricLabel::Cluster), cluster)),
(
"endpoint",
label_if(selected.is_enabled(MetricLabel::Endpoint), endpoint),
),
(
"status_class",
label_if(
selected.is_enabled(MetricLabel::StatusClass),
SharedString::const_str(status_class),
),
),
];
counter!(UPSTREAM_REQUESTS_TOTAL, selected_labels(&pairs)).increment(1);
return;
}
counter!(
UPSTREAM_REQUESTS_TOTAL,
"cluster" => cluster,
"endpoint" => endpoint,
"status_class" => status_class
)
.increment(1);
}
pub(crate) fn record_upstream_connect_failure(cluster: SharedString) {
if !is_recorder_installed() {
return;
}
if metric_labels().is_enabled(MetricLabel::Cluster) {
counter!(UPSTREAM_CONNECT_FAILURES_TOTAL, "cluster" => cluster).increment(1);
} else {
counter!(UPSTREAM_CONNECT_FAILURES_TOTAL).increment(1);
}
}
pub(crate) fn record_upstream_retry(cluster: SharedString, result: &'static str) {
if !is_recorder_installed() {
return;
}
if metric_labels().is_enabled(MetricLabel::Cluster) {
counter!(UPSTREAM_RETRIES_TOTAL, "cluster" => cluster, "result" => result).increment(1);
} else {
counter!(UPSTREAM_RETRIES_TOTAL, "result" => result).increment(1);
}
}
pub(crate) fn set_upstream_endpoint_gauges(cluster: SharedString, healthy: usize, total: usize) {
if !is_recorder_installed() {
return;
}
#[expect(clippy::cast_precision_loss, reason = "endpoint counts fit f64 exactly below 2^53")]
{
gauge!(UPSTREAM_HEALTHY_ENDPOINTS, "cluster" => cluster.clone()).set(healthy as f64);
gauge!(UPSTREAM_TOTAL_ENDPOINTS, "cluster" => cluster).set(total as f64);
}
}
pub fn clear_stale_upstream_health_gauges<'a, P: IntoIterator<Item = &'a str>, C: IntoIterator<Item = &'a str>>(
previous_health_clusters: P,
current_health_clusters: C,
) {
if !is_recorder_installed() {
return;
}
let current: std::collections::HashSet<&str> = current_health_clusters.into_iter().collect();
for name in previous_health_clusters {
if !current.contains(name) {
set_upstream_endpoint_gauges(SharedString::from(name.to_owned()), 0, 0);
}
}
}
pub fn seed_upstream_health_gauges(registry: &praxis_core::health::HealthRegistry) {
if !is_recorder_installed() {
return;
}
for (name, state) in registry.iter() {
let (healthy, total) = state.endpoint_counts();
set_upstream_endpoint_gauges(SharedString::from(name.as_ref().to_owned()), healthy, total);
}
}
pub(crate) fn record_health_transition(cluster: SharedString, result: &'static str, healthy: usize, total: usize) {
if !is_recorder_installed() {
return;
}
if metric_labels().is_enabled(MetricLabel::Cluster) {
counter!(
UPSTREAM_HEALTH_TRANSITIONS_TOTAL,
"cluster" => cluster.clone(),
"result" => result
)
.increment(1);
} else {
counter!(UPSTREAM_HEALTH_TRANSITIONS_TOTAL, "result" => result).increment(1);
}
set_upstream_endpoint_gauges(cluster, healthy, total);
}
pub(crate) fn count_healthy_endpoints(health: &praxis_core::health::ClusterHealthEntry) -> (usize, usize) {
health.endpoint_counts()
}
pub fn record_config_reload_success() {
if !is_recorder_installed() {
return;
}
counter!(CONFIG_RELOAD_TOTAL, "result" => RELOAD_RESULT_SUCCESS).increment(1);
let ts = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(0.0, |d| d.as_secs_f64());
gauge!(CONFIG_RELOAD_LAST_SUCCESS_TIMESTAMP).set(ts);
}
pub fn record_config_reload_failure() {
if !is_recorder_installed() {
return;
}
counter!(CONFIG_RELOAD_TOTAL, "result" => RELOAD_RESULT_FAILURE).increment(1);
}
pub(crate) fn cluster_none() -> SharedString {
SharedString::const_str("none")
}
pub(crate) fn route_unknown() -> SharedString {
SharedString::const_str("unknown")
}
#[cfg(test)]
#[expect(clippy::allow_attributes, reason = "blanket test suppressions")]
#[allow(clippy::unwrap_used, clippy::expect_used, clippy::indexing_slicing, reason = "tests")]
mod tests {
use super::*;
#[test]
fn status_class_1xx() {
assert_eq!(status_class(100), "1xx", "100 should be 1xx");
assert_eq!(status_class(199), "1xx", "199 should be 1xx");
}
#[test]
fn status_class_2xx() {
assert_eq!(status_class(200), "2xx", "200 should be 2xx");
assert_eq!(status_class(204), "2xx", "204 should be 2xx");
assert_eq!(status_class(299), "2xx", "299 should be 2xx");
}
#[test]
fn status_class_3xx() {
assert_eq!(status_class(301), "3xx", "301 should be 3xx");
assert_eq!(status_class(399), "3xx", "399 should be 3xx");
}
#[test]
fn status_class_4xx() {
assert_eq!(status_class(400), "4xx", "400 should be 4xx");
assert_eq!(status_class(404), "4xx", "404 should be 4xx");
assert_eq!(status_class(499), "4xx", "499 should be 4xx");
}
#[test]
fn status_class_5xx() {
assert_eq!(status_class(500), "5xx", "500 should be 5xx");
assert_eq!(status_class(503), "5xx", "503 should be 5xx");
assert_eq!(status_class(599), "5xx", "599 should be 5xx");
}
#[test]
fn status_class_zero_is_unknown() {
assert_eq!(status_class(0), "unknown", "0 should be unknown");
}
#[test]
fn status_class_out_of_range_is_unknown() {
assert_eq!(status_class(600), "unknown", "600 should be unknown");
assert_eq!(status_class(99), "unknown", "99 should be unknown");
}
#[test]
fn method_label_standard_methods() {
for m in [
"GET", "POST", "PUT", "DELETE", "PATCH", "HEAD", "OPTIONS", "TRACE", "CONNECT",
] {
assert_eq!(method_label(m), m, "{m} should pass through");
}
}
#[test]
fn method_label_custom_methods_collapse_to_other() {
assert_eq!(method_label("PURGE"), "OTHER", "PURGE should be OTHER");
assert_eq!(method_label("FOOBAR"), "OTHER", "FOOBAR should be OTHER");
assert_eq!(method_label(""), "OTHER", "empty should be OTHER");
}
#[test]
fn record_utilities_noop_without_recorder() {
record_overload_reject(OVERLOAD_REASON_MEMORY);
record_upstream_connect_failure(cluster_none());
record_error(ERROR_TYPE_INTERNAL);
record_upstream_request(cluster_none(), SharedString::const_str("10.0.0.1:80"), "2xx");
record_upstream_retry(cluster_none(), RETRY_RESULT_SUCCESS);
record_upstream_connect_duration(cluster_none(), 0.01);
set_upstream_endpoint_gauges(cluster_none(), 1, 2);
record_health_transition(cluster_none(), HEALTH_RESULT_HEALTHY, 1, 2);
record_config_reload_success();
record_config_reload_failure();
clear_stale_upstream_health_gauges(["gone"], std::iter::empty::<&str>());
let _request_guard = ActiveRequestGuard::acquire(SharedString::const_str("test"));
}
#[test]
fn active_request_guard_returns_to_zero_on_drop() {
install_prometheus_recorder();
let listener = SharedString::const_str("active-request-guard-listener");
let guard = ActiveRequestGuard::acquire(listener.clone());
let held = render_prometheus().expect("recorder should render");
assert!(
held.contains("praxis_http_active_requests{listener=\"active-request-guard-listener\"} 1"),
"gauge should read 1 while the guard is held:\n{held}"
);
drop(guard);
let released = render_prometheus().expect("recorder should render");
assert!(
released.contains("praxis_http_active_requests{listener=\"active-request-guard-listener\"} 0"),
"gauge should return to 0 once the guard drops:\n{released}"
);
}
#[test]
fn overload_reject_reasons_appear_in_scrape() {
install_prometheus_recorder();
record_overload_reject(OVERLOAD_REASON_MEMORY);
record_overload_reject(OVERLOAD_REASON_GLOBAL_CONNECTIONS);
record_overload_reject(OVERLOAD_REASON_LISTENER_CONNECTIONS);
record_overload_reject(OVERLOAD_REASON_FILE_DESCRIPTORS);
let body = render_prometheus().expect("recorder should render");
for reason in [
OVERLOAD_REASON_MEMORY,
OVERLOAD_REASON_GLOBAL_CONNECTIONS,
OVERLOAD_REASON_LISTENER_CONNECTIONS,
OVERLOAD_REASON_FILE_DESCRIPTORS,
] {
let needle = format!("praxis_overload_rejects_total{{reason=\"{reason}\"}}");
assert!(body.contains(&needle), "expected `{needle}` in scrape:\n{body}");
}
}
#[test]
fn process_fd_gauges_appear_in_scrape() {
install_prometheus_recorder();
set_process_fd_gauges(praxis_core::fd::FdUsage { limit: 1_024, open: 37 });
let body = render_prometheus().expect("recorder should render");
assert!(
body.contains("praxis_process_open_fds 37"),
"open descriptors gauge expected in scrape:\n{body}"
);
assert!(
body.contains("praxis_process_max_fds 1024"),
"descriptor limit gauge expected in scrape:\n{body}"
);
}
#[test]
fn upstream_requests_carry_cluster_endpoint_and_status_class() {
install_prometheus_recorder();
record_upstream_request(
SharedString::const_str("api"),
SharedString::const_str("10.0.0.7:8080"),
"5xx",
);
let body = render_prometheus().expect("recorder should render");
assert!(
body.contains(
"praxis_upstream_requests_total{cluster=\"api\",endpoint=\"10.0.0.7:8080\",status_class=\"5xx\"} 1"
),
"counter should carry all three labels:\n{body}"
);
}
#[test]
fn selected_labels_drops_disabled_dimensions() {
let pairs = [
("method", Some(SharedString::const_str("GET"))),
("route", None),
("cluster", Some(SharedString::const_str("api"))),
];
let emitted = selected_labels(&pairs);
let names: Vec<&str> = emitted.iter().map(Label::key).collect();
assert_eq!(names, vec!["method", "cluster"], "a disabled dimension must be absent");
}
#[test]
fn selected_labels_preserves_order_and_values() {
let pairs = [
("cluster", Some(SharedString::const_str("api"))),
("endpoint", Some(SharedString::const_str("10.0.0.1:80"))),
];
let emitted = selected_labels(&pairs);
let rendered: Vec<(&str, &str)> = emitted.iter().map(|l| (l.key(), l.value())).collect();
assert_eq!(
rendered,
vec![("cluster", "api"), ("endpoint", "10.0.0.1:80")],
"enabled dimensions keep their order and values"
);
}
#[test]
fn label_if_gates_on_the_flag() {
assert_eq!(
label_if(true, SharedString::const_str("x")).as_deref(),
Some("x"),
"an enabled dimension keeps its value"
);
assert_eq!(
label_if(false, SharedString::const_str("x")),
None,
"a disabled dimension yields no value"
);
}
#[test]
fn metric_labels_default_to_all_enabled() {
assert!(
metric_labels().all_enabled(),
"without an explicit install every dimension must stay on, so the \
recorders keep their allocation-free fast path"
);
}
#[test]
fn error_types_appear_in_scrape() {
install_prometheus_recorder();
for error_type in [
ERROR_TYPE_FILTER_REJECT,
ERROR_TYPE_TIMEOUT,
ERROR_TYPE_UPSTREAM_UNAVAILABLE,
ERROR_TYPE_UPSTREAM_PROTOCOL,
ERROR_TYPE_DOWNSTREAM,
ERROR_TYPE_INTERNAL,
] {
record_error(error_type);
let body = render_prometheus().expect("recorder should render");
let needle = format!("praxis_errors_total{{type=\"{error_type}\"}}");
assert!(body.contains(&needle), "expected `{needle}` in scrape:\n{body}");
}
}
#[test]
fn error_type_for_maps_pingora_errors_to_bounded_values() {
use ::pingora_core::{ErrorSource, ErrorType};
assert_eq!(
error_type_for(&ErrorType::ConnectTimedout, &ErrorSource::Upstream),
ERROR_TYPE_TIMEOUT,
"connect timeout is a timeout"
);
assert_eq!(
error_type_for(&ErrorType::ConnectRefused, &ErrorSource::Upstream),
ERROR_TYPE_UPSTREAM_UNAVAILABLE,
"a refused connect means the upstream was unreachable"
);
assert_eq!(
error_type_for(&ErrorType::ReadError, &ErrorSource::Upstream),
ERROR_TYPE_UPSTREAM_PROTOCOL,
"a mid-exchange read error is a protocol failure"
);
assert_eq!(
error_type_for(&ErrorType::ReadTimedout, &ErrorSource::Downstream),
ERROR_TYPE_DOWNSTREAM,
"downstream source wins over the error kind"
);
assert_eq!(
error_type_for(&ErrorType::InternalError, &ErrorSource::Internal),
ERROR_TYPE_INTERNAL,
"internal source is an internal fault"
);
}
#[test]
fn body_size_histograms_use_byte_buckets() {
install_prometheus_recorder();
record_body_size_metrics("GET", "2xx", cluster_none(), 500, 4_000);
let body = render_prometheus().expect("recorder should render");
assert!(
body.contains("praxis_http_request_body_bytes_bucket") && body.contains("le=\"1024\""),
"request body histogram should use byte buckets, not duration defaults:\n{body}"
);
assert!(
body.contains("praxis_http_response_body_bytes_bucket") && body.contains("le=\"4096\""),
"response body histogram should use byte buckets:\n{body}"
);
assert!(
!body.contains("praxis_http_request_body_bytes_bucket{le=\"0.005\"}")
&& !body.contains("praxis_http_request_body_bytes_bucket{method=\"GET\",status_class=\"2xx\",cluster=\"\",le=\"0.005\"}"),
"request body histogram must not use duration default buckets:\n{body}"
);
}
#[test]
fn clear_stale_upstream_health_gauges_zeros_removed_clusters() {
install_prometheus_recorder();
set_upstream_endpoint_gauges(SharedString::from("old-cluster".to_owned()), 2, 3);
set_upstream_endpoint_gauges(SharedString::from("kept-cluster".to_owned()), 1, 1);
clear_stale_upstream_health_gauges(["old-cluster", "kept-cluster"], ["kept-cluster"]);
let body = render_prometheus().expect("recorder should render");
assert!(
body.contains("praxis_upstream_healthy_endpoints{cluster=\"old-cluster\"} 0"),
"removed cluster healthy gauge should be zeroed:\n{body}"
);
assert!(
body.contains("praxis_upstream_total_endpoints{cluster=\"old-cluster\"} 0"),
"removed cluster total gauge should be zeroed:\n{body}"
);
assert!(
body.contains("praxis_upstream_healthy_endpoints{cluster=\"kept-cluster\"} 1"),
"kept cluster should retain its value:\n{body}"
);
}
#[test]
fn collect_stats_metrics_sums_cluster_counters() {
let text = r#"
praxis_http_active_requests{listener="web"} 2
praxis_tcp_active_connections{listener="tcp-in"} 1
praxis_upstream_requests_total{cluster="backend",endpoint="127.0.0.1:1",status_class="2xx"} 3
praxis_upstream_requests_total{cluster="backend",endpoint="127.0.0.1:2",status_class="5xx"} 1
praxis_upstream_connect_failures_total{cluster="backend"} 2
"#;
let snap = collect_stats_metrics(text);
assert_eq!(
snap.http_active_by_listener.get("web"),
Some(&2),
"HTTP active per listener should parse"
);
assert_eq!(
snap.tcp_active_by_listener.get("tcp-in"),
Some(&1),
"TCP active per listener should parse"
);
assert_eq!(
snap.upstream_requests_by_cluster.get("backend"),
Some(&4),
"upstream requests should sum by cluster"
);
assert_eq!(
snap.connect_failures_by_cluster.get("backend"),
Some(&2),
"connect failures should parse by cluster"
);
}
#[test]
fn parse_prometheus_labels_handles_commas_inside_quoted_values() {
let labels = parse_prometheus_labels(r#"tag="a,b",listener="web""#);
assert_eq!(labels.get("tag"), Some(&"a,b".to_owned()), "comma inside quotes");
assert_eq!(labels.get("listener"), Some(&"web".to_owned()), "second label");
}
#[test]
fn collect_stats_metrics_parses_unlabeled_upstream_counters() {
let text = r#"
praxis_upstream_requests_total{status_class="2xx"} 5
praxis_upstream_connect_failures_total 2
"#;
let snap = collect_stats_metrics(text);
assert_eq!(
snap.upstream_requests_aggregate,
Some(5),
"unlabeled upstream requests should aggregate"
);
assert_eq!(
snap.connect_failures_aggregate,
Some(2),
"unlabeled connect failures should aggregate"
);
}
#[test]
fn seed_upstream_health_gauges_publishes_registry_counts() {
use std::sync::Arc;
use praxis_core::health::{ClusterHealthEntry, EndpointHealth};
install_prometheus_recorder();
let endpoints = vec![EndpointHealth::new(), EndpointHealth::new()];
endpoints[0].mark_unhealthy();
let entry = Arc::new(ClusterHealthEntry::new(
endpoints,
vec![Arc::from("a:1"), Arc::from("b:1")],
None,
None,
));
let registry = Arc::new([(Arc::from("backend"), entry)].into_iter().collect());
seed_upstream_health_gauges(®istry);
let body = render_prometheus().expect("recorder should render");
assert!(
body.contains("praxis_upstream_healthy_endpoints{cluster=\"backend\"} 1"),
"seed should publish healthy count:\n{body}"
);
assert!(
body.contains("praxis_upstream_total_endpoints{cluster=\"backend\"} 2"),
"seed should publish total count:\n{body}"
);
}
#[test]
fn count_healthy_endpoints_counts_correctly() {
use std::sync::Arc;
use praxis_core::health::{ClusterHealthEntry, EndpointHealth};
let endpoints = vec![EndpointHealth::new(), EndpointHealth::new(), EndpointHealth::new()];
endpoints[1].mark_unhealthy();
let entry = ClusterHealthEntry::new(
endpoints,
vec![Arc::from("a:1"), Arc::from("b:1"), Arc::from("c:1")],
None,
None,
);
let (healthy, total) = count_healthy_endpoints(&entry);
assert_eq!(total, 3, "total should be 3");
assert_eq!(healthy, 2, "two endpoints should be healthy");
}
}