gru/RustProxy
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.34.2

Browse Source
This commit is contained in:
Mateusz Gruszczyński committed 2026-10-19 15:40:37 +02:00
1 parent 34fb534e02
commit b2b5b45d8a
22 files changed
+2146 -107
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/admin/api_token_access.rs
+2
View File
@@ -20,6 +20,8 @@ fn general_api_token_allowed(method: &Method, path: &str, permission: &str) -> b
| "/api/file-logs"
| "/api/file-logs/query"
| "/api/metrics"
| "/api/metrics/json"
| "/api/metrics/summary"
| "/api/cache/bans"
);
let write = permission == "rw"
src/admin/logging_api.rs
+12 -2
View File
@@ -298,6 +298,12 @@ async fn audit(State(state): State<Arc<App>>) -> Result<Json<Value>, ApiError> {
async fn metrics(State(state): State<Arc<App>>) -> Response {
crate::prometheus::response(&state)
}
async fn metrics_json(State(state): State<Arc<App>>) -> Response {
crate::prometheus::json_response(&state)
}
async fn metrics_summary(State(state): State<Arc<App>>) -> Response {
crate::prometheus::summary_response(&state)
}
#[derive(Default, Deserialize)]
#[serde(default, deny_unknown_fields)]
struct EventsQuery {
@@ -311,7 +317,10 @@ async fn events(
) -> Result<Response, ApiError> {
let interval = query.interval.unwrap_or(3);
if ![1, 2, 3, 5, 10].contains(&interval) {
return Err(ApiError(StatusCode::UNPROCESSABLE_ENTITY, "invalid SSE interval".into()));
return Err(ApiError(
StatusCode::UNPROCESSABLE_ENTITY,
"invalid SSE interval".into(),
));
}
let secret = auth::cookie_secret(&headers)
.ok_or_else(|| ApiError(StatusCode::UNAUTHORIZED, "auth.required".into()))?;
@@ -349,7 +358,8 @@ async fn events(
_ => return None,
}
let payload =
json!({"status":live_status_payload(&state), "logs":logs_payload(&state)}).to_string();
json!({"status":live_status_payload(&state), "logs":logs_payload(&state)})
.to_string();
let event = Event::default().event("snapshot").data(payload);
Some((
Ok::<Event, Infallible>(event),
src/admin/openapi.json
+692 -1
View File
@@ -6801,6 +6801,557 @@
"database_path"
],
"additionalProperties": false
},
"MonitoringMetricSample": {
"type": "object",
"description": "Single metric sample. Values are JSON numbers; non-finite Prometheus values are null. Unlabelled samples use an empty labels object.",
"properties": {
"labels": {
"type": "object",
"additionalProperties": {
"type": "string"
}
},
"value": {
"type": "number",
"nullable": true
}
},
"required": [
"labels",
"value"
],
"additionalProperties": false
},
"MonitoringMetricFamilies": {
"type": "object",
"description": "All numeric samples returned by the same collectors as the Prometheus exporter, including histogram buckets, counts and sums.",
"properties": {
"types": {
"type": "object",
"additionalProperties": {
"type": "string"
},
"description": "Metric-family name to Prometheus TYPE."
},
"samples": {
"type": "object",
"additionalProperties": {
"type": "array",
"items": {
"$ref": "#/components/schemas/MonitoringMetricSample"
}
}
}
},
"required": [
"types",
"samples"
],
"additionalProperties": false
},
"MonitoringSnapshot": {
"type": "object",
"description": "Pull-only operational snapshot with a precomputed summary and all Prometheus samples. Summary per-second rates are computed over bounded polling intervals; the first sample has null rates. OS gauges may be null until the background collector samples.",
"properties": {
"version": {
"type": "string"
},
"collected_at_ms": {
"type": "integer",
"format": "int64"
},
"uptime_seconds": {
"type": "integer",
"format": "int64"
},
"revision": {
"type": "integer",
"format": "int64"
},
"vhosts": {
"type": "object",
"properties": {
"total": {
"type": "integer"
},
"enabled": {
"type": "integer"
},
"disabled": {
"type": "integer"
},
"items": {
"type": "array",
"items": {
"type": "object"
}
}
},
"required": [
"total",
"enabled",
"disabled",
"items"
]
},
"redirect_hosts": {
"type": "object"
},
"not_found_hosts": {
"type": "object"
},
"requests": {
"type": "object",
"properties": {
"total": {
"type": "integer",
"format": "int64"
},
"active": {
"type": "integer",
"format": "int64"
},
"errors_total": {
"type": "integer",
"format": "int64"
},
"status": {
"type": "object",
"additionalProperties": {
"type": "integer",
"format": "int64"
}
}
},
"required": [
"total",
"active",
"errors_total",
"status"
]
},
"system": {
"type": "object"
},
"cache": {
"type": "object"
},
"backends": {
"type": "object"
},
"web_server": {
"type": "object"
},
"streams": {
"type": "object"
},
"docker": {
"type": "object"
},
"traffic": {
"type": "object"
},
"transform": {
"type": "object"
},
"logging": {
"type": "object"
},
"capacity": {
"type": "object"
},
"prometheus": {
"$ref": "#/components/schemas/MonitoringMetricFamilies"
},
"summary": {
"$ref": "#/components/schemas/MonitoringSummary"
}
},
"required": [
"summary",
"version",
"collected_at_ms",
"uptime_seconds",
"revision",
"vhosts",
"requests",
"system",
"cache",
"backends",
"web_server",
"streams",
"docker",
"traffic",
"transform",
"logging",
"capacity",
"prometheus"
]
},
"MonitoringSummary": {
"type": "object",
"description": "Compact read-only pull snapshot with computed counts, health states, bytes, percentages and rates. No Prometheus samples, paths, client addresses or secrets. Counters reset on process restart.",
"properties": {
"collected_at_ms": {
"type": "integer"
},
"uptime_seconds": {
"type": "integer"
},
"vhosts": {
"type": "object",
"properties": {
"total": {
"type": "integer"
},
"enabled": {
"type": "integer"
},
"disabled": {
"type": "integer"
},
"serving_estimate": {
"type": "integer",
"description": "Enabled primary Web Server or ready backend pool; not an HTTP end-to-end health check."
},
"backend_ready": {
"type": "integer"
},
"backend_not_ready": {
"type": "integer"
},
"docker_enabled": {
"type": "integer"
}
},
"required": [
"total",
"enabled",
"disabled",
"serving_estimate",
"backend_ready",
"backend_not_ready",
"docker_enabled"
],
"additionalProperties": false
},
"backends": {
"type": "object",
"properties": {
"total": {
"type": "integer",
"description": "All currently compiled backend pool members, including Docker-generated pools."
},
"configured_static": {
"type": "integer"
},
"up": {
"type": "integer"
},
"down": {
"type": "integer"
},
"checking": {
"type": "integer"
},
"draining": {
"type": "integer"
},
"disabled": {
"type": "integer"
},
"unknown": {
"type": "integer"
},
"pools_total": {
"type": "integer"
},
"pools_ready": {
"type": "integer"
},
"active_exchanges": {
"type": "integer"
},
"traffic_bytes_per_second": {
"type": "number",
"nullable": true,
"description": "Sum of backend moving rates; excludes static/Web Server and local responses."
}
},
"required": [
"total",
"configured_static",
"up",
"down",
"checking",
"draining",
"disabled",
"unknown",
"pools_total",
"pools_ready",
"active_exchanges",
"traffic_bytes_per_second"
],
"additionalProperties": false
},
"traffic": {
"type": "object",
"properties": {
"requests_total": {
"type": "integer"
},
"requests_completed_total": {
"type": "integer"
},
"requests_active": {
"type": "integer"
},
"requests_per_second": {
"type": "number",
"nullable": true,
"description": "Rate between server-side polling snapshots; null until at least 1 second has elapsed or after a gap/reset."
},
"requests_per_second_average": {
"type": "number",
"nullable": true
},
"errors_total": {
"type": "integer"
},
"errors_per_second": {
"type": "number",
"nullable": true
},
"error_percent": {
"type": "number",
"nullable": true,
"description": "errors_total / completed_total * 100; lifetime ratio."
},
"status_2xx_total": {
"type": "integer"
},
"status_4xx_total": {
"type": "integer"
},
"status_5xx_total": {
"type": "integer"
},
"bytes_in_total": {
"type": "integer"
},
"bytes_out_total": {
"type": "integer"
},
"bytes_total": {
"type": "integer"
},
"megabytes_total": {
"type": "number",
"description": "Decimal MB = (bytes_in + bytes_out)/1,000,000."
},
"bytes_in_per_second": {
"type": "number",
"nullable": true
},
"bytes_out_per_second": {
"type": "number",
"nullable": true
},
"bytes_per_second": {
"type": "number",
"nullable": true
},
"sample_interval_seconds": {
"type": "number",
"nullable": true
},
"network_rx_bytes_per_second": {
"type": "number",
"nullable": true,
"description": "Host network rate, not RustProxy-only."
},
"network_tx_bytes_per_second": {
"type": "number",
"nullable": true,
"description": "Host network rate, not RustProxy-only."
}
},
"required": [
"requests_total",
"requests_completed_total",
"requests_active",
"requests_per_second",
"requests_per_second_average",
"errors_total",
"errors_per_second",
"error_percent",
"status_2xx_total",
"status_4xx_total",
"status_5xx_total",
"bytes_in_total",
"bytes_out_total",
"bytes_total",
"megabytes_total",
"bytes_in_per_second",
"bytes_out_per_second",
"bytes_per_second",
"sample_interval_seconds",
"network_rx_bytes_per_second",
"network_tx_bytes_per_second"
],
"additionalProperties": false
},
"system": {
"type": "object",
"properties": {
"cpu_percent": {
"type": "number",
"nullable": true
},
"process_cpu_percent": {
"type": "number",
"nullable": true
},
"memory_percent": {
"type": "number",
"nullable": true
},
"memory_used_bytes": {
"type": "integer",
"nullable": true
},
"memory_total_bytes": {
"type": "integer",
"nullable": true
},
"process_rss_bytes": {
"type": "integer",
"nullable": true
},
"sampled_at_ms": {
"type": "integer"
}
},
"required": [
"cpu_percent",
"process_cpu_percent",
"memory_percent",
"memory_used_bytes",
"memory_total_bytes",
"process_rss_bytes",
"sampled_at_ms"
],
"additionalProperties": false
},
"cache": {
"type": "object",
"properties": {
"hits_total": {
"type": "integer"
},
"misses_total": {
"type": "integer"
},
"hit_percent": {
"type": "number",
"nullable": true
},
"hot_entries": {
"type": "integer"
},
"cold_entries": {
"type": "integer"
},
"hot_bytes": {
"type": "integer"
},
"cold_bytes": {
"type": "integer"
}
},
"required": [
"hits_total",
"misses_total",
"hit_percent",
"hot_entries",
"cold_entries",
"hot_bytes",
"cold_bytes"
],
"additionalProperties": false
},
"web_server": {
"type": "object",
"properties": {
"services_total": {
"type": "integer"
},
"requests_total": {
"type": "integer"
}
},
"required": [
"services_total",
"requests_total"
],
"additionalProperties": false
},
"streams": {
"type": "object",
"properties": {
"total": {
"type": "integer"
},
"enabled": {
"type": "integer"
},
"active_sessions": {
"type": "integer"
}
},
"required": [
"total",
"enabled",
"active_sessions"
],
"additionalProperties": false
},
"capacity": {
"type": "object",
"properties": {
"requests_limit": {
"type": "integer"
},
"requests_used": {
"type": "integer"
},
"requests_used_percent": {
"type": "number",
"nullable": true
}
},
"required": [
"requests_limit",
"requests_used",
"requests_used_percent"
],
"additionalProperties": false
}
},
"required": [
"collected_at_ms",
"uptime_seconds",
"vhosts",
"backends",
"traffic",
"system",
"cache",
"web_server",
"streams",
"capacity"
],
"additionalProperties": false
}
}
},
@@ -9492,6 +10043,76 @@
]
}
},
"/api/metrics/json": {
"get": {
"summary": "Read-only JSON metrics snapshot for dashboards and integrations",
"tags": [
"Monitoring"
],
"operationId": "get_metrics_json",
"responses": {
"200": {
"description": "Bounded operational JSON snapshot, not a streaming feed",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/MonitoringSnapshot"
}
}
}
},
"401": {
"description": "Authentication required or revoked",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
},
"403": {
"description": "Host, origin or CSRF validation failed",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
},
"429": {
"description": "Bounded worker/session/rate limit",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
},
"503": {
"description": "Control-plane storage/worker unavailable",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
}
},
"description": "Single GET JSON snapshot with summary (aggregated counts, percentages and bounded rates), Overview counters/gauges and every sample from the Prometheus exporter (under prometheus.types/samples). Same admin session or CIDR-scoped general RO/RW API token as /api/metrics. No SSE, request paths, IPs or raw User-Agent values. Prometheus exporter sibling: {configured metrics path}.json; the same exporter credentials/CIDRs apply.",
"security": [
{
"AdminSession": []
},
{
"GeneralApiBearer": []
}
]
}
},
"/docs/openapi.json": {
"get": {
"summary": "This protected OpenAPI document",
@@ -12841,7 +13462,77 @@
}
}
}
},
"/api/metrics/summary": {
"get": {
"summary": "Compact computed live metrics summary for external dashboards",
"tags": [
"Monitoring"
],
"operationId": "get_metrics_summary",
"responses": {
"200": {
"description": "Small JSON-only computed metrics summary",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/MonitoringSummary"
}
}
}
},
"401": {
"description": "Authentication required or revoked",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
},
"403": {
"description": "Host, origin or CSRF validation failed",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
},
"429": {
"description": "Bounded worker/session/rate limit",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
},
"503": {
"description": "Control-plane storage/worker unavailable",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
}
},
"description": "Read-only compact JSON summary of vhosts, backend pools, total HTTP traffic, rolling per-second request and byte rates, CPU, RAM, cache, Web Server and streams. A first read or gap/reset produces null rolling rates; repeat GET at 2-5 second intervals. No SSE, no detailed Prometheus samples. Auth: admin session or CIDR-scoped General API Bearer (ro/rw). The separate enabled exporter offers {configured metrics path}.summary.json, protected by its own bearer/CIDR policy.",
"security": [
{
"AdminSession": []
},
{
"GeneralApiBearer": []
}
]
}
}
},
"tags": []
}
}
src/admin/router.rs
+12 -2
View File
@@ -200,9 +200,17 @@ pub fn router(state: Arc<App>) -> Router {
"/api/certificates/{certificate_id}/reload",
post(certificate_reload_one).layer(DefaultBodyLimit::max(4096)),
)
.route("/api/api-tokens", get(list_api_tokens).post(create_api_token))
.route(
"/api/api-tokens",
get(list_api_tokens).post(create_api_token),
)
.route("/api/api-tokens/{id}", delete(revoke_api_token))
.route("/api/prometheus-token", get(prometheus_token_status).post(rotate_prometheus_token).delete(revoke_prometheus_token))
.route(
"/api/prometheus-token",
get(prometheus_token_status)
.post(rotate_prometheus_token)
.delete(revoke_prometheus_token),
)
.route(
"/api/certificate-api",
get(certificate_api_status)
@@ -267,6 +275,8 @@ pub fn router(state: Arc<App>) -> Router {
.route("/api/events", get(events))
.route("/metrics", get(metrics))
.route("/api/metrics", get(metrics))
.route("/api/metrics/json", get(metrics_json))
.route("/api/metrics/summary", get(metrics_summary))
.fallback(|| async { reject(StatusCode::NOT_FOUND, "not_found") })
.layer(DefaultBodyLimit::max(
crate::config::MAX_CONFIG_BYTES + 1024,
src/admin/tests.rs
+79
View File
@@ -2250,6 +2250,7 @@ mod tests {
"/api/acme",
"/api/auth/session",
"/api/metrics",
"/api/metrics/json",
"/metrics",
"/docs/openapi.json",
"/docs/content.json",
@@ -3664,6 +3665,84 @@ mod tests {
store.close().await;
}
#[tokio::test]
async fn json_metrics_api_is_authenticated_and_matches_prometheus_families() {
let (router, store, _dir, secret, _) = fixture().await;
let cookie = format!("{}={secret}", auth::COOKIE_NAME);
let snapshot = router.clone().oneshot(
request("/api/metrics/json").header("cookie", &cookie)
.body(Body::empty()).unwrap()
).await.unwrap();
assert_eq!(snapshot.status(), StatusCode::OK);
assert!(snapshot.headers()["content-type"].to_str().unwrap().starts_with("application/json"));
let snapshot = json_body(snapshot).await;
for field in ["summary", "vhosts", "requests", "system", "cache", "backends", "web_server",
"streams", "docker", "traffic", "transform", "logging", "capacity", "prometheus"] {
assert!(snapshot.get(field).is_some(), "missing: {field}");
}
assert_eq!(snapshot["vhosts"]["total"], 0);
assert_eq!(snapshot["vhosts"]["enabled"], 0);
assert_eq!(snapshot["vhosts"]["disabled"], 0);
assert_eq!(snapshot["summary"]["vhosts"]["enabled"], 0);
assert_eq!(snapshot["summary"]["backends"]["total"], 0);
assert!(snapshot["summary"]["traffic"]["bytes_total"].is_number());
assert!(snapshot["prometheus"]["samples"]["rust_proxy_requests_total"].is_array());
assert_eq!(snapshot["prometheus"]["types"]["rust_proxy_requests_total"], "counter");
assert!(general_api_token_allowed(&Method::GET, "/api/metrics/json", "ro"));
assert!(general_api_token_allowed(&Method::GET, "/api/metrics/summary", "ro"));
assert!(!general_api_token_allowed(&Method::POST, "/api/metrics/json", "rw"));
assert!(!general_api_token_allowed(&Method::POST, "/api/metrics/summary", "rw"));
let summary = router.clone().oneshot(
request("/api/metrics/summary").header("cookie", &cookie)
.body(Body::empty()).unwrap()
).await.unwrap();
assert_eq!(summary.status(), StatusCode::OK);
assert_eq!(summary.headers()["cache-control"], "no-store");
let summary = json_body(summary).await;
assert!(summary.get("prometheus").is_none());
assert_eq!(summary["vhosts"]["enabled"], snapshot["summary"]["vhosts"]["enabled"]);
assert_eq!(summary["backends"]["total"], snapshot["summary"]["backends"]["total"]);
store.close().await;
}
#[tokio::test]
async fn json_metrics_accepts_ro_bearer_only_from_allowed_peer() {
let (router, store, _dir, _, _) = fixture().await;
let token = "T".repeat(64);
let cidrs = vec!["127.0.0.1/32".to_string()];
assert!(store.create_api_token(
"api-monitor", &auth::secret_hash(&token), "ro", "monitor test", &cidrs, "owner"
).await.unwrap());
let build = |peer: Option<&str>| {
let mut request = request("/api/metrics/json")
.header("authorization", format!("Bearer {token}"));
if let Some(peer) = peer {
request = request.extension(PeerAddr(peer.parse().unwrap()));
}
request.body(Body::empty()).unwrap()
};
let response = router.clone().oneshot(build(None)).await.unwrap();
assert_eq!(response.status(), StatusCode::FORBIDDEN);
let response = router.clone().oneshot(build(Some("127.0.0.1:12345"))).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let response = json_body(response).await;
assert!(response["prometheus"]["samples"].is_object());
assert!(response.get("config").is_none());
let summarize = |peer: Option<&str>| {
let mut req = request("/api/metrics/summary")
.header("authorization", format!("Bearer {token}"));
if let Some(peer) = peer { req = req.extension(PeerAddr(peer.parse().unwrap())); }
req.body(Body::empty()).unwrap()
};
assert_eq!(router.clone().oneshot(summarize(None)).await.unwrap().status(), StatusCode::FORBIDDEN);
let summary = router.oneshot(summarize(Some("127.0.0.1:12345"))).await.unwrap();
assert_eq!(summary.status(), StatusCode::OK);
let summary = json_body(summary).await;
assert!(summary["traffic"]["requests_total"].is_number());
assert!(summary.get("prometheus").is_none());
store.close().await;
}
#[test]
fn storage_metadata_is_exposed_only_through_configuration_get_to_api_tokens() {
assert!(general_api_token_allowed(&Method::GET, "/api/config", "ro"));
src/metrics.rs
+1
View File
@@ -1,4 +1,5 @@
//! Aggregate metrics: no attacker-controlled labels or unbounded series.
include!("metrics/global.rs");
include!("metrics/rate_sampling.rs");
include!("metrics/vhost.rs");
src/metrics/global.rs
+1
View File
@@ -2,6 +2,7 @@ use serde::Serialize;
use std::sync::atomic::{AtomicU64, Ordering};
#[derive(Default)]
pub struct Metrics {
pub summary_rate_sampler: RateSampler,
pub admin_requests: AtomicU64,
pub admin_response_headers: AtomicU64,
pub admin_header_duration_us: AtomicU64,
src/metrics/rate_sampling.rs
+100
View File
@@ -0,0 +1,100 @@
// Bounded pull-driven rate sampling for the JSON summary. No per-client state,
// background task, or unbounded history. A cached sample is shared by readers.
use std::sync::Mutex;
use std::time::{Duration, Instant};
#[derive(Debug, Clone, Copy, Default, Serialize)]
pub struct SummaryRates {
pub requests_per_second: Option<f64>,
pub errors_per_second: Option<f64>,
pub bytes_in_per_second: Option<f64>,
pub bytes_out_per_second: Option<f64>,
pub sample_interval_seconds: Option<f64>,
}
#[derive(Debug, Clone, Copy)]
struct SummaryCounters {
requests: u64,
errors: u64,
bytes_in: u64,
bytes_out: u64,
}
#[derive(Default)]
struct RateState {
baseline: Option<(Instant, SummaryCounters)>,
latest: SummaryRates,
}
#[derive(Default)]
pub struct RateSampler(Mutex<RateState>);
impl RateSampler {
fn sample(&self, counters: SummaryCounters, now: Instant) -> SummaryRates {
let mut state = self.0.lock().unwrap_or_else(|error| error.into_inner());
if let Some((previous_at, previous)) = state.baseline {
let elapsed = now.saturating_duration_since(previous_at);
// Do not turn rapid parallel polls into zero/spiky rates. Reuse
// the last measured rate until a useful sampling interval passes.
if elapsed < Duration::from_secs(1) {
return state.latest;
}
let continuous = elapsed <= Duration::from_secs(30)
&& counters.requests >= previous.requests
&& counters.errors >= previous.errors
&& counters.bytes_in >= previous.bytes_in
&& counters.bytes_out >= previous.bytes_out;
if continuous {
let seconds = elapsed.as_secs_f64();
state.latest = SummaryRates {
requests_per_second: Some((counters.requests - previous.requests) as f64 / seconds),
errors_per_second: Some((counters.errors - previous.errors) as f64 / seconds),
bytes_in_per_second: Some((counters.bytes_in - previous.bytes_in) as f64 / seconds),
bytes_out_per_second: Some((counters.bytes_out - previous.bytes_out) as f64 / seconds),
sample_interval_seconds: Some(seconds),
};
} else {
// Long polling gaps or counter resets invalidate the baseline.
state.latest = SummaryRates::default();
}
}
state.baseline = Some((now, counters));
state.latest
}
}
impl Metrics {
pub fn summary_rates(&self) -> SummaryRates {
self.summary_rate_sampler.sample(SummaryCounters {
requests: self.requests.load(Ordering::Relaxed),
errors: self.errors.load(Ordering::Relaxed),
bytes_in: self.bytes_in.load(Ordering::Relaxed),
bytes_out: self.bytes_out.load(Ordering::Relaxed),
}, Instant::now())
}
}
#[cfg(test)]
mod summary_rate_tests {
use super::*;
#[test]
fn sampled_rates_handle_first_read_short_intervals_long_gaps_and_resets() {
let sampler = RateSampler::default();
let at = Instant::now();
let counts = |requests, errors, bytes_in, bytes_out| SummaryCounters {
requests, errors, bytes_in, bytes_out
};
assert!(sampler.sample(counts(10, 2, 100, 200), at).requests_per_second.is_none());
let rate = sampler.sample(counts(20, 4, 300, 500), at + Duration::from_secs(2));
assert_eq!(rate.requests_per_second, Some(5.0));
assert_eq!(rate.errors_per_second, Some(1.0));
assert_eq!(rate.bytes_in_per_second, Some(100.0));
assert_eq!(rate.bytes_out_per_second, Some(150.0));
assert_eq!(rate.sample_interval_seconds, Some(2.0));
assert_eq!(sampler.sample(counts(21, 4, 310, 520), at + Duration::from_millis(2300)).requests_per_second, Some(5.0));
assert!(sampler.sample(counts(22, 4, 330, 550), at + Duration::from_secs(40)).requests_per_second.is_none());
assert!(sampler.sample(counts(1, 0, 0, 0), at + Duration::from_secs(42)).requests_per_second.is_none());
assert_eq!(sampler.sample(counts(5, 0, 200, 300), at + Duration::from_secs(44)).requests_per_second, Some(2.0));
}
}
src/pages.rs
+27 -9
View File
@@ -8,7 +8,8 @@ use std::collections::BTreeMap;
pub const MAX_HTML_BYTES: usize = 32 * 1024;
pub const ERROR_CODES: [u16; 11] = [400, 401, 403, 404, 408, 413, 429, 500, 502, 503, 504];
pub const BUILTIN_ERROR_CODES: [u16; 12] = [400, 401, 403, 404, 408, 410, 413, 429, 500, 502, 503, 504];
pub const BUILTIN_ERROR_CODES: [u16; 12] =
[400, 401, 403, 404, 408, 410, 413, 429, 500, 502, 503, 504];
/// Built-in operational gateway pages. They are deliberately not marked as
/// `PageServed`, so an administrator-configured vhost/global error page can
@@ -30,7 +31,7 @@ pub fn builtin_gateway_error(status: StatusCode) -> Option<&'static str> {
_ => None,
}
}
fn builtin_application_page(status: StatusCode) -> String {
pub(crate) fn builtin_application_page(status: StatusCode) -> String {
if status == StatusCode::OK {
return include_str!("admin/web/status_200.html").to_owned();
}
@@ -41,7 +42,8 @@ fn builtin_application_page(status: StatusCode) -> String {
// as the packaged pages, rather than the former bright bare 200/4xx page.
// Code and canonical reason originate from http::StatusCode, never a request.
let reason = status.canonical_reason().unwrap_or("Response");
format!(r#"<!doctype html>
format!(
r#"<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
@@ -59,7 +61,10 @@ fn builtin_application_page(status: StatusCode) -> String {
<p>RustProxy generated this response.</p>
</main>
</body>
</html>"#, code = status.as_u16(), reason = reason)
</html>"#,
code = status.as_u16(),
reason = reason
)
}
#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize)]
@@ -184,8 +189,15 @@ impl Page {
.status
.and_then(|v| StatusCode::from_u16(v).ok())
.unwrap_or(default_status);
let bodyless = matches!(status, StatusCode::NO_CONTENT | StatusCode::RESET_CONTENT | StatusCode::NOT_MODIFIED);
let body = if bodyless { String::new() } else { self.body(status) };
let bodyless = matches!(
status,
StatusCode::NO_CONTENT | StatusCode::RESET_CONTENT | StatusCode::NOT_MODIFIED
);
let body = if bodyless {
String::new()
} else {
self.body(status)
};
let length = body.len();
let mut response = Response::new(Body::from(body));
*response.status_mut() = status;
@@ -259,9 +271,13 @@ impl VhostPages {
validate_errors(&self.errors)?;
ensure!(
self.intercept_statuses.len() <= ERROR_CODES.len()
&& self.intercept_statuses.iter().all(|status| ERROR_CODES.contains(status))
&& self
.intercept_statuses
.iter()
.all(|status| ERROR_CODES.contains(status))
&& {
let distinct: std::collections::HashSet<_> = self.intercept_statuses.iter().collect();
let distinct: std::collections::HashSet<_> =
self.intercept_statuses.iter().collect();
distinct.len() == self.intercept_statuses.len()
},
"intercepted application status codes must be unique supported HTTP error codes"
@@ -312,7 +328,9 @@ pub fn intercept_application_error(
status: None,
html: String::new(),
};
let selected = vhost_pages.errors.get(&code)
let selected = vhost_pages
.errors
.get(&code)
.or_else(|| global_pages.errors.get(&code))
.unwrap_or(&default_page);
let mut body_only = selected.clone();
src/prometheus.rs
+2
View File
@@ -6,6 +6,8 @@ include!("prometheus/traffic.rs");
include!("prometheus/web_php.rs");
include!("prometheus/vhosts_backends.rs");
include!("prometheus/docker_misc.rs");
mod json;
pub use json::{response as json_response, summary_response};
#[cfg(test)]
include!("prometheus/tests.rs");
src/prometheus/json.rs
+758
View File
@@ -0,0 +1,758 @@
// Pull-friendly JSON telemetry: a compact chart snapshot plus every numeric
// sample that the Prometheus text exporter currently emits. Both formats use
// the same collector switches and security boundaries.
use super::App;
use axum::response::{IntoResponse, Response};
use serde::Serialize;
use serde_json::{json, Value};
use std::{
collections::BTreeMap,
sync::atomic::Ordering,
time::{SystemTime, UNIX_EPOCH},
};
#[derive(Default, Serialize)]
struct MetricSamples {
// Metadata is stored separately because histogram/summary samples have
// suffixes (_bucket/_sum/_count) that share one Prometheus TYPE family.
types: BTreeMap<String, String>,
samples: BTreeMap<String, Vec<Sample>>,
}
#[derive(Serialize)]
struct Sample {
labels: BTreeMap<String, String>,
value: Value,
}
// Read only our own Prometheus exposition, never an untrusted remote payload.
// Preserve large u64 counters exactly; JSON cannot represent NaN or infinity,
// so these values become null rather than breaking the whole API response.
fn sample_value(raw: &str) -> Value {
if let Ok(value) = raw.parse::<u64>() {
return json!(value);
}
if let Ok(value) = raw.parse::<i64>() {
return json!(value);
}
raw.parse::<f64>()
.ok()
.filter(|v| v.is_finite())
.map_or(Value::Null, |value| json!(value))
}
// Prometheus label escaping: \\, \", \n. Scan the quoted value rather than
// splitting on commas (a configured label may contain commas or quotes).
fn parse_labels(raw: &str) -> Option<BTreeMap<String, String>> {
let bytes = raw.as_bytes();
let mut pos = 0;
let mut labels = BTreeMap::new();
if raw.is_empty() {
return Some(labels);
}
while pos < bytes.len() {
let equal = raw[pos..].find("=\"")? + pos;
let key = &raw[pos..equal];
if key.is_empty() || !key.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_') {
return None;
}
pos = equal + 2;
let mut start = pos;
let mut value = String::new();
let mut closed = false;
while pos < bytes.len() {
match bytes[pos] {
b'\\' => {
value.push_str(&raw[start..pos]);
pos += 1;
let escaped = *bytes.get(pos)?;
value.push(match escaped {
b'n' => '\n',
b'\\' => '\\',
b'"' => '"',
_ => return None,
});
pos += 1;
start = pos;
}
b'"' => {
value.push_str(&raw[start..pos]);
pos += 1;
closed = true;
break;
}
_ => pos += 1,
}
}
if !closed || labels.insert(key.to_owned(), value).is_some() {
return None;
}
if pos == bytes.len() {
break;
}
if bytes[pos] != b',' {
return None;
}
pos += 1;
if pos == bytes.len() {
return None;
}
}
Some(labels)
}
fn parse_exposition(exposition: &str) -> MetricSamples {
let mut result = MetricSamples::default();
for line in exposition.lines() {
if let Some(spec) = line.strip_prefix("# TYPE ") {
if let Some((name, kind)) = spec.split_once(' ') {
result.types.insert(name.to_owned(), kind.to_owned());
}
continue;
}
if line.is_empty() || line.starts_with('#') {
continue;
}
let Some((selector, raw_value)) = line.rsplit_once(' ') else {
continue;
};
let (name, labels) = match selector.find('{') {
Some(open) if selector.ends_with('}') => {
let Some(labels) = parse_labels(&selector[open + 1..selector.len() - 1]) else {
continue;
};
(&selector[..open], labels)
}
Some(_) => continue,
None => (selector, BTreeMap::new()),
};
if name.is_empty() || !name.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_') {
continue;
}
result
.samples
.entry(name.to_owned())
.or_default()
.push(Sample {
labels,
value: sample_value(raw_value),
});
}
result
}
fn counts(total: usize, enabled: usize) -> Value {
json!({"total": total, "enabled": enabled, "disabled": total - enabled})
}
// A small, field-stable projection that an external dashboard can consume
// without parsing metric families or knowing RustProxy's widget layout.
fn percentage(part: u64, total: u64) -> Option<f64> {
(total > 0).then(|| part as f64 * 100.0 / total as f64)
}
fn summary_from_snapshot(detail: &Value, rates: crate::metrics::SummaryRates) -> Value {
let number = |value: &Value| value.as_u64().unwrap_or(0);
let requests = &detail["requests"];
let vhosts = &detail["vhosts"];
let system = &detail["system"];
let cache = &detail["cache"];
let backends = &detail["backends"];
let mut up = 0usize;
let mut down = 0usize;
let mut checking = 0usize;
let mut draining = 0usize;
let mut disabled = 0usize;
let mut unknown = 0usize;
let mut backend_bytes_per_second = 0.0;
let mut backend_members = 0usize;
let mut ready_pools = 0usize;
if let Some(pools) = backends["pools"].as_array() {
for pool in pools {
if pool["ready"].as_bool() == Some(true) {
ready_pools += 1;
}
if let Some(members) = pool["members"].as_array() {
backend_members += members.len();
for member in members {
match member["status"].as_str().unwrap_or("unknown") {
"up" => up += 1,
"down" => down += 1,
"checking" => checking += 1,
"draining" => draining += 1,
"disabled" => disabled += 1,
_ => unknown += 1,
}
if let Some(value) = member["traffic_bytes_per_second"]
.as_f64()
.filter(|value| value.is_finite() && *value >= 0.0)
{
backend_bytes_per_second += value;
}
}
}
}
}
let enabled = number(&vhosts["enabled"]);
let serving_estimate = vhosts["items"].as_array().map_or(0, |items| {
items
.iter()
.filter(|item| item["serving_estimate"].as_bool() == Some(true))
.count()
});
let backend_vhosts_ready = vhosts["items"].as_array().map_or(0, |items| {
items
.iter()
.filter(|item| {
item["enabled"].as_bool() == Some(true)
&& item["backend_ready"].as_bool() == Some(true)
})
.count()
});
let backend_vhosts_unready = vhosts["items"].as_array().map_or(0, |items| {
items
.iter()
.filter(|item| {
item["enabled"].as_bool() == Some(true)
&& item["backend_ready"].as_bool() == Some(false)
})
.count()
});
let bytes_in = number(&requests["bytes_in_total"]);
let bytes_out = number(&requests["bytes_out_total"]);
let bytes_total = bytes_in.saturating_add(bytes_out);
let uptime = number(&detail["uptime_seconds"]);
let memory_percent = system["memory_used_bytes"]
.as_u64()
.zip(system["memory_total_bytes"].as_u64())
.and_then(|(used, total)| percentage(used, total));
let request_error_percent = percentage(
number(&requests["errors_total"]),
number(&requests["completed_total"]),
);
let web_requests = detail["web_server"]["services"]
.as_array()
.map_or(0u64, |items| {
items
.iter()
.map(|item| number(&item["web"]["requests"]))
.fold(0u64, u64::saturating_add)
});
let stream_active = detail["streams"]["items"].as_array().map_or(0u64, |items| {
items
.iter()
.map(|item| number(&item["metrics"]["active"]))
.fold(0u64, u64::saturating_add)
});
let capacity_limit = number(&detail["capacity"]["requests_limit"]);
let capacity_available = number(&detail["capacity"]["requests_available"]);
let backend_bytes_per_second = backend_bytes_per_second
.is_finite()
.then_some(backend_bytes_per_second);
let vhosts_summary = json!({
"total": vhosts["total"], "enabled": enabled, "disabled": vhosts["disabled"],
"serving_estimate": serving_estimate,
"backend_ready": backend_vhosts_ready, "backend_not_ready": backend_vhosts_unready,
"docker_enabled": detail["docker"]["vhosts"]["enabled"]
});
let backends_summary = json!({
"total": backend_members, "configured_static": backends["total"],
"up": up, "down": down, "checking": checking,
"draining": draining, "disabled": disabled, "unknown": unknown,
"pools_total": backends["pools"].as_array().map_or(0, Vec::len),
"pools_ready": ready_pools,
"active_exchanges": backends["active"],
"traffic_bytes_per_second": backend_bytes_per_second
});
let traffic_summary = json!({
"requests_total": requests["total"],
"requests_completed_total": requests["completed_total"],
"requests_active": requests["active"],
"requests_per_second": rates.requests_per_second,
"requests_per_second_average": (uptime > 0).then(|| number(&requests["total"]) as f64 / uptime as f64),
"errors_total": requests["errors_total"],
"errors_per_second": rates.errors_per_second,
"error_percent": request_error_percent,
"status_2xx_total": requests["status"]["2xx"],
"status_4xx_total": requests["status"]["4xx"],
"status_5xx_total": requests["status"]["5xx"],
"bytes_in_total": bytes_in,
"bytes_out_total": bytes_out,
"bytes_total": bytes_total,
"megabytes_total": bytes_total as f64 / 1_000_000.0,
"bytes_in_per_second": rates.bytes_in_per_second,
"bytes_out_per_second": rates.bytes_out_per_second,
"bytes_per_second": rates.bytes_in_per_second.zip(rates.bytes_out_per_second).map(|(a, b)| a + b),
"sample_interval_seconds": rates.sample_interval_seconds,
"network_rx_bytes_per_second": system["network_rx_bytes_per_second"],
"network_tx_bytes_per_second": system["network_tx_bytes_per_second"]
});
let system_summary = json!({
"cpu_percent": system["cpu_percent"],
"process_cpu_percent": system["process_cpu_percent"],
"memory_percent": memory_percent,
"memory_used_bytes": system["memory_used_bytes"],
"memory_total_bytes": system["memory_total_bytes"],
"process_rss_bytes": system["process_rss_bytes"],
"sampled_at_ms": system["sampled_at_ms"]
});
let cache_summary = json!({
"hits_total": cache["hits"], "misses_total": cache["misses"],
"hit_percent": cache["hit_ratio"].as_f64().map(|v| v * 100.0),
"hot_entries": cache["entries"], "cold_entries": cache["cold_entries"],
"hot_bytes": cache["resident_bytes"], "cold_bytes": cache["cold_bytes"]
});
json!({
"collected_at_ms": detail["collected_at_ms"],
"uptime_seconds": uptime,
"vhosts": vhosts_summary,
"backends": backends_summary,
"traffic": traffic_summary,
"system": system_summary,
"cache": cache_summary,
"web_server": {
"services_total": detail["web_server"]["total"],
"requests_total": web_requests
},
"streams": {
"total": detail["streams"]["total"],
"enabled": detail["streams"]["enabled"],
"active_sessions": stream_active
},
"capacity": {
"requests_limit": capacity_limit,
"requests_used": capacity_limit.saturating_sub(capacity_available),
"requests_used_percent": percentage(capacity_limit.saturating_sub(capacity_available), capacity_limit)
}
})
}
fn snapshot(app: &App) -> Value {
let snapshot = app.config.load_full();
let config = &snapshot.envelope.config;
let metrics = app.metrics.view();
let system = app.diagnostics.view();
let cache = app.cache.view();
let vhosts = config
.vhosts
.iter()
.map(|host| {
let live = snapshot.vhost_metrics.get(&host.id).map(|m| m.view());
let pool_ready = snapshot
.pools
.iter()
.find(|pool| pool.id == host.id)
.map(|pool| pool.ready());
let serving_estimate =
host.enabled && (pool_ready == Some(true) || !host.source.is_reverse_proxy());
json!({"id": host.id, "enabled": host.enabled, "metrics": live,
"backend_ready": pool_ready, "serving_estimate": serving_estimate})
})
.collect::<Vec<_>>();
let mut backend_up = 0usize;
let mut backend_down = 0usize;
let mut backend_active = 0u64;
let pools = snapshot
.pools
.iter()
.map(|pool| {
let view = pool.view();
let members = view["members"]
.as_array()
.map(|items| {
items.iter().map(|member| {
let status = member["status"].as_str().unwrap_or("unknown");
if status == "up" { backend_up += 1; }
if status == "down" { backend_down += 1; }
backend_active = backend_active.saturating_add(member["active"].as_u64().unwrap_or(0));
json!({
"id": member["id"], "status": status, "healthy": member["healthy"],
"active": member["active"], "requests_total": member["requests"],
"errors_total": member["errors"], "latency_ms": member["latency_ms"],
"traffic_bytes_per_second": member["traffic_bytes_per_second"]
})
}).collect::<Vec<_>>()
})
.unwrap_or_default();
json!({"vhost": pool.id, "ready": pool.ready(), "members": members})
})
.collect::<Vec<_>>();
let services = config
.web_services
.iter()
.filter_map(|service| {
let compiled = snapshot.web_php_services.get(&service.id)?;
let php = compiled.php.as_ref().map(|handler| {
let available = handler.permits.available_permits();
json!({"metrics": handler.metrics.view(), "capacity_limit": handler.max_concurrent,
"capacity_used": handler.max_concurrent.saturating_sub(available),
"capacity_available": available})
});
Some(json!({"id": service.id, "web": compiled.web_metrics.view(), "php": php}))
})
.collect::<Vec<_>>();
let streams = config
.streams
.iter()
.map(|stream| {
let live = app.stream_metrics.get(&stream.id).map(|m| {
let view = m.view();
// The dashboard's full stream view includes configured backend
// socket addresses and health internals. Metrics consumers only
// need counters/status, never infrastructure addresses.
let backends = view
.backends
.iter()
.map(|backend| {
json!({
"id": backend.id,
"enabled": backend.enabled,
"status": backend.status,
"active": backend.active,
"sessions_total": backend.sessions_total,
"sessions_completed": backend.sessions_completed,
"errors": backend.errors,
"rejected": backend.rejected,
"bytes_in": backend.bytes_in,
"bytes_out": backend.bytes_out,
"packets_in": backend.packets_in,
"packets_out": backend.packets_out
})
})
.collect::<Vec<_>>();
json!({
"sessions_total": view.sessions_total,
"sessions_completed": view.sessions_completed,
"active": view.active,
"rejected": view.rejected,
"errors": view.errors,
"bytes_in": view.bytes_in,
"bytes_out": view.bytes_out,
"packets_in": view.packets_in,
"packets_out": view.packets_out,
"backends": backends
})
});
json!({"id": stream.id, "enabled": stream.enabled, "metrics": live})
})
.collect::<Vec<_>>();
let traffic = app.traffic.view(None);
let data = &traffic.data;
let methods = data
.method_families
.iter()
.filter(|x| {
super::safe_traffic_state(
&x.value,
&[
"GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "CONNECT", "TRACE",
"OTHER",
],
)
})
.collect::<Vec<_>>();
let statuses = data
.statuses
.iter()
.filter(|x| super::safe_traffic_state(&x.value, &["1xx", "2xx", "3xx", "4xx", "5xx"]))
.collect::<Vec<_>>();
let cache_results = data
.cache
.iter()
.filter(|x| super::safe_traffic_state(&x.value, &["HIT", "MISS", "BYPASS", "-"]))
.collect::<Vec<_>>();
let sources = data
.sources
.iter()
.filter(|x| {
super::safe_traffic_state(
&x.value,
&[
"reverse_proxy",
"web_server",
"local_response",
"unresolved",
],
)
})
.collect::<Vec<_>>();
let agents = &config.docker_agents.agents;
let docker_vhosts = agents
.iter()
.map(|agent| agent.assignments.len())
.sum::<usize>();
let docker_enabled = agents
.iter()
.filter(|agent| agent.enabled)
.flat_map(|agent| &agent.assignments)
.filter(|assignment| assignment.enabled)
.count();
let now_ms = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64;
// Split the JSON builders: a single giant json! macro would exceed the
// default macro recursion depth on supported stable Rust toolchains.
let requests = json!({
"total": metrics.requests, "completed_total": metrics.completed,
"active": metrics.active, "errors_total": metrics.errors,
"rejected_total": metrics.rejected_requests,
"status": {
"1xx": app.metrics.status_1xx.load(Ordering::Relaxed),
"2xx": app.metrics.status_2xx.load(Ordering::Relaxed),
"3xx": app.metrics.status_3xx.load(Ordering::Relaxed),
"4xx": app.metrics.status_4xx.load(Ordering::Relaxed),
"5xx": app.metrics.status_5xx.load(Ordering::Relaxed)
},
"bytes_in_total": app.metrics.bytes_in.load(Ordering::Relaxed),
"bytes_out_total": app.metrics.bytes_out.load(Ordering::Relaxed),
"mean_duration_ms": metrics.mean_duration_ms,
"connections_total": metrics.connections,
"connections_active": metrics.active_connections,
"websockets_total": metrics.websocket_total,
"websockets_active": metrics.websocket_active,
"http1_total": metrics.http1_requests,
"http2_total": metrics.http2_requests,
"http3_total": metrics.http3_requests
});
let system = json!({
"sampled_at_ms": system.sampled_at_ms,
"cpu_percent": system.cpu_percent,
"process_cpu_percent": system.process_cpu_percent,
"logical_cpus": system.logical_cpus,
"system_uptime_seconds": system.system_uptime_seconds,
"memory_total_bytes": system.memory_total_bytes,
"memory_used_bytes": system.memory_used_bytes,
"process_rss_bytes": system.process_rss_bytes,
"process_virtual_memory_bytes": system.process_virtual_memory_bytes,
"process_threads": system.process_threads,
"open_fds": system.open_fds,
"max_fds": system.max_fds,
"process_read_bytes": system.process_read_bytes,
"process_write_bytes": system.process_write_bytes,
"cgroup_memory_current_bytes": system.cgroup_memory_current_bytes,
"cgroup_memory_max_bytes": system.cgroup_memory_max_bytes,
"swap_total_bytes": system.system_swap_total_bytes,
"swap_used_bytes": system.system_swap_used_bytes,
"network_rx_bytes_per_second": system.network_rx_bytes_per_second,
"network_tx_bytes_per_second": system.network_tx_bytes_per_second,
"network_rx_bytes_total": system.network_rx_bytes,
"network_tx_bytes_total": system.network_tx_bytes,
"network_interfaces": system.network_interfaces,
"load_average": system.load_average
});
let traffic = json!({
"requests_total": data.requests,
"errors_total": data.errors,
"request_bytes_total": data.request_bytes,
"response_bytes_total": data.response_bytes,
"mean_duration_ms": data.mean_duration_ms,
"queue_dropped_total": traffic.queue_dropped,
"approximate_top_lists": traffic.approximate_top_lists,
"methods": methods, "statuses": statuses,
"cache": cache_results, "sources": sources
});
let mut detail = json!({
"version": env!("CARGO_PKG_VERSION"),
"collected_at_ms": now_ms,
"uptime_seconds": app.started.elapsed().as_secs(),
"revision": snapshot.envelope.revision,
"vhosts": {
"total": config.vhosts.len(),
"enabled": config.vhosts.iter().filter(|h| h.enabled).count(),
"disabled": config.vhosts.iter().filter(|h| !h.enabled).count(),
"items": vhosts
},
"redirect_hosts": counts(config.redirects.len(), config.redirects.iter().filter(|h| h.enabled).count()),
"not_found_hosts": counts(config.not_found_hosts.len(), config.not_found_hosts.iter().filter(|h| h.enabled).count()),
"requests": requests,
"system": system,
"cache": cache,
"backends": {
"total": config.vhosts.iter().map(|h| h.backends.len()).sum::<usize>(),
"up": backend_up, "down": backend_down, "active": backend_active,
"pools": pools
},
"web_server": {"total": config.web_services.len(), "services": services},
"streams": {
"total": config.streams.len(),
"enabled": config.streams.iter().filter(|s| s.enabled).count(),
"disabled": config.streams.iter().filter(|s| !s.enabled).count(),
"items": streams
},
"docker": {
"agents": counts(agents.len(), agents.iter().filter(|a| a.enabled).count()),
"vhosts": counts(docker_vhosts, docker_enabled)
},
"traffic": traffic,
"transform": app.transform_metrics.total(),
"logging": {
"dropped_total": app.logs.dropped.load(Ordering::Relaxed),
"sinks": app.logs.stats()
},
"capacity": {
"requests_limit": app.max_inflight,
"requests_available": app.inflight.available_permits(),
"listeners_active": app.active_listeners.len()
}
});
// Share the exact same computed summary between the full and small APIs.
detail["summary"] = summary_from_snapshot(&detail, app.metrics.summary_rates());
detail
}
pub fn summary_payload(app: &App) -> Value {
snapshot(app)["summary"].clone()
}
pub fn summary_response(app: &App) -> Response {
(
[
("cache-control", "no-store"),
("x-content-type-options", "nosniff"),
],
axum::Json(summary_payload(app)),
)
.into_response()
}
pub fn payload(app: &App) -> Value {
let mut result = snapshot(app);
// Use one exporter implementation, so new Prometheus families automatically
// appear here without implementing the same counters a second time.
result["prometheus"] = json!(parse_exposition(&super::render(app)));
result
}
pub fn response(app: &App) -> Response {
(
[
("cache-control", "no-store"),
("x-content-type-options", "nosniff"),
],
axum::Json(payload(app)),
)
.into_response()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn summary_computes_backend_states_rps_totals_and_percentages() {
let detail: Value = serde_json::from_str(
r#"{
"collected_at_ms": 456, "uptime_seconds": 100,
"vhosts": {"total": 4, "enabled": 3, "disabled": 1, "items": [
{"enabled": true, "backend_ready": true, "serving_estimate": true},
{"enabled": true, "backend_ready": false, "serving_estimate": false},
{"enabled": true, "backend_ready": null, "serving_estimate": true},
{"enabled": false, "backend_ready": true, "serving_estimate": false}
]},
"docker": {"vhosts": {"enabled": 2}},
"backends": {"total": 3, "active": 5, "pools": [
{"ready": true, "members": [
{"status": "up", "traffic_bytes_per_second": 1000.0},
{"status": "down", "traffic_bytes_per_second": 50.0},
{"status": "checking", "traffic_bytes_per_second": null}
]},
{"ready": false, "members": [
{"status": "disabled", "traffic_bytes_per_second": 0.0},
{"status": "draining", "traffic_bytes_per_second": 250.0}
]}
]},
"requests": {"total": 100, "completed_total": 80, "active": 2,
"errors_total": 4, "bytes_in_total": 20000000,
"bytes_out_total": 80000000,
"status": {"2xx": 70, "4xx": 5, "5xx": 5}},
"system": {"cpu_percent": 25.0, "process_cpu_percent": 4.0,
"memory_used_bytes": 200, "memory_total_bytes": 800,
"process_rss_bytes": 32, "sampled_at_ms": 111,
"network_rx_bytes_per_second": 10.0,
"network_tx_bytes_per_second": 20.0},
"cache": {"hits": 3, "misses": 1, "hit_ratio": 0.75,
"entries": 1, "cold_entries": 2, "resident_bytes": 300, "cold_bytes": 500},
"web_server": {"total": 2, "services": [
{"web": {"requests": 4}}, {"web": {"requests": 5}}]},
"streams": {"total": 2, "enabled": 1, "items": [
{"metrics": {"active": 2}}, {"metrics": null}]},
"capacity": {"requests_limit": 10, "requests_available": 7}
}"#,
)
.unwrap();
let summary = summary_from_snapshot(
&detail,
crate::metrics::SummaryRates {
requests_per_second: Some(7.5),
errors_per_second: Some(0.5),
bytes_in_per_second: Some(1000.0),
bytes_out_per_second: Some(2000.0),
sample_interval_seconds: Some(2.0),
},
);
assert_eq!(summary["vhosts"]["serving_estimate"], 2);
assert_eq!(summary["vhosts"]["backend_ready"], 1);
assert_eq!(summary["vhosts"]["backend_not_ready"], 1);
assert_eq!(summary["backends"]["total"], 5);
assert_eq!(summary["backends"]["configured_static"], 3);
assert_eq!(summary["backends"]["up"], 1);
assert_eq!(summary["backends"]["down"], 1);
assert_eq!(summary["backends"]["checking"], 1);
assert_eq!(summary["backends"]["draining"], 1);
assert_eq!(summary["backends"]["disabled"], 1);
assert_eq!(summary["backends"]["traffic_bytes_per_second"], 1300.0);
assert_eq!(summary["traffic"]["requests_per_second"], 7.5);
assert_eq!(summary["traffic"]["bytes_per_second"], 3000.0);
assert_eq!(summary["traffic"]["bytes_total"], 100_000_000);
assert_eq!(summary["traffic"]["megabytes_total"], 100.0);
assert_eq!(summary["traffic"]["error_percent"], 5.0);
assert_eq!(summary["system"]["memory_percent"], 25.0);
assert_eq!(summary["cache"]["hit_percent"], 75.0);
assert_eq!(summary["web_server"]["requests_total"], 9);
assert_eq!(summary["streams"]["active_sessions"], 2);
assert_eq!(summary["capacity"]["requests_used"], 3);
assert_eq!(summary["capacity"]["requests_used_percent"], 30.0);
}
#[test]
fn json_samples_preserve_labels_large_counters_and_histograms() {
let labels = super::super::labels(&[
("vhost", "a,\\\"b\nwith newline"),
("domain", "example.test"),
]);
let source = format!(
"# TYPE rust_proxy_requests_total counter\nrust_proxy_requests_total 9007199254740993\n\
# TYPE rust_proxy_vhost_exchange_duration_seconds histogram\n\
rust_proxy_vhost_exchange_duration_seconds_bucket{{{labels},le=\"+Inf\"}} 2\n\
rust_proxy_vhost_exchange_duration_seconds_sum{{{labels}}} 1.25\n"
);
let parsed = parse_exposition(&source);
assert_eq!(parsed.types["rust_proxy_requests_total"], "counter");
assert_eq!(
parsed.types["rust_proxy_vhost_exchange_duration_seconds"],
"histogram"
);
assert_eq!(
parsed.samples["rust_proxy_requests_total"][0]
.value
.as_u64(),
Some(9007199254740993)
);
let bucket = &parsed.samples["rust_proxy_vhost_exchange_duration_seconds_bucket"][0];
assert_eq!(bucket.labels["vhost"], "a,\\\"b\nwith newline");
assert_eq!(bucket.labels["le"], "+Inf");
assert_eq!(bucket.value.as_u64(), Some(2));
assert_eq!(
parsed.samples["rust_proxy_vhost_exchange_duration_seconds_sum"][0]
.value
.as_f64(),
Some(1.25)
);
assert!(parse_labels("vhost=\"broken").is_none());
assert!(parse_labels("vhost=\"one\",vhost=\"two\"").is_none());
assert_eq!(sample_value("+Inf"), serde_json::Value::Null);
}
}
src/prometheus/model.rs
+13 -2
View File
@@ -140,10 +140,15 @@ pub fn router(app: Arc<App>, token: Option<Zeroizing<String>>) -> Router {
async fn scrape(State(exporter): State<Arc<Exporter>>, request: Request) -> Response {
let config = exporter.app.config.load_full();
let settings = &config.envelope.config.prometheus;
// JSON is a sibling of the configured path, with identical transport auth.
let json_path = format!("{}.json", settings.path);
let wants_json = request.uri().path() == json_path.as_str();
let summary_path = format!("{}.summary.json", settings.path);
let wants_summary = request.uri().path() == summary_path.as_str();
// Revocation is immediate even when changing bind/credentials needs restart.
if !settings.enabled
|| settings.restart_differs(&exporter.app.active_prometheus)
|| request.uri().path() != settings.path
|| (request.uri().path() != settings.path && !wants_json && !wants_summary)
|| request.uri().query().is_some()
{
return StatusCode::NOT_FOUND.into_response();
@@ -201,7 +206,13 @@ async fn scrape(State(exporter): State<Arc<Exporter>>, request: Request) -> Resp
let Ok(_permit) = exporter.slots.try_acquire() else {
return StatusCode::TOO_MANY_REQUESTS.into_response();
};
response(&exporter.app)
if wants_json {
json_response(&exporter.app)
} else if wants_summary {
summary_response(&exporter.app)
} else {
response(&exporter.app)
}
}
pub fn response(app: &App) -> Response {
(
src/prometheus/tests.rs
+164
View File
@@ -56,4 +56,168 @@ mod tests {
assert_eq!(text.matches("# TYPE x").count(), 1);
assert!(text.ends_with('\n'));
}
#[tokio::test]
async fn json_exporter_enforces_peer_bearer_and_exact_path() {
use crate::{
auth,
config::{self, Config, ValidationPolicy},
db::Store,
};
use axum::body::Body;
use http::Request;
use std::{sync::Arc, time::Duration};
use tower::ServiceExt;
let dir = tempfile::tempdir().unwrap();
let mut config = Config::default();
config.prometheus.enabled = true;
let policy = ValidationPolicy {
tls_addr: None,
cert_dir: None,
data_dir: None,
proxy_addr: "127.0.0.1:8080".parse().unwrap(),
admin_addr: "127.0.0.1:9090".parse().unwrap(),
};
let store = Store::open(&dir.path().join("db"), &config).await.unwrap();
store
.initialize_admin("owner", "unused-test-hash", "test")
.await
.unwrap();
let auth = auth::Auth::new(store.clone(), dir.path().to_path_buf())
.await
.unwrap();
let snapshot = config::compile(config, 0, &policy).unwrap();
let state = App::new(
store.clone(),
snapshot,
policy,
auth,
10,
Duration::from_secs(30),
Duration::from_secs(30),
Duration::from_secs(300),
);
let token = "test-secret-1234567890-abcdefghijklmnopqrstuvwxyz";
state
.prometheus_token_hash
.store(Some(Arc::new(auth::secret_hash(token))));
let router = router(state);
let make = |path: &str, peer: Option<&str>, bearer: Option<&str>| {
let mut builder = Request::builder().uri(path);
if let Some(peer) = peer {
builder = builder.extension(PeerAddr(peer.parse().unwrap()));
}
if let Some(bearer) = bearer {
builder = builder.header("authorization", format!("Bearer {bearer}"));
}
builder.body(Body::empty()).unwrap()
};
for (path, peer, bearer, expected) in [
("/metrics.json", None, Some(token), StatusCode::FORBIDDEN),
(
"/metrics.json",
Some("203.0.113.1:5555"),
Some(token),
StatusCode::FORBIDDEN,
),
(
"/metrics.json",
Some("127.0.0.1:5555"),
None,
StatusCode::UNAUTHORIZED,
),
(
"/metrics.json",
Some("127.0.0.1:5555"),
Some("invalid"),
StatusCode::UNAUTHORIZED,
),
(
"/metrics.json?no=1",
Some("127.0.0.1:5555"),
Some(token),
StatusCode::NOT_FOUND,
),
(
"/metrics.json/",
Some("127.0.0.1:5555"),
Some(token),
StatusCode::NOT_FOUND,
),
(
"/metrics.summary.json",
None,
Some(token),
StatusCode::FORBIDDEN,
),
(
"/metrics.summary.json",
Some("127.0.0.1:5555"),
None,
StatusCode::UNAUTHORIZED,
),
(
"/metrics.summary.json?any=1",
Some("127.0.0.1:5555"),
Some(token),
StatusCode::NOT_FOUND,
),
(
"/metrics.summary.json/",
Some("127.0.0.1:5555"),
Some(token),
StatusCode::NOT_FOUND,
),
] {
let response = router
.clone()
.oneshot(make(path, peer, bearer))
.await
.unwrap();
assert_eq!(response.status(), expected, "{path} {peer:?}");
}
let response = router
.clone()
.oneshot(make("/metrics.json", Some("127.0.0.1:5555"), Some(token)))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(response.headers()["cache-control"], "no-store");
assert!(response.headers()["content-type"]
.to_str()
.unwrap()
.starts_with("application/json"));
let bytes = axum::body::to_bytes(response.into_body(), 8_000_000)
.await
.unwrap();
let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
assert_eq!(body["vhosts"]["total"], 0);
assert!(body["requests"]["total"].is_number());
assert!(body["prometheus"]["samples"]["rust_proxy_requests_total"].is_array());
let response = router
.clone()
.oneshot(make(
"/metrics.summary.json",
Some("127.0.0.1:5555"),
Some(token),
))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(response.headers()["cache-control"], "no-store");
let bytes = axum::body::to_bytes(response.into_body(), 1_000_000)
.await
.unwrap();
let summary: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
assert_eq!(summary["vhosts"]["total"], 0);
assert!(summary["traffic"]["bytes_total"].is_number());
assert!(summary.get("prometheus").is_none());
let response = router
.oneshot(make("/metrics", Some("127.0.0.1:5555"), Some(token)))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
store.close().await;
}
}
src/web_server/response/headers.rs
+34 -33
View File
@@ -21,7 +21,8 @@ use tokio::io::{AsyncReadExt, AsyncSeekExt};
use tokio_util::io::ReaderStream;
pub(super) fn status_response(status: StatusCode, detail: &'static str) -> Response {
// Keep internal diagnostics in server logs, not in public HTTP bodies.
// Keep implementation diagnostics in logs; use the existing embedded
// application-style pages for every Web Server-generated error.
if status.is_server_error() {
tracing::warn!(
status = status.as_u16(),
@@ -29,46 +30,46 @@ pub(super) fn status_response(status: StatusCode, detail: &'static str) -> Respo
"Web Server error"
);
}
let message = format!(
"{} {}\n",
status.as_u16(),
status.canonical_reason().unwrap_or("Error")
);
let mut response = Response::new(Body::from(message.clone()));
*response.status_mut() = status;
response.headers_mut().insert(
CONTENT_TYPE,
HeaderValue::from_static("text/plain; charset=utf-8"),
);
response.headers_mut().insert(
CONTENT_LENGTH,
HeaderValue::from_str(&message.len().to_string()).unwrap(),
);
response
.headers_mut()
.insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
response.headers_mut().insert(
"x-content-type-options",
HeaderValue::from_static("nosniff"),
);
response
// The public serve() facade strips bodies from HEAD responses while
// retaining the same Content-Length as the corresponding GET.
builtin_error_response(status, &Method::GET)
}
pub(super) fn builtin_error_response(status: StatusCode, method: &Method) -> Response {
let html = crate::pages::builtin_gateway_error(status)
.unwrap_or("<!doctype html><html><body><h1>Error</h1></body></html>");
let mut response = Response::new(if *method == Method::HEAD {
Body::empty()
} else {
Body::from(html)
});
// Reuse the bundled pages wherever one exists. For statuses such as 405,
// use the shared styled application fallback instead of a plain-text or
// unstyled generic error. Do not mark these as PageServed so normal Vhost
// error-page policy can still replace them.
let (body, length) = match crate::pages::builtin_gateway_error(status) {
Some(html) => (
if *method == Method::HEAD {
Body::empty()
} else {
Body::from(html)
},
html.len(),
),
None => {
let html = crate::pages::builtin_application_page(status);
let length = html.len();
(
if *method == Method::HEAD {
Body::empty()
} else {
Body::from(html)
},
length,
)
}
};
let mut response = Response::new(body);
*response.status_mut() = status;
response.headers_mut().insert(
CONTENT_TYPE,
HeaderValue::from_static("text/html; charset=utf-8"),
);
if let Ok(length) = HeaderValue::from_str(&html.len().to_string()) {
response.headers_mut().insert(CONTENT_LENGTH, length);
if let Ok(value) = HeaderValue::from_str(&length.to_string()) {
response.headers_mut().insert(CONTENT_LENGTH, value);
}
response
.headers_mut()
src/web_server/tests.rs
+99 -26
View File
@@ -119,10 +119,13 @@ async fn missing_files_and_disabled_directory_listing_return_normal_errors() {
.await;
assert_eq!(missing.status(), StatusCode::NOT_FOUND);
assert_eq!(missing.headers()[CACHE_CONTROL], "no-store");
assert_eq!(missing.headers()[CONTENT_TYPE], "text/plain; charset=utf-8");
assert_eq!(missing.headers()[CONTENT_TYPE], "text/html; charset=utf-8");
assert_eq!(missing.headers()["x-content-type-options"], "nosniff");
assert_eq!(
missing.into_body().collect().await.unwrap().to_bytes(),
"404 Not Found\n"
missing.into_body().collect().await.unwrap().to_bytes().as_ref(),
crate::pages::builtin_gateway_error(StatusCode::NOT_FOUND)
.unwrap()
.as_bytes()
);
let missing_head = serve_request(
@@ -132,7 +135,14 @@ async fn missing_files_and_disabled_directory_listing_return_normal_errors() {
)
.await;
assert_eq!(missing_head.status(), StatusCode::NOT_FOUND);
assert_eq!(missing_head.headers()[CONTENT_LENGTH], "14");
let missing_length = crate::pages::builtin_gateway_error(StatusCode::NOT_FOUND)
.unwrap()
.len()
.to_string();
assert_eq!(
missing_head.headers()[CONTENT_LENGTH],
HeaderValue::from_str(&missing_length).unwrap()
);
assert!(missing_head
.into_body()
.collect()
@@ -171,42 +181,105 @@ async fn missing_files_and_disabled_directory_listing_return_normal_errors() {
.await;
assert_eq!(method.status(), StatusCode::METHOD_NOT_ALLOWED);
assert_eq!(method.headers()[ALLOW], "GET, HEAD");
assert_eq!(
method.into_body().collect().await.unwrap().to_bytes(),
"405 Method Not Allowed\n"
);
assert_eq!(method.headers()[CONTENT_TYPE], "text/html; charset=utf-8");
assert_eq!(method.headers()[CACHE_CONTROL], "no-store");
let method_body = method.into_body().collect().await.unwrap().to_bytes();
let method_html = std::str::from_utf8(&method_body).unwrap();
assert!(method_html.contains("<h1>405 - Method Not Allowed</h1>"));
assert!(method_html.contains("background: #131615"));
}
#[tokio::test]
async fn internal_filesystem_diagnostics_never_appear_in_public_error_responses() {
use std::io::ErrorKind;
for (kind, expected_status, expected_text) in [
(
ErrorKind::NotFound,
StatusCode::NOT_FOUND,
"404 Not Found\n",
),
(
ErrorKind::PermissionDenied,
StatusCode::FORBIDDEN,
"403 Forbidden\n",
),
(
ErrorKind::Other,
StatusCode::INTERNAL_SERVER_ERROR,
"500 Internal Server Error\n",
),
for (kind, expected_status) in [
(ErrorKind::NotFound, StatusCode::NOT_FOUND),
(ErrorKind::PermissionDenied, StatusCode::FORBIDDEN),
(ErrorKind::Other, StatusCode::INTERNAL_SERVER_ERROR),
] {
let error = std::io::Error::from(kind);
let response = super::response::io_response(&error);
assert_eq!(response.status(), expected_status);
assert_eq!(response.headers()[CACHE_CONTROL], "no-store");
assert_eq!(response.headers()[CONTENT_TYPE], "text/html; charset=utf-8");
assert_eq!(response.headers()["x-content-type-options"], "nosniff");
let body = response.into_body().collect().await.unwrap().to_bytes();
assert_eq!(body.as_ref(), expected_text.as_bytes());
assert!(!body.starts_with(b"RustProxy:"));
assert_eq!(
body.as_ref(),
crate::pages::builtin_gateway_error(expected_status)
.unwrap()
.as_bytes()
);
assert!(!body.windows(b"RustProxy:".len()).any(|w| w == b"RustProxy:"));
}
}
#[tokio::test]
async fn web_server_validation_and_unmapped_error_statuses_are_styled() {
let tmp = TempDir::new().unwrap();
let service = service(tmp.path(), false);
// Path validation follows the same existing built-in page as other 400s.
let invalid = serve_request(&service, &request(Method::GET, "/"), "/../secret").await;
assert_eq!(invalid.status(), StatusCode::BAD_REQUEST);
assert_eq!(invalid.headers()[CONTENT_TYPE], "text/html; charset=utf-8");
assert_eq!(
invalid.into_body().collect().await.unwrap().to_bytes().as_ref(),
crate::pages::builtin_gateway_error(StatusCode::BAD_REQUEST)
.unwrap()
.as_bytes()
);
// 405 has no dedicated packaged HTML file: reuse the embedded styled
// fallback, while preserving Allow and the actual HTTP status.
let method = serve_request(&service, &request(Method::PUT, "/"), "/").await;
assert_eq!(method.status(), StatusCode::METHOD_NOT_ALLOWED);
assert_eq!(method.headers()[ALLOW], "GET, HEAD");
assert_eq!(method.headers()[CONTENT_TYPE], "text/html; charset=utf-8");
let body = method.into_body().collect().await.unwrap().to_bytes();
assert!(std::str::from_utf8(&body)
.unwrap()
.contains("405 - Method Not Allowed"));
// An operator-configured status without a dedicated HTML template still
// uses the common styled fallback, including for HEAD requests.
let fallback_service = Service {
id: "teapot".into(),
name: "Teapot".into(),
document_root: tmp.path().to_path_buf(),
try_files: vec!["=418".into()],
..Default::default()
}
.compile()
.unwrap();
let fallback = serve_request(
&fallback_service,
&request(Method::GET, "/missing"),
"/missing",
)
.await;
assert_eq!(fallback.status(), StatusCode::IM_A_TEAPOT);
assert_eq!(fallback.headers()[CONTENT_TYPE], "text/html; charset=utf-8");
assert_eq!(fallback.headers()[CACHE_CONTROL], "no-store");
assert!(fallback.extensions().get::<crate::pages::PageServed>().is_none());
let fallback_html = fallback.into_body().collect().await.unwrap().to_bytes();
let fallback_html = std::str::from_utf8(&fallback_html).unwrap();
assert!(fallback_html.contains("<h1>418 -"));
assert!(fallback_html.contains("background: #131615"));
let head = serve_request(
&fallback_service,
&request(Method::HEAD, "/missing"),
"/missing",
)
.await;
assert_eq!(head.status(), StatusCode::IM_A_TEAPOT);
assert_eq!(
head.headers()[CONTENT_LENGTH],
HeaderValue::from_str(&fallback_html.len().to_string()).unwrap()
);
assert!(head.into_body().collect().await.unwrap().to_bytes().is_empty());
}
#[tokio::test]
async fn last_modified_condition_returns_304() {
let tmp = TempDir::new().unwrap();