diff --git a/src/api/web_runtime.rs b/src/api/web_runtime.rs index b64ad84..b9ccae6 100644 --- a/src/api/web_runtime.rs +++ b/src/api/web_runtime.rs @@ -13,12 +13,13 @@ use super::model::ApiFailure; use super::{ALLOW_GET, ALLOW_POST, ApiShared}; use crate::config::ProxyConfig; use crate::web::control::{WebRuntimeLifecycle, WebRuntimePublication}; -use crate::web::manager::{ - ControlError, OperatorLifecycleError, SessionDetail, WebProcessRuntime, -}; +use crate::web::manager::{ControlError, OperatorLifecycleError, SessionDetail, WebProcessRuntime}; // Exact JSON DTOs and strict query parsing stay independent from route dispatch. mod request; +// Ingress, capacity, and decoy telemetry remain separate availability planes. +mod observability; +use observability::{WebCapacityStatus, WebDecoyUpstreamStatus, WebIngressStatus}; use request::{ CloseRequest, DrainRequest, RuntimeInstanceRequest, parse_session_query, parse_session_ref, valid_runtime_instance, @@ -73,7 +74,7 @@ pub(super) async fn handle( reject_query(query)?; let publication = shared.web_runtime_rx.borrow().clone(); let runtime = publication.runtime.upgrade(); - let data = WebStatusData::new(publication, runtime.as_deref(), config.web.enabled); + let data = WebStatusData::new(publication, runtime.as_deref(), config); Ok(success_response(StatusCode::OK, data, revision)) } ("GET", SESSIONS_PATH) => { @@ -265,6 +266,9 @@ struct WebStatusData { reason: Option<&'static str>, listeners: Vec, effective_config_enabled: bool, + ingress: WebIngressStatus, + capacity: WebCapacityStatus, + decoy_upstream: WebDecoyUpstreamStatus, #[serde(skip_serializing_if = "Option::is_none")] operator_lifecycle: Option, #[serde(skip_serializing_if = "Option::is_none")] @@ -275,7 +279,7 @@ impl WebStatusData { fn new( publication: WebRuntimePublication, runtime: Option<&WebProcessRuntime>, - effective_config_enabled: bool, + config: &ProxyConfig, ) -> Self { let available = runtime.is_some() && matches!( @@ -295,6 +299,9 @@ impl WebStatusData { }) }; let operator_lifecycle = runtime.map(WebProcessRuntime::operator_lifecycle_status); + let ingress = WebIngressStatus::new(&publication, runtime.is_some()); + let capacity = WebCapacityStatus::new(&publication, runtime, config); + let decoy_upstream = WebDecoyUpstreamStatus::new(&publication); Self { lifecycle: publication.lifecycle.as_str(), lifecycle_epoch: publication.epoch, @@ -306,7 +313,10 @@ impl WebStatusData { .iter() .map(ToString::to_string) .collect(), - effective_config_enabled, + effective_config_enabled: config.web.enabled, + ingress, + capacity, + decoy_upstream, operator_lifecycle, runtime: runtime.map(WebProcessRuntime::try_status), } diff --git a/src/api/web_runtime/observability.rs b/src/api/web_runtime/observability.rs new file mode 100644 index 0000000..f327f93 --- /dev/null +++ b/src/api/web_runtime/observability.rs @@ -0,0 +1,161 @@ +use serde::Serialize; + +use crate::config::{ProxyConfig, WebHttpConnectionCapacityAction}; +use crate::web::control::{WebRuntimeLifecycle, WebRuntimePublication}; +use crate::web::manager::{WebCapacityResourceStatus, WebCapacitySnapshot, WebProcessRuntime}; +use crate::web::telemetry::{WebOutcomeCounter, WebRejectionCounter}; + +/// Private WEB ingress state owned by this Telemt process. +#[derive(Serialize)] +pub(super) struct WebIngressStatus { + configured_listeners: usize, + live_acceptors: usize, + accepting_connections: bool, + #[serde(skip_serializing_if = "Option::is_none")] + reason: Option<&'static str>, + tcp_accept_total: u64, + tcp_accept_error_total: u64, +} + +impl WebIngressStatus { + pub(super) fn new(publication: &WebRuntimePublication, runtime_available: bool) -> Self { + let configured_listeners = publication.listeners.len(); + let live_acceptors = publication.telemetry.live_acceptors(); + let accepting_connections = publication.lifecycle == WebRuntimeLifecycle::Running + && runtime_available + && configured_listeners != 0 + && live_acceptors == configured_listeners; + let reason = if accepting_connections { + None + } else { + Some(match publication.lifecycle { + WebRuntimeLifecycle::Starting => "starting", + WebRuntimeLifecycle::NoWebListener => "no_web_listener", + WebRuntimeLifecycle::Draining => "ingress_draining", + WebRuntimeLifecycle::Drained => "ingress_drained", + WebRuntimeLifecycle::DeadlineExceeded => "deadline_exceeded", + WebRuntimeLifecycle::Running if !runtime_available => "runtime_released", + WebRuntimeLifecycle::Running if configured_listeners == 0 => "no_web_listener", + WebRuntimeLifecycle::Running => "acceptor_unavailable", + }) + }; + Self { + configured_listeners, + live_acceptors, + accepting_connections, + reason, + tcp_accept_total: publication.telemetry.accepted(), + tcp_accept_error_total: publication.telemetry.accept_errors(), + } + } +} + +/// Bounded process-wide WEB capacity and terminal rejection view. +#[derive(Serialize)] +pub(super) struct WebCapacityStatus { + http_connection_capacity_action: WebHttpConnectionCapacityAction, + max_http_overload_connections: usize, + http_overload_timeout_ms: u64, + resources: Vec, + saturated_resources: Vec<&'static str>, + partial: Vec<&'static str>, + rejections: Vec, + http_connection_overload_outcomes: Vec, +} + +impl WebCapacityStatus { + pub(super) fn new( + publication: &WebRuntimePublication, + runtime: Option<&WebProcessRuntime>, + config: &ProxyConfig, + ) -> Self { + let snapshot = runtime + .map(WebProcessRuntime::capacity_snapshot) + .unwrap_or_else(runtime_unavailable_snapshot); + Self { + http_connection_capacity_action: config.web.http_connection_capacity_action, + max_http_overload_connections: config.web.limits.max_http_overload_connections, + http_overload_timeout_ms: config.web.timeouts.http_overload_timeout_ms, + resources: snapshot.resources, + saturated_resources: snapshot.saturated_resources, + partial: snapshot.partial, + rejections: publication.telemetry.rejection_counters(), + http_connection_overload_outcomes: publication.telemetry.overload_counters(), + } + } +} + +fn runtime_unavailable_snapshot() -> WebCapacitySnapshot { + WebCapacitySnapshot { + resources: Vec::new(), + saturated_resources: Vec::new(), + partial: vec!["runtime"], + } +} + +/// Passive health of Telemt's internal plain-HTTP decoy origin hop. +#[derive(Serialize)] +pub(super) struct WebDecoyUpstreamStatus { + outcomes: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + last_outcome: Option<&'static str>, + #[serde(skip_serializing_if = "Option::is_none")] + last_outcome_age_ms: Option, +} + +impl WebDecoyUpstreamStatus { + pub(super) fn new(publication: &WebRuntimePublication) -> Self { + let last = publication.telemetry.last_decoy(); + Self { + outcomes: publication.telemetry.decoy_counters(), + last_outcome: last.map(|value| value.0), + last_outcome_age_ms: last.map(|value| value.1), + } + } +} + +#[cfg(test)] +mod tests { + use crate::config::ProxyConfig; + use crate::web::control::WebRuntimeControl; + + #[test] + fn starting_ingress_does_not_claim_external_availability() { + let control = WebRuntimeControl::new(); + let publication = control.subscribe().borrow().clone(); + let value = + serde_json::to_value(super::WebIngressStatus::new(&publication, false)).unwrap(); + assert_eq!(value["configured_listeners"], 0); + assert_eq!(value["live_acceptors"], 0); + assert_eq!(value["accepting_connections"], false); + assert_eq!(value["reason"], "starting"); + } + + #[test] + fn unavailable_runtime_keeps_fixed_counter_sets_visible() { + let control = WebRuntimeControl::new(); + let publication = control.subscribe().borrow().clone(); + let config = ProxyConfig::default(); + let capacity = + serde_json::to_value(super::WebCapacityStatus::new(&publication, None, &config)) + .unwrap(); + let decoy = serde_json::to_value(super::WebDecoyUpstreamStatus::new(&publication)).unwrap(); + + assert_eq!( + capacity["rejections"].as_array().unwrap().len(), + crate::web::telemetry::WebRejectionReason::ALL.len() + ); + assert_eq!( + capacity["http_connection_overload_outcomes"] + .as_array() + .unwrap() + .len(), + crate::web::telemetry::WebHttpConnectionOverloadOutcome::ALL.len() + ); + assert_eq!( + decoy["outcomes"].as_array().unwrap().len(), + crate::web::telemetry::WebDecoyUpstreamOutcome::ALL.len() + ); + assert_eq!(capacity["partial"][0], "runtime"); + } +} diff --git a/src/config/load.rs b/src/config/load.rs index 45d03e9..66876e9 100644 --- a/src/config/load.rs +++ b/src/config/load.rs @@ -216,6 +216,11 @@ impl ProxyConfig { runtime_web::rebuild(self) } + /// Revalidates decoy separation after restart-only listener fields are resolved. + pub(crate) fn validate_web_decoy_listener_separation(&self) -> Result<()> { + validate_web::validate_decoy_listener_separation(self) + } + pub(crate) fn runtime_user_auth(&self) -> Option<&UserAuthSnapshot> { self.runtime_user_auth.as_deref() } diff --git a/src/config/load/strict_keys.rs b/src/config/load/strict_keys.rs index 1f1a90e..08b9ee5 100644 --- a/src/config/load/strict_keys.rs +++ b/src/config/load/strict_keys.rs @@ -265,6 +265,7 @@ const WEB_CONFIG_KEYS: &[&str] = &[ "carriers", "carrier_learning", "carrier_negotiation_aggressiveness", + "http_connection_capacity_action", "debug", "limits", "timeouts", @@ -278,6 +279,7 @@ const WEB_LIMITS_CONFIG_KEYS: &[&str] = &[ "carrier_batch_bytes", "max_frames_per_body", "max_http_connections", + "max_http_overload_connections", "max_http_handlers", "max_lane_open_waits_per_session", "pending_bytes_per_lane", @@ -354,6 +356,7 @@ const WEB_TIMEOUTS_CONFIG_KEYS: &[&str] = &[ "bootstrap_lifetime_secs", "reconnect_grace_secs", "http_idle_secs", + "http_overload_timeout_ms", "shutdown_secs", "decoy_header_secs", ]; diff --git a/src/config/load/validate_web.rs b/src/config/load/validate_web.rs index 0863298..51b3433 100644 --- a/src/config/load/validate_web.rs +++ b/src/config/load/validate_web.rs @@ -83,9 +83,59 @@ pub(super) fn validate(config: &mut ProxyConfig) -> Result<()> { timeouts::validate(&config.web.timeouts)?; websocket::validate(&carriers, &config.web.limits, &config.web.timeouts)?; validate_vhosts(config)?; + validate_decoy_listener_separation(config)?; Ok(()) } +/// Rejects a direct decoy recursion into an effective WEB listener. +pub(super) fn validate_decoy_listener_separation(config: &ProxyConfig) -> Result<()> { + let web_listeners = config + .server + .listeners + .iter() + .filter(|listener| listener.transport == ListenerTransport::Web) + .filter(|listener| { + (listener.ip.is_ipv4() && config.network.ipv4) + || (listener.ip.is_ipv6() && config.network.ipv6 != Some(false)) + }) + .map(|listener| SocketAddr::new(listener.ip, listener.port.unwrap_or(config.server.port))) + .collect::>(); + for (vhost_idx, vhost) in config.web.vhosts.iter().enumerate() { + let WebDecoyConfig::HttpUpstream { upstream } = &vhost.decoy else { + continue; + }; + let parsed = url::Url::parse(upstream).map_err(|error| { + ProxyError::Config(format!( + "web.vhosts[{vhost_idx}].decoy.upstream is invalid: {error}" + )) + })?; + let Some(port) = parsed.port_or_known_default() else { + continue; + }; + let upstream_ip = match parsed.host() { + Some(url::Host::Ipv4(ip)) => IpAddr::V4(ip), + Some(url::Host::Ipv6(ip)) => IpAddr::V6(ip), + _ => continue, + }; + let upstream_addr = SocketAddr::new(upstream_ip, port); + if web_listeners + .iter() + .any(|listener| listener_covers(*listener, upstream_addr)) + { + return config_error(&format!( + "web.vhosts[{vhost_idx}].decoy upstream overlaps WEB listener {upstream_addr}" + )); + } + } + Ok(()) +} + +fn listener_covers(listener: SocketAddr, target: SocketAddr) -> bool { + listener.port() == target.port() + && (listener.ip() == target.ip() + || (listener.ip().is_unspecified() && listener.is_ipv4() == target.is_ipv4())) +} + fn validate_web_listener( config: &ProxyConfig, idx: usize, @@ -169,6 +219,10 @@ fn validate_limits(limits: &WebLimitsConfig) -> Result<()> { let positive = [ ("max_http_connections", limits.max_http_connections), + ( + "max_http_overload_connections", + limits.max_http_overload_connections, + ), ("max_http_handlers", limits.max_http_handlers), ( "max_lane_open_waits_per_session", @@ -222,6 +276,10 @@ fn validate_limits(limits: &WebLimitsConfig) -> Result<()> { } for (field, value) in [ ("max_http_connections", limits.max_http_connections), + ( + "max_http_overload_connections", + limits.max_http_overload_connections, + ), ("max_http_handlers", limits.max_http_handlers), ("max_body_readers", limits.max_body_readers), ("max_body_bytes_global", limits.max_body_bytes_global), diff --git a/src/config/load/validate_web/memory.rs b/src/config/load/validate_web/memory.rs index 81262ac..7f5ef4b 100644 --- a/src/config/load/validate_web/memory.rs +++ b/src/config/load/validate_web/memory.rs @@ -5,6 +5,7 @@ const WEB_DEBUG_STATUS_PAGE_BYTES: usize = 8 * 1024 * 1024; const WEB_DEBUG_GROUP_SCRATCH_BYTES: usize = 4 * 1024 * 1024; const WEB_CARRIER_LEARNING_ENTRY_BYTES: usize = 512; const WEB_LANE_STATE_BYTES: usize = 512; +const WEB_OVERLOAD_CONNECTION_BYTES: usize = 4 * 1024; /// Validates process-wide body, header, queue, static, and debug reservations. pub(super) fn validate(limits: &WebLimitsConfig) -> Result<()> { @@ -30,6 +31,12 @@ pub(super) fn validate(limits: &WebLimitsConfig) -> Result<()> { .ok_or_else(|| { ProxyError::Config("web.limits HTTP header reservations overflow usize".to_string()) })?; + let overload_connection_reservation = limits + .max_http_overload_connections + .checked_mul(WEB_OVERLOAD_CONNECTION_BYTES) + .ok_or_else(|| { + ProxyError::Config("web.limits HTTP overload reservations overflow usize".to_string()) + })?; let debug_ring_index = limits .debug_records_capacity .checked_mul(std::mem::size_of::()) @@ -77,6 +84,7 @@ pub(super) fn validate(limits: &WebLimitsConfig) -> Result<()> { .and_then(|value| value.checked_add(carrier_learning_reservation)) .and_then(|value| value.checked_add(lane_state_reservation)) .and_then(|value| value.checked_add(http_header_reservation)) + .and_then(|value| value.checked_add(overload_connection_reservation)) .ok_or_else(|| ProxyError::Config("web.limits byte ceilings overflow usize".to_string()))?; if reserved > limits.memory_envelope_bytes || limits.memory_envelope_bytes > MAX_WEB_MEMORY_ENVELOPE_BYTES diff --git a/src/config/load/validate_web/timeouts.rs b/src/config/load/validate_web/timeouts.rs index 4c9877c..66fa6e9 100644 --- a/src/config/load/validate_web/timeouts.rs +++ b/src/config/load/validate_web/timeouts.rs @@ -2,6 +2,9 @@ use super::*; /// Validates WEB request, learning, and lifecycle timeouts. pub(super) fn validate(timeouts: &WebTimeoutsConfig) -> Result<()> { + if !(1..=60_000).contains(&timeouts.http_overload_timeout_ms) { + return config_error("web.timeouts.http_overload_timeout_ms must be within [1, 60000]"); + } let values = [ ("header_secs", timeouts.header_secs), ("body_secs", timeouts.body_secs), diff --git a/src/config/tests/load_basic_tests/web_tests.rs b/src/config/tests/load_basic_tests/web_tests.rs index 17a5c05..e8ed6b4 100644 --- a/src/config/tests/load_basic_tests/web_tests.rs +++ b/src/config/tests/load_basic_tests/web_tests.rs @@ -56,6 +56,78 @@ fn web_config_builds_canonical_runtime_snapshot() { ); } +#[test] +fn web_http_connection_capacity_policy_is_bounded_and_configurable() { + let configured = WEB_CONFIG + .replace( + "carrier = \"https-lanes\"", + "carrier = \"https-lanes\"\nhttp_connection_capacity_action = \"wait\"", + ) + .replace( + "[[web.vhosts]]", + "[web.limits]\nmax_http_overload_connections = 23\n\n[web.timeouts]\nhttp_overload_timeout_ms = 731\n\n[[web.vhosts]]", + ); + let config = load_config_from_temp_toml(&configured); + + assert_eq!( + config.web.http_connection_capacity_action, + WebHttpConnectionCapacityAction::Wait + ); + assert_eq!(config.web.limits.max_http_overload_connections, 23); + assert_eq!(config.web.timeouts.http_overload_timeout_ms, 731); + + let defaults = ProxyConfig::default(); + assert_eq!( + defaults.web.http_connection_capacity_action, + WebHttpConnectionCapacityAction::Drop + ); + assert_eq!(defaults.web.limits.max_http_overload_connections, 64); + assert_eq!(defaults.web.timeouts.http_overload_timeout_ms, 250); +} + +#[test] +fn web_http_connection_capacity_policy_rejects_unknown_or_unbounded_values() { + let unknown = WEB_CONFIG.replace( + "carrier = \"https-lanes\"", + "carrier = \"https-lanes\"\nhttp_connection_capacity_action = \"queue\"", + ); + assert!(load_config_error_from_temp_toml(&unknown).contains("http_connection_capacity_action")); + + for timeout in [0, 60_001] { + let invalid = WEB_CONFIG.replace( + "[[web.vhosts]]", + &format!("[web.timeouts]\nhttp_overload_timeout_ms = {timeout}\n\n[[web.vhosts]]"), + ); + assert!( + load_config_error_from_temp_toml(&invalid) + .contains("web.timeouts.http_overload_timeout_ms") + ); + } + + let no_overload_slots = WEB_CONFIG.replace( + "[[web.vhosts]]", + "[web.limits]\nmax_http_overload_connections = 0\n\n[[web.vhosts]]", + ); + assert!( + load_config_error_from_temp_toml(&no_overload_slots) + .contains("web.limits.max_http_overload_connections") + ); +} + +#[test] +fn web_decoy_rejects_direct_and_wildcard_listener_loops() { + let direct = WEB_CONFIG.replace("http://127.0.0.1:18081", "http://127.0.0.1:18080"); + assert!( + load_config_error_from_temp_toml(&direct).contains("decoy upstream overlaps WEB listener") + ); + + let wildcard = direct.replace("ip = \"127.0.0.1\"", "ip = \"0.0.0.0\""); + assert!( + load_config_error_from_temp_toml(&wildcard) + .contains("decoy upstream overlaps WEB listener") + ); +} + #[test] fn web_profile_user_labels_are_bounded_for_runtime_status() { let user = "a".repeat(65); diff --git a/src/config/types.rs b/src/config/types.rs index 562f54e..d5f8d61 100644 --- a/src/config/types.rs +++ b/src/config/types.rs @@ -53,8 +53,9 @@ pub use server::{ }; #[allow(unused_imports)] pub use web::{ - WebCarrierNegotiationAggressiveness, WebConfig, WebDecoyConfig, WebLimitsConfig, - WebProfileConfig, WebSecretMode, WebTimeoutsConfig, WebVhostConfig, + WebCarrierNegotiationAggressiveness, WebConfig, WebDecoyConfig, + WebHttpConnectionCapacityAction, WebLimitsConfig, WebProfileConfig, WebSecretMode, + WebTimeoutsConfig, WebVhostConfig, }; pub(crate) use web::{ WebRuntimeConfig, WebRuntimeDecoy, WebRuntimeProfile, WebRuntimeVhost, WebStaticAsset, diff --git a/src/config/types/web.rs b/src/config/types/web.rs index 42ef24f..857d579 100644 --- a/src/config/types/web.rs +++ b/src/config/types/web.rs @@ -12,6 +12,9 @@ use super::web_debug::WebDebugConfig; // Serialized WEB defaults remain separate from the runtime data model. mod defaults; use defaults::*; +// Accepted-socket overload policy remains separate from the bulky WEB data model. +mod overload; +pub use overload::WebHttpConnectionCapacityAction; /// Client-facing secret representation used to derive a WEB capability. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] @@ -95,6 +98,9 @@ pub struct WebLimitsConfig { /// Process-wide accepted WEB HTTP connection ceiling. #[serde(default = "default_web_max_http_connections")] pub max_http_connections: usize, + /// Accepted overload sockets allowed to wait or emit a retryable response. + #[serde(default = "default_web_max_http_overload_connections")] + pub max_http_overload_connections: usize, /// Process-wide concurrently executing HTTP handler ceiling. #[serde(default = "default_web_max_http_handlers")] pub max_http_handlers: usize, @@ -226,6 +232,7 @@ impl Default for WebLimitsConfig { carrier_batch_bytes: default_web_carrier_batch_bytes(), max_frames_per_body: default_web_max_frames_per_body(), max_http_connections: default_web_max_http_connections(), + max_http_overload_connections: default_web_max_http_overload_connections(), max_http_handlers: default_web_max_http_handlers(), max_lane_open_waits_per_session: default_web_max_lane_open_waits_per_session(), pending_bytes_per_lane: default_web_pending_bytes_per_lane(), @@ -333,6 +340,9 @@ pub struct WebTimeoutsConfig { /// Maximum idle lifetime of a WEB HTTP keep-alive connection. #[serde(default = "default_web_http_idle_secs")] pub http_idle_secs: u64, + /// Per-phase wait or response deadline for accepted HTTP overload sockets. + #[serde(default = "default_web_http_overload_timeout_ms")] + pub http_overload_timeout_ms: u64, /// Maximum graceful wait for WEB connections and process-owned tasks. #[serde(default = "default_web_shutdown_secs")] pub shutdown_secs: u64, @@ -364,6 +374,7 @@ impl Default for WebTimeoutsConfig { bootstrap_lifetime_secs: default_web_bootstrap_lifetime_secs(), reconnect_grace_secs: default_web_reconnect_grace_secs(), http_idle_secs: default_web_http_idle_secs(), + http_overload_timeout_ms: default_web_http_overload_timeout_ms(), shutdown_secs: default_web_shutdown_secs(), decoy_header_secs: default_web_decoy_header_timeout_secs(), } @@ -401,6 +412,9 @@ pub struct WebConfig { /// Controls the evidence thresholds used by automatic carrier ranking. #[serde(default)] pub carrier_negotiation_aggressiveness: WebCarrierNegotiationAggressiveness, + /// Action applied when accepted HTTP connection capacity is exhausted. + #[serde(default)] + pub http_connection_capacity_action: WebHttpConnectionCapacityAction, /// Hard process and protocol limits. #[serde(default)] pub limits: WebLimitsConfig, @@ -447,6 +461,7 @@ impl Default for WebConfig { carriers: WebCarriers::default(), carrier_learning: default_web_carrier_learning(), carrier_negotiation_aggressiveness: WebCarrierNegotiationAggressiveness::default(), + http_connection_capacity_action: WebHttpConnectionCapacityAction::default(), limits: WebLimitsConfig::default(), debug: WebDebugConfig::default(), timeouts: WebTimeoutsConfig::default(), diff --git a/src/config/types/web/defaults.rs b/src/config/types/web/defaults.rs index 42db74f..1616e98 100644 --- a/src/config/types/web/defaults.rs +++ b/src/config/types/web/defaults.rs @@ -40,6 +40,7 @@ usize_default!(default_web_max_frame_payload_bytes, 1024 * 1024); usize_default!(default_web_carrier_batch_bytes, 2 * 1024 * 1024); usize_default!(default_web_max_frames_per_body, 4096); usize_default!(default_web_max_http_connections, 1024); +usize_default!(default_web_max_http_overload_connections, 64); usize_default!(default_web_max_http_handlers, 512); usize_default!(default_web_max_lane_open_waits_per_session, 16); usize_default!(default_web_pending_bytes_per_lane, 8 * 1024 * 1024); @@ -105,5 +106,6 @@ pub(super) fn default_web_carrier_learning() -> bool { u64_default!(default_web_bootstrap_lifetime_secs, 120); u64_default!(default_web_reconnect_grace_secs, 120); u64_default!(default_web_http_idle_secs, 75); +u64_default!(default_web_http_overload_timeout_ms, 250); u64_default!(default_web_shutdown_secs, 15); u64_default!(default_web_decoy_header_timeout_secs, 30); diff --git a/src/config/types/web/overload.rs b/src/config/types/web/overload.rs new file mode 100644 index 0000000..cd00995 --- /dev/null +++ b/src/config/types/web/overload.rs @@ -0,0 +1,14 @@ +use serde::{Deserialize, Serialize}; + +/// Action applied after an accepted WEB socket finds HTTP connection capacity exhausted. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum WebHttpConnectionCapacityAction { + /// Close the accepted socket without emitting an HTTP response. + #[default] + Drop, + /// Wait for ordinary HTTP connection capacity under the overload deadline. + Wait, + /// Emit a bounded retryable HTTP response without parsing the request. + Respond, +} diff --git a/src/maestro/listeners.rs b/src/maestro/listeners.rs index 9d065a7..2fbec10 100644 --- a/src/maestro/listeners.rs +++ b/src/maestro/listeners.rs @@ -6,6 +6,7 @@ //! - `bind` prepares and activates sockets without partial startup binding. //! - `accept` runs cancellation-aware TCP accept loops. //! - `control` coordinates reversible listener transitions and shutdown. +//! - `web_overload` handles accepted WEB sockets outside ordinary capacity. mod accept; mod bind; @@ -13,6 +14,7 @@ mod control; mod plan; #[cfg(unix)] mod unix; +mod web_overload; pub(crate) use bind::bind_listeners; pub(crate) use control::{ListenerManager, PreparedListenerTransition}; diff --git a/src/maestro/listeners/accept.rs b/src/maestro/listeners/accept.rs index 75e768d..9385b3e 100644 --- a/src/maestro/listeners/accept.rs +++ b/src/maestro/listeners/accept.rs @@ -13,9 +13,11 @@ use crate::config::{ListenerTransport, RstOnCloseMode}; use crate::proxy::ClientHandler; use crate::transport::socket::set_linger_zero; use crate::web::manager::WebProcessRuntime; +use crate::web::telemetry::{WebAcceptorGuard, WebHttpConnectionOverloadOutcome}; use super::bind::BoundTcpListener; use super::plan::ListenerBindSpec; +use super::web_overload; use crate::maestro::generation::RuntimeGeneration; use crate::maestro::helpers::{ expected_handshake_close_description, is_expected_handshake_eof, peer_close_description, @@ -190,6 +192,7 @@ async fn run_accept_loop( web_runtime: Option>, connections: TaskTracker, cancellation: CancellationToken, + _web_acceptor_guard: Option, ) { loop { let accepted = tokio::select! { @@ -204,8 +207,45 @@ async fn run_accept_loop( error!(addr = %spec.addr, "WEB listener has no process runtime"); return; }; + web_runtime.telemetry().record_accept(); let Some(connection_permit) = web_runtime.try_http_connection() else { - drop(stream); + let config = web_runtime.active_generation().config(); + let action = config.web.http_connection_capacity_action; + let phase_timeout = + Duration::from_millis(config.web.timeouts.http_overload_timeout_ms); + drop(config); + if action == crate::config::WebHttpConnectionCapacityAction::Drop { + web_runtime.telemetry().record_rejection( + crate::web::telemetry::WebRejectionReason::HttpConnectionCapacity, + ); + web_runtime + .telemetry() + .record_overload(WebHttpConnectionOverloadOutcome::Dropped); + drop(stream); + continue; + } + let Some(overload_permit) = web_runtime.try_http_overload_connection() + else { + web_runtime.telemetry().record_rejection( + crate::web::telemetry::WebRejectionReason::HttpConnectionCapacity, + ); + web_runtime.telemetry().record_overload( + WebHttpConnectionOverloadOutcome::OverflowCapacityDrop, + ); + drop(stream); + continue; + }; + connections.spawn(web_overload::serve( + stream, + peer_addr, + spec.web_client_ip_source, + Arc::clone(&spec.web_trusted_proxy_cidrs), + Arc::clone(web_runtime), + cancellation.clone(), + overload_permit, + action, + phase_timeout, + )); continue; }; connections.spawn(crate::web::http::serve_connection( @@ -245,6 +285,9 @@ async fn run_accept_loop( } } Err(error_value) => { + if let Some(web_runtime) = &web_runtime { + web_runtime.telemetry().record_accept_error(); + } error!(addr = %spec.addr, error = %error_value, "TCP accept error"); tokio::select! { biased; @@ -262,8 +305,16 @@ impl ListenerSlot { active_runtime: Arc>, web_runtime: Option>, ) -> Self { + let web_runtime = if bound.spec.transport == ListenerTransport::Web { + web_runtime + } else { + None + }; let cancellation = CancellationToken::new(); let connections = TaskTracker::new(); + let web_acceptor_guard = web_runtime + .as_ref() + .map(|runtime| runtime.telemetry().acceptor_guard()); let task = tokio::spawn(run_accept_loop( bound.listener.clone(), bound.spec.clone(), @@ -271,6 +322,7 @@ impl ListenerSlot { web_runtime.clone(), connections.clone(), cancellation.clone(), + web_acceptor_guard, )); Self { spec: bound.spec, @@ -364,6 +416,10 @@ impl ListenerSlot { self.active_runtime = active_runtime.clone(); self.cancellation = CancellationToken::new(); self.connections = TaskTracker::new(); + let web_acceptor_guard = self + .web_runtime + .as_ref() + .map(|runtime| runtime.telemetry().acceptor_guard()); self.task = Some(tokio::spawn(run_accept_loop( self.listener.clone(), self.spec.clone(), @@ -371,6 +427,7 @@ impl ListenerSlot { self.web_runtime.clone(), self.connections.clone(), self.cancellation.clone(), + web_acceptor_guard, ))); } } diff --git a/src/maestro/listeners/control.rs b/src/maestro/listeners/control.rs index 46cfd17..b0a1380 100644 --- a/src/maestro/listeners/control.rs +++ b/src/maestro/listeners/control.rs @@ -58,8 +58,13 @@ impl ListenerManager { .map(|listener| listener.spec.addr) .collect(); let has_web = !web_listeners.is_empty(); - let web_runtime = - has_web.then(|| WebProcessRuntime::start_with_trace(active_runtime.clone(), trace)); + let web_runtime = has_web.then(|| { + WebProcessRuntime::start_with_trace( + active_runtime.clone(), + trace, + web_control.telemetry(), + ) + }); let mut slots = BTreeMap::new(); for listener in bound.listeners { let addr = listener.spec.addr; @@ -461,4 +466,30 @@ mod tests { manager.shutdown().await.unwrap(); runtime.stop_sessions().await; } + + #[tokio::test] + async fn acceptor_liveness_counts_only_web_listeners() { + let runtime = test_runtime_generation(1, ProxyConfig::default()); + let active_runtime = Arc::new(ArcSwap::from(runtime.clone())); + let (native_listener, _native_addr) = bound_listener().await; + let (mut web_listener, _web_addr) = bound_listener().await; + web_listener.spec.transport = ListenerTransport::Web; + let bound = BoundListeners { + listeners: vec![native_listener, web_listener], + #[cfg(unix)] + unix_listener: None, + }; + let trace = WebTraceStore::new( + runtime.config().web.debug.clone(), + &runtime.config().web.limits, + ); + let control = WebRuntimeControl::new(); + let receiver = control.subscribe(); + let mut manager = ListenerManager::start(bound, active_runtime, trace, control); + + assert_eq!(receiver.borrow().telemetry.live_acceptors(), 1); + + manager.shutdown().await.unwrap(); + runtime.stop_sessions().await; + } } diff --git a/src/maestro/listeners/web_overload.rs b/src/maestro/listeners/web_overload.rs new file mode 100644 index 0000000..240739f --- /dev/null +++ b/src/maestro/listeners/web_overload.rs @@ -0,0 +1,277 @@ +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; + +use ipnetwork::IpNetwork; +use tokio::io::AsyncWriteExt; +use tokio::net::TcpStream; +use tokio::sync::OwnedSemaphorePermit; +use tokio_util::sync::CancellationToken; + +use crate::config::{WebClientIpSource, WebHttpConnectionCapacityAction}; +use crate::web::manager::WebProcessRuntime; +use crate::web::telemetry::{WebHttpConnectionOverloadOutcome, WebRejectionReason}; + +pub(super) const SERVICE_UNAVAILABLE_RESPONSE: &[u8] = b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nCache-Control: no-store\r\nRetry-After: 1\r\nConnection: close\r\n\r\n"; + +/// Handles one accepted WEB socket outside ordinary connection capacity. +#[allow(clippy::too_many_arguments)] +pub(super) async fn serve( + stream: TcpStream, + peer: SocketAddr, + client_ip_source: WebClientIpSource, + trusted_proxy_cidrs: Arc<[IpNetwork]>, + runtime: Arc, + cancellation: CancellationToken, + overload_permit: OwnedSemaphorePermit, + action: WebHttpConnectionCapacityAction, + phase_timeout: Duration, +) { + match action { + WebHttpConnectionCapacityAction::Drop => unreachable!("drop is handled before spawn"), + WebHttpConnectionCapacityAction::Respond => { + let outcome = respond(stream, &cancellation, phase_timeout).await; + record_final_capacity_rejection(&runtime, outcome); + runtime.telemetry().record_overload(outcome); + } + WebHttpConnectionCapacityAction::Wait => { + let connection_permit = tokio::select! { + biased; + _ = cancellation.cancelled() => { + runtime + .telemetry() + .record_overload(WebHttpConnectionOverloadOutcome::ShutdownDrop); + return; + } + permit = tokio::time::timeout(phase_timeout, runtime.acquire_http_connection()) => { + permit.ok().flatten() + } + }; + let Some(connection_permit) = connection_permit else { + if runtime.is_shutdown() { + runtime + .telemetry() + .record_overload(WebHttpConnectionOverloadOutcome::ShutdownDrop); + return; + } + let outcome = match respond(stream, &cancellation, phase_timeout).await { + WebHttpConnectionOverloadOutcome::Responded503 => { + WebHttpConnectionOverloadOutcome::WaitTimeout503 + } + other => other, + }; + record_final_capacity_rejection(&runtime, outcome); + runtime.telemetry().record_overload(outcome); + return; + }; + runtime + .telemetry() + .record_overload(WebHttpConnectionOverloadOutcome::WaitAdmitted); + drop(overload_permit); + crate::web::http::serve_connection( + stream, + peer, + client_ip_source, + trusted_proxy_cidrs, + runtime, + cancellation, + connection_permit, + ) + .await; + } + } +} + +fn record_final_capacity_rejection( + runtime: &WebProcessRuntime, + outcome: WebHttpConnectionOverloadOutcome, +) { + if matches!( + outcome, + WebHttpConnectionOverloadOutcome::Responded503 + | WebHttpConnectionOverloadOutcome::WaitTimeout503 + | WebHttpConnectionOverloadOutcome::ResponseErrorDrop + ) { + runtime + .telemetry() + .record_rejection(WebRejectionReason::HttpConnectionCapacity); + } +} + +async fn respond( + stream: TcpStream, + cancellation: &CancellationToken, + phase_timeout: Duration, +) -> WebHttpConnectionOverloadOutcome { + tokio::select! { + biased; + _ = cancellation.cancelled() => WebHttpConnectionOverloadOutcome::ShutdownDrop, + written = write_service_unavailable(stream, phase_timeout) => { + if written { + WebHttpConnectionOverloadOutcome::Responded503 + } else { + WebHttpConnectionOverloadOutcome::ResponseErrorDrop + } + } + } +} + +async fn write_service_unavailable(mut stream: TcpStream, deadline: Duration) -> bool { + tokio::time::timeout(deadline, async { + stream.write_all(SERVICE_UNAVAILABLE_RESPONSE).await?; + stream.shutdown().await + }) + .await + .is_ok_and(|result| result.is_ok()) +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + use std::time::Duration; + + use arc_swap::ArcSwap; + use tokio::io::AsyncReadExt; + use tokio::net::{TcpListener, TcpStream}; + use tokio_util::sync::CancellationToken; + + use crate::config::{ProxyConfig, WebClientIpSource, WebHttpConnectionCapacityAction}; + use crate::maestro::generation::test_runtime_generation; + use crate::web::manager::WebProcessRuntime; + use crate::web::telemetry::{WebHttpConnectionOverloadOutcome, WebRejectionReason}; + + async fn tcp_pair() -> (TcpStream, TcpStream) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let client = TcpStream::connect(addr); + let server = listener.accept(); + let (client, server) = tokio::join!(client, server); + (server.unwrap().0, client.unwrap()) + } + + #[tokio::test] + async fn overload_response_is_exact_retryable_http() { + let (server, mut client) = tcp_pair().await; + assert!(super::write_service_unavailable(server, Duration::from_secs(1)).await); + + let mut bytes = Vec::new(); + client.read_to_end(&mut bytes).await.unwrap(); + assert_eq!(bytes, super::SERVICE_UNAVAILABLE_RESPONSE); + } + + #[tokio::test] + async fn wait_timeout_is_one_rejection_and_one_retryable_response() { + let (runtime, generation) = runtime(); + let held = runtime.try_http_connection().unwrap(); + let overload = runtime.try_http_overload_connection().unwrap(); + let (server, mut client) = tcp_pair().await; + let peer = server.peer_addr().unwrap(); + + super::serve( + server, + peer, + WebClientIpSource::XForwardedFor, + trusted_loopback(), + Arc::clone(&runtime), + CancellationToken::new(), + overload, + WebHttpConnectionCapacityAction::Wait, + Duration::from_millis(10), + ) + .await; + drop(held); + + let mut bytes = Vec::new(); + client.read_to_end(&mut bytes).await.unwrap(); + assert_eq!(bytes, super::SERVICE_UNAVAILABLE_RESPONSE); + assert_eq!( + runtime + .telemetry() + .overload_total(WebHttpConnectionOverloadOutcome::WaitTimeout503,), + 1 + ); + assert_eq!( + runtime + .telemetry() + .rejection_total(WebRejectionReason::HttpConnectionCapacity), + 1 + ); + stop(runtime, generation).await; + } + + #[tokio::test] + async fn admitted_wait_is_not_counted_as_a_rejection() { + let (runtime, generation) = runtime(); + let held = runtime.try_http_connection().unwrap(); + let overload = runtime.try_http_overload_connection().unwrap(); + let (server, _client) = tcp_pair().await; + let peer = server.peer_addr().unwrap(); + let cancellation = CancellationToken::new(); + let task = tokio::spawn(super::serve( + server, + peer, + WebClientIpSource::XForwardedFor, + trusted_loopback(), + Arc::clone(&runtime), + cancellation.clone(), + overload, + WebHttpConnectionCapacityAction::Wait, + Duration::from_secs(1), + )); + tokio::task::yield_now().await; + drop(held); + for _ in 0..100 { + if runtime + .telemetry() + .overload_total(WebHttpConnectionOverloadOutcome::WaitAdmitted) + == 1 + { + break; + } + tokio::task::yield_now().await; + } + cancellation.cancel(); + tokio::time::timeout(Duration::from_secs(1), task) + .await + .unwrap() + .unwrap(); + + assert_eq!( + runtime + .telemetry() + .overload_total(WebHttpConnectionOverloadOutcome::WaitAdmitted), + 1 + ); + assert_eq!( + runtime + .telemetry() + .rejection_total(WebRejectionReason::HttpConnectionCapacity), + 0 + ); + stop(runtime, generation).await; + } + + fn runtime() -> ( + Arc, + Arc, + ) { + let mut config = ProxyConfig::default(); + config.web.limits.max_http_connections = 1; + let generation = test_runtime_generation(1, config); + let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation)))); + (runtime, generation) + } + + fn trusted_loopback() -> Arc<[ipnetwork::IpNetwork]> { + Arc::from(["127.0.0.1/32".parse().unwrap()]) + } + + async fn stop( + runtime: Arc, + generation: Arc, + ) { + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; + } +} diff --git a/src/maestro/orchestrator.rs b/src/maestro/orchestrator.rs index 2e7a4cd..a862ae8 100644 --- a/src/maestro/orchestrator.rs +++ b/src/maestro/orchestrator.rs @@ -315,6 +315,7 @@ pub(super) async fn run_telemt_core( &runtime.config, &startup_tracker, active_runtime.clone(), + web_runtime_control.subscribe(), ) .await; diff --git a/src/maestro/runtime_build.rs b/src/maestro/runtime_build.rs index 8593487..14af642 100644 --- a/src/maestro/runtime_build.rs +++ b/src/maestro/runtime_build.rs @@ -45,6 +45,9 @@ pub(crate) async fn prepare_runtime( quota_store: Arc, runtime_log_filter: RuntimeLogFilter, ) -> Result { + config + .validate_web_decoy_listener_separation() + .map_err(|error| error.to_string())?; let started_at_epoch_secs = SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap_or_default() diff --git a/src/maestro/runtime_build_tests.rs b/src/maestro/runtime_build_tests.rs index 3ded7c5..080ccfc 100644 --- a/src/maestro/runtime_build_tests.rs +++ b/src/maestro/runtime_build_tests.rs @@ -230,3 +230,34 @@ fn synlimited_endpoint_move_remains_restart_only() { assert_eq!(resolved.effective.server.listeners[0].port, Some(443)); assert!(!resolved.runtime_changed); } + +#[test] +fn deferred_listener_identity_cannot_create_an_effective_decoy_loop() { + let mut old = ProxyConfig::default(); + old.server.listeners = vec![test_listener(18080)]; + old.server.listeners[0].transport = crate::config::ListenerTransport::Web; + let mut desired = old.clone(); + desired.server.listeners[0].port = Some(18081); + desired.server.listen_backlog = desired.server.listen_backlog.saturating_add(1); + desired.web.vhosts = vec![ + serde_json::from_value(serde_json::json!({ + "host": "proxy.example", + "public_addr": "203.0.113.10:443", + "decoy": { + "mode": "http_upstream", + "upstream": "http://127.0.0.1:18080" + }, + "profiles": [] + })) + .unwrap(), + ]; + + assert!(desired.validate_web_decoy_listener_separation().is_ok()); + let resolved = resolve_reload_config(&old, &desired); + assert!( + resolved + .effective + .validate_web_decoy_listener_separation() + .is_err() + ); +} diff --git a/src/maestro/runtime_tasks.rs b/src/maestro/runtime_tasks.rs index 632da2c..826335d 100644 --- a/src/maestro/runtime_tasks.rs +++ b/src/maestro/runtime_tasks.rs @@ -405,6 +405,7 @@ pub(crate) async fn spawn_metrics_if_configured( config: &Arc, startup_tracker: &Arc, active_runtime: Arc>, + web_runtime_rx: tokio::sync::watch::Receiver, ) { // metrics_listen takes precedence; fall back to metrics_port for backward compat. let metrics_target: Option<(u16, Option)> = @@ -437,7 +438,7 @@ pub(crate) async fn spawn_metrics_if_configured( let active_runtime = active_runtime.clone(); let listen_backlog = config.server.listen_backlog; tokio::spawn(async move { - metrics::serve(port, listen, listen_backlog, active_runtime).await; + metrics::serve(port, listen, listen_backlog, active_runtime, web_runtime_rx).await; }); startup_tracker .complete_component( diff --git a/src/metrics.rs b/src/metrics.rs index 9bd36df..d5e3500 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -26,6 +26,9 @@ use crate::tls_front::cache; use crate::tls_front::fetcher; use crate::transport::{ListenOptions, create_listener}; +// Process-owned WEB metrics stay isolated from the legacy renderer body. +mod web; + // Keeps `/metrics` response size bounded when per-user telemetry is enabled. const USER_LABELED_METRICS_MAX_USERS: usize = 4096; // Keeps TLS-front per-domain health series bounded for large generated configs. @@ -38,6 +41,7 @@ pub async fn serve( listen: Option, listen_backlog: u32, active_runtime: Arc>, + web_runtime_rx: tokio::sync::watch::Receiver, ) { // If `metrics_listen` is set, bind on that single address only. if let Some(ref listen_addr) = listen { @@ -54,7 +58,7 @@ pub async fn serve( match bind_metrics_listener(addr, ipv6_only, listen_backlog) { Ok(listener) => { info!("Metrics endpoint: http://{}/metrics and /beobachten", addr); - serve_listener(listener, active_runtime).await; + serve_listener(listener, active_runtime, web_runtime_rx).await; } Err(e) => { warn!(error = %e, "Failed to bind metrics on {}", addr); @@ -100,14 +104,15 @@ pub async fn serve( warn!("Metrics listener is unavailable on both IPv4 and IPv6"); } (Some(listener), None) | (None, Some(listener)) => { - serve_listener(listener, active_runtime).await; + serve_listener(listener, active_runtime, web_runtime_rx).await; } (Some(listener4), Some(listener6)) => { let active_runtime_v6 = active_runtime.clone(); + let web_runtime_rx_v6 = web_runtime_rx.clone(); tokio::spawn(async move { - serve_listener(listener6, active_runtime_v6).await; + serve_listener(listener6, active_runtime_v6, web_runtime_rx_v6).await; }); - serve_listener(listener4, active_runtime).await; + serve_listener(listener4, active_runtime, web_runtime_rx).await; } } } @@ -127,7 +132,11 @@ fn bind_metrics_listener( TcpListener::from_std(socket.into()) } -async fn serve_listener(listener: TcpListener, active_runtime: Arc>) { +async fn serve_listener( + listener: TcpListener, + active_runtime: Arc>, + web_runtime_rx: tokio::sync::watch::Receiver, +) { let connection_permits = Arc::new(Semaphore::new(METRICS_MAX_CONTROL_CONNECTIONS)); loop { @@ -165,28 +174,13 @@ async fn serve_listener(listener: TcpListener, active_runtime: Arc( req: Request, - stats: &Stats, - beobachten: &BeobachtenStore, - shared_state: &ProxySharedState, - ip_tracker: &UserIpTracker, - tls_cache: Option<&TlsFrontCache>, - config: &ProxyConfig, + runtime: &RuntimeGeneration, + web_publication: &crate::web::control::WebRuntimePublication, ) -> Result>, Infallible> { + let stats = &runtime.stats; + let beobachten = &runtime.beobachten; + let shared_state = &runtime.proxy_shared; + let ip_tracker = &runtime.ip_tracker; + let tls_cache = runtime.tls_cache.as_deref(); + let config = runtime.config(); + if req.uri().path() == "/metrics" { - let body = render_metrics(stats, shared_state, config, ip_tracker, tls_cache).await; + let body = render_metrics( + stats, + shared_state, + &config, + ip_tracker, + tls_cache, + web_publication, + ) + .await; let resp = Response::builder() .status(StatusCode::OK) .header("content-type", "text/plain; version=0.0.4; charset=utf-8") @@ -230,7 +235,7 @@ async fn handle( } if req.uri().path() == "/beobachten" { - let body = render_beobachten(stats, beobachten, config); + let body = render_beobachten(stats, beobachten, &config); let resp = Response::builder() .status(StatusCode::OK) .header("content-type", "text/plain; charset=utf-8") @@ -432,6 +437,7 @@ async fn render_metrics( config: &ProxyConfig, ip_tracker: &UserIpTracker, tls_cache: Option<&TlsFrontCache>, + web_publication: &crate::web::control::WebRuntimePublication, ) -> String { use std::fmt::Write; let mut out = String::with_capacity(4096); @@ -3856,6 +3862,7 @@ async fn render_metrics( unique_ip_suppressed ); + web::render(&mut out, web_publication, config); out } @@ -3870,6 +3877,11 @@ mod tests { CachedTlsData, ParsedServerHello, TlsBehaviorProfile, TlsCertPayload, TlsProfileSource, }; + fn test_web_publication() -> crate::web::control::WebRuntimePublication { + let control = crate::web::control::WebRuntimeControl::new(); + control.subscribe().borrow().clone() + } + #[tokio::test] async fn test_render_metrics_format() { let stats = Arc::new(Stats::new()); @@ -3937,7 +3949,15 @@ mod tests { .await .unwrap(); - let output = render_metrics(&stats, shared_state.as_ref(), &config, &tracker, None).await; + let output = render_metrics( + &stats, + shared_state.as_ref(), + &config, + &tracker, + None, + &test_web_publication(), + ) + .await; assert!(output.contains(&format!( "telemt_build_info{{version=\"{}\"}} 1", @@ -4065,7 +4085,15 @@ mod tests { ) .await; - let output = render_metrics(&stats, &shared_state, &config, &tracker, Some(&cache)).await; + let output = render_metrics( + &stats, + &shared_state, + &config, + &tracker, + Some(&cache), + &test_web_publication(), + ) + .await; assert!(output.contains("telemt_tls_front_profile_domains{status=\"configured\"} 2")); assert!(output.contains("telemt_tls_front_profile_domains{status=\"emitted\"} 2")); @@ -4113,7 +4141,15 @@ mod tests { let shared_state = ProxySharedState::new(); let tracker = UserIpTracker::new(); let config = ProxyConfig::default(); - let output = render_metrics(&stats, &shared_state, &config, &tracker, None).await; + let output = render_metrics( + &stats, + &shared_state, + &config, + &tracker, + None, + &test_web_publication(), + ) + .await; assert!(output.contains("telemt_connections_total 0")); assert!(output.contains("telemt_connections_bad_total 0")); assert!(output.contains("telemt_handshake_timeouts_total 0")); @@ -4137,7 +4173,15 @@ mod tests { let mut config = ProxyConfig::default(); config.access.user_max_unique_ips_global_each = 2; - let output = render_metrics(&stats, &shared_state, &config, &tracker, None).await; + let output = render_metrics( + &stats, + &shared_state, + &config, + &tracker, + None, + &test_web_publication(), + ) + .await; assert!(output.contains("telemt_user_unique_ips_limit{user=\"alice\"} 2")); assert!(output.contains("telemt_user_unique_ips_utilization{user=\"alice\"} 0.500000")); @@ -4149,7 +4193,15 @@ mod tests { let shared_state = ProxySharedState::new(); let tracker = UserIpTracker::new(); let config = ProxyConfig::default(); - let output = render_metrics(&stats, &shared_state, &config, &tracker, None).await; + let output = render_metrics( + &stats, + &shared_state, + &config, + &tracker, + None, + &test_web_publication(), + ) + .await; assert!(output.contains("# TYPE telemt_uptime_seconds gauge")); assert!(output.contains("# TYPE telemt_connections_total counter")); assert!(output.contains("# TYPE telemt_connections_bad_total counter")); @@ -4214,27 +4266,17 @@ mod tests { #[tokio::test] async fn test_endpoint_integration() { - let stats = Arc::new(Stats::new()); - let beobachten = Arc::new(BeobachtenStore::new()); - let shared_state = ProxySharedState::new(); - let tracker = UserIpTracker::new(); let mut config = ProxyConfig::default(); - stats.increment_connects_all(); - stats.increment_connects_all(); - stats.increment_connects_all(); + config.general.beobachten = true; + config.general.beobachten_minutes = 10; + let runtime = crate::maestro::generation::test_runtime_generation(1, config); + let web_publication = test_web_publication(); + runtime.stats.increment_connects_all(); + runtime.stats.increment_connects_all(); + runtime.stats.increment_connects_all(); let req = Request::builder().uri("/metrics").body(()).unwrap(); - let resp = handle( - req, - &stats, - &beobachten, - shared_state.as_ref(), - &tracker, - None, - &config, - ) - .await - .unwrap(); + let resp = handle(req, &runtime, &web_publication).await.unwrap(); assert_eq!(resp.status(), StatusCode::OK); let body = resp.into_body().collect().await.unwrap().to_bytes(); assert!( @@ -4251,25 +4293,13 @@ mod tests { )) ); - config.general.beobachten = true; - config.general.beobachten_minutes = 10; - beobachten.record( + runtime.beobachten.record( "TLS-scanner", "203.0.113.10".parse::().unwrap(), Duration::from_secs(600), ); let req_beob = Request::builder().uri("/beobachten").body(()).unwrap(); - let resp_beob = handle( - req_beob, - &stats, - &beobachten, - shared_state.as_ref(), - &tracker, - None, - &config, - ) - .await - .unwrap(); + let resp_beob = handle(req_beob, &runtime, &web_publication).await.unwrap(); assert_eq!(resp_beob.status(), StatusCode::OK); let body_beob = resp_beob.into_body().collect().await.unwrap().to_bytes(); let beob_text = std::str::from_utf8(body_beob.as_ref()).unwrap(); @@ -4277,17 +4307,7 @@ mod tests { assert!(beob_text.contains("203.0.113.10-1")); let req404 = Request::builder().uri("/other").body(()).unwrap(); - let resp404 = handle( - req404, - &stats, - &beobachten, - shared_state.as_ref(), - &tracker, - None, - &config, - ) - .await - .unwrap(); + let resp404 = handle(req404, &runtime, &web_publication).await.unwrap(); assert_eq!(resp404.status(), StatusCode::NOT_FOUND); } } diff --git a/src/metrics/web.rs b/src/metrics/web.rs new file mode 100644 index 0000000..23f6b5d --- /dev/null +++ b/src/metrics/web.rs @@ -0,0 +1,365 @@ +use std::fmt::Write; + +use crate::config::{ProxyConfig, WebHttpConnectionCapacityAction}; +use crate::web::control::{WebRuntimeLifecycle, WebRuntimePublication}; +use crate::web::manager::OperatorLifecycleState; +use crate::web::telemetry::{ + WebDecoyUpstreamOutcome, WebHttpConnectionOverloadOutcome, WebRejectionReason, +}; + +/// Renders fixed-cardinality process-owned WEB observability families. +pub(super) fn render(out: &mut String, publication: &WebRuntimePublication, config: &ProxyConfig) { + let runtime = publication.runtime.upgrade(); + let configured_listeners = publication.listeners.len(); + let live_acceptors = publication.telemetry.live_acceptors(); + let accepting_connections = publication.lifecycle == WebRuntimeLifecycle::Running + && runtime.is_some() + && configured_listeners != 0 + && live_acceptors == configured_listeners; + + let _ = writeln!( + out, + "# HELP telemt_web_ingress_lifecycle_state Current process-owned WEB ingress lifecycle" + ); + let _ = writeln!(out, "# TYPE telemt_web_ingress_lifecycle_state gauge"); + for state in WebRuntimeLifecycle::ALL { + let _ = writeln!( + out, + "telemt_web_ingress_lifecycle_state{{state=\"{}\"}} {}", + state.as_str(), + flag(publication.lifecycle == state) + ); + } + + let operator_status = runtime + .as_deref() + .map(crate::web::manager::WebProcessRuntime::operator_lifecycle_status); + let _ = writeln!( + out, + "# HELP telemt_web_operator_lifecycle_state Current reversible WEB operator lifecycle" + ); + let _ = writeln!(out, "# TYPE telemt_web_operator_lifecycle_state gauge"); + for state in OPERATOR_STATES { + let active = match operator_status.as_ref() { + Some(status) => operator_state_token(status.state) == state, + None => state == "unavailable", + }; + let _ = writeln!( + out, + "telemt_web_operator_lifecycle_state{{state=\"{state}\"}} {}", + flag(active) + ); + } + + let operator_admission_open = operator_status + .as_ref() + .is_some_and(|status| status.admission_open); + let effective_new_work_admission = operator_status + .as_ref() + .is_some_and(|status| status.effective_new_work_admission); + let _ = writeln!( + out, + "# HELP telemt_web_ingress_state Independent WEB ingress and admission flags" + ); + let _ = writeln!(out, "# TYPE telemt_web_ingress_state gauge"); + for (name, value) in [ + ("runtime_available", runtime.is_some()), + ("accepting_connections", accepting_connections), + ("config_enabled", config.web.enabled), + ("operator_admission_open", operator_admission_open), + ("effective_new_work_admission", effective_new_work_admission), + ] { + let _ = writeln!( + out, + "telemt_web_ingress_state{{flag=\"{name}\"}} {}", + flag(value) + ); + } + + let _ = writeln!( + out, + "# HELP telemt_web_listeners Process-owned WEB listener and acceptor counts" + ); + let _ = writeln!(out, "# TYPE telemt_web_listeners gauge"); + let _ = writeln!( + out, + "telemt_web_listeners{{status=\"configured\"}} {configured_listeners}" + ); + let _ = writeln!( + out, + "telemt_web_listeners{{status=\"acceptors_live\"}} {live_acceptors}" + ); + + let _ = writeln!( + out, + "# HELP telemt_web_tcp_accept_total Accepted sockets and accept syscall errors" + ); + let _ = writeln!(out, "# TYPE telemt_web_tcp_accept_total counter"); + let _ = writeln!( + out, + "telemt_web_tcp_accept_total{{result=\"accepted\"}} {}", + publication.telemetry.accepted() + ); + let _ = writeln!( + out, + "telemt_web_tcp_accept_total{{result=\"error\"}} {}", + publication.telemetry.accept_errors() + ); + + let _ = writeln!( + out, + "# HELP telemt_web_rejections_total WEB operational rejection decisions by fixed reason" + ); + let _ = writeln!(out, "# TYPE telemt_web_rejections_total counter"); + for reason in WebRejectionReason::ALL { + let _ = writeln!( + out, + "telemt_web_rejections_total{{reason=\"{}\"}} {}", + reason.as_str(), + publication.telemetry.rejection_total(reason) + ); + } + + let _ = writeln!( + out, + "# HELP telemt_web_http_connection_overload_total Accepted saturated sockets by terminal outcome" + ); + let _ = writeln!( + out, + "# TYPE telemt_web_http_connection_overload_total counter" + ); + for outcome in WebHttpConnectionOverloadOutcome::ALL { + let _ = writeln!( + out, + "telemt_web_http_connection_overload_total{{outcome=\"{}\"}} {}", + outcome.as_str(), + publication.telemetry.overload_total(outcome) + ); + } + + let action = config.web.http_connection_capacity_action; + let _ = writeln!( + out, + "# HELP telemt_web_http_connection_capacity_action Effective overload action" + ); + let _ = writeln!( + out, + "# TYPE telemt_web_http_connection_capacity_action gauge" + ); + for (token, variant) in [ + ("drop", WebHttpConnectionCapacityAction::Drop), + ("wait", WebHttpConnectionCapacityAction::Wait), + ("respond", WebHttpConnectionCapacityAction::Respond), + ] { + let _ = writeln!( + out, + "telemt_web_http_connection_capacity_action{{action=\"{token}\"}} {}", + flag(action == variant) + ); + } + let _ = writeln!( + out, + "# HELP telemt_web_http_overload_timeout_milliseconds Effective overload phase timeout" + ); + let _ = writeln!( + out, + "# TYPE telemt_web_http_overload_timeout_milliseconds gauge" + ); + let _ = writeln!( + out, + "telemt_web_http_overload_timeout_milliseconds {}", + config.web.timeouts.http_overload_timeout_ms + ); + + if let Some(runtime) = runtime.as_deref() { + render_capacity(out, &runtime.capacity_snapshot()); + } + + let _ = writeln!( + out, + "# HELP telemt_web_decoy_upstream_requests_total Internal plain-HTTP decoy origin outcomes" + ); + let _ = writeln!( + out, + "# TYPE telemt_web_decoy_upstream_requests_total counter" + ); + for outcome in WebDecoyUpstreamOutcome::ALL { + let _ = writeln!( + out, + "telemt_web_decoy_upstream_requests_total{{outcome=\"{}\"}} {}", + outcome.as_str(), + publication.telemetry.decoy_total(outcome) + ); + } + + render_aggregate_totals(out, publication); +} + +fn render_capacity(out: &mut String, snapshot: &crate::web::manager::WebCapacitySnapshot) { + let _ = writeln!( + out, + "# HELP telemt_web_capacity_snapshot_partial Whether a non-blocking capacity plane was omitted" + ); + let _ = writeln!(out, "# TYPE telemt_web_capacity_snapshot_partial gauge"); + let _ = writeln!( + out, + "telemt_web_capacity_snapshot_partial{{plane=\"budget\"}} {}", + flag(snapshot.partial.contains(&"budget")) + ); + for (unit, family) in [ + ("slots", "telemt_web_capacity_slots"), + ("bytes", "telemt_web_capacity_bytes"), + ("items", "telemt_web_capacity_items"), + ] { + let _ = writeln!( + out, + "# HELP {family} Current process-wide WEB capacity in {unit}" + ); + let _ = writeln!(out, "# TYPE {family} gauge"); + for status in snapshot + .resources + .iter() + .filter(|status| status.unit == unit) + { + for (kind, value) in [ + ("used", status.used), + ("available", status.available), + ("limit", status.limit), + ] { + let _ = writeln!( + out, + "{family}{{resource=\"{}\",kind=\"{kind}\"}} {value}", + status.resource + ); + } + } + } + let _ = writeln!( + out, + "# HELP telemt_web_capacity_closed Whether terminal shutdown closed a WEB capacity authority" + ); + let _ = writeln!(out, "# TYPE telemt_web_capacity_closed gauge"); + let _ = writeln!( + out, + "# HELP telemt_web_capacity_saturated Whether a WEB resource has no immediately available capacity" + ); + let _ = writeln!(out, "# TYPE telemt_web_capacity_saturated gauge"); + for status in &snapshot.resources { + let _ = writeln!( + out, + "telemt_web_capacity_closed{{resource=\"{}\"}} {}", + status.resource, + flag(status.closed) + ); + let _ = writeln!( + out, + "telemt_web_capacity_saturated{{resource=\"{}\"}} {}", + status.resource, + flag(status.available == 0 && !status.closed) + ); + } +} + +fn render_aggregate_totals(out: &mut String, publication: &WebRuntimePublication) { + let totals = publication.telemetry.aggregates(); + let _ = writeln!( + out, + "# HELP telemt_web_session_incarnations_total Process-owned WEB session lifecycle totals" + ); + let _ = writeln!(out, "# TYPE telemt_web_session_incarnations_total counter"); + let _ = writeln!( + out, + "telemt_web_session_incarnations_total{{event=\"created\"}} {}", + totals.sessions_created + ); + let _ = writeln!( + out, + "telemt_web_session_incarnations_total{{event=\"closed\"}} {}", + totals.sessions_closed + ); + let _ = writeln!( + out, + "# HELP telemt_web_streams_total Process-owned WEB logical stream totals" + ); + let _ = writeln!(out, "# TYPE telemt_web_streams_total counter"); + let _ = writeln!( + out, + "telemt_web_streams_total{{event=\"opened\"}} {}", + totals.streams_opened + ); + let _ = writeln!( + out, + "telemt_web_streams_total{{event=\"rejected\"}} {}", + totals.streams_rejected + ); + let _ = writeln!( + out, + "# HELP telemt_web_carrier_bytes_total Process-owned WEB carrier payload bytes" + ); + let _ = writeln!(out, "# TYPE telemt_web_carrier_bytes_total counter"); + let _ = writeln!( + out, + "telemt_web_carrier_bytes_total{{direction=\"up\"}} {}", + totals.bytes_up + ); + let _ = writeln!( + out, + "telemt_web_carrier_bytes_total{{direction=\"down\"}} {}", + totals.bytes_down + ); +} + +fn operator_state_token(state: OperatorLifecycleState) -> &'static str { + match state { + OperatorLifecycleState::Running => "running", + OperatorLifecycleState::Paused => "paused", + OperatorLifecycleState::Draining => "draining", + OperatorLifecycleState::ForceClosing => "force_closing", + OperatorLifecycleState::Drained => "drained", + } +} + +const OPERATOR_STATES: [&str; 6] = [ + "unavailable", + "running", + "paused", + "draining", + "force_closing", + "drained", +]; + +const fn flag(value: bool) -> u8 { + if value { 1 } else { 0 } +} + +#[cfg(test)] +mod tests { + use crate::config::ProxyConfig; + use crate::web::control::WebRuntimeControl; + + #[test] + fn renderer_emits_complete_zeroed_fixed_counter_sets() { + let control = WebRuntimeControl::new(); + let publication = control.subscribe().borrow().clone(); + let mut output = String::new(); + super::render(&mut output, &publication, &ProxyConfig::default()); + + assert_eq!( + output.matches("telemt_web_rejections_total{").count(), + crate::web::telemetry::WebRejectionReason::ALL.len() + ); + assert_eq!( + output + .matches("telemt_web_http_connection_overload_total{") + .count(), + crate::web::telemetry::WebHttpConnectionOverloadOutcome::ALL.len() + ); + assert_eq!( + output + .matches("telemt_web_decoy_upstream_requests_total{") + .count(), + crate::web::telemetry::WebDecoyUpstreamOutcome::ALL.len() + ); + assert!(output.contains("telemt_web_ingress_lifecycle_state{state=\"starting\"} 1")); + } +} diff --git a/src/web/control.rs b/src/web/control.rs index fa7b9cf..f0ad1e2 100644 --- a/src/web/control.rs +++ b/src/web/control.rs @@ -7,6 +7,7 @@ use std::time::Instant; use tokio::sync::watch; use super::manager::WebProcessRuntime; +use super::telemetry::WebTelemetry; /// Process-owned WEB ingress lifecycle state. #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -26,6 +27,16 @@ pub(crate) enum WebRuntimeLifecycle { } impl WebRuntimeLifecycle { + /// Complete fixed lifecycle set used by one-hot metrics. + pub(crate) const ALL: [Self; 6] = [ + Self::Starting, + Self::NoWebListener, + Self::Running, + Self::Draining, + Self::Drained, + Self::DeadlineExceeded, + ]; + /// Returns the stable API token for this lifecycle state. pub(crate) const fn as_str(self) -> &'static str { match self { @@ -52,28 +63,34 @@ pub(crate) struct WebRuntimePublication { pub(crate) listeners: Arc<[SocketAddr]>, /// Weak runtime access that never extends data-plane ownership. pub(crate) runtime: Weak, + /// Process-owned counters that remain readable after runtime release. + pub(crate) telemetry: Arc, } /// Single-writer process lifecycle publisher for WEB ingress. #[derive(Clone)] pub(crate) struct WebRuntimeControl { epoch: Arc, + telemetry: Arc, tx: watch::Sender, } impl WebRuntimeControl { /// Creates the process channel in the pre-listener `starting` state. pub(crate) fn new() -> Self { + let telemetry = WebTelemetry::new(); let publication = WebRuntimePublication { epoch: 1, lifecycle: WebRuntimeLifecycle::Starting, since: Instant::now(), listeners: Arc::from([]), runtime: Weak::new(), + telemetry: Arc::clone(&telemetry), }; let (tx, _rx) = watch::channel(publication); Self { epoch: Arc::new(AtomicU64::new(1)), + telemetry, tx, } } @@ -83,6 +100,11 @@ impl WebRuntimeControl { self.tx.subscribe() } + /// Returns the shared process-owned WEB telemetry handle. + pub(crate) fn telemetry(&self) -> Arc { + Arc::clone(&self.telemetry) + } + /// Publishes one lifecycle transition and optional weak runtime reference. pub(crate) fn publish( &self, @@ -97,6 +119,7 @@ impl WebRuntimeControl { since: Instant::now(), listeners, runtime, + telemetry: Arc::clone(&self.telemetry), }); } } diff --git a/src/web/http/decoy.rs b/src/web/http/decoy.rs index f7652d4..32873de 100644 --- a/src/web/http/decoy.rs +++ b/src/web/http/decoy.rs @@ -15,6 +15,7 @@ use super::{ }; use crate::config::{WebRuntimeDecoy, WebRuntimeVhost}; use crate::web::manager::WebProcessRuntime; +use crate::web::telemetry::WebDecoyUpstreamOutcome; /// Serves the configured ordinary site after optionally removing carrier material. /// Transport-sanitized static fallbacks remain uncacheable after query removal. @@ -195,16 +196,18 @@ async fn proxy_to_upstream( .map(|value| value.as_str()) .unwrap_or("/"); let Ok(uri) = path_and_query.parse::() else { - return bad_gateway(); + return decoy_failure(runtime, WebDecoyUpstreamOutcome::RequestError); }; *request.uri_mut() = uri; let _deadline_lease = match lease_deadline(request_deadline.as_ref(), header_timeout) { Ok(lease) => lease, - Err(()) => return bad_gateway(), + Err(()) => { + return decoy_failure(runtime, WebDecoyUpstreamOutcome::DeadlineExhausted); + } }; - let stream = match tokio::time::timeout(header_timeout, TcpStream::connect(addr)).await { - Ok(Ok(stream)) => stream, - _ => return bad_gateway(), + let stream = match connect_upstream(addr, header_timeout).await { + Ok(stream) => stream, + Err(outcome) => return decoy_failure(runtime, outcome), }; drop(_deadline_lease); let max_header_bytes = runtime @@ -217,12 +220,19 @@ async fn proxy_to_upstream( builder.max_buf_size(max_header_bytes); let _deadline_lease = match lease_deadline(request_deadline.as_ref(), header_timeout) { Ok(lease) => lease, - Err(()) => return bad_gateway(), + Err(()) => { + return decoy_failure(runtime, WebDecoyUpstreamOutcome::DeadlineExhausted); + } }; let (mut sender, connection) = match tokio::time::timeout(header_timeout, builder.handshake(TokioIo::new(stream))).await { Ok(Ok(parts)) => parts, - _ => return bad_gateway(), + Ok(Err(_)) => { + return decoy_failure(runtime, WebDecoyUpstreamOutcome::HttpHandshakeError); + } + Err(_) => { + return decoy_failure(runtime, WebDecoyUpstreamOutcome::HttpHandshakeTimeout); + } }; drop(_deadline_lease); runtime.spawn_auxiliary(async move { @@ -230,14 +240,24 @@ async fn proxy_to_upstream( }); let _deadline_lease = match lease_deadline(request_deadline.as_ref(), header_timeout) { Ok(lease) => lease, - Err(()) => return bad_gateway(), + Err(()) => { + return decoy_failure(runtime, WebDecoyUpstreamOutcome::DeadlineExhausted); + } }; let mut response = match tokio::time::timeout(header_timeout, sender.send_request(request)).await { Ok(Ok(response)) => response, - _ => return bad_gateway(), + Ok(Err(_)) => { + return decoy_failure(runtime, WebDecoyUpstreamOutcome::RequestError); + } + Err(_) => { + return decoy_failure(runtime, WebDecoyUpstreamOutcome::ResponseHeadTimeout); + } }; drop(_deadline_lease); + runtime + .telemetry() + .record_decoy(WebDecoyUpstreamOutcome::Success); remove_hop_by_hop(response.headers_mut()); response.map(|body| { body.map_err(|error| -> BoxError { Box::new(error) }) @@ -245,6 +265,25 @@ async fn proxy_to_upstream( }) } +async fn connect_upstream( + addr: SocketAddr, + timeout: Duration, +) -> Result { + match tokio::time::timeout(timeout, TcpStream::connect(addr)).await { + Ok(Ok(stream)) => Ok(stream), + Ok(Err(error)) if error.kind() == std::io::ErrorKind::ConnectionRefused => { + Err(WebDecoyUpstreamOutcome::ConnectRefused) + } + Ok(Err(_)) => Err(WebDecoyUpstreamOutcome::ConnectError), + Err(_) => Err(WebDecoyUpstreamOutcome::ConnectTimeout), + } +} + +fn decoy_failure(runtime: &WebProcessRuntime, outcome: WebDecoyUpstreamOutcome) -> HttpResponse { + runtime.telemetry().record_decoy(outcome); + bad_gateway() +} + fn lease_deadline( deadline: Option<&super::activity::RequestDeadlineHandle>, timeout: Duration, @@ -315,4 +354,16 @@ mod tests { assert!(resolve_static_path("/../index.html", &site).is_none()); assert!(resolve_static_path("//index.html", &site).is_none()); } + + #[tokio::test] + async fn closed_loopback_origin_is_classified_as_connect_refused() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + drop(listener); + + assert_eq!( + connect_upstream(addr, Duration::from_secs(1)).await.err(), + Some(WebDecoyUpstreamOutcome::ConnectRefused) + ); + } } diff --git a/src/web/http/operator_lifecycle_tests.rs b/src/web/http/operator_lifecycle_tests.rs index 9cd3275..e579d59 100644 --- a/src/web/http/operator_lifecycle_tests.rs +++ b/src/web/http/operator_lifecycle_tests.rs @@ -46,10 +46,7 @@ async fn live_runtime() -> ( Arc, TcpListener, ) { - let generation = test_runtime_generation( - 1, - runtime_config([71; 32], WebCarrier::Https), - ); + let generation = test_runtime_generation(1, runtime_config([71; 32], WebCarrier::Https)); let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation)))); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); (runtime, generation, listener) @@ -100,12 +97,7 @@ async fn pause_preserves_decoy_retry_and_exact_session_replay() { let hello = frame::encode(FrameType::Hello, 0, &[1]); runtime.pause_operator().await.unwrap(); - let paused_create = request( - &listener, - &runtime, - create_request(&bootstrap, &hello), - ) - .await; + let paused_create = request(&listener, &runtime, create_request(&bootstrap, &hello)).await; let (paused_headers, _) = split_response(&paused_create); assert!(paused_headers.starts_with(b"HTTP/1.1 503")); assert_eq!(response_header(paused_headers, "retry-after"), "1"); @@ -131,26 +123,20 @@ async fn pause_preserves_decoy_retry_and_exact_session_replay() { let decoy = request(&listener, &runtime, decoy).await; let (decoy_headers, decoy_body) = split_response(&decoy); assert!(decoy_headers.starts_with(b"HTTP/1.1 200")); - assert!(!decoy_body.windows(11).any(|window| window == b"bootstrap=\"")); + assert!( + !decoy_body + .windows(11) + .any(|window| window == b"bootstrap=\"") + ); runtime.resume_operator().await.unwrap(); - let created = request( - &listener, - &runtime, - create_request(&bootstrap, &hello), - ) - .await; + let created = request(&listener, &runtime, create_request(&bootstrap, &hello)).await; let (created_headers, _) = split_response(&created); assert!(created_headers.starts_with(b"HTTP/1.1 200")); let token = response_header(created_headers, "x-session-token").to_string(); runtime.pause_operator().await.unwrap(); - let replay = request( - &listener, - &runtime, - create_request(&bootstrap, &hello), - ) - .await; + let replay = request(&listener, &runtime, create_request(&bootstrap, &hello)).await; let (replay_headers, _) = split_response(&replay); assert!(replay_headers.starts_with(b"HTTP/1.1 200")); assert_eq!(response_header(replay_headers, "x-session-token"), token); @@ -230,7 +216,10 @@ async fn paused_replacement_preserves_old_session_and_attempt_replay() { let replay = request(&listener, &runtime, first_request).await; let (replay_headers, _) = split_response(&replay); assert!(replay_headers.starts_with(b"HTTP/1.1 200")); - assert_eq!(response_header(replay_headers, "x-session-token"), first_token); + assert_eq!( + response_header(replay_headers, "x-session-token"), + first_token + ); runtime.resume_operator().await.unwrap(); let replacement = request(&listener, &runtime, replacement_request).await; @@ -249,7 +238,10 @@ async fn client_delete_during_drain_completes_naturally() { let (runtime, generation, listener) = live_runtime().await; let bootstrap = issue_bootstrap(&runtime); let (_, token) = create_session(&listener, &runtime, &bootstrap).await; - runtime.drain_operator(Duration::from_secs(30)).await.unwrap(); + runtime + .drain_operator(Duration::from_secs(30)) + .await + .unwrap(); let delete = format!( "DELETE /api/v1/session HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.10\r\nAuthorization: Bearer {token}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" @@ -281,8 +273,14 @@ async fn deadline_force_closes_all_sessions_and_requires_explicit_resume() { let bootstrap = issue_bootstrap(&runtime); let (_, token) = create_session(&listener, &runtime, &bootstrap).await; - let accepted = runtime.drain_operator(Duration::from_secs(30)).await.unwrap(); - assert_eq!(serde_json::to_value(&accepted).unwrap()["state"], "draining"); + let accepted = runtime + .drain_operator(Duration::from_secs(30)) + .await + .unwrap(); + assert_eq!( + serde_json::to_value(&accepted).unwrap()["state"], + "draining" + ); assert!(matches!( runtime.drain_operator(Duration::from_secs(30)).await, Err(OperatorLifecycleError::OperationInProgress) @@ -323,7 +321,10 @@ async fn resume_before_deadline_cancels_drain_without_closing_existing_session() let bootstrap = issue_bootstrap(&runtime); let (_, token) = create_session(&listener, &runtime, &bootstrap).await; - runtime.drain_operator(Duration::from_secs(30)).await.unwrap(); + runtime + .drain_operator(Duration::from_secs(30)) + .await + .unwrap(); let still_draining = runtime.pause_operator().await.unwrap(); assert_eq!( serde_json::to_value(&still_draining).unwrap()["state"], diff --git a/src/web/http/session.rs b/src/web/http/session.rs index 1834d0a..a054bae 100644 --- a/src/web/http/session.rs +++ b/src/web/http/session.rs @@ -190,9 +190,9 @@ pub(super) async fn handle_session( } Err( error @ (ManagerError::Limit - | ManagerError::Backpressure - | ManagerError::Concurrent - | ManagerError::AdmissionPaused), + | ManagerError::Backpressure + | ManagerError::Concurrent + | ManagerError::AdmissionPaused), ) => { runtime.trace().record_profile_lifecycle( client_ip, diff --git a/src/web/http/websocket/driver/io.rs b/src/web/http/websocket/driver/io.rs index 5d3a350..220bed5 100644 --- a/src/web/http/websocket/driver/io.rs +++ b/src/web/http/websocket/driver/io.rs @@ -46,7 +46,7 @@ pub(super) async fn reserve_data( cancellation: &CancellationToken, timeout: Duration, ) -> Result { - tokio::time::timeout(timeout, async { + match tokio::time::timeout(timeout, async { loop { if cancellation.is_cancelled() { return Err(()); @@ -63,7 +63,15 @@ pub(super) async fn reserve_data( } }) .await - .map_err(|_| ())? + { + Ok(result) => result, + Err(_) => { + runtime.telemetry().record_rejection( + crate::web::telemetry::WebRejectionReason::WebSocketBytesCapacity, + ); + Err(()) + } + } } pub(super) async fn process_multiplex( diff --git a/src/web/manager.rs b/src/web/manager.rs index e66b189..028813f 100644 --- a/src/web/manager.rs +++ b/src/web/manager.rs @@ -1,7 +1,7 @@ use std::future::Future; use std::net::IpAddr; use std::sync::Arc; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::AtomicU64; use std::time::Duration; use arc_swap::ArcSwap; @@ -12,6 +12,7 @@ use tokio_util::task::TaskTracker; use crate::config::{WebCarrier, WebLimitsConfig}; use crate::maestro::generation::RuntimeGeneration; +use crate::web::telemetry::{WebRejectionReason, WebTelemetry}; use crate::web::trace::WebTraceStore; // Credential maps, quotas, and token-bucket helpers remain private to the manager. @@ -37,7 +38,7 @@ pub(crate) use lifecycle::WebShutdownOutcome; // Reversible operator admission stays independent from terminal process shutdown. mod operator_lifecycle; pub(crate) use operator_lifecycle::{ - OperatorLifecycleError, OperatorLifecycleStatus, + OperatorLifecycleError, OperatorLifecycleState, OperatorLifecycleStatus, }; // Queue and WebSocket allocations share one process-owned data-plane budget. mod budget; @@ -48,6 +49,9 @@ mod status; pub(crate) use status::{ SessionDetail, SessionFilter, SessionListRequest, SessionRefError, WebRuntimeStatus, }; +// Fixed-cardinality capacity snapshots serve API and Prometheus observability. +mod observability; +pub(crate) use observability::{WebCapacityResourceStatus, WebCapacitySnapshot}; // Asynchronous bounded close operations isolate mutation lifecycle from HTTP requests. mod control; pub(crate) use budget::WebSocketBudgetLease; @@ -152,6 +156,7 @@ pub(crate) struct WebProcessRuntime { stream_admission: Mutex, learning: Mutex, http_connections: Arc, + http_overload_connections: Arc, http_handlers: Arc, lane_polls: Arc, lane_aux_polls: Arc, @@ -169,13 +174,7 @@ pub(crate) struct WebProcessRuntime { next_control_operation_id: AtomicU64, shutdown: CancellationToken, tasks: TaskTracker, - sessions_created: AtomicU64, - sessions_closed: AtomicU64, - streams_opened: AtomicU64, - streams_rejected: AtomicU64, - bytes_up: AtomicU64, - bytes_down: AtomicU64, - limit_hits: AtomicU64, + telemetry: Arc, } impl WebProcessRuntime { @@ -184,13 +183,14 @@ impl WebProcessRuntime { pub(crate) fn start(active_runtime: Arc>) -> Arc { let config = active_runtime.load().config(); let trace = WebTraceStore::new(config.web.debug.clone(), &config.web.limits); - Self::start_with_trace(active_runtime, trace) + Self::start_with_trace(active_runtime, trace, WebTelemetry::new()) } /// Starts one process-scoped manager with a shared API-visible trace store. pub(crate) fn start_with_trace( active_runtime: Arc>, trace: Arc, + telemetry: Arc, ) -> Arc { let initial_generation = active_runtime.load_full(); let config = initial_generation.config(); @@ -209,8 +209,7 @@ impl WebProcessRuntime { .saturating_sub(limits.websocket_http_connection_reserve); let lane_poll_limit = limits.max_http_handlers / 2; let lane_aux_poll_limit = (lane_poll_limit / 2).max(1); - let runtime_instance: Arc = - Arc::from(format!("{:032x}", rand::random::())); + let runtime_instance: Arc = Arc::from(format!("{:032x}", rand::random::())); let runtime = Arc::new(Self { operator_lifecycle: operator_lifecycle::OperatorLifecycle::new(Arc::clone( &runtime_instance, @@ -219,6 +218,9 @@ impl WebProcessRuntime { active_runtime, trace, http_connections: Arc::new(Semaphore::new(limits.max_http_connections)), + http_overload_connections: Arc::new(Semaphore::new( + limits.max_http_overload_connections, + )), http_handlers: Arc::new(Semaphore::new(limits.max_http_handlers)), lane_polls: Arc::new(Semaphore::new(lane_poll_limit)), lane_aux_polls: Arc::new(Semaphore::new(lane_aux_poll_limit)), @@ -239,13 +241,7 @@ impl WebProcessRuntime { learning: Mutex::new(carrier_learning), shutdown: CancellationToken::new(), tasks: TaskTracker::new(), - sessions_created: AtomicU64::new(0), - sessions_closed: AtomicU64::new(0), - streams_opened: AtomicU64::new(0), - streams_rejected: AtomicU64::new(0), - bytes_up: AtomicU64::new(0), - bytes_down: AtomicU64::new(0), - limit_hits: AtomicU64::new(0), + telemetry, }); let weak = Arc::downgrade(&runtime); let shutdown = runtime.shutdown.clone(); @@ -287,6 +283,16 @@ impl WebProcessRuntime { &self.trace } + /// Returns the process-owned operational telemetry handle. + pub(crate) fn telemetry(&self) -> &Arc { + &self.telemetry + } + + /// Returns whether terminal process shutdown has started. + pub(crate) fn is_shutdown(&self) -> bool { + self.shutdown.is_cancelled() + } + /// Reserves one accepted HTTP connection. pub(crate) fn try_http_connection(&self) -> Option { let permit = Arc::clone(&self.http_connections).try_acquire_owned().ok(); @@ -296,11 +302,28 @@ impl WebProcessRuntime { permit } + /// Waits for one accepted HTTP connection slot after bounded overload admission. + pub(crate) async fn acquire_http_connection(&self) -> Option { + Arc::clone(&self.http_connections) + .acquire_owned() + .await + .ok() + } + + /// Reserves one accepted socket outside ordinary HTTP connection capacity. + pub(crate) fn try_http_overload_connection(&self) -> Option { + Arc::clone(&self.http_overload_connections) + .try_acquire_owned() + .ok() + } + /// Reserves one concurrently executing HTTP request handler. pub(crate) fn try_http_handler(&self) -> Option { let permit = Arc::clone(&self.http_handlers).try_acquire_owned().ok(); if permit.is_none() { self.record_limit_hit(); + self.telemetry + .record_rejection(WebRejectionReason::HttpHandlerCapacity); } permit } @@ -315,6 +338,11 @@ impl WebProcessRuntime { let permit = Arc::clone(slots).try_acquire_owned().ok(); if permit.is_none() { self.record_limit_hit(); + self.telemetry.record_rejection(if auxiliary { + WebRejectionReason::LaneAuxPollCapacity + } else { + WebRejectionReason::LanePollCapacity + }); } permit } @@ -323,7 +351,7 @@ impl WebProcessRuntime { pub(crate) fn try_stream_handshake(&self) -> Option { let permit = Arc::clone(&self.stream_handshakes).try_acquire_owned().ok(); if permit.is_none() { - self.record_stream_rejected(); + self.record_stream_rejected_reason(WebRejectionReason::StreamHandshakeCapacity); } permit } @@ -359,10 +387,14 @@ impl WebProcessRuntime { ) -> Option<(OwnedSemaphorePermit, OwnedSemaphorePermit)> { let Some(bytes) = u32::try_from(bytes).ok() else { self.record_limit_hit(); + self.telemetry + .record_rejection(WebRejectionReason::BodyBytesCapacity); return None; }; let Some(reader) = Arc::clone(&self.body_readers).try_acquire_owned().ok() else { self.record_limit_hit(); + self.telemetry + .record_rejection(WebRejectionReason::BodyReaderCapacity); return None; }; let Some(body) = Arc::clone(&self.body_bytes) @@ -370,6 +402,8 @@ impl WebProcessRuntime { .ok() else { self.record_limit_hit(); + self.telemetry + .record_rejection(WebRejectionReason::BodyBytesCapacity); return None; }; Some((reader, body)) @@ -383,6 +417,8 @@ impl WebProcessRuntime { .ok(); if permit.is_none() { self.record_limit_hit(); + self.telemetry + .record_rejection(WebRejectionReason::BodyBytesCapacity); } permit } @@ -401,6 +437,8 @@ impl WebProcessRuntime { .try_reserve_queue(owner, bytes, items, control, downlink) { self.record_limit_hit(); + self.telemetry + .record_rejection(WebRejectionReason::QueueGlobalCapacity); return false; } true @@ -473,15 +511,15 @@ impl WebProcessRuntime { /// Accounts one successfully committed carrier uplink body. pub(crate) fn record_up(&self, bytes: usize) { - self.bytes_up.fetch_add(bytes as u64, Ordering::Relaxed); + self.telemetry.record_up(bytes); } /// Accounts one emitted carrier downlink body. pub(crate) fn record_down(&self, bytes: usize) { - self.bytes_down.fetch_add(bytes as u64, Ordering::Relaxed); + self.telemetry.record_down(bytes); } fn record_limit_hit(&self) { - self.limit_hits.fetch_add(1, Ordering::Relaxed); + self.telemetry.record_limit_hit(); } } diff --git a/src/web/manager/admission.rs b/src/web/manager/admission.rs index 18ae6a6..a8a7190 100644 --- a/src/web/manager/admission.rs +++ b/src/web/manager/admission.rs @@ -1,9 +1,9 @@ use std::net::{IpAddr, SocketAddr}; -use std::sync::atomic::Ordering; use std::time::Instant; use super::state::{allocate_stream_port, allow_rate, decrement_map, release_stream_port}; use super::{ProfileKey, WebProcessRuntime}; +use crate::web::telemetry::WebRejectionReason; impl WebProcessRuntime { /// Reserves one process-wide and per-profile live logical-stream slot. @@ -17,14 +17,16 @@ impl WebProcessRuntime { let _operator_admission = match self.try_operator_admission() { Ok(admission) => admission, Err(error) => { - self.streams_rejected.fetch_add(1, Ordering::Relaxed); + self.telemetry.record_stream_rejected(); return Err(error); } }; let now = Instant::now(); let mut state = self.stream_admission.lock(); if state.closed { - self.streams_rejected.fetch_add(1, Ordering::Relaxed); + self.telemetry.record_stream_rejected(); + self.telemetry + .record_rejection(WebRejectionReason::RuntimeClosed); return Err(super::ManagerError::Closed); } if state.streams_live >= self.limits.max_streams_global @@ -34,25 +36,26 @@ impl WebProcessRuntime { .copied() .unwrap_or(0) >= max_streams - || !allow_rate( - &mut state.stream_rate, - now, - self.limits.new_streams_per_minute, - self.limits.new_streams_burst, - ) { - self.streams_rejected.fetch_add(1, Ordering::Relaxed); - self.limit_hits.fetch_add(1, Ordering::Relaxed); + self.record_stream_rejected_reason(WebRejectionReason::StreamCapacity); + return Err(super::ManagerError::Limit); + } + if !allow_rate( + &mut state.stream_rate, + now, + self.limits.new_streams_per_minute, + self.limits.new_streams_burst, + ) { + self.record_stream_rejected_reason(WebRejectionReason::StreamRate); return Err(super::ManagerError::Limit); } let Some(peer_port) = allocate_stream_port(&mut state, client_ip, public_addr) else { - self.streams_rejected.fetch_add(1, Ordering::Relaxed); - self.limit_hits.fetch_add(1, Ordering::Relaxed); + self.record_stream_rejected_reason(WebRejectionReason::StreamTupleExhausted); return Err(super::ManagerError::Limit); }; state.streams_live += 1; *state.streams_per_profile.entry(profile_key).or_insert(0) += 1; - self.streams_opened.fetch_add(1, Ordering::Relaxed); + self.telemetry.record_stream_opened(); Ok(peer_port) } @@ -76,9 +79,15 @@ impl WebProcessRuntime { /// Records a logical stream rejected outside manager quota acquisition. pub(crate) fn record_stream_rejected(&self) { - self.streams_rejected.fetch_add(1, Ordering::Relaxed); + self.telemetry.record_stream_rejected(); self.record_limit_hit(); } + + /// Records one logical stream rejection with its operational cause. + pub(crate) fn record_stream_rejected_reason(&self, reason: WebRejectionReason) { + self.record_stream_rejected(); + self.telemetry.record_rejection(reason); + } } #[cfg(test)] diff --git a/src/web/manager/credentials.rs b/src/web/manager/credentials.rs index 5e081df..60e84fb 100644 --- a/src/web/manager/credentials.rs +++ b/src/web/manager/credentials.rs @@ -1,6 +1,5 @@ use std::net::IpAddr; use std::sync::Arc; -use std::sync::atomic::Ordering; use std::time::{Duration, Instant}; use sha2::{Digest, Sha256}; @@ -14,6 +13,7 @@ use super::{BootstrapResult, ManagerError, TOKEN_BYTES, TokenHash, WebProcessRun use crate::config::WebRuntimeProfile; use crate::maestro::generation::RuntimeGeneration; use crate::web::session::WebSession; +use crate::web::telemetry::WebRejectionReason; impl WebProcessRuntime { /// Issues a one-use bootstrap credential for an active compatible profile. @@ -63,7 +63,14 @@ impl WebProcessRuntime { .as_ref() .and_then(|runtime| matching_profile(runtime, &profile)) .ok_or(ManagerError::Authentication)?; - if !config.web.enabled || !generation.proxy_shared.is_user_enabled(&profile.user) { + if !config.web.enabled { + self.telemetry + .record_rejection(WebRejectionReason::ConfigDisabled); + return Err(ManagerError::Closed); + } + if !generation.proxy_shared.is_user_enabled(&profile.user) { + self.telemetry + .record_rejection(WebRejectionReason::UserDisabled); return Err(ManagerError::Closed); } let _operator_admission = self.try_operator_admission()?; @@ -71,32 +78,47 @@ impl WebProcessRuntime { let mut state = self.state.lock(); remove_expired_locked(&mut state, now); state.apply_issuance_policy(generation.id, config.web.enabled); - if state.closed - || !state.issuance_enabled - || state - .bootstraps_per_ip - .get(&client_ip) - .copied() - .unwrap_or(0) - >= self.limits.max_bootstraps_per_ip - || !allow_rate( - &mut state.bootstrap_rate, - now, - self.limits.new_bootstraps_per_minute, - self.limits.new_bootstraps_burst, - ) + if state.closed || !state.issuance_enabled { + self.record_limit_hit(); + self.telemetry + .record_rejection(WebRejectionReason::RuntimeClosed); + return Err(ManagerError::Limit); + } + if state + .bootstraps_per_ip + .get(&client_ip) + .copied() + .unwrap_or(0) + >= self.limits.max_bootstraps_per_ip { - self.limit_hits.fetch_add(1, Ordering::Relaxed); + self.record_limit_hit(); + self.telemetry + .record_rejection(WebRejectionReason::BootstrapCapacity); + return Err(ManagerError::Limit); + } + if !allow_rate( + &mut state.bootstrap_rate, + now, + self.limits.new_bootstraps_per_minute, + self.limits.new_bootstraps_burst, + ) { + self.record_limit_hit(); + self.telemetry + .record_rejection(WebRejectionReason::BootstrapRate); return Err(ManagerError::Limit); } if state.bootstraps.len() >= self.limits.max_bootstraps_global && !evict_oldest_unused_bootstrap(&mut state) { - self.limit_hits.fetch_add(1, Ordering::Relaxed); + self.record_limit_hit(); + self.telemetry + .record_rejection(WebRejectionReason::BootstrapCapacity); return Err(ManagerError::Limit); } let Some((token, hash)) = new_unique_token(generation, &state) else { - self.limit_hits.fetch_add(1, Ordering::Relaxed); + self.record_limit_hit(); + self.telemetry + .record_rejection(WebRejectionReason::BootstrapCapacity); return Err(ManagerError::Limit); }; let trace_session_id = self.trace.next_session_id(); diff --git a/src/web/manager/lifecycle.rs b/src/web/manager/lifecycle.rs index 3853393..236df4b 100644 --- a/src/web/manager/lifecycle.rs +++ b/src/web/manager/lifecycle.rs @@ -1,5 +1,4 @@ use std::net::IpAddr; -use std::sync::atomic::Ordering; use std::time::{Duration, Instant}; use tokio::time::Instant as TokioInstant; @@ -79,7 +78,7 @@ impl WebProcessRuntime { for bootstrap_hash in bootstrap_hashes { remove_bootstrap_locked(&mut state, bootstrap_hash); } - self.sessions_closed.fetch_add(1, Ordering::Relaxed); + self.telemetry.record_session_closed(); drop(state); self.notify_operator_work_changed(); } @@ -93,6 +92,7 @@ impl WebProcessRuntime { self.close_websockets(); self.data_budget.close(); self.http_connections.close(); + self.http_overload_connections.close(); self.http_handlers.close(); self.lane_polls.close(); self.lane_aux_polls.close(); @@ -234,28 +234,29 @@ impl WebShutdownDrain { .saturating_duration_since(self.started) .as_millis() .min(u128::from(u64::MAX)) as u64; + let aggregates = self.runtime.telemetry.aggregates(); match outcome { WebShutdownOutcome::Drained => info!( target: "telemt::web", shutdown_drained = true, shutdown_budget_ms = budget_ms, shutdown_elapsed_ms = elapsed_ms, - sessions_created = self.runtime.sessions_created.load(Ordering::Relaxed), - sessions_closed = self.runtime.sessions_closed.load(Ordering::Relaxed), + sessions_created = aggregates.sessions_created, + sessions_closed = aggregates.sessions_closed, sessions_live, sessions_pending, session_tasks_live, auxiliary_tasks_live, - streams_opened = self.runtime.streams_opened.load(Ordering::Relaxed), - streams_rejected = self.runtime.streams_rejected.load(Ordering::Relaxed), + streams_opened = aggregates.streams_opened, + streams_rejected = aggregates.streams_rejected, streams_live, pending_bytes = budget.queue_bytes, pending_items = budget.queue_items, websocket_bytes = budget.websocket_bytes, data_high_water_bytes = budget.high_water_bytes, - bytes_up = self.runtime.bytes_up.load(Ordering::Relaxed), - bytes_down = self.runtime.bytes_down.load(Ordering::Relaxed), - limit_hits = self.runtime.limit_hits.load(Ordering::Relaxed), + bytes_up = aggregates.bytes_up, + bytes_down = aggregates.bytes_down, + limit_hits = aggregates.limit_hits, "WEB runtime stopped" ), WebShutdownOutcome::DeadlineExceeded => warn!( @@ -263,22 +264,22 @@ impl WebShutdownDrain { shutdown_drained = false, shutdown_budget_ms = budget_ms, shutdown_elapsed_ms = elapsed_ms, - sessions_created = self.runtime.sessions_created.load(Ordering::Relaxed), - sessions_closed = self.runtime.sessions_closed.load(Ordering::Relaxed), + sessions_created = aggregates.sessions_created, + sessions_closed = aggregates.sessions_closed, sessions_live, sessions_pending, session_tasks_live, auxiliary_tasks_live, - streams_opened = self.runtime.streams_opened.load(Ordering::Relaxed), - streams_rejected = self.runtime.streams_rejected.load(Ordering::Relaxed), + streams_opened = aggregates.streams_opened, + streams_rejected = aggregates.streams_rejected, streams_live, pending_bytes = budget.queue_bytes, pending_items = budget.queue_items, websocket_bytes = budget.websocket_bytes, data_high_water_bytes = budget.high_water_bytes, - bytes_up = self.runtime.bytes_up.load(Ordering::Relaxed), - bytes_down = self.runtime.bytes_down.load(Ordering::Relaxed), - limit_hits = self.runtime.limit_hits.load(Ordering::Relaxed), + bytes_up = aggregates.bytes_up, + bytes_down = aggregates.bytes_down, + limit_hits = aggregates.limit_hits, "WEB runtime shutdown deadline exceeded" ), } diff --git a/src/web/manager/observability.rs b/src/web/manager/observability.rs new file mode 100644 index 0000000..6f13066 --- /dev/null +++ b/src/web/manager/observability.rs @@ -0,0 +1,187 @@ +use serde::Serialize; + +use super::WebProcessRuntime; + +/// Current usage of one fixed process-wide WEB resource. +#[derive(Clone, Serialize)] +pub(crate) struct WebCapacityResourceStatus { + /// Stable closed-set resource token. + pub(crate) resource: &'static str, + /// Stable unit used by API consumers and metric-family routing. + pub(crate) unit: &'static str, + /// Currently retained capacity. + pub(crate) used: usize, + /// Unretained portion of the immutable ceiling before class-specific reserves. + pub(crate) available: usize, + /// Immutable process-wide ceiling. + pub(crate) limit: usize, + /// Whether terminal shutdown closed this allocation authority. + pub(crate) closed: bool, +} + +/// Non-blocking fixed-cardinality capacity snapshot. +#[derive(Clone, Serialize)] +pub(crate) struct WebCapacitySnapshot { + /// Exact fixed process-wide resources available without dynamic labels. + pub(crate) resources: Vec, + /// Resources with no immediately available capacity. + pub(crate) saturated_resources: Vec<&'static str>, + /// Snapshot planes omitted because their short lock was contended. + pub(crate) partial: Vec<&'static str>, +} + +impl WebProcessRuntime { + /// Captures fixed global resource usage without blocking the data plane. + pub(crate) fn capacity_snapshot(&self) -> WebCapacitySnapshot { + let websocket_capacity = self + .limits + .max_http_connections + .saturating_sub(self.limits.websocket_http_connection_reserve); + let mut resources = vec![ + semaphore_status( + "http_connections", + "slots", + &self.http_connections, + self.limits.max_http_connections, + ), + semaphore_status( + "http_overload_connections", + "slots", + &self.http_overload_connections, + self.limits.max_http_overload_connections, + ), + semaphore_status( + "http_handlers", + "slots", + &self.http_handlers, + self.limits.max_http_handlers, + ), + semaphore_status( + "lane_polls", + "slots", + &self.lane_polls, + self.limits.max_http_handlers / 2, + ), + semaphore_status( + "lane_aux_polls", + "slots", + &self.lane_aux_polls, + (self.limits.max_http_handlers / 4).max(1), + ), + semaphore_status( + "body_readers", + "slots", + &self.body_readers, + self.limits.max_body_readers, + ), + semaphore_status( + "body_bytes", + "bytes", + &self.body_bytes, + self.limits.max_body_bytes_global, + ), + semaphore_status( + "stream_handshakes", + "slots", + &self.stream_handshakes, + self.limits.max_stream_handshakes, + ), + semaphore_status( + "websocket_connections", + "slots", + &self.websocket_connections, + websocket_capacity, + ), + ]; + let mut partial = Vec::new(); + if let Some(budget) = self.data_budget.try_snapshot() { + resources.extend([ + bounded_status( + "pending_bytes", + "bytes", + budget.queue_bytes.saturating_add(budget.websocket_bytes), + self.limits.pending_bytes_global, + budget.closed, + ), + bounded_status( + "queue_items", + "items", + budget.queue_items, + self.limits.pending_items_global, + budget.closed, + ), + bounded_status( + "websocket_bytes", + "bytes", + budget.websocket_bytes, + self.limits.websocket_bytes_global, + budget.closed, + ), + ]); + } else { + partial.push("budget"); + } + let saturated_resources = resources + .iter() + .filter(|status| status.available == 0 && !status.closed) + .map(|status| status.resource) + .collect(); + WebCapacitySnapshot { + resources, + saturated_resources, + partial, + } + } +} + +fn semaphore_status( + resource: &'static str, + unit: &'static str, + semaphore: &tokio::sync::Semaphore, + limit: usize, +) -> WebCapacityResourceStatus { + let available = semaphore.available_permits().min(limit); + WebCapacityResourceStatus { + resource, + unit, + used: limit.saturating_sub(available), + available, + limit, + closed: semaphore.is_closed(), + } +} + +fn bounded_status( + resource: &'static str, + unit: &'static str, + used: usize, + limit: usize, + closed: bool, +) -> WebCapacityResourceStatus { + let used = used.min(limit); + WebCapacityResourceStatus { + resource, + unit, + used, + available: limit.saturating_sub(used), + limit, + closed, + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use tokio::sync::Semaphore; + + #[test] + fn semaphore_status_reports_usage_and_available_capacity() { + let semaphore = Arc::new(Semaphore::new(3)); + let _permit = semaphore.clone().try_acquire_owned().unwrap(); + let status = super::semaphore_status("test", "slots", &semaphore, 3); + assert_eq!(status.used, 1); + assert_eq!(status.available, 2); + assert_eq!(status.limit, 3); + } +} diff --git a/src/web/manager/operator_lifecycle.rs b/src/web/manager/operator_lifecycle.rs index 672bd73..d2b9fe8 100644 --- a/src/web/manager/operator_lifecycle.rs +++ b/src/web/manager/operator_lifecycle.rs @@ -16,6 +16,9 @@ pub(crate) use status::{ OperatorDrainOutcome, OperatorDrainState, OperatorDrainStatus, OperatorLifecycleState, OperatorLifecycleStatus, }; +// Mutable and published lifecycle state storage. +mod state; +use state::{ActiveDrain, OperatorLifecycleInner, OperatorSnapshot, WorkCounts}; const OPERATOR_ADMISSION_CLOSED: usize = 1 << (usize::BITS - 1); const OPERATOR_REGISTRATION_COUNT: usize = OPERATOR_ADMISSION_CLOSED - 1; @@ -30,19 +33,6 @@ pub(crate) enum OperatorLifecycleError { OperationInProgress, } -#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] -struct WorkCounts { - sessions: usize, - streams: usize, - websockets: usize, -} - -impl WorkCounts { - fn is_zero(self) -> bool { - self.sessions == 0 && self.streams == 0 && self.websockets == 0 - } -} - struct OperatorAdmission { state: AtomicUsize, registrations_drained: Notify, @@ -114,29 +104,6 @@ impl Drop for OperatorRegistration<'_> { } } -struct ActiveDrain { - sequence: u64, - cancellation: CancellationToken, -} - -struct OperatorLifecycleInner { - state: OperatorLifecycleState, - epoch: u64, - since: Instant, - terminal: bool, - active: Option, - drain: Option, -} - -#[derive(Clone)] -struct OperatorSnapshot { - state: OperatorLifecycleState, - epoch: u64, - since: Instant, - terminal: bool, - drain: Option, -} - pub(super) struct OperatorLifecycle { runtime_instance: Arc, admission: OperatorAdmission, @@ -187,8 +154,8 @@ impl OperatorLifecycle { pub(super) fn status(&self, config_enabled: bool) -> OperatorLifecycleStatus { let snapshot = self.published.load(); - let admission_open = !snapshot.terminal - && snapshot.state == OperatorLifecycleState::Running; + let admission_open = + !snapshot.terminal && snapshot.state == OperatorLifecycleState::Running; OperatorLifecycleStatus { state: snapshot.state, epoch: snapshot.epoch, @@ -203,6 +170,30 @@ impl OperatorLifecycle { self.published.load().terminal } + fn rejection_reason(&self) -> crate::web::telemetry::WebRejectionReason { + let inner = self.inner.lock(); + if inner.terminal { + return crate::web::telemetry::WebRejectionReason::RuntimeClosed; + } + match inner.state { + OperatorLifecycleState::Paused => { + crate::web::telemetry::WebRejectionReason::OperatorPaused + } + OperatorLifecycleState::Draining => { + crate::web::telemetry::WebRejectionReason::OperatorDraining + } + OperatorLifecycleState::ForceClosing => { + crate::web::telemetry::WebRejectionReason::OperatorForceClosing + } + OperatorLifecycleState::Drained => { + crate::web::telemetry::WebRejectionReason::OperatorDrained + } + OperatorLifecycleState::Running => { + crate::web::telemetry::WebRejectionReason::RuntimeClosed + } + } + } + fn publish_locked(&self, inner: &OperatorLifecycleInner) { self.published.store(Arc::new(OperatorSnapshot { state: inner.state, @@ -213,11 +204,7 @@ impl OperatorLifecycle { })); } - fn transition_locked( - &self, - inner: &mut OperatorLifecycleInner, - state: OperatorLifecycleState, - ) { + fn transition_locked(&self, inner: &mut OperatorLifecycleInner, state: OperatorLifecycleState) { if inner.state != state { inner.state = state; inner.epoch = inner.epoch.saturating_add(1); @@ -460,9 +447,14 @@ impl WebProcessRuntime { pub(super) fn try_operator_admission( &self, ) -> Result, super::ManagerError> { - self.operator_lifecycle - .try_register() - .ok_or(super::ManagerError::AdmissionPaused) + match self.operator_lifecycle.try_register() { + Some(registration) => Ok(registration), + None => { + self.telemetry + .record_rejection(self.operator_lifecycle.rejection_reason()); + Err(super::ManagerError::AdmissionPaused) + } + } } pub(super) fn notify_operator_work_changed(&self) { diff --git a/src/web/manager/operator_lifecycle/state.rs b/src/web/manager/operator_lifecycle/state.rs new file mode 100644 index 0000000..f0b324e --- /dev/null +++ b/src/web/manager/operator_lifecycle/state.rs @@ -0,0 +1,62 @@ +use std::time::Instant; + +use tokio_util::sync::CancellationToken; + +use super::{OperatorDrainStatus, OperatorLifecycleState}; + +/// Exact live-work counts sampled while an operator drain is active. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(super) struct WorkCounts { + /// Live WEB sessions. + pub(super) sessions: usize, + /// Live logical streams. + pub(super) streams: usize, + /// Live session-owned WebSockets. + pub(super) websockets: usize, +} + +impl WorkCounts { + /// Returns whether every tracked work class reached zero. + pub(super) fn is_zero(self) -> bool { + self.sessions == 0 && self.streams == 0 && self.websockets == 0 + } +} + +/// Single active drain identity and cancellation authority. +pub(super) struct ActiveDrain { + /// Process-local operation sequence. + pub(super) sequence: u64, + /// Cancellation token owned by resume or terminal shutdown. + pub(super) cancellation: CancellationToken, +} + +/// Mutex-protected lifecycle state used for atomic transitions. +pub(super) struct OperatorLifecycleInner { + /// Current reversible lifecycle state. + pub(super) state: OperatorLifecycleState, + /// Monotonic transition epoch. + pub(super) epoch: u64, + /// Monotonic transition timestamp. + pub(super) since: Instant, + /// Whether terminal process shutdown closed the state machine. + pub(super) terminal: bool, + /// Currently active drain, if any. + pub(super) active: Option, + /// Active or retained latest drain status. + pub(super) drain: Option, +} + +/// Lock-free read snapshot published after lifecycle mutations. +#[derive(Clone)] +pub(super) struct OperatorSnapshot { + /// Current reversible lifecycle state. + pub(super) state: OperatorLifecycleState, + /// Monotonic transition epoch. + pub(super) epoch: u64, + /// Monotonic transition timestamp. + pub(super) since: Instant, + /// Whether terminal process shutdown closed the state machine. + pub(super) terminal: bool, + /// Active or retained latest drain status. + pub(super) drain: Option, +} diff --git a/src/web/manager/operator_lifecycle/tests.rs b/src/web/manager/operator_lifecycle/tests.rs index 6ce2ce8..a4d7dce 100644 --- a/src/web/manager/operator_lifecycle/tests.rs +++ b/src/web/manager/operator_lifecycle/tests.rs @@ -106,7 +106,10 @@ async fn pause_fence_leaves_no_late_admission_commits_under_scheduler_pressure() #[tokio::test] async fn empty_drain_completes_gracefully_and_stays_closed_until_resume() { let (runtime, generation) = test_runtime(); - let accepted = runtime.drain_operator(Duration::from_secs(30)).await.unwrap(); + let accepted = runtime + .drain_operator(Duration::from_secs(30)) + .await + .unwrap(); assert_eq!(accepted.state, OperatorLifecycleState::Draining); let completed = tokio::time::timeout(Duration::from_secs(1), async { diff --git a/src/web/manager/session_admission.rs b/src/web/manager/session_admission.rs index 6b271c3..4efc8f5 100644 --- a/src/web/manager/session_admission.rs +++ b/src/web/manager/session_admission.rs @@ -1,10 +1,10 @@ use std::net::IpAddr; -use std::sync::atomic::Ordering; use std::time::Instant; use super::state::{ManagerState, allow_rate}; use super::{ProfileKey, WebProcessRuntime}; use crate::config::WebRuntimeProfile; +use crate::web::telemetry::WebRejectionReason; /// Applies process, address, profile, and rate ceilings to one initial session. pub(super) fn admit_initial( @@ -15,23 +15,33 @@ pub(super) fn admit_initial( profile_key: ProfileKey, profile: &WebRuntimeProfile, ) -> bool { - let admitted = state.sessions.len() < runtime.limits.max_sessions_global - && state.sessions_per_ip.get(&client_ip).copied().unwrap_or(0) - < runtime.limits.max_sessions_per_ip - && state + if state.sessions.len() >= runtime.limits.max_sessions_global + || state.sessions_per_ip.get(&client_ip).copied().unwrap_or(0) + >= runtime.limits.max_sessions_per_ip + || state .sessions_per_profile .get(&profile_key) .copied() .unwrap_or(0) - < profile.max_sessions - && allow_rate( - &mut state.session_rate, - now, - runtime.limits.new_sessions_per_minute, - runtime.limits.new_sessions_burst, - ); - if !admitted { - runtime.limit_hits.fetch_add(1, Ordering::Relaxed); + >= profile.max_sessions + { + runtime.record_limit_hit(); + runtime + .telemetry + .record_rejection(WebRejectionReason::SessionCapacity); + return false; } - admitted + if !allow_rate( + &mut state.session_rate, + now, + runtime.limits.new_sessions_per_minute, + runtime.limits.new_sessions_burst, + ) { + runtime.record_limit_hit(); + runtime + .telemetry + .record_rejection(WebRejectionReason::SessionRate); + return false; + } + true } diff --git a/src/web/manager/session_creation.rs b/src/web/manager/session_creation.rs index e0474ab..9f9512b 100644 --- a/src/web/manager/session_creation.rs +++ b/src/web/manager/session_creation.rs @@ -1,6 +1,5 @@ use std::net::IpAddr; use std::sync::Arc; -use std::sync::atomic::Ordering; use std::time::{Duration, Instant}; use sha2::{Digest, Sha256}; @@ -278,7 +277,9 @@ impl WebProcessRuntime { return Err(ManagerError::Limit); } let Some((session_token, session_hash)) = new_unique_token(&generation, &state) else { - self.limit_hits.fetch_add(1, Ordering::Relaxed); + self.record_limit_hit(); + self.telemetry + .record_rejection(crate::web::telemetry::WebRejectionReason::SessionCapacity); return Err(ManagerError::Limit); }; let carrier_deadline_at = carrier_request @@ -341,7 +342,7 @@ impl WebProcessRuntime { ) }; decrement_map(&mut state.bootstraps_per_ip, &issuance_ip); - self.sessions_created.fetch_add(1, Ordering::Relaxed); + self.telemetry.record_session_created(); let identity = session.trace_identity(); let result = CreateResult { token: session_token, diff --git a/src/web/manager/session_creation/replacement.rs b/src/web/manager/session_creation/replacement.rs index 6f73020..fdcaab4 100644 --- a/src/web/manager/session_creation/replacement.rs +++ b/src/web/manager/session_creation/replacement.rs @@ -48,7 +48,9 @@ impl WebProcessRuntime { return Err(ManagerError::Closed); } let Some((session_token, session_hash)) = new_unique_token(&generation, &state) else { - self.limit_hits.fetch_add(1, Ordering::Relaxed); + self.record_limit_hit(); + self.telemetry + .record_rejection(crate::web::telemetry::WebRejectionReason::SessionCapacity); drop(state); self.cancel_replacement(bootstrap_hash, &replacement.old_session); return Err(ManagerError::Limit); @@ -112,8 +114,8 @@ impl WebProcessRuntime { { *slot = Some(replacement.old_session.carrier()); } - self.sessions_created.fetch_add(1, Ordering::Relaxed); - self.sessions_closed.fetch_add(1, Ordering::Relaxed); + self.telemetry.record_session_created(); + self.telemetry.record_session_closed(); let result = CreateResult { token: session_token, carrier: replacement.carrier, diff --git a/src/web/manager/status.rs b/src/web/manager/status.rs index 45991a7..3bdf96d 100644 --- a/src/web/manager/status.rs +++ b/src/web/manager/status.rs @@ -1,7 +1,6 @@ use std::net::IpAddr; use std::ops::Bound::{Excluded, Unbounded}; use std::sync::Arc; -use std::sync::atomic::Ordering; use std::time::Instant; use serde::Serialize; @@ -298,6 +297,7 @@ impl WebProcessRuntime { .limits .max_http_connections .saturating_sub(self.limits.websocket_http_connection_reserve); + let aggregates = self.telemetry.aggregates(); WebRuntimeStatus { runtime_instance: self.runtime_instance().to_string(), generation_id, @@ -313,6 +313,13 @@ impl WebProcessRuntime { "http_connections", permits(&self.http_connections, self.limits.max_http_connections), ), + ( + "http_overload_connections", + permits( + &self.http_overload_connections, + self.limits.max_http_overload_connections, + ), + ), ( "http_handlers", permits(&self.http_handlers, self.limits.max_http_handlers), @@ -346,13 +353,13 @@ impl WebProcessRuntime { ), ], auxiliary_tasks: self.tasks.len(), - session_incarnations_created: self.sessions_created.load(Ordering::Relaxed), - session_incarnations_closed: self.sessions_closed.load(Ordering::Relaxed), - streams_opened: self.streams_opened.load(Ordering::Relaxed), - streams_rejected: self.streams_rejected.load(Ordering::Relaxed), - bytes_up: self.bytes_up.load(Ordering::Relaxed), - bytes_down: self.bytes_down.load(Ordering::Relaxed), - limit_hits: self.limit_hits.load(Ordering::Relaxed), + session_incarnations_created: aggregates.sessions_created, + session_incarnations_closed: aggregates.sessions_closed, + streams_opened: aggregates.streams_opened, + streams_rejected: aggregates.streams_rejected, + bytes_up: aggregates.bytes_up, + bytes_down: aggregates.bytes_down, + limit_hits: aggregates.limit_hits, partial, } } diff --git a/src/web/manager/websocket.rs b/src/web/manager/websocket.rs index 7a36073..3c7dc28 100644 --- a/src/web/manager/websocket.rs +++ b/src/web/manager/websocket.rs @@ -8,6 +8,7 @@ use tokio::sync::OwnedSemaphorePermit; use tokio_util::sync::CancellationToken; use super::{ManagerError, ProfileKey, WebProcessRuntime, WebSocketBudgetLease}; +use crate::web::telemetry::WebRejectionReason; // Deterministic victim ordering remains isolated from registry mutation. mod policy; @@ -208,7 +209,7 @@ pub(super) async fn admit( return Err(ManagerError::Closed); } let liveness_interval_ms = liveness_interval.as_millis().min(u128::from(u64::MAX)) as u64; - match try_admit( + let capacity_reason = match try_admit( runtime, owner, session_id, @@ -220,12 +221,23 @@ pub(super) async fn admit( &parent_cancellation, ) { Ok(connection) => return Ok(connection), - Err(TryAdmitError::Conflict) => return Err(ManagerError::Concurrent), - Err(TryAdmitError::Closed) => return Err(ManagerError::Closed), - Err(TryAdmitError::Capacity) => {} - } + Err(TryAdmitError::Conflict) => { + runtime + .telemetry() + .record_rejection(WebRejectionReason::Concurrent); + return Err(ManagerError::Concurrent); + } + Err(TryAdmitError::Closed) => { + runtime + .telemetry() + .record_rejection(WebRejectionReason::RuntimeClosed); + return Err(ManagerError::Closed); + } + Err(TryAdmitError::Capacity(reason)) => reason, + }; let Some(victim) = select_victim(runtime, owner, session_id, client_ip, None, true) else { runtime.record_limit_hit(); + runtime.telemetry().record_rejection(capacity_reason); return Err(ManagerError::Limit); }; let released = victim.released.cancelled(); @@ -249,17 +261,28 @@ pub(super) async fn admit( &parent_cancellation, ) { Ok(connection) => Ok(connection), - Err(TryAdmitError::Conflict) => Err(ManagerError::Concurrent), - Err(TryAdmitError::Closed) => Err(ManagerError::Closed), - Err(TryAdmitError::Capacity) => { + Err(TryAdmitError::Conflict) => { + runtime + .telemetry() + .record_rejection(WebRejectionReason::Concurrent); + Err(ManagerError::Concurrent) + } + Err(TryAdmitError::Closed) => { + runtime + .telemetry() + .record_rejection(WebRejectionReason::RuntimeClosed); + Err(ManagerError::Closed) + } + Err(TryAdmitError::Capacity(reason)) => { runtime.record_limit_hit(); + runtime.telemetry().record_rejection(reason); Err(ManagerError::Limit) } } } enum TryAdmitError { - Capacity, + Capacity(WebRejectionReason), Conflict, Closed, } @@ -292,10 +315,13 @@ fn try_admit( } let slot = Arc::clone(&runtime.websocket_connections) .try_acquire_owned() - .map_err(|_| TryAdmitError::Capacity)?; - let base_budget = runtime - .try_websocket_base_budget(owner, base_bytes) - .ok_or(TryAdmitError::Capacity)?; + .map_err(|_| TryAdmitError::Capacity(WebRejectionReason::WebSocketConnectionCapacity))?; + let base_budget = + runtime + .try_websocket_base_budget(owner, base_bytes) + .ok_or(TryAdmitError::Capacity( + WebRejectionReason::WebSocketBytesCapacity, + ))?; if parent_cancellation.is_cancelled() { return Err(TryAdmitError::Closed); } @@ -304,7 +330,7 @@ fn try_admit( .fetch_update(Ordering::AcqRel, Ordering::Acquire, |value| { value.checked_add(1) }) - .map_err(|_| TryAdmitError::Capacity)?; + .map_err(|_| TryAdmitError::Capacity(WebRejectionReason::WebSocketConnectionCapacity))?; let now = runtime.websocket_tick(); let entry = Arc::new(WebSocketEntry { id, diff --git a/src/web/mod.rs b/src/web/mod.rs index e149e0b..71ef5f0 100644 --- a/src/web/mod.rs +++ b/src/web/mod.rs @@ -14,5 +14,7 @@ pub(crate) mod manager; pub(crate) mod session; /// AsyncRead and AsyncWrite adapter for one logical MTProxy stream. pub(crate) mod stream; +/// Process-owned fixed-cardinality WEB operational telemetry. +pub(crate) mod telemetry; /// Process-owned bounded WEB debugging records and capture lifecycle. pub(crate) mod trace; diff --git a/src/web/session/backend.rs b/src/web/session/backend.rs index bd3d20f..c9ea00f 100644 --- a/src/web/session/backend.rs +++ b/src/web/session/backend.rs @@ -31,6 +31,9 @@ impl WebSession { }; let generation = manager.active_generation(); if !*generation.admission_rx.borrow() { + manager.record_stream_rejected_reason( + crate::web::telemetry::WebRejectionReason::GenerationAdmissionClosed, + ); self.trace_lifecycle( crate::web::trace::TraceLifecycleEvent::StreamRejected, Some(stream.id), @@ -70,6 +73,9 @@ impl WebSession { } }; if let Err(future) = generation.try_spawn_session(future) { + manager.record_stream_rejected_reason( + crate::web::telemetry::WebRejectionReason::GenerationScopeClosed, + ); retain_rejected.store(retain_reservation_on_reject, Ordering::Release); self.trace_lifecycle( crate::web::trace::TraceLifecycleEvent::StreamRejected, @@ -257,7 +263,9 @@ async fn run_stream( return; }; let Ok(_connection_permit) = connection_permits.try_acquire_owned() else { - manager.record_stream_rejected(); + manager.record_stream_rejected_reason( + crate::web::telemetry::WebRejectionReason::GenerationConnectionCapacity, + ); session.trace_lifecycle( crate::web::trace::TraceLifecycleEvent::StreamRejected, Some(stream_identity.id), diff --git a/src/web/session/downlink.rs b/src/web/session/downlink.rs index 13a749d..1147296 100644 --- a/src/web/session/downlink.rs +++ b/src/web/session/downlink.rs @@ -183,6 +183,11 @@ impl WebSession { && data_items <= item_limit - items }; if !fits { + if let Some(manager) = self.manager.upgrade() { + manager.telemetry().record_rejection( + crate::web::telemetry::WebRejectionReason::QueueSessionCapacity, + ); + } return false; } let Some(manager) = self.manager.upgrade() else { diff --git a/src/web/session/uplink.rs b/src/web/session/uplink.rs index 49da439..c224f0f 100644 --- a/src/web/session/uplink.rs +++ b/src/web/session/uplink.rs @@ -284,10 +284,13 @@ impl WebSession { } fn reserve_stream_locked(&self, state: &mut SessionState) -> Option { + let manager = self.manager.upgrade()?; if state.active_peer_ports.len() >= self.profile.max_streams_per_session { + manager.record_stream_rejected_reason( + crate::web::telemetry::WebRejectionReason::StreamSessionCapacity, + ); return None; } - let manager = self.manager.upgrade()?; let peer_port = manager .try_acquire_stream( self.profile_key, diff --git a/src/web/telemetry.rs b/src/web/telemetry.rs new file mode 100644 index 0000000..c23b08c --- /dev/null +++ b/src/web/telemetry.rs @@ -0,0 +1,541 @@ +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::time::Instant; + +use serde::Serialize; + +/// Stable operational rejection reason recorded at the decision point. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[repr(usize)] +pub(crate) enum WebRejectionReason { + /// Ordinary accepted HTTP connection slots were exhausted. + HttpConnectionCapacity, + /// Concurrent HTTP request handler slots were exhausted. + HttpHandlerCapacity, + /// Primary long-poll slots were exhausted. + LanePollCapacity, + /// Auxiliary lane-open wait slots were exhausted. + LaneAuxPollCapacity, + /// Concurrent collected-body reader slots were exhausted. + BodyReaderCapacity, + /// Process-wide collected-body bytes were exhausted. + BodyBytesCapacity, + /// Bootstrap credential quota or identifier space was exhausted. + BootstrapCapacity, + /// Bootstrap issuance rate policy rejected the operation. + BootstrapRate, + /// Live session quota or identifier space was exhausted. + SessionCapacity, + /// Session creation rate policy rejected the operation. + SessionRate, + /// One session reached its logical-stream ceiling. + StreamSessionCapacity, + /// Global or profile logical-stream capacity was exhausted. + StreamCapacity, + /// Logical-stream creation rate policy rejected the operation. + StreamRate, + /// No synthetic source port remained for the exact relay tuple. + StreamTupleExhausted, + /// Concurrent inner-handshake slots were exhausted. + StreamHandshakeCapacity, + /// The active relay generation had no connection permit. + GenerationConnectionCapacity, + /// One session reached its queued data ceiling. + QueueSessionCapacity, + /// The process-wide queued data ceiling was exhausted. + QueueGlobalCapacity, + /// Process-wide WebSocket connection slots were exhausted. + WebSocketConnectionCapacity, + /// Shared WebSocket byte capacity was exhausted. + WebSocketBytesCapacity, + /// Bounded WebSocket replacement capacity was exhausted. + WebSocketEvictionCapacity, + /// Effective WEB configuration disabled new work. + ConfigDisabled, + /// Effective user policy disabled new work. + UserDisabled, + /// Operator admission was paused. + OperatorPaused, + /// Operator admission was gracefully draining. + OperatorDraining, + /// Operator drain had committed forced close signals. + OperatorForceClosing, + /// Operator drain had reached confirmed zero. + OperatorDrained, + /// The active relay generation closed new admission. + GenerationAdmissionClosed, + /// The active relay generation could not own a new task. + GenerationScopeClosed, + /// Terminal runtime closure rejected the operation. + RuntimeClosed, + /// A bounded operation could not acquire or complete its deadline. + Deadline, + /// Concurrent ownership rejected the operation. + Concurrent, +} + +impl WebRejectionReason { + /// Complete fixed rejection set in stable metric order. + pub(crate) const ALL: [Self; 32] = [ + Self::HttpConnectionCapacity, + Self::HttpHandlerCapacity, + Self::LanePollCapacity, + Self::LaneAuxPollCapacity, + Self::BodyReaderCapacity, + Self::BodyBytesCapacity, + Self::BootstrapCapacity, + Self::BootstrapRate, + Self::SessionCapacity, + Self::SessionRate, + Self::StreamSessionCapacity, + Self::StreamCapacity, + Self::StreamRate, + Self::StreamTupleExhausted, + Self::StreamHandshakeCapacity, + Self::GenerationConnectionCapacity, + Self::QueueSessionCapacity, + Self::QueueGlobalCapacity, + Self::WebSocketConnectionCapacity, + Self::WebSocketBytesCapacity, + Self::WebSocketEvictionCapacity, + Self::ConfigDisabled, + Self::UserDisabled, + Self::OperatorPaused, + Self::OperatorDraining, + Self::OperatorForceClosing, + Self::OperatorDrained, + Self::GenerationAdmissionClosed, + Self::GenerationScopeClosed, + Self::RuntimeClosed, + Self::Deadline, + Self::Concurrent, + ]; + + /// Returns the stable API and Prometheus label token. + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::HttpConnectionCapacity => "http_connection_capacity", + Self::HttpHandlerCapacity => "http_handler_capacity", + Self::LanePollCapacity => "lane_poll_capacity", + Self::LaneAuxPollCapacity => "lane_aux_poll_capacity", + Self::BodyReaderCapacity => "body_reader_capacity", + Self::BodyBytesCapacity => "body_bytes_capacity", + Self::BootstrapCapacity => "bootstrap_capacity", + Self::BootstrapRate => "bootstrap_rate", + Self::SessionCapacity => "session_capacity", + Self::SessionRate => "session_rate", + Self::StreamSessionCapacity => "stream_session_capacity", + Self::StreamCapacity => "stream_capacity", + Self::StreamRate => "stream_rate", + Self::StreamTupleExhausted => "stream_tuple_exhausted", + Self::StreamHandshakeCapacity => "stream_handshake_capacity", + Self::GenerationConnectionCapacity => "generation_connection_capacity", + Self::QueueSessionCapacity => "queue_session_capacity", + Self::QueueGlobalCapacity => "queue_global_capacity", + Self::WebSocketConnectionCapacity => "websocket_connection_capacity", + Self::WebSocketBytesCapacity => "websocket_bytes_capacity", + Self::WebSocketEvictionCapacity => "websocket_eviction_capacity", + Self::ConfigDisabled => "config_disabled", + Self::UserDisabled => "user_disabled", + Self::OperatorPaused => "operator_paused", + Self::OperatorDraining => "operator_draining", + Self::OperatorForceClosing => "operator_force_closing", + Self::OperatorDrained => "operator_drained", + Self::GenerationAdmissionClosed => "generation_admission_closed", + Self::GenerationScopeClosed => "generation_scope_closed", + Self::RuntimeClosed => "runtime_closed", + Self::Deadline => "deadline", + Self::Concurrent => "concurrent", + } + } +} + +/// Terminal result for one accepted socket that found normal HTTP capacity exhausted. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[repr(usize)] +pub(crate) enum WebHttpConnectionOverloadOutcome { + /// Legacy policy closed the accepted socket immediately. + Dropped, + /// Bounded waiting acquired ordinary connection capacity. + WaitAdmitted, + /// Bounded waiting expired and emitted the retryable response. + WaitTimeout503, + /// Immediate response policy emitted the retryable response. + Responded503, + /// The bounded overload socket pool was already full. + OverflowCapacityDrop, + /// Writing or closing the retryable response failed. + ResponseErrorDrop, + /// Listener or process shutdown cancelled overload handling. + ShutdownDrop, +} + +impl WebHttpConnectionOverloadOutcome { + /// Complete fixed accepted-socket outcome set in stable metric order. + pub(crate) const ALL: [Self; 7] = [ + Self::Dropped, + Self::WaitAdmitted, + Self::WaitTimeout503, + Self::Responded503, + Self::OverflowCapacityDrop, + Self::ResponseErrorDrop, + Self::ShutdownDrop, + ]; + + /// Returns the stable API and Prometheus label token. + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::Dropped => "dropped", + Self::WaitAdmitted => "wait_admitted", + Self::WaitTimeout503 => "wait_timeout_503", + Self::Responded503 => "responded_503", + Self::OverflowCapacityDrop => "overflow_capacity_drop", + Self::ResponseErrorDrop => "response_error_drop", + Self::ShutdownDrop => "shutdown_drop", + } + } +} + +/// Passive outcome for one plain-HTTP decoy upstream request. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[repr(usize)] +pub(crate) enum WebDecoyUpstreamOutcome { + /// The internal decoy origin produced a response head. + Success, + /// The request activity plane could not grant a deadline lease. + DeadlineExhausted, + /// The configured origin actively refused the TCP connection. + ConnectRefused, + /// The bounded TCP connect phase timed out. + ConnectTimeout, + /// The TCP connect failed for another I/O reason. + ConnectError, + /// The bounded HTTP client handshake timed out. + HttpHandshakeTimeout, + /// The HTTP client handshake failed. + HttpHandshakeError, + /// The origin did not produce a response head before the deadline. + ResponseHeadTimeout, + /// URI preparation or the HTTP request failed. + RequestError, +} + +impl WebDecoyUpstreamOutcome { + /// Complete fixed decoy-origin outcome set in stable metric order. + pub(crate) const ALL: [Self; 9] = [ + Self::Success, + Self::DeadlineExhausted, + Self::ConnectRefused, + Self::ConnectTimeout, + Self::ConnectError, + Self::HttpHandshakeTimeout, + Self::HttpHandshakeError, + Self::ResponseHeadTimeout, + Self::RequestError, + ]; + + /// Returns the stable API and Prometheus label token. + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::Success => "success", + Self::DeadlineExhausted => "deadline_exhausted", + Self::ConnectRefused => "connect_refused", + Self::ConnectTimeout => "connect_timeout", + Self::ConnectError => "connect_error", + Self::HttpHandshakeTimeout => "http_handshake_timeout", + Self::HttpHandshakeError => "http_handshake_error", + Self::ResponseHeadTimeout => "response_head_timeout", + Self::RequestError => "request_error", + } + } +} + +/// API-safe typed rejection counter. +#[derive(Clone, Serialize)] +pub(crate) struct WebRejectionCounter { + /// Stable closed-set rejection token. + pub(crate) reason: &'static str, + /// Monotonic process-lifetime count. + pub(crate) total: u64, +} + +/// API-safe typed outcome counter. +#[derive(Clone, Serialize)] +pub(crate) struct WebOutcomeCounter { + /// Stable closed-set outcome token. + pub(crate) outcome: &'static str, + /// Monotonic process-lifetime count. + pub(crate) total: u64, +} + +/// Existing aggregate WEB totals retained with their original semantics. +#[derive(Clone, Copy)] +pub(crate) struct WebAggregateSnapshot { + /// Created session incarnations. + pub(crate) sessions_created: u64, + /// Closed session incarnations. + pub(crate) sessions_closed: u64, + /// Admitted logical streams. + pub(crate) streams_opened: u64, + /// Rejected logical streams covered by the legacy aggregate. + pub(crate) streams_rejected: u64, + /// Accepted carrier uplink payload bytes. + pub(crate) bytes_up: u64, + /// Emitted carrier downlink payload bytes. + pub(crate) bytes_down: u64, + /// Legacy aggregate capacity-hit count. + pub(crate) limit_hits: u64, +} + +/// Process-owned operational counters shared by ingress, API, and metrics. +pub(crate) struct WebTelemetry { + started: Instant, + live_acceptors: AtomicUsize, + accepted: AtomicU64, + accept_errors: AtomicU64, + rejections: [AtomicU64; WebRejectionReason::ALL.len()], + overload_outcomes: [AtomicU64; WebHttpConnectionOverloadOutcome::ALL.len()], + decoy_outcomes: [AtomicU64; WebDecoyUpstreamOutcome::ALL.len()], + last_decoy_outcome: AtomicUsize, + last_decoy_elapsed_ms: AtomicU64, + sessions_created: AtomicU64, + sessions_closed: AtomicU64, + streams_opened: AtomicU64, + streams_rejected: AtomicU64, + bytes_up: AtomicU64, + bytes_down: AtomicU64, + limit_hits: AtomicU64, +} + +impl WebTelemetry { + /// Creates zeroed process-lifetime WEB telemetry. + pub(crate) fn new() -> Arc { + Arc::new(Self { + started: Instant::now(), + live_acceptors: AtomicUsize::new(0), + accepted: AtomicU64::new(0), + accept_errors: AtomicU64::new(0), + rejections: std::array::from_fn(|_| AtomicU64::new(0)), + overload_outcomes: std::array::from_fn(|_| AtomicU64::new(0)), + decoy_outcomes: std::array::from_fn(|_| AtomicU64::new(0)), + last_decoy_outcome: AtomicUsize::new(usize::MAX), + last_decoy_elapsed_ms: AtomicU64::new(0), + sessions_created: AtomicU64::new(0), + sessions_closed: AtomicU64::new(0), + streams_opened: AtomicU64::new(0), + streams_rejected: AtomicU64::new(0), + bytes_up: AtomicU64::new(0), + bytes_down: AtomicU64::new(0), + limit_hits: AtomicU64::new(0), + }) + } + + /// Registers one live accept loop before its task can be cancelled unpolled. + pub(crate) fn acceptor_guard(self: &Arc) -> WebAcceptorGuard { + self.live_acceptors.fetch_add(1, Ordering::AcqRel); + WebAcceptorGuard { + telemetry: Arc::clone(self), + } + } + + /// Returns currently registered WEB accept loops. + pub(crate) fn live_acceptors(&self) -> usize { + self.live_acceptors.load(Ordering::Acquire) + } + + /// Records one successfully accepted WEB socket. + pub(crate) fn record_accept(&self) { + self.accepted.fetch_add(1, Ordering::Relaxed); + } + + /// Returns successfully accepted WEB sockets. + pub(crate) fn accepted(&self) -> u64 { + self.accepted.load(Ordering::Relaxed) + } + + /// Records one WEB listener `accept` error. + pub(crate) fn record_accept_error(&self) { + self.accept_errors.fetch_add(1, Ordering::Relaxed); + } + + /// Returns WEB listener `accept` errors. + pub(crate) fn accept_errors(&self) -> u64 { + self.accept_errors.load(Ordering::Relaxed) + } + + /// Records one operational rejection at its admission decision point. + pub(crate) fn record_rejection(&self, reason: WebRejectionReason) { + self.rejections[reason as usize].fetch_add(1, Ordering::Relaxed); + } + + /// Returns one fixed rejection counter. + pub(crate) fn rejection_total(&self, reason: WebRejectionReason) -> u64 { + self.rejections[reason as usize].load(Ordering::Relaxed) + } + + /// Captures the complete fixed rejection set for API serialization. + pub(crate) fn rejection_counters(&self) -> Vec { + WebRejectionReason::ALL + .into_iter() + .map(|reason| WebRejectionCounter { + reason: reason.as_str(), + total: self.rejection_total(reason), + }) + .collect() + } + + /// Records one terminal accepted-socket overload outcome. + pub(crate) fn record_overload(&self, outcome: WebHttpConnectionOverloadOutcome) { + self.overload_outcomes[outcome as usize].fetch_add(1, Ordering::Relaxed); + } + + /// Returns one fixed accepted-socket overload counter. + pub(crate) fn overload_total(&self, outcome: WebHttpConnectionOverloadOutcome) -> u64 { + self.overload_outcomes[outcome as usize].load(Ordering::Relaxed) + } + + /// Captures the complete fixed overload outcome set for API serialization. + pub(crate) fn overload_counters(&self) -> Vec { + WebHttpConnectionOverloadOutcome::ALL + .into_iter() + .map(|outcome| WebOutcomeCounter { + outcome: outcome.as_str(), + total: self.overload_total(outcome), + }) + .collect() + } + + /// Records one internal plain-HTTP decoy origin outcome. + pub(crate) fn record_decoy(&self, outcome: WebDecoyUpstreamOutcome) { + self.decoy_outcomes[outcome as usize].fetch_add(1, Ordering::Relaxed); + let elapsed_ms = self.started.elapsed().as_millis().min(u128::from(u64::MAX)) as u64; + self.last_decoy_elapsed_ms + .store(elapsed_ms.saturating_add(1), Ordering::Relaxed); + self.last_decoy_outcome + .store(outcome as usize, Ordering::Release); + } + + /// Returns one fixed internal decoy origin counter. + pub(crate) fn decoy_total(&self, outcome: WebDecoyUpstreamOutcome) -> u64 { + self.decoy_outcomes[outcome as usize].load(Ordering::Relaxed) + } + + /// Captures the complete fixed decoy outcome set for API serialization. + pub(crate) fn decoy_counters(&self) -> Vec { + WebDecoyUpstreamOutcome::ALL + .into_iter() + .map(|outcome| WebOutcomeCounter { + outcome: outcome.as_str(), + total: self.decoy_total(outcome), + }) + .collect() + } + + /// Returns the last decoy outcome and its monotonic age in milliseconds. + pub(crate) fn last_decoy(&self) -> Option<(&'static str, u64)> { + let raw = self.last_decoy_outcome.load(Ordering::Acquire); + let outcome = WebDecoyUpstreamOutcome::ALL.get(raw).copied()?; + let recorded = self.last_decoy_elapsed_ms.load(Ordering::Relaxed); + let now = self.started.elapsed().as_millis().min(u128::from(u64::MAX)) as u64; + Some(( + outcome.as_str(), + now.saturating_sub(recorded.saturating_sub(1)), + )) + } + + /// Records one created session incarnation. + pub(crate) fn record_session_created(&self) { + self.sessions_created.fetch_add(1, Ordering::Relaxed); + } + + /// Records one closed session incarnation. + pub(crate) fn record_session_closed(&self) { + self.sessions_closed.fetch_add(1, Ordering::Relaxed); + } + + /// Records one admitted logical stream. + pub(crate) fn record_stream_opened(&self) { + self.streams_opened.fetch_add(1, Ordering::Relaxed); + } + + /// Records one logical stream in the legacy rejection aggregate. + pub(crate) fn record_stream_rejected(&self) { + self.streams_rejected.fetch_add(1, Ordering::Relaxed); + } + + /// Adds accepted carrier uplink payload bytes. + pub(crate) fn record_up(&self, bytes: usize) { + self.bytes_up.fetch_add(bytes as u64, Ordering::Relaxed); + } + + /// Adds emitted carrier downlink payload bytes. + pub(crate) fn record_down(&self, bytes: usize) { + self.bytes_down.fetch_add(bytes as u64, Ordering::Relaxed); + } + + /// Records one legacy aggregate capacity hit. + pub(crate) fn record_limit_hit(&self) { + self.limit_hits.fetch_add(1, Ordering::Relaxed); + } + + /// Captures legacy process-owned aggregate totals. + pub(crate) fn aggregates(&self) -> WebAggregateSnapshot { + WebAggregateSnapshot { + sessions_created: self.sessions_created.load(Ordering::Relaxed), + sessions_closed: self.sessions_closed.load(Ordering::Relaxed), + streams_opened: self.streams_opened.load(Ordering::Relaxed), + streams_rejected: self.streams_rejected.load(Ordering::Relaxed), + bytes_up: self.bytes_up.load(Ordering::Relaxed), + bytes_down: self.bytes_down.load(Ordering::Relaxed), + limit_hits: self.limit_hits.load(Ordering::Relaxed), + } + } +} + +/// Cancellation-safe live-acceptor registration. +pub(crate) struct WebAcceptorGuard { + telemetry: Arc, +} + +impl Drop for WebAcceptorGuard { + fn drop(&mut self) { + self.telemetry.live_acceptors.fetch_sub(1, Ordering::AcqRel); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn fixed_counter_sets_and_acceptor_guard_are_exact() { + let telemetry = WebTelemetry::new(); + let guard = telemetry.acceptor_guard(); + assert_eq!(telemetry.live_acceptors(), 1); + telemetry.record_rejection(WebRejectionReason::HttpConnectionCapacity); + telemetry.record_overload(WebHttpConnectionOverloadOutcome::Dropped); + telemetry.record_decoy(WebDecoyUpstreamOutcome::ConnectRefused); + assert_eq!( + telemetry.rejection_counters().len(), + WebRejectionReason::ALL.len() + ); + assert_eq!( + telemetry.overload_counters().len(), + WebHttpConnectionOverloadOutcome::ALL.len() + ); + assert_eq!( + telemetry.decoy_counters().len(), + WebDecoyUpstreamOutcome::ALL.len() + ); + assert_eq!( + telemetry.rejection_total(WebRejectionReason::HttpConnectionCapacity), + 1 + ); + assert_eq!( + telemetry.last_decoy().map(|value| value.0), + Some("connect_refused") + ); + drop(guard); + assert_eq!(telemetry.live_acceptors(), 0); + } +}