From afc02ada8857d9ab40ce2f4fb59fe76a489fb657 Mon Sep 17 00:00:00 2001 From: Max Goisser Date: Thu, 15 Oct 2026 11:52:19 +0200 Subject: [PATCH] Add support for Prometheus metrics Related Linear issue: ERMCP-41 --- Cargo.lock | 184 +++++++++++++++++ Cargo.toml | 3 + README.md | 87 ++++++++ generated/config-reference.json | 15 ++ src/config.rs | 140 +++++++++++++ src/config_reference.rs | 25 +++ src/lib.rs | 1 + src/main.rs | 237 +++++++++++++++++++++- src/metrics.rs | 342 ++++++++++++++++++++++++++++++++ src/runtime.rs | 145 ++++++++++++-- 10 files changed, 1158 insertions(+), 21 deletions(-) create mode 100644 src/metrics.rs diff --git a/Cargo.lock b/Cargo.lock index e333ffe..3a2d741 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -375,12 +375,15 @@ dependencies = [ "deno_webidl", "facet", "getrandom 0.3.4", + "metrics", + "metrics-exporter-prometheus", "reqwest", "rmcp", "schemars", "serde", "serde_json", "tokio", + "tokio-metrics", "toml", ] @@ -446,6 +449,21 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "crossbeam-epoch" +version = "0.9.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc74980687109a3b14c72fd458107bf0baa1da1a1a805e178d15501ba9b86d9d" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-utils" +version = "0.8.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a31eee39dddec8330830986fcd7625edb5a24ec90ea038215273bbc3adb08ac6" + [[package]] name = "cssparser" version = "0.36.0" @@ -1025,6 +1043,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" dependencies = [ "allocator-api2", + "foldhash", ] [[package]] @@ -1483,6 +1502,47 @@ version = "2.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" +[[package]] +name = "metrics" +version = "0.24.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "89550ee9f79e88fef3119de263694973a8adb26c21d75322164fb8c493039fe2" +dependencies = [ + "portable-atomic", + "rapidhash", +] + +[[package]] +name = "metrics-exporter-prometheus" +version = "0.17.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b166dea96003ee2531cf14833efedced545751d800f03535801d833313f8c15" +dependencies = [ + "base64 0.22.1", + "indexmap", + "metrics", + "metrics-util", + "quanta", + "thiserror", +] + +[[package]] +name = "metrics-util" +version = "0.20.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96f8722f8562635f92f8ed992f26df0532266eb03d5202607c20c0d7e9745e13" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", + "hashbrown 0.16.1", + "metrics", + "quanta", + "rand 0.9.5", + "rand_xoshiro", + "rapidhash", + "sketches-ddsketch", +] + [[package]] name = "mime" version = "0.3.17" @@ -1683,6 +1743,12 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "portable-atomic" +version = "1.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05c8b63e8d9609db387f0324918f81d68fe27748f084ef092fb35954d0539a85" + [[package]] name = "potential_utf" version = "0.1.5" @@ -1694,6 +1760,15 @@ dependencies = [ "zerovec", ] +[[package]] +name = "ppv-lite86" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" +dependencies = [ + "zerocopy", +] + [[package]] name = "prettyplease" version = "0.2.37" @@ -1713,6 +1788,21 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "quanta" +version = "0.12.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3ab5a9d756f0d97bdc89019bd2e4ea098cf9cde50ee7564dde6b81ccc8f06c7" +dependencies = [ + "crossbeam-utils", + "libc", + "once_cell", + "raw-cpuid", + "wasi", + "web-sys", + "winapi", +] + [[package]] name = "quinn" version = "0.11.11" @@ -1805,6 +1895,16 @@ dependencies = [ "rand_core 0.6.4", ] +[[package]] +name = "rand" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9ef1d0d795eb7d84685bca4f72f3649f064e6641543d3a8c415898726a57b41" +dependencies = [ + "rand_chacha", + "rand_core 0.9.5", +] + [[package]] name = "rand" version = "0.10.2" @@ -1816,12 +1916,31 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core 0.9.5", +] + [[package]] name = "rand_core" version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" +[[package]] +name = "rand_core" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" +dependencies = [ + "getrandom 0.3.4", +] + [[package]] name = "rand_core" version = "0.10.1" @@ -1837,6 +1956,33 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "rand_xoshiro" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f703f4665700daf5512dcca5f43afa6af89f09db47fb56be587f80636bda2d41" +dependencies = [ + "rand_core 0.9.5", +] + +[[package]] +name = "rapidhash" +version = "4.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5da7e78a036ce858e8d55b7e7dc8ba3a88b78350fd2155d3591bbd966b58589e" +dependencies = [ + "rustversion", +] + +[[package]] +name = "raw-cpuid" +version = "11.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "498cd0dc59d73224351ee52a95fee0f1a617a2eae0e7d9d720cc622c73a54186" +dependencies = [ + "bitflags", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -2224,6 +2370,12 @@ version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ee5873ec9cce0195efcb7a4e9507a04cd49aec9c83d0389df45b1ef7ba2e649" +[[package]] +name = "sketches-ddsketch" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c6f73aeb92d671e0cc4dca167e59b2deb6387c375391bc99ee743f326994a2b" + [[package]] name = "slab" version = "0.4.12" @@ -2536,6 +2688,18 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "tokio-metrics" +version = "0.4.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e0410015c6db7b67b9c9ab2a3af4d74a942d637ff248d0d055073750deac6f9" +dependencies = [ + "futures-util", + "pin-project-lite", + "tokio", + "tokio-stream", +] + [[package]] name = "tokio-rustls" version = "0.26.4" @@ -3164,6 +3328,26 @@ dependencies = [ "synstructure", ] +[[package]] +name = "zerocopy" +version = "0.8.63" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5fe1f8f1b06191a00962174c61aa5005e0bb391a6d80d07e24d115c01a92ed8" +dependencies = [ + "zerocopy-derive", +] + +[[package]] +name = "zerocopy-derive" +version = "0.8.63" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "863ad3ac83293fb4d740aedbfdc9240dd8d1a50c1099acd76ce80ce7c7230c7f" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "zerofrom" version = "0.1.8" diff --git a/Cargo.toml b/Cargo.toml index 12aa553..1ae1011 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,12 +14,15 @@ deno_web = "0.286.0" deno_webidl = "0.255.0" facet = "0.44.3" getrandom = "0.3.4" +metrics = "0.24.3" +metrics-exporter-prometheus = { version = "0.17.2", default-features = false } reqwest = { version = "0.12.24", default-features = false, features = ["gzip", "rustls-tls"] } rmcp = { version = "3.4.1", default-features = false, features = ["macros", "schemars", "server", "transport-streamable-http-server"] } schemars = "1.1.0" serde = { version = "1.0.228", features = ["derive"] } serde_json = "1.0.145" tokio = { version = "1.48.0", features = ["fs", "io-util", "macros", "net", "rt-multi-thread", "signal", "sync", "time"] } +tokio-metrics = "0.4.7" toml = "0.9.11" [build-dependencies] diff --git a/README.md b/README.md index 9390a1c..71b388f 100644 --- a/README.md +++ b/README.md @@ -476,3 +476,90 @@ cargo clippy --all-targets --all-features -- -D warnings This project is licensed under the [MIT License](LICENSE). Third-party dependencies retain their respective licenses. + +## Prometheus metrics + +Metrics are runtime opt-in; no rebuild is needed. Defaults (also used for omitted +keys in a partial section): + +```toml +[integrations.prometheus] +enabled = false +host = "127.0.0.1" +port = 3131 +``` + +To allow scraping from outside a container: + +```toml +[integrations.prometheus] +enabled = true +host = "0.0.0.0" +port = 3131 +``` + +| TOML key | Default | Environment override | +| --- | --- | --- | +| `integrations.prometheus.enabled` | `false` | `CODE_INTERPRETER_MCP_PROMETHEUS_ENABLED` (`true`/`false`) | +| `integrations.prometheus.host` | `"127.0.0.1"` | `CODE_INTERPRETER_MCP_PROMETHEUS_HOST` | +| `integrations.prometheus.port` | `3131` | `CODE_INTERPRETER_MCP_PROMETHEUS_PORT` | + +Environment overrides apply after TOML merging. When enabled, the host must be +an IPv4 or IPv6 address (e.g. `::1`), the port must be 1–65535 and must differ from +`bind_addr`'s port, even when the addresses differ. Invalid settings, bind failures, +and recorder-installation failures stop startup with an error. + +Scrape `http://127.0.0.1:3131/metrics` (or `http://[::1]:3131/metrics` for IPv6). +The fixed `/metrics` route uses a dedicated listener, independent of MCP Host +verification and sessions, with no authentication. Expose its port only to intended +scrapers using your deployment's network controls. The listener starts before Python +initialization: **metrics availability does not imply readiness**. Continue using +`/readyz` for readiness and `/livez` for liveness. Disabled metrics install no +recorder, open no listener, and spawn no reporter. Scrapes read recorded samples +without acquiring the execution queue lock or running Python/network fetches. + +All ten application metric families below have the `code_interpreter_mcp_` prefix. +Counters and gauges start at zero (queue capacity uses the configured value); +histograms are registered without synthetic observations. Labels never contain code, +paths, package names, URLs, session IDs, or error messages. + +| Metric suffix | Type | Labels and meaning | +| --- | --- | --- | +| `tool_calls_total` | Counter | `tool`, `outcome`: each completed or cancelled handler once | +| `tool_call_duration_seconds` | Histogram | `tool`, `outcome`: handler entry to completion, including queue/init | +| `execution_queue_depth` | Gauge | Jobs waiting for a worker | +| `execution_queue_capacity` | Gauge | Configured maximum waiting jobs | +| `execution_queue_wait_seconds` | Histogram | `outcome`: `dispatched`, `cancelled`, `shutdown`; enqueue until removal | +| `workers` | Gauge | `state`: `idle`, `initializing`, `executing`; mutually exclusive live worker states | +| `python_execution_duration_seconds` | Histogram | `outcome`: `success`, `python_error`, `timeout`, `runtime_failure`; excludes queue/init | +| `isolate_initialization_duration_seconds` | Histogram | `phase`: `startup`, `on_demand`, `replenishment`; `outcome`: `success`, `runtime_failure` | +| `execution_rejections_total` | Counter | `reason`: `validation`, `queue_full`, `shutdown` | +| `worker_unexpected_exits_total` | Counter | Worker exits outside normal shutdown, including panics | + +Tool labels are `run_python_code`, `list_code_snippets`, `get_code_snippet`. +Tool outcomes are `success`, `validation_error`, `overload`, `python_error`, +`timeout`, `cancelled`, `runtime_failure`. Structured tool results determine +outcomes, not HTTP status. These metrics cover handlers entered after MCP argument +parsing; malformed protocol requests rejected before handler entry are not tool calls. +A missing snippet is a successful lookup returning null. Rejections intentionally +count a subset of tool calls separately. Cancellation removes queued jobs, but +already-running Python may finish after its caller has gone; execution observations +remain separate. Shutdown removes queued jobs and lets active workers drain. + +All durations use `Duration::as_secs_f64()` in seconds (25 ms = 0.025). +Python execution buckets are `0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, +1, 2.5, 5, 10, 30, 60, 120`. Queue wait, isolate initialization, and handler +histograms use those boundaries plus `300, 600`. All include the implicit `+Inf` +bucket. Buckets control histogram resolution, not recorded duration precision. + +The same switch enables a five-second `tokio-metrics` reporter for the **main Tokio +runtime**, not the dedicated Python worker threads (covered by `workers` and the +execution/init metrics). Stable-runtime gauges are `tokio_workers_count`, +`tokio_live_tasks_count`, `tokio_global_queue_depth`, `tokio_total_park_count`, +`tokio_min_park_count`, `tokio_max_park_count`, `tokio_total_busy_duration_seconds`, +`tokio_min_busy_duration_seconds`, `tokio_max_busy_duration_seconds`, and +`tokio_busy_ratio`. Park counts and busy durations describe the last reporting +interval; busy ratio is the fraction of available worker time spent busy. The +monitor's durations are converted to fractional seconds rather than using its +built-in reporter's integer microseconds. No unstable Tokio compiler flags, +per-metric toggles, or configurable histogram buckets are required. diff --git a/generated/config-reference.json b/generated/config-reference.json index 7b10d8e..a5968f5 100644 --- a/generated/config-reference.json +++ b/generated/config-reference.json @@ -14,6 +14,21 @@ "http_allowed_hosts.[]": {}, "input_limits.max_file_kb": {}, "input_limits.max_total_file_kb": {}, + "integrations.prometheus.enabled": { + "default": false, + "environment": "CODE_INTERPRETER_MCP_PROMETHEUS_ENABLED", + "example": true + }, + "integrations.prometheus.host": { + "default": "127.0.0.1", + "environment": "CODE_INTERPRETER_MCP_PROMETHEUS_HOST", + "example": "0.0.0.0" + }, + "integrations.prometheus.port": { + "default": 3131, + "environment": "CODE_INTERPRETER_MCP_PROMETHEUS_PORT", + "example": 3131 + }, "package_cache_dir": {}, "preinstall_packages.[]": {}, "runtime_package_installation": {}, diff --git a/src/config.rs b/src/config.rs index 4ed3b60..3578bef 100644 --- a/src/config.rs +++ b/src/config.rs @@ -40,6 +40,70 @@ pub struct AppConfig { pub search: SearchConfig, #[serde(default)] pub input_limits: InputLimitsConfig, + #[serde(default)] + pub integrations: IntegrationsConfig, +} + +#[derive(Debug, Clone, Default, Deserialize, Facet)] +#[serde(default)] +pub struct IntegrationsConfig { + pub prometheus: PrometheusConfig, +} + +#[derive(Debug, Clone, Deserialize, Facet)] +#[serde(default)] +pub struct PrometheusConfig { + pub enabled: bool, + pub host: String, + pub port: i64, +} + +impl Default for PrometheusConfig { + fn default() -> Self { + Self { + enabled: false, + host: "127.0.0.1".into(), + port: 3131, + } + } +} + +impl PrometheusConfig { + pub fn apply_env(&mut self, get: impl Fn(&str) -> Option) -> Result<()> { + if let Some(value) = get("CODE_INTERPRETER_MCP_PROMETHEUS_ENABLED") { + self.enabled = value + .parse() + .context("CODE_INTERPRETER_MCP_PROMETHEUS_ENABLED must be true or false")?; + } + if let Some(value) = get("CODE_INTERPRETER_MCP_PROMETHEUS_HOST") { + self.host = value; + } + if let Some(value) = get("CODE_INTERPRETER_MCP_PROMETHEUS_PORT") { + self.port = value + .parse() + .context("CODE_INTERPRETER_MCP_PROMETHEUS_PORT must be an integer in 1–65535")?; + } + Ok(()) + } + + pub fn listen_address(&self, bind_addr: &str) -> Result> { + if !self.enabled { + return Ok(None); + } + let host: std::net::IpAddr = self + .host + .parse() + .context("integrations.prometheus.host must be an IPv4 or IPv6 address")?; + let port = u16::try_from(self.port) + .ok() + .filter(|port| *port != 0) + .context("integrations.prometheus.port must be in 1–65535")?; + let main: std::net::SocketAddr = bind_addr.parse().context("invalid bind_addr")?; + if port == main.port() { + bail!("integrations.prometheus.port must differ from the bind_addr port"); + } + Ok(Some(std::net::SocketAddr::new(host, port))) + } } #[derive(Debug, Clone, Deserialize, Facet)] @@ -174,6 +238,14 @@ impl AppConfig { "CODE_INTERPRETER_MCP_MAX_TOTAL_FILE_KB must be a positive integer in KB (1 KB = 1024 bytes)", )?; } + config + .integrations + .prometheus + .apply_env(|key| env::var(key).ok())?; + config + .integrations + .prometheus + .listen_address(&config.bind_addr)?; config.input_limits.validate()?; Ok(config) } @@ -310,6 +382,74 @@ fn merge(base: &mut toml::Value, incoming: toml::Value) { mod tests { use super::*; + #[test] + fn prometheus_defaults_merge_environment_and_validation() { + let mut base: toml::Value = + toml::from_str("[integrations.prometheus]\nenabled = true\nhost = '::1'").unwrap(); + merge( + &mut base, + toml::from_str("[integrations.prometheus]\nport = 4000").unwrap(), + ); + let config: AppConfig = base.try_into().unwrap(); + let mut prometheus = config.integrations.prometheus; + assert_eq!( + prometheus + .listen_address("127.0.0.1:8000") + .unwrap() + .unwrap() + .to_string(), + "[::1]:4000" + ); + prometheus + .apply_env(|key| match key { + "CODE_INTERPRETER_MCP_PROMETHEUS_ENABLED" => Some("true".into()), + "CODE_INTERPRETER_MCP_PROMETHEUS_HOST" => Some("0.0.0.0".into()), + "CODE_INTERPRETER_MCP_PROMETHEUS_PORT" => Some("5000".into()), + _ => None, + }) + .unwrap(); + assert_eq!( + prometheus + .listen_address("127.0.0.1:8000") + .unwrap() + .unwrap() + .to_string(), + "0.0.0.0:5000" + ); + assert!(prometheus.listen_address("[::1]:5000").is_err()); + for port in [-1, 0, 65536] { + prometheus.port = port; + assert!(prometheus.listen_address("127.0.0.1:8000").is_err()); + } + prometheus.port = 3131; + prometheus.host = "localhost".into(); + assert!(prometheus.listen_address("127.0.0.1:8000").is_err()); + prometheus.enabled = false; + assert!(prometheus.listen_address("invalid").unwrap().is_none()); + for (key, value) in [ + ("CODE_INTERPRETER_MCP_PROMETHEUS_ENABLED", "yes"), + ("CODE_INTERPRETER_MCP_PROMETHEUS_PORT", "abc"), + ] { + assert!( + prometheus + .apply_env(|name| (name == key).then(|| value.into())) + .is_err() + ); + } + for source in [ + "", + "[integrations]", + "[integrations.prometheus]", + "[integrations.prometheus]\nenabled = false", + ] { + let config: AppConfig = toml::from_str(source).unwrap(); + let p = config.integrations.prometheus; + assert!(!p.enabled); + assert_eq!(p.host, "127.0.0.1"); + assert_eq!(p.port, 3131); + } + } + #[test] fn worker_memory_limit_defaults_and_overrides() { for source in ["", "[workers]", "[workers]\nmax_workers = 4"] { diff --git a/src/config_reference.rs b/src/config_reference.rs index 3e85df8..396cdc4 100644 --- a/src/config_reference.rs +++ b/src/config_reference.rs @@ -8,6 +8,31 @@ use crate::config::AppConfig; pub fn generate_config_reference() -> Value { let mut entries = BTreeMap::new(); collect_paths(AppConfig::SHAPE, &mut Vec::new(), &mut entries); + for (key, default, environment, example) in [ + ( + "enabled", + serde_json::json!(false), + "CODE_INTERPRETER_MCP_PROMETHEUS_ENABLED", + serde_json::json!(true), + ), + ( + "host", + serde_json::json!("127.0.0.1"), + "CODE_INTERPRETER_MCP_PROMETHEUS_HOST", + serde_json::json!("0.0.0.0"), + ), + ( + "port", + serde_json::json!(3131), + "CODE_INTERPRETER_MCP_PROMETHEUS_PORT", + serde_json::json!(3131), + ), + ] { + entries.insert( + format!("integrations.prometheus.{key}"), + serde_json::json!({"default": default, "environment": environment, "example": example}), + ); + } serde_json::to_value(entries).expect("config reference is serializable") } diff --git a/src/lib.rs b/src/lib.rs index 22d6ce6..1c27f7c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,4 +1,5 @@ pub mod config; pub mod config_reference; pub mod input_limits; +pub mod metrics; pub mod snippet_search; diff --git a/src/main.rs b/src/main.rs index 406903a..43f45b1 100644 --- a/src/main.rs +++ b/src/main.rs @@ -34,6 +34,7 @@ use rmcp::{ }; use serde::{Deserialize, Serialize}; +use code_interpreter_mcp::metrics as app_metrics; mod runtime; mod tokio_workers; use code_interpreter_mcp::config::{AppConfig, CodeSnippetsConfig}; @@ -121,6 +122,21 @@ pub(crate) struct RunPythonCodeOutput { } impl RunPythonCodeOutput { + pub(crate) fn outcome(&self) -> &'static str { + if self.success { + return "success"; + } + if self.timed_out { + return "timeout"; + } + match self.error.as_ref().map(|error| error.kind.as_str()) { + Some("validation") => "validation_error", + Some("overloaded") => "overload", + Some("python") => "python_error", + _ => "runtime_failure", + } + } + fn validation(message: impl Into) -> Self { Self::failure("validation", message, None, false) } @@ -207,6 +223,19 @@ pub(crate) struct ValidatedFile { pub(crate) content_base64: String, } +// Removing a queued request also runs when rmcp drops the handler future. +struct PendingExecution<'a> { + runtime: &'a PyodideRuntime, + id: u64, + cancelled: Arc, +} +impl Drop for PendingExecution<'_> { + fn drop(&mut self) { + self.cancelled.store(true, Ordering::Release); + self.runtime.cancel_queued(self.id); + } +} + #[derive(Clone)] struct CodeInterpreterMcpServer { runtime: PyodideRuntime, @@ -263,9 +292,12 @@ impl CodeInterpreterMcpServer { Parameters(request): Parameters, context: RequestContext, ) -> Json { + let mut call = app_metrics::ToolCall::new("run_python_code"); let request = match validate_request(request, self.input_limits) { Ok(request) => request, Err(error) => { + app_metrics::rejection("validation"); + call.finish("validation_error"); return Json(RunPythonCodeOutput::validation(error.to_string())); } }; @@ -274,13 +306,22 @@ impl CodeInterpreterMcpServer { let (job_id, mut response) = match self.runtime.submit(request, cancellation.clone()) { Ok(submitted) => submitted, Err(_) if !self.runtime.health().is_ready() => { + call.finish("runtime_failure"); return Json(RunPythonCodeOutput::runtime( "Worker pool is not accepting executions", )); } - Err(_) => return Json(RunPythonCodeOutput::overloaded()), + Err(_) => { + call.finish("overload"); + return Json(RunPythonCodeOutput::overloaded()); + } }; + let _pending = PendingExecution { + runtime: &self.runtime, + id: job_id, + cancelled: cancellation.clone(), + }; let mut progress_token = context.meta.get_progress_token(); let peer = context.peer; let request_cancellation = context.ct; @@ -291,7 +332,9 @@ impl CodeInterpreterMcpServer { loop { tokio::select! { result = &mut response => { - return Json(result.unwrap_or_else(|_| RunPythonCodeOutput::runtime("Worker ended before returning a result"))); + let output = result.unwrap_or_else(|_| RunPythonCodeOutput::runtime("Worker ended before returning a result")); + call.finish(output.outcome()); + return Json(output); } _ = request_cancellation.cancelled() => { cancellation.store(true, Ordering::Release); @@ -324,6 +367,8 @@ impl CodeInterpreterMcpServer { &self, Parameters(request): Parameters, ) -> Json { + let mut call = app_metrics::ToolCall::new("list_code_snippets"); + call.finish("success"); let snippets = code_interpreter_mcp::snippet_search::search_snippets( &self.code_snippets.snippets, request.search.as_deref(), @@ -348,6 +393,8 @@ impl CodeInterpreterMcpServer { &self, Parameters(request): Parameters, ) -> Json { + let mut call = app_metrics::ToolCall::new("get_code_snippet"); + call.finish("success"); let snippet = self.code_snippets .snippets @@ -681,6 +728,7 @@ async fn async_main() -> Result<()> { .parse() .with_context(|| format!("invalid bind_addr: {}", config.bind_addr))?; + let _metrics = app_metrics::init(&config).await?; let server = CodeInterpreterMcpServer::new(&config).await?; eprintln!( "Input size limits: {} decoded bytes per file, {} decoded bytes total; HTTP request body ceiling: {} bytes", @@ -750,6 +798,191 @@ mod tests { ); } + // A child process owns the global recorder; parallel unit tests use local recorders. + #[tokio::test] + async fn prometheus_http_and_execution_outcomes() { + const CHILD: &str = "ERMCP_METRICS_TEST_CHILD"; + if std::env::var_os(CHILD).is_none() { + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "tests::prometheus_http_and_execution_outcomes", + "--nocapture", + ]) + .env(CHILD, "1") + .output() + .unwrap(); + assert!( + output.status.success(), + "{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + return; + } + let mut config: AppConfig = + toml::from_str("[workers]\nmax_workers = 1\nprewarmed_workers = 1").unwrap(); + let reservation = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let metrics_address = reservation.local_addr().unwrap(); + config.integrations.prometheus.enabled = true; + config.integrations.prometheus.port = metrics_address.port().into(); + drop(reservation); + let _metrics = app_metrics::init(&config).await.unwrap(); + let client = reqwest::Client::builder() + .no_proxy() + .timeout(Duration::from_secs(30)) + .build() + .unwrap(); + let scrape = || { + client + .get(format!("http://{metrics_address}/metrics")) + .header("Host", "independent.example") + }; + // Available before any Python worker exists; no MCP session/Host checks. + let idle = scrape().send().await.unwrap(); + assert_eq!(idle.status(), StatusCode::OK); + assert!( + idle.headers()["content-type"] + .to_str() + .unwrap() + .contains("version=0.0.4") + ); + let idle = idle.text().await.unwrap(); + for name in [ + app_metrics::CALLS, + app_metrics::CALL_DURATION, + app_metrics::DEPTH, + app_metrics::CAPACITY, + app_metrics::WAIT, + app_metrics::WORKERS, + app_metrics::EXECUTION, + app_metrics::INIT, + app_metrics::REJECTIONS, + app_metrics::EXITS, + ] { + assert!(idle.contains(&format!("# TYPE {name} ")), "{name}: {idle}"); + } + assert_eq!( + client + .get(format!("http://{metrics_address}/mcp")) + .send() + .await + .unwrap() + .status(), + StatusCode::NOT_FOUND + ); + let server = CodeInterpreterMcpServer::new(&config).await.unwrap(); + let runtime = server.runtime.clone(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}/mcp", listener.local_addr().unwrap()); + let router = http_router(&config, server); + let task = tokio::spawn(async move { + axum::serve(listener, router).await.unwrap(); + }); + let post = || { + client + .post(&url) + .header("Content-Type", "application/json") + .header("Accept", "application/json, text/event-stream") + }; + let response = post().body(r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-11-25","capabilities":{},"clientInfo":{"name":"metrics-test","version":"1"}}}"#).send().await.unwrap(); + let session = response.headers()["mcp-session-id"] + .to_str() + .unwrap() + .to_owned(); + let _ = response.text().await.unwrap(); + let _ = post() + .header("Mcp-Session-Id", &session) + .header("MCP-Protocol-Version", "2025-11-25") + .body(r#"{"jsonrpc":"2.0","method":"notifications/initialized"}"#) + .send() + .await + .unwrap(); + for (code, timeout, outcome) in [ + ("40 + 2", 30000, "success"), + ("raise ValueError('test')", 30000, "python_error"), + ("while True: pass", 10, "timeout"), + ("pass", 0, "validation_error"), + ] { + let request = RunPythonCodeRequest { + code: code.into(), + files: vec![], + timeout_ms: Some(timeout), + }; + let response = post() + .header("Mcp-Session-Id", &session) + .header("MCP-Protocol-Version", "2025-11-25") + .body(serde_json::to_vec(&rpc_request(&request)).unwrap()) + .send() + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let _ = response.text().await.unwrap(); + let output = scrape().send().await.unwrap().text().await.unwrap(); + assert!(output.contains(&format!("code_interpreter_mcp_tool_calls_total{{tool=\"run_python_code\",outcome=\"{outcome}\"}} 1")), "{output}"); + } + let output = scrape().send().await.unwrap().text().await.unwrap(); + for outcome in ["success", "python_error", "timeout"] { + assert!(output.contains(&format!("code_interpreter_mcp_python_execution_duration_seconds_count{{outcome=\"{outcome}\"}} 1")), "{output}"); + } + assert!( + output.contains( + "code_interpreter_mcp_execution_rejections_total{reason=\"validation\"} 1" + ) + ); + assert!(output.contains("tokio_workers_count")); + runtime.shutdown(); + task.abort(); + // A second listener must fail recorder installation, not silently serve empty data. + let reservation = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + config.integrations.prometheus.port = reservation.local_addr().unwrap().port().into(); + drop(reservation); + assert!( + app_metrics::init(&config) + .await + .err() + .unwrap() + .to_string() + .contains("global recorder") + ); + } + + #[test] + fn dropping_pending_execution_removes_queue_and_counts_cancellation_once() { + let recorder = app_metrics::builder().unwrap().build_recorder(); + let handle = recorder.handle(); + metrics::with_local_recorder(&recorder, || { + app_metrics::register(1); + let (runtime, worker) = PyodideRuntime::test_pool(); + let call = app_metrics::ToolCall::new("run_python_code"); + let cancelled = Arc::new(AtomicBool::new(false)); + let (id, response) = runtime + .submit( + RuntimeRequest { + code: "pass".into(), + files: vec![], + timeout_ms: 1000, + }, + cancelled.clone(), + ) + .ok() + .unwrap(); + drop(PendingExecution { + runtime: &runtime, + id, + cancelled, + }); + assert!(runtime.queue_position(id).is_none()); + drop(response); + drop(call); + runtime.shutdown(); + drop(worker); + }); + let output = handle.render(); + assert!(output.contains("code_interpreter_mcp_execution_queue_depth 0")); + assert!(output.contains("code_interpreter_mcp_tool_calls_total{tool=\"run_python_code\",outcome=\"cancelled\"} 1")); + } + #[tokio::test] async fn probes_accept_pod_hosts_and_report_saturation_and_worker_loss() { let (runtime, worker) = PyodideRuntime::test_pool(); diff --git a/src/metrics.rs b/src/metrics.rs new file mode 100644 index 0000000..e631350 --- /dev/null +++ b/src/metrics.rs @@ -0,0 +1,342 @@ +//! Bounded application metrics. Scrapes only read the recorder, never worker state. +use crate::config::AppConfig; +use anyhow::{Context, Result}; +use metrics::{ + Unit, counter, describe_counter, describe_gauge, describe_histogram, gauge, histogram, +}; +use metrics_exporter_prometheus::{Matcher, PrometheusBuilder}; +use std::time::Instant; + +pub const CALLS: &str = "code_interpreter_mcp_tool_calls_total"; +pub const CALL_DURATION: &str = "code_interpreter_mcp_tool_call_duration_seconds"; +pub const DEPTH: &str = "code_interpreter_mcp_execution_queue_depth"; +pub const CAPACITY: &str = "code_interpreter_mcp_execution_queue_capacity"; +pub const WAIT: &str = "code_interpreter_mcp_execution_queue_wait_seconds"; +pub const WORKERS: &str = "code_interpreter_mcp_workers"; +pub const EXECUTION: &str = "code_interpreter_mcp_python_execution_duration_seconds"; +pub const INIT: &str = "code_interpreter_mcp_isolate_initialization_duration_seconds"; +pub const REJECTIONS: &str = "code_interpreter_mcp_execution_rejections_total"; +pub const EXITS: &str = "code_interpreter_mcp_worker_unexpected_exits_total"; +pub const OUTCOMES: &[&str] = &[ + "success", + "validation_error", + "overload", + "python_error", + "timeout", + "cancelled", + "runtime_failure", +]; +pub const BUCKETS: &[f64] = &[ + 0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1., 2.5, 5., 10., 30., 60., 120., 300., 600., +]; +pub const EXECUTION_BUCKETS: &[f64] = &[ + 0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1., 2.5, 5., 10., 30., 60., 120., +]; + +pub fn builder() -> Result { + let mut builder = PrometheusBuilder::new(); + for name in [CALL_DURATION, WAIT, INIT, EXECUTION] { + builder = builder.set_buckets_for_metric( + Matcher::Full(name.into()), + if name == EXECUTION { + EXECUTION_BUCKETS + } else { + BUCKETS + }, + )?; + } + Ok(builder) +} + +/// Owned by application startup so listener/reporting tasks stop on shutdown or startup failure. +pub struct MetricsTasks(Vec>); +impl Drop for MetricsTasks { + fn drop(&mut self) { + for task in &self.0 { + task.abort(); + } + } +} + +pub async fn init(config: &AppConfig) -> Result> { + let Some(address) = config + .integrations + .prometheus + .listen_address(&config.bind_addr)? + else { + return Ok(None); + }; + let listener = tokio::net::TcpListener::bind(address) + .await + .with_context(|| format!("failed to bind Prometheus listener to {address}"))?; + let recorder = builder()?.build_recorder(); + let handle = recorder.handle(); + metrics::set_global_recorder(recorder) + .context("failed to install Prometheus recorder: a global recorder is already installed")?; + register(config.workers.queue_capacity); + let upkeep_handle = handle.clone(); + let router = axum::Router::new().route( + "/metrics", + axum::routing::get(move || { + let handle = handle.clone(); + async move { + ( + [( + axum::http::header::CONTENT_TYPE, + "text/plain; version=0.0.4; charset=utf-8", + )], + handle.render(), + ) + } + }), + ); + let exporter = tokio::spawn(async move { + if let Err(error) = axum::serve(listener, router).await { + eprintln!("Prometheus listener failed: {error}"); + } + }); + let reporter = tokio::spawn(report_tokio(upkeep_handle)); + eprintln!("Prometheus metrics enabled at http://{address}/metrics (not a readiness probe)"); + Ok(Some(MetricsTasks(vec![exporter, reporter]))) +} + +// The built-in metrics-rs reporter converts durations to integer microseconds. +// Use the stable monitor directly to preserve fractional seconds consistently. +async fn report_tokio(handle: metrics_exporter_prometheus::PrometheusHandle) { + let monitor = tokio_metrics::RuntimeMonitor::new(&tokio::runtime::Handle::current()); + for (name, description) in [ + ("tokio_workers_count", "Main runtime worker threads"), + ("tokio_live_tasks_count", "Main runtime live tasks"), + ( + "tokio_global_queue_depth", + "Main runtime global queue depth", + ), + ( + "tokio_total_park_count", + "Worker parks during the reporting interval", + ), + ( + "tokio_min_park_count", + "Minimum worker parks during the reporting interval", + ), + ( + "tokio_max_park_count", + "Maximum worker parks during the reporting interval", + ), + ] { + describe_gauge!(name, Unit::Count, description); + } + for (name, description) in [ + ( + "tokio_total_busy_duration_seconds", + "Sum of worker busy time during the reporting interval", + ), + ( + "tokio_min_busy_duration_seconds", + "Minimum worker busy time during the reporting interval", + ), + ( + "tokio_max_busy_duration_seconds", + "Maximum worker busy time during the reporting interval", + ), + ] { + describe_gauge!(name, Unit::Seconds, description); + } + describe_gauge!( + "tokio_busy_ratio", + "Fraction of main runtime worker time spent busy" + ); + let mut interval = tokio::time::interval(std::time::Duration::from_secs(5)); + for snapshot in monitor.intervals() { + handle.run_upkeep(); + gauge!("tokio_workers_count").set(snapshot.workers_count as f64); + gauge!("tokio_live_tasks_count").set(snapshot.live_tasks_count as f64); + gauge!("tokio_global_queue_depth").set(snapshot.global_queue_depth as f64); + gauge!("tokio_total_park_count").set(snapshot.total_park_count as f64); + gauge!("tokio_min_park_count").set(snapshot.min_park_count as f64); + gauge!("tokio_max_park_count").set(snapshot.max_park_count as f64); + gauge!("tokio_total_busy_duration_seconds").set(snapshot.total_busy_duration.as_secs_f64()); + gauge!("tokio_min_busy_duration_seconds").set(snapshot.min_busy_duration.as_secs_f64()); + gauge!("tokio_max_busy_duration_seconds").set(snapshot.max_busy_duration.as_secs_f64()); + let available = snapshot.elapsed.as_secs_f64() * snapshot.workers_count as f64; + gauge!("tokio_busy_ratio").set(if available > 0. { + snapshot.total_busy_duration.as_secs_f64() / available + } else { + 0. + }); + interval.tick().await; + } +} + +pub fn register(capacity: usize) { + describe_counter!( + CALLS, + Unit::Count, + "Completed tool calls by tool and outcome" + ); + describe_counter!( + REJECTIONS, + Unit::Count, + "Execution requests refused by reason" + ); + describe_counter!( + EXITS, + Unit::Count, + "Unexpected worker exits reducing capacity" + ); + describe_gauge!(DEPTH, Unit::Count, "Jobs waiting for a worker"); + describe_gauge!(CAPACITY, Unit::Count, "Configured waiting queue capacity"); + describe_gauge!( + WORKERS, + Unit::Count, + "Live workers by mutually exclusive state" + ); + for (name, description) in [ + ( + CALL_DURATION, + "Tool handler time including queue and initialization", + ), + (WAIT, "Time queued until dispatch or removal"), + (INIT, "Isolate initialization time by phase and outcome"), + ( + EXECUTION, + "Python execution time excluding queue and initialization", + ), + ] { + describe_histogram!(name, Unit::Seconds, description); + } + for tool in ["run_python_code", "list_code_snippets", "get_code_snippet"] { + for &outcome in OUTCOMES { + counter!(CALLS, "tool" => tool, "outcome" => outcome).absolute(0); + let _ = histogram!(CALL_DURATION, "tool" => tool, "outcome" => outcome); + } + } + for reason in ["validation", "queue_full", "shutdown"] { + counter!(REJECTIONS, "reason" => reason).absolute(0); + } + for outcome in ["dispatched", "cancelled", "shutdown"] { + let _ = histogram!(WAIT, "outcome" => outcome); + } + for outcome in ["success", "python_error", "timeout", "runtime_failure"] { + let _ = histogram!(EXECUTION, "outcome" => outcome); + } + for phase in ["startup", "on_demand", "replenishment"] { + for outcome in ["success", "runtime_failure"] { + let _ = histogram!(INIT, "phase" => phase, "outcome" => outcome); + } + } + counter!(EXITS).absolute(0); + gauge!(DEPTH).set(0.); + gauge!(CAPACITY).set(capacity as f64); + for state in ["idle", "initializing", "executing"] { + gauge!(WORKERS, "state" => state).set(0.); + } +} + +pub fn rejection(reason: &'static str) { + counter!(REJECTIONS, "reason" => reason).increment(1); +} +pub fn queue_removed(start: Instant, outcome: &'static str) { + gauge!(DEPTH).decrement(1.); + histogram!(WAIT, "outcome" => outcome).record(start.elapsed().as_secs_f64()); +} + +/// Default cancellation also covers dropped handler futures; explicit completion counts once. +pub struct ToolCall { + start: Instant, + tool: &'static str, + outcome: &'static str, +} +impl ToolCall { + pub fn new(tool: &'static str) -> Self { + Self { + start: Instant::now(), + tool, + outcome: "cancelled", + } + } + pub fn finish(&mut self, outcome: &'static str) { + self.outcome = outcome; + } +} +impl Drop for ToolCall { + fn drop(&mut self) { + counter!(CALLS, "tool" => self.tool, "outcome" => self.outcome).increment(1); + histogram!(CALL_DURATION, "tool" => self.tool, "outcome" => self.outcome) + .record(self.start.elapsed().as_secs_f64()); + } +} + +pub struct WorkerState(&'static str); +impl WorkerState { + pub fn new() -> Self { + gauge!(WORKERS, "state" => "idle").increment(1.); + Self("idle") + } + pub fn set(&mut self, state: &'static str) { + gauge!(WORKERS, "state" => self.0).decrement(1.); + gauge!(WORKERS, "state" => state).increment(1.); + self.0 = state; + } +} +impl Default for WorkerState { + fn default() -> Self { + Self::new() + } +} +impl Drop for WorkerState { + fn drop(&mut self) { + gauge!(WORKERS, "state" => self.0).decrement(1.); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn records_once_preserves_fractional_seconds_and_balances_worker_states() { + let recorder = builder().unwrap().build_recorder(); + let handle = recorder.handle(); + metrics::with_local_recorder(&recorder, || { + register(8); + let mut call = ToolCall::new("run_python_code"); + call.finish("success"); + drop(call); + drop(ToolCall::new("run_python_code")); + let mut worker = WorkerState::new(); + worker.set("initializing"); + worker.set("executing"); + drop(worker); + histogram!(EXECUTION, "outcome" => "success").record(0.025); + }); + let output = handle.render(); + assert!(output.contains("# TYPE code_interpreter_mcp_tool_calls_total counter")); + assert!(output.contains("# HELP code_interpreter_mcp_execution_queue_depth")); + for outcome in ["success", "cancelled"] { + assert!(output.contains(&format!("code_interpreter_mcp_tool_calls_total{{tool=\"run_python_code\",outcome=\"{outcome}\"}} 1")), "{output}"); + } + assert!(output.contains( + "code_interpreter_mcp_python_execution_duration_seconds_sum{outcome=\"success\"} 0.025" + )); + for state in ["idle", "initializing", "executing"] { + assert!(output.contains(&format!( + "code_interpreter_mcp_workers{{state=\"{state}\"}} 0" + ))); + } + } + + #[tokio::test] + async fn disabled_and_bind_failure_do_not_install_recorder() { + let mut config: AppConfig = toml::from_str("").unwrap(); + assert!(init(&config).await.unwrap().is_none()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + config.integrations.prometheus.enabled = true; + config.integrations.prometheus.port = listener.local_addr().unwrap().port().into(); + let error = init(&config).await.err().unwrap().to_string(); + assert!( + error.contains("failed to bind Prometheus listener"), + "{error}" + ); + } +} diff --git a/src/runtime.rs b/src/runtime.rs index 49c371c..3908b75 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -10,7 +10,7 @@ use std::{ atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}, mpsc, }, - time::Duration, + time::{Duration, Instant}, }; use anyhow::{Context, Result, bail}; @@ -23,6 +23,7 @@ use deno_error::JsErrorBox; use crate::{RunPythonCodeOutput, RuntimeRequest}; use code_interpreter_mcp::config::AppConfig; +use code_interpreter_mcp::metrics as app_metrics; use serde::{Deserialize, Serialize}; use tokio::sync::oneshot; @@ -65,6 +66,7 @@ struct PackageLayerFile { } struct QueuedRequest { + enqueued: Instant, id: u64, request: RuntimeRequest, cancelled: Arc, @@ -100,17 +102,20 @@ impl WorkerHealth { } /// Lives for the entire worker thread, including startup and panic unwinding. -pub(crate) struct WorkerGuard(Arc); +pub(crate) struct WorkerGuard(Arc, app_metrics::WorkerState); impl WorkerGuard { fn new(queue: Arc) -> Self { queue.health.live_workers.fetch_add(1, Ordering::AcqRel); - Self(queue) + Self(queue, app_metrics::WorkerState::new()) } } impl Drop for WorkerGuard { fn drop(&mut self) { + if !self.0.stopping.load(Ordering::Acquire) || std::thread::panicking() { + metrics::counter!(app_metrics::EXITS).increment(1); + } let last = self.0.health.live_workers.fetch_sub(1, Ordering::AcqRel) == 1; if self.0.state.is_poisoned() || (last && !self.0.stopping.load(Ordering::Acquire)) { self.0.health.fatal.store(true, Ordering::Release); @@ -146,17 +151,26 @@ impl WorkerQueue { cancelled: Arc, ) -> Result<(u64, oneshot::Receiver), RuntimeRequest> { if !self.health.is_ready() { + app_metrics::rejection("shutdown"); return Err(request); } let Ok(mut state) = self.state.lock() else { + app_metrics::rejection("shutdown"); return Err(request); }; - if state.closed || !self.health.is_ready() || state.requests.len() >= self.capacity { + if state.closed || !self.health.is_ready() { + app_metrics::rejection("shutdown"); + return Err(request); + } + if state.requests.len() >= self.capacity { + app_metrics::rejection("queue_full"); return Err(request); } let id = self.next_id.fetch_add(1, Ordering::Relaxed); let (response, receiver) = oneshot::channel(); + metrics::gauge!(app_metrics::DEPTH).increment(1.); state.requests.push_back(QueuedRequest { + enqueued: Instant::now(), id, request, cancelled, @@ -170,6 +184,15 @@ impl WorkerQueue { let mut state = self.state.lock().expect("worker queue mutex poisoned"); loop { if let Some(request) = state.requests.pop_front() { + let cancelled = + request.cancelled.load(Ordering::Acquire) || request.response.is_closed(); + app_metrics::queue_removed( + request.enqueued, + if cancelled { "cancelled" } else { "dispatched" }, + ); + if cancelled { + continue; + } return Some(request); } if state.closed { @@ -190,10 +213,11 @@ impl WorkerQueue { } fn cancel(&self, id: u64) { - let mut state = self.state.lock().expect("worker queue mutex poisoned"); + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); if let Some(index) = state.requests.iter().position(|request| request.id == id) && let Some(request) = state.requests.remove(index) { + app_metrics::queue_removed(request.enqueued, "cancelled"); request.cancelled.store(true, Ordering::Release); } } @@ -205,6 +229,7 @@ impl WorkerQueue { let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); state.closed = true; for job in state.requests.drain(..) { + app_metrics::queue_removed(job.enqueued, "shutdown"); let _ = job.response.send(RunPythonCodeOutput::runtime( "Worker pool stopped before queued execution could start", )); @@ -463,7 +488,7 @@ fn worker_thread( prewarmed: bool, started: tokio::sync::mpsc::Sender>, ) { - let _guard = WorkerGuard::new(queue.clone()); + let mut guard = WorkerGuard::new(queue.clone()); let runtime = match tokio::runtime::Builder::new_current_thread() .enable_all() .build() @@ -475,7 +500,7 @@ fn worker_thread( } }; let mut prepared = if prewarmed { - match runtime.block_on(create_initialized_isolate(&config, false)) { + match initialize_worker(&runtime, &config, &mut guard.1, "startup") { Ok(isolate) => Some(isolate), Err(error) => { let _ = started.blocking_send(Err(format!("worker prewarm failed: {error:#}"))); @@ -489,27 +514,31 @@ fn worker_thread( drop(started); loop { + guard.1.set("idle"); let Some(job) = queue.next() else { break; }; if job.cancelled.load(Ordering::Acquire) { continue; } - let timeout_ms = job.request.timeout_ms; let result = match prepared.take() { Some(isolate) => Ok(isolate), - None => runtime.block_on(create_initialized_isolate(&config, false)), + None => initialize_worker(&runtime, &config, &mut guard.1, "on_demand"), } - .and_then(|mut isolate| runtime.block_on(execute_python(&mut isolate, &job.request))); + .and_then(|mut isolate| { + guard.1.set("executing"); + let start = Instant::now(); + let result = runtime.block_on(execute_python(&mut isolate, &job.request)); + let outcome = match &result { + Ok(output) => output.outcome(), + Err(_) => "runtime_failure", + }; + metrics::histogram!(app_metrics::EXECUTION, "outcome" => outcome) + .record(start.elapsed().as_secs_f64()); + result + }); let output = match result { Ok(output) => output, - Err(error) - if error - .chain() - .any(|cause| cause.to_string().contains("execution terminated")) => - { - RunPythonCodeOutput::timed_out(timeout_ms) - } Err(error) => RunPythonCodeOutput::runtime(format!( "The embedded Deno/Pyodide runtime failed: {error:#}" )), @@ -519,7 +548,7 @@ fn worker_thread( // The request isolate is discarded. Build a clean replacement before // this worker becomes warm for its next request. if prewarmed && !queue.stopping.load(Ordering::Acquire) { - match runtime.block_on(create_initialized_isolate(&config, false)) { + match initialize_worker(&runtime, &config, &mut guard.1, "replenishment") { Ok(isolate) => prepared = Some(isolate), Err(error) => { eprintln!("Python worker could not replenish its warm isolate: {error:#}") @@ -529,6 +558,19 @@ fn worker_thread( } } +fn initialize_worker( + runtime: &tokio::runtime::Runtime, + config: &RuntimeConfig, + state: &mut app_metrics::WorkerState, + phase: &'static str, +) -> Result { + state.set("initializing"); + let start = Instant::now(); + let result = runtime.block_on(create_initialized_isolate(config, false)); + metrics::histogram!(app_metrics::INIT, "phase" => phase, "outcome" => if result.is_ok() { "success" } else { "runtime_failure" }).record(start.elapsed().as_secs_f64()); + result +} + fn create_runtime(config: &RuntimeConfig) -> Result { let web_blob_store = Arc::new(deno_web::BlobStore::default()) as Arc; @@ -646,8 +688,14 @@ async fn execute_python( let isolate_handle = runtime.v8_isolate().thread_safe_handle(); let (finished_tx, finished_rx) = mpsc::sync_channel(1); let timeout = Duration::from_millis(request.timeout_ms); + let timed_out = Arc::new(AtomicBool::new(false)); + let watchdog_timed_out = timed_out.clone(); let watchdog = std::thread::spawn(move || { - if finished_rx.recv_timeout(timeout).is_err() { + if matches!( + finished_rx.recv_timeout(timeout), + Err(mpsc::RecvTimeoutError::Timeout) + ) { + watchdog_timed_out.store(true, Ordering::Release); isolate_handle.terminate_execution(); } }); @@ -668,6 +716,9 @@ async fn execute_python( }; let _ = finished_tx.send(()); let _ = watchdog.join(); + if timed_out.load(Ordering::Acquire) { + return Ok(RunPythonCodeOutput::timed_out(request.timeout_ms)); + } let value = result?; deno_core::scope!(scope, runtime); let local = v8::Local::new(scope, value); @@ -1005,6 +1056,62 @@ mod tests { } } + #[test] + fn queue_metrics_cover_dispatch_cancel_overload_shutdown_and_worker_loss() { + let recorder = app_metrics::builder().unwrap().build_recorder(); + let handle = recorder.handle(); + metrics::with_local_recorder(&recorder, || { + app_metrics::register(1); + let (runtime, worker) = PyodideRuntime::test_pool(); + let (id, receiver) = runtime + .submit(request(), Arc::new(AtomicBool::new(false))) + .ok() + .unwrap(); + assert!( + runtime + .submit(request(), Arc::new(AtomicBool::new(false))) + .is_err() + ); + runtime.cancel_queued(id); + drop(receiver); + let (_, receiver) = runtime + .submit(request(), Arc::new(AtomicBool::new(false))) + .ok() + .unwrap(); + let job = runtime.pool.queue.next().unwrap(); + drop(job); + drop(receiver); + let (_, receiver) = runtime + .submit(request(), Arc::new(AtomicBool::new(false))) + .ok() + .unwrap(); + // Last worker exits unexpectedly; its guard drains the queue. + drop(worker); + assert!( + runtime + .submit(request(), Arc::new(AtomicBool::new(false))) + .is_err() + ); + drop(receiver); + }); + let output = handle.render(); + for sample in [ + "execution_queue_depth 0", + "workers{state=\"idle\"} 0", + "worker_unexpected_exits_total 1", + "execution_rejections_total{reason=\"queue_full\"} 1", + "execution_rejections_total{reason=\"shutdown\"} 1", + "execution_queue_wait_seconds_count{outcome=\"dispatched\"} 1", + "execution_queue_wait_seconds_count{outcome=\"cancelled\"} 1", + "execution_queue_wait_seconds_count{outcome=\"shutdown\"} 1", + ] { + assert!( + output.contains(&format!("code_interpreter_mcp_{sample}")), + "{sample}: {output}" + ); + } + } + #[test] fn probes_ignore_queue_lock_and_saturation_and_shutdown_cancels_queue() { let (runtime, worker) = PyodideRuntime::test_pool();