diff --git a/Cargo.lock b/Cargo.lock index 308b24b..fa7a114 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -37,6 +37,7 @@ dependencies = [ "crank-test-support", "crank-trace", "metrics", + "metrics-util", "opentelemetry", "opentelemetry_sdk", "rand 0.10.2", @@ -654,9 +655,9 @@ dependencies = [ "async-trait", "axum", "crank-core", + "crank-metrics", "crank-trace", "futures-util", - "metrics", "opentelemetry", "opentelemetry_sdk", "reqwest 0.12.28", @@ -697,6 +698,7 @@ dependencies = [ "crank-adapter-rest", "crank-core", "crank-mapping", + "crank-metrics", "crank-observability", "crank-registry", "crank-runtime", @@ -704,7 +706,7 @@ dependencies = [ "crank-test-support", "crank-trace", "futures-util", - "metrics", + "metrics-util", "opentelemetry", "opentelemetry_sdk", "reqwest 0.12.28", @@ -758,11 +760,20 @@ dependencies = [ "thiserror 2.0.18", ] +[[package]] +name = "crank-metrics" +version = "0.3.1" +dependencies = [ + "metrics", + "metrics-util", +] + [[package]] name = "crank-observability" version = "0.3.1" dependencies = [ "axum", + "crank-metrics", "metrics", "metrics-exporter-prometheus", "opentelemetry", @@ -816,11 +827,13 @@ dependencies = [ "crank-adapter-rest", "crank-core", "crank-mapping", + "crank-metrics", "crank-schema", "crank-trace", "futures-util", "hkdf 0.12.4", "metrics", + "metrics-util", "redis", "serde", "serde_json", @@ -1075,6 +1088,12 @@ dependencies = [ "serde", ] +[[package]] +name = "endian-type" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c34f04666d835ff5d62e058c3995147c06f42fe86ff053337632bca83e42702d" + [[package]] name = "equivalent" version = "1.0.2" @@ -2009,6 +2028,7 @@ dependencies = [ "crank-test-support", "futures-util", "metrics", + "metrics-util", "opentelemetry", "opentelemetry-proto", "opentelemetry_sdk", @@ -2075,11 +2095,15 @@ version = "0.20.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96f8722f8562635f92f8ed992f26df0532266eb03d5202607c20c0d7e9745e13" dependencies = [ + "aho-corasick", "crossbeam-epoch", "crossbeam-utils", "hashbrown 0.16.1", + "indexmap 2.14.0", "metrics", + "ordered-float", "quanta", + "radix_trie", "rand 0.9.4", "rand_xoshiro", "rapidhash", @@ -2112,6 +2136,15 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "nibble_vec" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77a5d83df9f36fe23f0c3648c6bbb8b0298bb5f1939c8f2704431371f4b84d43" +dependencies = [ + "smallvec", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" @@ -2296,6 +2329,15 @@ dependencies = [ "thiserror 2.0.18", ] +[[package]] +name = "ordered-float" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7d950ca161dc355eaf28f82b11345ed76c6e1f6eb1f4f4479e0323b9e2fbd0e" +dependencies = [ + "num-traits", +] + [[package]] name = "parking" version = "2.2.1" @@ -2591,6 +2633,16 @@ version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "radix_trie" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c069c179fcdc6a2fe24d8d18305cf085fdbd4f922c041943e203685d6a1c58fd" +dependencies = [ + "endian-type", + "nibble_vec", +] + [[package]] name = "rand" version = "0.9.4" diff --git a/Cargo.toml b/Cargo.toml index e135715..d65b669 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,6 +8,7 @@ members = [ "crates/crank-import", "crates/crank-schema", "crates/crank-mapping", + "crates/crank-metrics", "crates/crank-observability", "crates/crank-registry", "crates/crank-runtime", diff --git a/apps/admin-api/Cargo.toml b/apps/admin-api/Cargo.toml index da0e1db..adbad03 100644 --- a/apps/admin-api/Cargo.toml +++ b/apps/admin-api/Cargo.toml @@ -24,7 +24,6 @@ crank-registry = { path = "../../crates/crank-registry" } crank-runtime = { path = "../../crates/crank-runtime" } crank-schema = { path = "../../crates/crank-schema" } crank-trace = { path = "../../crates/crank-trace" } -metrics.workspace = true rand.workspace = true serde.workspace = true serde_json.workspace = true @@ -41,6 +40,8 @@ uuid.workspace = true [dev-dependencies] async-trait = "0.1" crank-test-support = { path = "../../crates/crank-test-support" } +metrics.workspace = true +metrics-util = "0.20.4" opentelemetry.workspace = true opentelemetry_sdk.workspace = true reqwest.workspace = true diff --git a/apps/admin-api/src/lib.rs b/apps/admin-api/src/lib.rs index b556a1c..b68dce2 100644 --- a/apps/admin-api/src/lib.rs +++ b/apps/admin-api/src/lib.rs @@ -3,6 +3,7 @@ pub mod auth; pub mod dto; pub mod error; pub mod import_guidance; +pub mod pool_metrics; pub mod rate_limit; pub mod request_context; pub mod routes; diff --git a/apps/admin-api/src/main.rs b/apps/admin-api/src/main.rs index 92fa1bb..6dc9c42 100644 --- a/apps/admin-api/src/main.rs +++ b/apps/admin-api/src/main.rs @@ -3,6 +3,7 @@ use std::{env, net::SocketAddr, path::PathBuf, time::Duration}; use admin_api::{ app::build_app, auth::{AuthSettings, BootstrapAdminConfig}, + pool_metrics::spawn_postgres_pool_metrics, service::AdminServiceBuilder, state::AppState, }; @@ -16,7 +17,7 @@ use crank_runtime::{ RequestRateLimitConfig, RequestRateLimiter, RuntimeCacheConfig, RuntimeCacheStores, RuntimeLimits, SecretCrypto, }; -use sqlx::{PgPool, postgres::PgConnectOptions}; +use sqlx::postgres::PgConnectOptions; use tokio::net::TcpListener; use tracing::{info, warn}; @@ -196,16 +197,6 @@ fn spawn_invocation_log_cleanup(service: admin_api::service::AdminService, reten }); } -fn spawn_postgres_pool_metrics(pool: PgPool) { - tokio::spawn(async move { - let mut interval = tokio::time::interval(Duration::from_secs(5)); - loop { - interval.tick().await; - crank_observability::record_db_pool_connections(pool.size(), pool.num_idle()); - } - }); -} - fn env_flag(name: &str) -> bool { matches!( env::var(name) diff --git a/apps/admin-api/src/pool_metrics.rs b/apps/admin-api/src/pool_metrics.rs new file mode 100644 index 0000000..ba08ea8 --- /dev/null +++ b/apps/admin-api/src/pool_metrics.rs @@ -0,0 +1,13 @@ +use std::time::Duration; + +use sqlx::PgPool; + +pub fn spawn_postgres_pool_metrics(pool: PgPool) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + let mut interval = tokio::time::interval(Duration::from_secs(5)); + loop { + interval.tick().await; + crank_observability::record_db_pool_connections(pool.size(), pool.num_idle()); + } + }) +} diff --git a/apps/admin-api/tests/product_metrics.rs b/apps/admin-api/tests/product_metrics.rs new file mode 100644 index 0000000..cf8dd6f --- /dev/null +++ b/apps/admin-api/tests/product_metrics.rs @@ -0,0 +1,89 @@ +#[path = "integration/common.rs"] +mod common; + +use std::{collections::BTreeSet, time::Duration}; + +use common::*; +use metrics_util::debugging::DebuggingRecorder; + +#[tokio::test(flavor = "multi_thread")] +async fn real_admin_routes_and_postgres_sampler_emit_bounded_metrics() { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + recorder + .install() + .expect("isolated integration test recorder"); + + let registry = test_registry().await; + let sampler = admin_api::pool_metrics::spawn_postgres_pool_metrics(registry.pool().clone()); + let server = spawn_admin_api(build_test_app( + registry, + test_storage_root("product_metrics"), + )) + .await; + let client = authorized_client(&server).await; + client + .get(format!("{server}/operations")) + .send() + .await + .unwrap() + .error_for_status() + .unwrap(); + + let snapshot = wait_for_metrics(&snapshotter).await; + assert!(snapshot.iter().any(|(key, _, _, _)| { + key.key().name() == "crank_http_requests_total" + && has_label( + key.key(), + "route", + "/api/admin/workspaces/{workspace_id}/operations", + ) + && has_label(key.key(), "method", "GET") + && has_label(key.key(), "status_class", "2xx") + })); + let pool_states = snapshot + .iter() + .filter(|(key, _, _, _)| key.key().name() == "crank_db_pool_connections") + .filter_map(|(key, _, _, _)| { + key.key() + .labels() + .find(|label| label.key() == "state") + .map(|label| label.value().to_owned()) + }) + .collect::>(); + assert_eq!( + pool_states, + BTreeSet::from(["idle".to_owned(), "used".to_owned()]) + ); + + sampler.abort(); +} + +async fn wait_for_metrics( + snapshotter: &metrics_util::debugging::Snapshotter, +) -> Vec<( + metrics_util::CompositeKey, + Option, + Option, + metrics_util::debugging::DebugValue, +)> { + for _ in 0..50 { + let snapshot = snapshotter.snapshot().into_vec(); + let has_http = snapshot + .iter() + .any(|(key, _, _, _)| key.key().name() == "crank_http_requests_total"); + let has_pool = snapshot + .iter() + .any(|(key, _, _, _)| key.key().name() == "crank_db_pool_connections"); + if has_http && has_pool { + return snapshot; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + panic!("product metrics did not become visible"); +} + +fn has_label(key: &metrics::Key, name: &str, value: &str) -> bool { + key.labels() + .any(|label| label.key() == name && label.value() == value) +} diff --git a/apps/mcp-server/Cargo.toml b/apps/mcp-server/Cargo.toml index 048e8cf..55e5d3b 100644 --- a/apps/mcp-server/Cargo.toml +++ b/apps/mcp-server/Cargo.toml @@ -21,7 +21,6 @@ crank-registry = { path = "../../crates/crank-registry" } crank-runtime = { path = "../../crates/crank-runtime" } crank-schema = { path = "../../crates/crank-schema" } futures-util = "0.3" -metrics.workspace = true serde.workspace = true serde_json.workspace = true sha2.workspace = true @@ -37,6 +36,7 @@ uuid.workspace = true crank-mapping = { path = "../../crates/crank-mapping" } crank-schema = { path = "../../crates/crank-schema" } crank-test-support = { path = "../../crates/crank-test-support" } +metrics.workspace = true opentelemetry.workspace = true opentelemetry-proto.workspace = true opentelemetry_sdk.workspace = true @@ -44,3 +44,4 @@ prost.workspace = true reqwest.workspace = true tower.workspace = true tracing-opentelemetry.workspace = true +metrics-util = "0.20.4" diff --git a/apps/mcp-server/src/lib.rs b/apps/mcp-server/src/lib.rs new file mode 100644 index 0000000..eee4213 --- /dev/null +++ b/apps/mcp-server/src/lib.rs @@ -0,0 +1 @@ +pub mod pool_metrics; diff --git a/apps/mcp-server/src/main.rs b/apps/mcp-server/src/main.rs index 6ad3e48..f3ffb61 100644 --- a/apps/mcp-server/src/main.rs +++ b/apps/mcp-server/src/main.rs @@ -13,7 +13,8 @@ use crank_runtime::{ RequestRateLimitConfig, RequestRateLimiter, RuntimeCacheConfig, RuntimeCacheStores, RuntimeLimits, SecretCrypto, }; -use sqlx::{PgPool, postgres::PgConnectOptions}; +use mcp_server::pool_metrics::spawn_postgres_pool_metrics; +use sqlx::postgres::PgConnectOptions; use tokio::net::TcpListener; use tracing::info; @@ -162,13 +163,3 @@ fn mcp_api_rate_limit_config_from_env() -> Result tokio::task::JoinHandle<()> { + tokio::spawn(async move { + let mut interval = tokio::time::interval(Duration::from_secs(5)); + loop { + interval.tick().await; + crank_observability::record_db_pool_connections(pool.size(), pool.num_idle()); + } + }) +} diff --git a/apps/mcp-server/tests/product_metrics.rs b/apps/mcp-server/tests/product_metrics.rs new file mode 100644 index 0000000..be1858c --- /dev/null +++ b/apps/mcp-server/tests/product_metrics.rs @@ -0,0 +1,345 @@ +#[path = "integration/common.rs"] +mod common; + +use std::{collections::BTreeSet, time::Duration}; + +use common::*; +use crank_core::PlatformApiKeyScope; +use crank_registry::PublishRequest; +use metrics_util::debugging::{DebugValue, DebuggingRecorder, Snapshotter}; +use serde_json::json; +use time::{OffsetDateTime, format_description::well_known::Rfc3339}; + +#[tokio::test] +async fn real_mcp_runtime_upstream_postgres_and_catalog_paths_emit_bounded_metrics() { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + recorder + .install() + .expect("isolated integration test recorder"); + + let registry = test_registry().await; + let pool_sampler = + mcp_server::pool_metrics::spawn_postgres_pool_metrics(registry.pool().clone()); + let upstream_base_url = spawn_upstream_server().await; + let operation = test_operation(&upstream_base_url, "crm_create_lead"); + registry + .create_operation(&test_workspace_id(), &operation, Some("metrics-test")) + .await + .unwrap(); + registry + .publish_operation(PublishRequest { + workspace_id: &test_workspace_id(), + operation_id: &operation.id, + version: 1, + published_at: &OffsetDateTime::parse("2026-07-31T00:00:00Z", &Rfc3339).unwrap(), + published_by: Some("metrics-test"), + }) + .await + .unwrap(); + publish_agent_for_operation(®istry, &operation, "metrics-agent").await; + let api_key = create_platform_api_key( + ®istry, + "metrics-agent", + "metrics-client", + &[PlatformApiKeyScope::Read, PlatformApiKeyScope::Write], + ) + .await; + let mut varied_products = Vec::new(); + for index in 0..3 { + let upstream_url = spawn_upstream_server().await; + let operation_name = format!("customer_operation_{index}"); + let operation = test_operation(&upstream_url, &operation_name); + registry + .create_operation(&test_workspace_id(), &operation, Some("metrics-test")) + .await + .unwrap(); + registry + .publish_operation(PublishRequest { + workspace_id: &test_workspace_id(), + operation_id: &operation.id, + version: 1, + published_at: &OffsetDateTime::parse("2026-07-31T00:00:00Z", &Rfc3339).unwrap(), + published_by: Some("metrics-test"), + }) + .await + .unwrap(); + let agent_slug = format!("customer-agent-{index}"); + publish_agent_for_operation(®istry, &operation, &agent_slug).await; + let key = create_platform_api_key( + ®istry, + &agent_slug, + &format!("metrics-client-{index}"), + &[PlatformApiKeyScope::Read, PlatformApiKeyScope::Write], + ) + .await; + varied_products.push((agent_slug, key, operation_name, upstream_url)); + } + + let base_url = spawn_mcp_server(build_test_app( + registry.clone(), + Duration::ZERO, + Some("https://crank.example.com".to_owned()), + )) + .await; + let client = reqwest::Client::new(); + let mcp_url = agent_mcp_url(&base_url, "metrics-agent"); + let malformed = client + .post(&mcp_url) + .header(reqwest::header::ACCEPT, "application/json") + .header(reqwest::header::AUTHORIZATION, format!("Bearer {api_key}")) + .header(reqwest::header::CONTENT_TYPE, "application/json") + .body("{") + .send() + .await + .unwrap(); + assert_eq!(malformed.status(), reqwest::StatusCode::BAD_REQUEST); + + let invalid_session_header = client + .post(&mcp_url) + .header( + reqwest::header::ACCEPT, + "application/json, text/event-stream", + ) + .header(reqwest::header::AUTHORIZATION, format!("Bearer {api_key}")) + .header( + "MCP-Session-Id", + reqwest::header::HeaderValue::from_bytes(b"\xff").unwrap(), + ) + .json(&json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": {"protocolVersion": "2025-11-25", "capabilities": {}} + })) + .send() + .await + .unwrap(); + assert_eq!( + invalid_session_header.status(), + reqwest::StatusCode::BAD_REQUEST + ); + + let session_id = initialize_session(&client, &mcp_url, &api_key).await; + assert_gauge(&snapshotter, "crank_mcp_active_sessions", 1.0).await; + let _ = post_jsonrpc( + &client, + &mcp_url, + &api_key, + Some(&session_id), + json!({"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}), + ) + .await; + call_tool( + &client, + &mcp_url, + &api_key, + &session_id, + "crm_create_lead", + "first@example.com", + "req-first", + ) + .await; + delete_session(&client, &mcp_url, &api_key, &session_id).await; + assert_gauge(&snapshotter, "crank_mcp_active_sessions", 0.0).await; + let session_id = initialize_session(&client, &mcp_url, &api_key).await; + assert_gauge(&snapshotter, "crank_mcp_active_sessions", 1.0).await; + for (index, (agent_slug, key, operation_name, _)) in varied_products.iter().enumerate() { + let url = agent_mcp_url(&base_url, agent_slug); + let session = initialize_session(&client, &url, key).await; + call_tool( + &client, + &url, + key, + &session, + operation_name, + &format!("variant-{index}@example.com"), + &format!("req-product-{index}"), + ) + .await; + delete_session(&client, &url, key, &session).await; + } + send_invalid_workspace_request(&client, &base_url, &api_key, "customer-workspace-seed").await; + tokio::time::sleep(Duration::from_millis(25)).await; + + let first_snapshot = snapshotter.snapshot().into_vec(); + let first_series = series(&first_snapshot); + assert!(first_snapshot.iter().any(|(key, _, _, _)| { + key.key().name() == "crank_mcp_requests_total" + && has_label(key.key(), "method", "invalid") + && has_label(key.key(), "outcome", "client_error") + })); + let names = first_snapshot + .iter() + .map(|(key, _, _, _)| key.key().name()) + .collect::>(); + for expected in [ + "crank_mcp_requests_total", + "crank_mcp_active_sessions", + "crank_tool_invocations_total", + "crank_tool_invocation_duration_seconds", + "crank_upstream_requests_total", + "crank_upstream_request_duration_seconds", + "crank_db_pool_connections", + "crank_catalog_tools", + "crank_catalog_estimated_context_tokens", + "crank_catalog_warnings", + ] { + assert!( + names.contains(expected), + "missing product metric {expected}" + ); + } + + for index in 0..32 { + call_tool( + &client, + &mcp_url, + &api_key, + &session_id, + "crm_create_lead", + &format!("user-{index}@example.com"), + &format!("req-metrics-{index}"), + ) + .await; + send_invalid_workspace_request( + &client, + &base_url, + &api_key, + &format!("customer-workspace-{index}"), + ) + .await; + } + tokio::time::sleep(Duration::from_millis(25)).await; + + let diverse_snapshot = snapshotter.snapshot().into_vec(); + assert_eq!(series(&diverse_snapshot), first_series); + delete_session(&client, &mcp_url, &api_key, &session_id).await; + assert_gauge(&snapshotter, "crank_mcp_active_sessions", 0.0).await; + let rendered = format!("{diverse_snapshot:?}"); + let mut forbidden = vec![ + "metrics-agent", + "crm_create_lead", + "user-31@example.com", + "req-metrics-31", + &upstream_base_url, + ]; + for (agent_slug, _, operation_name, upstream_url) in &varied_products { + forbidden.extend([ + agent_slug.as_str(), + operation_name.as_str(), + upstream_url.as_str(), + ]); + } + for forbidden in forbidden { + assert!( + !rendered.contains(forbidden), + "untrusted value leaked into metric labels: {forbidden}" + ); + } + pool_sampler.abort(); +} + +async fn delete_session(client: &reqwest::Client, mcp_url: &str, api_key: &str, session_id: &str) { + let deleted = client + .delete(mcp_url) + .header(reqwest::header::AUTHORIZATION, format!("Bearer {api_key}")) + .header("MCP-Session-Id", session_id) + .send() + .await + .unwrap(); + assert_eq!(deleted.status(), reqwest::StatusCode::NO_CONTENT); +} + +async fn call_tool( + client: &reqwest::Client, + mcp_url: &str, + api_key: &str, + session_id: &str, + tool_name: &str, + email: &str, + request_id: &str, +) { + let response = post_jsonrpc_response( + client, + mcp_url, + api_key, + Some(session_id), + Some(request_id), + json!({ + "jsonrpc": "2.0", + "id": request_id, + "method": "tools/call", + "params": { + "name": tool_name, + "arguments": {"email": email} + } + }), + ) + .await; + assert_eq!(response.status(), reqwest::StatusCode::OK); +} + +async fn send_invalid_workspace_request( + client: &reqwest::Client, + base_url: &str, + api_key: &str, + workspace_slug: &str, +) { + let response = client + .post(format!("{base_url}/v1/{workspace_slug}/metrics-agent")) + .header(reqwest::header::ACCEPT, "application/json") + .header(reqwest::header::AUTHORIZATION, format!("Bearer {api_key}")) + .json(&json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": {"protocolVersion": "2025-11-25", "capabilities": {}} + })) + .send() + .await + .unwrap(); + assert!(response.status().is_client_error()); +} + +fn series( + snapshot: &[( + metrics_util::CompositeKey, + Option, + Option, + metrics_util::debugging::DebugValue, + )], +) -> BTreeSet { + snapshot + .iter() + .map(|(key, _, _, _)| format!("{key:?}")) + .collect() +} + +fn has_label(key: &metrics::Key, name: &str, value: &str) -> bool { + key.labels() + .any(|label| label.key() == name && label.value() == value) +} + +async fn assert_gauge(snapshotter: &Snapshotter, name: &str, expected: f64) { + for _ in 0..50 { + let value = snapshotter + .snapshot() + .into_vec() + .into_iter() + .find_map(|(key, _, _, value)| { + if key.key().name() != name { + return None; + } + match value { + DebugValue::Gauge(value) => Some(value.into_inner()), + _ => None, + } + }); + if value == Some(expected) { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + panic!("{name} did not become {expected}"); +} diff --git a/crates/crank-adapter-rest/Cargo.toml b/crates/crank-adapter-rest/Cargo.toml index d3ef6ec..e94a956 100644 --- a/crates/crank-adapter-rest/Cargo.toml +++ b/crates/crank-adapter-rest/Cargo.toml @@ -9,9 +9,9 @@ version.workspace = true [dependencies] async-trait = "0.1" crank-core = { path = "../crank-core" } +crank-metrics = { path = "../crank-metrics" } crank-trace = { path = "../crank-trace" } futures-util = "0.3" -metrics.workspace = true opentelemetry.workspace = true reqwest = { workspace = true, features = ["stream"] } serde.workspace = true diff --git a/crates/crank-adapter-rest/src/client.rs b/crates/crank-adapter-rest/src/client.rs index 7234246..30d9bf4 100644 --- a/crates/crank-adapter-rest/src/client.rs +++ b/crates/crank-adapter-rest/src/client.rs @@ -7,6 +7,7 @@ use std::{ }; use crank_core::{HttpMethod, RestTarget}; +use crank_metrics::{UpstreamOperationKind, UpstreamOutcome, UpstreamRequestMetrics}; use crank_trace::{ErrorCategory, Stage, StageOutcome}; use futures_util::StreamExt; use opentelemetry::{global, propagation::Injector, trace::TraceContextExt}; @@ -71,24 +72,13 @@ impl RestAdapter { target: &RestTarget, request: &RestRequest, ) -> Result { - let started_at = std::time::Instant::now(); + let request_metrics = UpstreamRequestMetrics::start(UpstreamOperationKind::Rest); let result = self.execute_inner(target, request).await; let outcome = match &result { - Ok(_) => "success", + Ok(_) => UpstreamOutcome::Success, Err(error) => upstream_outcome(error), }; - metrics::counter!( - "crank_upstream_requests_total", - "operation_kind" => "rest", - "outcome" => outcome - ) - .increment(1); - metrics::histogram!( - "crank_upstream_request_duration_seconds", - "operation_kind" => "rest", - "outcome" => outcome - ) - .record(started_at.elapsed().as_secs_f64()); + request_metrics.complete(outcome); result } @@ -149,27 +139,27 @@ impl RestAdapter { } } -fn upstream_outcome(error: &RestAdapterError) -> &'static str { +fn upstream_outcome(error: &RestAdapterError) -> UpstreamOutcome { match error { RestAdapterError::UnexpectedStatus { status, .. } if (400..500).contains(status) => { - "client_error" + UpstreamOutcome::ClientError } RestAdapterError::UnexpectedStatus { status, .. } if (500..600).contains(status) => { - "server_error" + UpstreamOutcome::ServerError } - RestAdapterError::UnexpectedStatus { .. } => "unexpected_status", - RestAdapterError::Transport(error) if error.is_timeout() => "timeout", - RestAdapterError::Transport(_) => "transport_error", - RestAdapterError::ResponseTooLarge { .. } => "response_too_large", - RestAdapterError::TargetNotAllowed { .. } => "rejected", - RestAdapterError::WindowExpired => "window_expired", - RestAdapterError::InvalidSseEvent => "invalid_response", + RestAdapterError::UnexpectedStatus { .. } => UpstreamOutcome::UnexpectedStatus, + RestAdapterError::Transport(error) if error.is_timeout() => UpstreamOutcome::Timeout, + RestAdapterError::Transport(_) => UpstreamOutcome::TransportError, + RestAdapterError::ResponseTooLarge { .. } => UpstreamOutcome::ResponseTooLarge, + RestAdapterError::TargetNotAllowed { .. } => UpstreamOutcome::Rejected, + RestAdapterError::WindowExpired => UpstreamOutcome::WindowExpired, + RestAdapterError::InvalidSseEvent => UpstreamOutcome::InvalidResponse, RestAdapterError::InvalidBaseUrl { .. } | RestAdapterError::InvalidPathParameter { .. } | RestAdapterError::InvalidQueryParameter { .. } | RestAdapterError::InvalidHeaderName { .. } - | RestAdapterError::InvalidHeaderValue { .. } => "invalid_request", - RestAdapterError::InvalidConfiguration { .. } => "configuration", + | RestAdapterError::InvalidHeaderValue { .. } => UpstreamOutcome::InvalidRequest, + RestAdapterError::InvalidConfiguration { .. } => UpstreamOutcome::Configuration, } } diff --git a/crates/crank-community-mcp/Cargo.toml b/crates/crank-community-mcp/Cargo.toml index bcf11f3..3df0963 100644 --- a/crates/crank-community-mcp/Cargo.toml +++ b/crates/crank-community-mcp/Cargo.toml @@ -12,13 +12,13 @@ axum.workspace = true base64.workspace = true crank-adapter-rest = { path = "../crank-adapter-rest" } crank-core = { path = "../crank-core" } +crank-metrics = { path = "../crank-metrics" } crank-observability = { path = "../crank-observability" } crank-registry = { path = "../crank-registry" } crank-runtime = { path = "../crank-runtime" } crank-schema = { path = "../crank-schema" } crank-trace = { path = "../crank-trace" } futures-util = "0.3" -metrics.workspace = true reqwest.workspace = true serde.workspace = true serde_json.workspace = true @@ -33,6 +33,7 @@ uuid.workspace = true [dev-dependencies] crank-mapping = { path = "../crank-mapping" } crank-test-support = { path = "../crank-test-support" } +metrics-util = "0.20.4" opentelemetry.workspace = true opentelemetry_sdk.workspace = true tracing-opentelemetry.workspace = true diff --git a/crates/crank-community-mcp/src/app.rs b/crates/crank-community-mcp/src/app.rs index ea56b63..750f2cf 100644 --- a/crates/crank-community-mcp/src/app.rs +++ b/crates/crank-community-mcp/src/app.rs @@ -7,7 +7,7 @@ use std::{ use axum::{ Json, Router, - extract::{Extension, Path, State}, + extract::{Extension, Path, State, rejection::JsonRejection}, http::{HeaderMap, StatusCode}, response::{IntoResponse, Response, sse::Event}, routing::{get, post}, @@ -31,7 +31,7 @@ use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use time::OffsetDateTime; use tokio::sync::Semaphore; -use tracing::{Instrument, info, warn}; +use tracing::{Instrument, info}; use crate::{ access::{ @@ -48,23 +48,23 @@ use crate::{ manifest::catalog_tool_definitions, rate_limit::{rate_limited_jsonrpc_response, rate_limited_status_response}, request_context::{RequestContext, apply_request_context}, - session::{SessionState, SharedSessionStore}, + session::{ActiveSessionMetrics, SessionState, SharedSessionStore, spawn_session_cleanup}, tool_error::{ ToolErrorContract, generic_tool_error_contract, runtime_error_code, tool_error_contract_from_runtime, tool_error_text, tool_error_value, }, tool_search::handle_catalog_tool_call, transport::{ - AllowedOrigins, HEADER_MCP_SESSION_ID, ResponseMode, json_response, - negotiate_post_response_mode, protocol_version_from_headers, session_id_from_headers, - sse_response, transport_response, validate_get_accept_header, validate_origin, - validate_session_protocol_version, with_request_id_header, + AllowedOrigins, ResponseMode, json_response, negotiate_post_response_mode, + protocol_version_from_headers, session_id_from_headers, sse_response, transport_response, + validate_get_accept_header, validate_origin, validate_session_protocol_version, + with_request_id_header, }, }; mod invocation_history; mod metrics; mod stages; -use self::metrics::{ActiveSessionGuard, McpRequestMetrics}; +use self::metrics::{ActiveStreamGuard, McpRequestMetrics}; use self::stages::{ enforce_traced_rate_limit, require_traced_approval_access, require_traced_machine_access, }; @@ -84,6 +84,7 @@ pub(super) struct AppState { pub(super) api_rate_limiter: RequestRateLimiter, secret_crypto: SecretCrypto, sessions: SharedSessionStore, + session_metrics: ActiveSessionMetrics, session_slots: Arc, pub(super) credential_verifier: SharedMachineCredentialVerifier, allowed_origins: AllowedOrigins, @@ -230,6 +231,7 @@ fn build_app_inner( max_concurrent_sessions: usize, start_background_workers: bool, ) -> Router { + let session_metrics = ActiveSessionMetrics::start(Arc::clone(&sessions)); let state = Arc::new(AppState { registry: registry.clone(), catalog: PublishedToolCatalog::new(registry, refresh_interval, coordination_store), @@ -237,13 +239,18 @@ fn build_app_inner( api_rate_limiter, secret_crypto, sessions, + session_metrics, session_slots: Arc::new(Semaphore::new(max_concurrent_sessions)), credential_verifier, allowed_origins: AllowedOrigins::new(public_base_url), }); if start_background_workers { spawn_approval_recovery(Arc::clone(&state)); - spawn_session_cleanup(Arc::clone(&state)); + spawn_session_cleanup( + Arc::clone(&state.sessions), + state.session_metrics.clone(), + SESSION_CLEANUP_INTERVAL, + ); } Router::new() @@ -283,28 +290,6 @@ async fn health() -> Json { })) } -fn spawn_session_cleanup(state: Arc) { - tokio::spawn(async move { - let mut interval = tokio::time::interval(SESSION_CLEANUP_INTERVAL); - loop { - interval.tick().await; - match state - .sessions - .cleanup_expired(OffsetDateTime::now_utc()) - .await - { - Ok(removed) if removed > 0 => { - info!(name: "mcp.session_cleanup.completed", removed); - } - Ok(_) => {} - Err(_) => { - warn!(name: "mcp.session_cleanup.failed", error_category = "session_store"); - } - } - } - }); -} - async fn readiness(State(state): State>) -> Response { match state.registry.ping().await { Ok(()) => Json(json!({ @@ -664,7 +649,7 @@ async fn mcp_get( return status.into_response(); } - let Ok(permit) = ActiveSessionGuard::try_acquire(&state.session_slots) else { + let Ok(permit) = ActiveStreamGuard::try_acquire(&state.session_slots) else { return StatusCode::TOO_MANY_REQUESTS.into_response(); }; @@ -704,7 +689,10 @@ async fn mcp_delete( && session.agent_slug == path.agent_slug => { match state.sessions.delete(&session_id).await { - Ok(true) => StatusCode::NO_CONTENT.into_response(), + Ok(true) => { + state.session_metrics.refresh(); + StatusCode::NO_CONTENT.into_response() + } Ok(false) => StatusCode::NOT_FOUND.into_response(), Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), } @@ -723,8 +711,17 @@ async fn mcp_post( State(state): State>, Extension(request_context): Extension, headers: HeaderMap, - Json(message): Json, + payload: Result, JsonRejection>, ) -> Response { + let message = match payload { + Ok(Json(message)) => message, + Err(rejection) => { + let mut request_metrics = McpRequestMetrics::invalid(); + let response = rejection.into_response(); + request_metrics.complete(&response); + return with_request_id_header(response, &request_context.request_id); + } + }; let mut request_metrics = McpRequestMetrics::new(&message); let transport_request_id = request_context.request_id; info!( @@ -736,78 +733,77 @@ async fn mcp_post( "mcp request received" ); - if let Err(status) = validate_origin(&state.allowed_origins, &headers) { - return with_request_id_header(status.into_response(), &transport_request_id); + let response = mcp_post_response(&path, state, &headers, &message, &transport_request_id).await; + request_metrics.complete(&response); + with_request_id_header(response, &transport_request_id) +} + +async fn mcp_post_response( + path: &AgentRoutePath, + state: Arc, + headers: &HeaderMap, + message: &Value, + transport_request_id: &str, +) -> Response { + if let Err(status) = validate_origin(&state.allowed_origins, headers) { + return status.into_response(); } - let response_mode = match negotiate_post_response_mode(&headers) { - Ok(mode) => request_metrics.set_response_mode(mode), - Err(status) => { - return with_request_id_header(status.into_response(), &transport_request_id); - } + let response_mode = match negotiate_post_response_mode(headers) { + Ok(mode) => mode, + Err(status) => return status.into_response(), }; - if is_response(&message) || is_notification(&message) && method_name(&message).is_none() { - return with_request_id_header(StatusCode::ACCEPTED.into_response(), &transport_request_id); + let request_session_id = match session_id_from_headers(headers) { + Ok(session_id) => session_id, + Err(status) => return status.into_response(), + }; + + if is_response(message) || is_notification(message) && method_name(message).is_none() { + return StatusCode::ACCEPTED.into_response(); } - let protocol_version = match protocol_version_from_headers(&headers) { + let protocol_version = match protocol_version_from_headers(headers) { Ok(value) => value, - Err(status) => { - return with_request_id_header(status.into_response(), &transport_request_id); - } + Err(status) => return status.into_response(), }; - let rate_limit_result = enforce_traced_rate_limit(&state, &path, &headers).await; + let rate_limit_result = enforce_traced_rate_limit(&state, path, headers).await; if let Err(error) = rate_limit_result { - return with_request_id_header( - rate_limited_jsonrpc_response(&message, response_mode, &protocol_version, error), - &transport_request_id, - ); + return rate_limited_jsonrpc_response(message, response_mode, &protocol_version, error); } - if let Some(session_id) = headers.get(HEADER_MCP_SESSION_ID) - && let Ok(session_id) = session_id.to_str() - { + if let Some(session_id) = request_session_id.as_deref() { let session = match state.sessions.get(session_id).await { Ok(session) => session, - Err(_) => { - return with_request_id_header( - StatusCode::INTERNAL_SERVER_ERROR.into_response(), - &transport_request_id, - ); - } + Err(_) => return StatusCode::INTERNAL_SERVER_ERROR.into_response(), }; if let Some(session) = session && let Err(status) = - validate_session_protocol_version(&headers, &session.protocol_version) + validate_session_protocol_version(headers, &session.protocol_version) { - return with_request_id_header(status.into_response(), &transport_request_id); + return status.into_response(); } } - let required_scope = match method_name(&message) { + let required_scope = match method_name(message) { Some("tools/call") => PlatformApiKeyScope::Write, _ => PlatformApiKeyScope::Read, }; - let access_result = - require_traced_machine_access(&state, &path, &headers, required_scope).await; + let access_result = require_traced_machine_access(&state, path, headers, required_scope).await; let credential = match access_result { Ok(credential) => credential, - Err(error) => { - return with_request_id_header(error.into_response(), &transport_request_id); - } + Err(error) => return error.into_response(), }; - let response = match method_name(&message) { - Some("initialize") if is_request(&message) => { - handle_initialize(state, &path, &message, response_mode).await + match method_name(message) { + Some("initialize") if is_request(message) => { + handle_initialize(state, path, message, response_mode).await } - Some("notifications/initialized") if is_notification(&message) => { - handle_initialized_notification(state, &path, &headers).await + Some("notifications/initialized") if is_notification(message) => { + handle_initialized_notification(state, path, headers).await } - Some("ping") if is_request(&message) => { - let session = match require_initialized_session(&state, &path, &headers, &message).await - { + Some("ping") if is_request(message) => { + let session = match require_initialized_session(&state, path, headers, message).await { Ok(session) => session, Err(response) => return response, }; @@ -815,7 +811,7 @@ async fn mcp_post( transport_response( StatusCode::OK, jsonrpc_result( - request_id(&message), + request_id(message), json!({ "protocolVersion": session.protocol_version }), ), response_mode, @@ -823,9 +819,8 @@ async fn mcp_post( Some(&session.protocol_version), ) } - Some("tools/list") if is_request(&message) => { - let session = match require_initialized_session(&state, &path, &headers, &message).await - { + Some("tools/list") if is_request(message) => { + let session = match require_initialized_session(&state, path, headers, message).await { Ok(session) => session, Err(response) => return response, }; @@ -840,27 +835,26 @@ async fn mcp_post( transport_response( StatusCode::OK, - jsonrpc_result(request_id(&message), json!({ "tools": definitions })), + jsonrpc_result(request_id(message), json!({ "tools": definitions })), response_mode, None, Some(&session.protocol_version), ) } - Err(error) => internal_jsonrpc_error(&message, error), + Err(error) => internal_jsonrpc_error(message, error), } } - Some("tools/call") if is_request(&message) => { - let session = match require_initialized_session(&state, &path, &headers, &message).await - { + Some("tools/call") if is_request(message) => { + let session = match require_initialized_session(&state, path, headers, message).await { Ok(session) => session, Err(response) => return response, }; - let tool_call_params: ToolCallParams = match serde_json::from_value(params(&message)) { + let tool_call_params: ToolCallParams = match serde_json::from_value(params(message)) { Ok(value) => value, Err(error) => { return transport_response( StatusCode::OK, - jsonrpc_error(request_id(&message), -32602, error.to_string()), + jsonrpc_error(request_id(message), -32602, error.to_string()), response_mode, None, Some(&session.protocol_version), @@ -883,27 +877,27 @@ async fn mcp_post( handle_catalog_tool_call( state.clone(), &session, - &message, + message, response_mode, &credential, &catalog, &tool_call_params.name, arguments, - &transport_request_id, + transport_request_id, ) .await } - Err(error) => internal_jsonrpc_error(&message, error), + Err(error) => internal_jsonrpc_error(message, error), } } - Some(method) if is_notification(&message) => { + Some(method) if is_notification(message) => { let _ = method; StatusCode::ACCEPTED.into_response() } Some(method) => transport_response( StatusCode::OK, jsonrpc_error( - request_id(&message), + request_id(message), -32601, format!("method {method} is not supported"), ), @@ -918,10 +912,7 @@ async fn mcp_post( None, Some(&protocol_version), ), - }; - - request_metrics.complete(response.status()); - with_request_id_header(response, &transport_request_id) + } } #[allow(clippy::too_many_arguments)] @@ -1432,6 +1423,7 @@ async fn handle_initialize( Ok(session_id) => session_id, Err(error) => return internal_jsonrpc_error(message, error), }; + state.session_metrics.refresh(); transport_response( StatusCode::OK, diff --git a/crates/crank-community-mcp/src/app/metrics.rs b/crates/crank-community-mcp/src/app/metrics.rs index e50f129..8e3d014 100644 --- a/crates/crank-community-mcp/src/app/metrics.rs +++ b/crates/crank-community-mcp/src/app/metrics.rs @@ -1,93 +1,107 @@ use std::sync::Arc; -use axum::http::StatusCode; +use axum::{http::header::CONTENT_TYPE, response::Response}; +use crank_metrics::{ + InFlightGuard, LimitStage, McpMethod, McpOutcome, McpResponseMode, record_limit_rejection, + record_mcp_request, +}; use serde_json::Value; use tokio::sync::{OwnedSemaphorePermit, Semaphore}; -use crate::{ - jsonrpc::{is_notification, is_response, method_name}, - transport::ResponseMode, -}; +use crate::jsonrpc::{is_notification, is_response, method_name}; pub(super) struct McpRequestMetrics { - method: &'static str, - response_mode: &'static str, - outcome: &'static str, + method: McpMethod, + response_mode: McpResponseMode, + outcome: McpOutcome, } impl McpRequestMetrics { - pub(super) fn new(message: &Value) -> Self { + pub(super) const fn invalid() -> Self { Self { - method: normalized_mcp_method(message), - response_mode: "unknown", - outcome: "rejected", + method: McpMethod::Invalid, + response_mode: McpResponseMode::Unknown, + outcome: McpOutcome::Aborted, } } - pub(super) fn set_response_mode(&mut self, mode: ResponseMode) -> ResponseMode { - self.response_mode = match mode { - ResponseMode::Json => "json", - ResponseMode::Sse => "sse", - }; - mode + pub(super) fn new(message: &Value) -> Self { + Self { + method: normalized_mcp_method(message), + response_mode: McpResponseMode::Unknown, + outcome: McpOutcome::Aborted, + } } - pub(super) fn complete(&mut self, status: StatusCode) { - self.outcome = match status.as_u16() { - 200..=299 => "success", - 400..=499 => "client_error", - 500..=599 => "server_error", - _ => "other", + pub(super) fn complete(&mut self, response: &Response) { + self.response_mode = response + .headers() + .get(CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .map_or(McpResponseMode::Unknown, |content_type| { + if content_type.starts_with("application/json") { + McpResponseMode::Json + } else if content_type.starts_with("text/event-stream") { + McpResponseMode::Sse + } else { + McpResponseMode::Unknown + } + }); + self.outcome = if response.status().is_success() { + response + .extensions() + .get::() + .copied() + .unwrap_or_else(|| McpOutcome::from_http_status(response.status().as_u16())) + } else { + McpOutcome::from_http_status(response.status().as_u16()) }; } + + #[cfg(test)] + pub(super) const fn outcome(&self) -> McpOutcome { + self.outcome + } + + #[cfg(test)] + pub(super) const fn response_mode(&self) -> McpResponseMode { + self.response_mode + } } impl Drop for McpRequestMetrics { fn drop(&mut self) { - ::metrics::counter!( - "crank_mcp_requests_total", - "method" => self.method, - "response_mode" => self.response_mode, - "outcome" => self.outcome - ) - .increment(1); + record_mcp_request(self.method, self.response_mode, self.outcome); } } -pub(super) fn normalized_mcp_method(message: &Value) -> &'static str { +pub(super) fn normalized_mcp_method(message: &Value) -> McpMethod { match method_name(message) { - Some("initialize") => "initialize", - Some("notifications/initialized") => "initialized", - Some("ping") => "ping", - Some("tools/list") => "tools_list", - Some("tools/call") => "tools_call", - Some(_) if is_notification(message) => "notification", - Some(_) => "unsupported", - None if is_response(message) => "response", - None => "invalid", + Some("initialize") => McpMethod::Initialize, + Some("notifications/initialized") => McpMethod::Initialized, + Some("ping") => McpMethod::Ping, + Some("tools/list") => McpMethod::ToolsList, + Some("tools/call") => McpMethod::ToolsCall, + Some(_) if is_notification(message) => McpMethod::Notification, + Some(_) => McpMethod::Unsupported, + None if is_response(message) => McpMethod::Response, + None => McpMethod::Invalid, } } -pub(super) struct ActiveSessionGuard { +pub(super) struct ActiveStreamGuard { _permit: OwnedSemaphorePermit, + _inflight: InFlightGuard, } -impl ActiveSessionGuard { +impl ActiveStreamGuard { pub(super) fn try_acquire(slots: &Arc) -> Result { let permit = Arc::clone(slots).try_acquire_owned().map_err(|_| { - ::metrics::counter!( - "crank_runtime_limit_rejections_total", - "stage" => "mcp_session" - ) - .increment(1); + record_limit_rejection(LimitStage::McpStream); })?; - ::metrics::gauge!("crank_mcp_active_sessions").increment(1.0); - Ok(Self { _permit: permit }) - } -} - -impl Drop for ActiveSessionGuard { - fn drop(&mut self) { - ::metrics::gauge!("crank_mcp_active_sessions").decrement(1.0); + Ok(Self { + _permit: permit, + _inflight: InFlightGuard::mcp_stream(), + }) } } diff --git a/crates/crank-community-mcp/src/app/tests.rs b/crates/crank-community-mcp/src/app/tests.rs index cb30c63..1c263c2 100644 --- a/crates/crank-community-mcp/src/app/tests.rs +++ b/crates/crank-community-mcp/src/app/tests.rs @@ -5,6 +5,7 @@ use std::{ use axum::body::to_bytes; use crank_core::InvocationStatus; +use crank_metrics::{McpMethod, McpOutcome, McpResponseMode}; use crank_observability::{ ObservabilityConfig, OperationalIncident, RedactionLimits, ServiceIdentity, operational_incident_total, @@ -16,16 +17,20 @@ use serde_json::{Value, json}; use tracing_subscriber::fmt::MakeWriter; use super::{ - ResponseMode, metrics::normalized_mcp_method, observe_invocation_history_outcome, - tool_error_response, + ResponseMode, + metrics::{McpRequestMetrics, normalized_mcp_method}, + observe_invocation_history_outcome, tool_error_response, }; -use crate::jsonrpc::CURRENT_PROTOCOL_VERSION; +use crate::jsonrpc::{CURRENT_PROTOCOL_VERSION, jsonrpc_error}; use crate::tool_error::generic_tool_error_contract; +use crate::transport::transport_response; #[tokio::test] async fn tool_error_response_includes_structured_context() { + let message = json!({"jsonrpc": "2.0", "id": "req-1", "method": "tools/call"}); + let mut request_metrics = McpRequestMetrics::new(&message); let response = tool_error_response( - &json!({"jsonrpc": "2.0", "id": "req-1"}), + &message, ResponseMode::Json, CURRENT_PROTOCOL_VERSION, generic_tool_error_contract( @@ -36,6 +41,12 @@ async fn tool_error_response_includes_structured_context() { Some("Проверьте параметры вызова инструмента."), ), ); + assert_eq!( + response.extensions().get::(), + Some(&McpOutcome::ToolError) + ); + request_metrics.complete(&response); + assert_eq!(request_metrics.outcome(), McpOutcome::ToolError); let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); let payload: Value = serde_json::from_slice(&body).unwrap(); @@ -53,6 +64,67 @@ async fn tool_error_response_includes_structured_context() { ); } +#[test] +fn jsonrpc_error_over_http_200_is_not_counted_as_success() { + let message = json!({"jsonrpc": "2.0", "id": 1, "method": "unsupported"}); + let mut request_metrics = McpRequestMetrics::new(&message); + let response = transport_response( + axum::http::StatusCode::OK, + jsonrpc_error(json!(1), -32601, "unsupported"), + ResponseMode::Json, + None, + Some(CURRENT_PROTOCOL_VERSION), + ); + + request_metrics.complete(&response); + + assert_eq!(request_metrics.outcome(), McpOutcome::JsonRpcError); +} + +#[test] +fn metric_uses_actual_json_response_when_sse_request_falls_back() { + let message = json!({"jsonrpc": "2.0", "id": 1, "method": "unsupported"}); + let mut request_metrics = McpRequestMetrics::new(&message); + let response = transport_response( + axum::http::StatusCode::INTERNAL_SERVER_ERROR, + jsonrpc_error(json!(1), -32603, "internal error"), + ResponseMode::Json, + None, + Some(CURRENT_PROTOCOL_VERSION), + ); + + request_metrics.complete(&response); + + assert_eq!(request_metrics.response_mode(), McpResponseMode::Json); + assert_eq!(request_metrics.outcome(), McpOutcome::ServerError); +} + +#[test] +fn transport_failure_takes_priority_over_jsonrpc_payload() { + let message = json!({"jsonrpc": "2.0", "id": 1, "method": "tools/call"}); + let mut request_metrics = McpRequestMetrics::new(&message); + let response = transport_response( + axum::http::StatusCode::TOO_MANY_REQUESTS, + jsonrpc_error(json!(1), -32000, "rate limited"), + ResponseMode::Json, + None, + Some(CURRENT_PROTOCOL_VERSION), + ); + + request_metrics.complete(&response); + + assert_eq!(request_metrics.outcome(), McpOutcome::ClientError); +} + +#[test] +fn unfinished_request_is_classified_as_aborted() { + let message = json!({"jsonrpc": "2.0", "id": 1, "method": "tools/call"}); + let request_metrics = McpRequestMetrics::new(&message); + + assert_eq!(request_metrics.response_mode(), McpResponseMode::Unknown); + assert_eq!(request_metrics.outcome(), McpOutcome::Aborted); +} + #[test] fn emits_bounded_history_loss_incident() { let writer = SharedLogWriter::default(); @@ -96,19 +168,19 @@ fn emits_bounded_history_loss_incident() { fn mcp_metric_method_is_always_from_a_closed_set() { assert_eq!( normalized_mcp_method(&json!({"jsonrpc": "2.0", "id": 1, "method": "tools/call"})), - "tools_call" + McpMethod::ToolsCall ); assert_eq!( normalized_mcp_method( &json!({"jsonrpc": "2.0", "id": 2, "method": "customer-controlled-method"}) ), - "unsupported" + McpMethod::Unsupported ); assert_eq!( normalized_mcp_method( &json!({"jsonrpc": "2.0", "method": "customer-controlled-notification"}) ), - "notification" + McpMethod::Notification ); } diff --git a/crates/crank-community-mcp/src/catalog.rs b/crates/crank-community-mcp/src/catalog.rs index 3346e39..1299bf5 100644 --- a/crates/crank-community-mcp/src/catalog.rs +++ b/crates/crank-community-mcp/src/catalog.rs @@ -339,10 +339,11 @@ fn record_catalog_metrics(metrics: impl Iterator) { aggregate }); - metrics::gauge!("crank_catalog_tools").set(aggregate.tool_count as f64); - metrics::gauge!("crank_catalog_estimated_context_tokens") - .set(aggregate.estimated_context_tokens as f64); - metrics::gauge!("crank_catalog_warnings").set(aggregate.warning_count as f64); + crank_metrics::set_catalog( + aggregate.tool_count, + aggregate.estimated_context_tokens, + aggregate.warning_count, + ); } fn now_unix_ms() -> u64 { diff --git a/crates/crank-community-mcp/src/session.rs b/crates/crank-community-mcp/src/session.rs index 4aa5402..124d77b 100644 --- a/crates/crank-community-mcp/src/session.rs +++ b/crates/crank-community-mcp/src/session.rs @@ -9,9 +9,12 @@ use sqlx::{ }; use thiserror::Error; use time::OffsetDateTime; -use tokio::sync::RwLock; +use tokio::sync::{RwLock, mpsc}; +use tracing::{info, warn}; use uuid::Uuid; +const ACTIVE_SESSION_COUNT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(2); + #[derive(Clone, Debug, PartialEq, Eq)] pub struct SessionState { pub id: String, @@ -54,10 +57,72 @@ pub trait TransportSessionStore: Send + Sync { async fn delete(&self, session_id: &str) -> Result; async fn cleanup_expired(&self, now: OffsetDateTime) -> Result; + + async fn active_count(&self, now: OffsetDateTime) -> Result; } pub type SharedSessionStore = Arc; +#[derive(Clone)] +pub struct ActiveSessionMetrics { + refresh_tx: mpsc::Sender<()>, +} + +impl ActiveSessionMetrics { + pub fn start(sessions: SharedSessionStore) -> Self { + let (refresh_tx, mut refresh_rx) = mpsc::channel(1); + tokio::spawn(async move { + while refresh_rx.recv().await.is_some() { + match tokio::time::timeout( + ACTIVE_SESSION_COUNT_TIMEOUT, + sessions.active_count(OffsetDateTime::now_utc()), + ) + .await + { + Ok(Ok(count)) => crank_metrics::set_mcp_active_sessions(count), + Ok(Err(_)) | Err(_) => { + warn!( + name: "mcp.active_session_metrics.refresh_failed", + error_category = "session_store", + "active session metrics refresh failed" + ); + } + } + } + }); + let metrics = Self { refresh_tx }; + metrics.refresh(); + metrics + } + + pub fn refresh(&self) { + let _ = self.refresh_tx.try_send(()); + } +} + +pub fn spawn_session_cleanup( + sessions: SharedSessionStore, + metrics: ActiveSessionMetrics, + cleanup_interval: std::time::Duration, +) { + tokio::spawn(async move { + let mut interval = tokio::time::interval(cleanup_interval); + loop { + interval.tick().await; + match sessions.cleanup_expired(OffsetDateTime::now_utc()).await { + Ok(removed) if removed > 0 => { + info!(name: "mcp.session_cleanup.completed", removed); + } + Ok(_) => {} + Err(_) => { + warn!(name: "mcp.session_cleanup.failed", error_category = "session_store"); + } + } + metrics.refresh(); + } + }); +} + #[derive(Clone, Debug)] pub struct PostgresTransportSessionStore { pool: PgPool, @@ -176,6 +241,17 @@ impl TransportSessionStore for InMemorySessionStore { guard.retain(|_, session| !is_expired(session, now)); Ok(u64::try_from(before.saturating_sub(guard.len())).unwrap_or(u64::MAX)) } + + async fn active_count(&self, now: OffsetDateTime) -> Result { + let guard = self.inner.read().await; + Ok(u64::try_from( + guard + .values() + .filter(|session| !is_expired(session, now)) + .count(), + ) + .unwrap_or(u64::MAX)) + } } #[async_trait] @@ -319,6 +395,22 @@ impl TransportSessionStore for PostgresTransportSessionStore { Ok(result.rows_affected()) } + + async fn active_count(&self, now: OffsetDateTime) -> Result { + let row = query( + "select count(*)::bigint as active_count + from mcp_transport_sessions + where expires_at is null or expires_at > $1::timestamptz", + ) + .bind(now) + .fetch_one(&self.pool) + .await + .map_err(|error| SessionStoreError { + details: error.to_string(), + })?; + let count = row.get::("active_count"); + Ok(u64::try_from(count).unwrap_or_default()) + } } async fn apply_postgres_migrations(pool: &PgPool) -> Result<(), SessionStoreError> { diff --git a/crates/crank-community-mcp/src/transport.rs b/crates/crank-community-mcp/src/transport.rs index 88d1606..eca0974 100644 --- a/crates/crank-community-mcp/src/transport.rs +++ b/crates/crank-community-mcp/src/transport.rs @@ -11,6 +11,7 @@ use axum::{ sse::{Event, KeepAlive, Sse}, }, }; +use crank_metrics::McpOutcome; use futures_util::stream; use reqwest::Url; use serde_json::Value; @@ -223,7 +224,11 @@ pub(super) fn json_response( session_id: Option<&str>, protocol_version: Option<&str>, ) -> Response { + let outcome = payload_outcome(&payload); let mut response = (status, Json(payload)).into_response(); + if let Some(outcome) = outcome { + response.extensions_mut().insert(outcome); + } if let Some(session_id) = session_id { response.headers_mut().insert( @@ -251,16 +256,35 @@ pub(super) fn transport_response( session_id: Option<&str>, protocol_version: Option<&str>, ) -> Response { + let outcome = payload_outcome(&payload); if status == StatusCode::OK && matches!(response_mode, ResponseMode::Sse) { let payload = payload.to_string(); let stream = stream::once(async move { Ok(Event::default().data(payload)) }); - return sse_response(status, stream, session_id, protocol_version); + let mut response = sse_response(status, stream, session_id, protocol_version); + if let Some(outcome) = outcome { + response.extensions_mut().insert(outcome); + } + return response; } json_response(status, payload, session_id, protocol_version) } +fn payload_outcome(payload: &Value) -> Option { + if payload.get("error").is_some() { + return Some(McpOutcome::JsonRpcError); + } + if payload + .pointer("/result/isError") + .and_then(Value::as_bool) + .unwrap_or(false) + { + return Some(McpOutcome::ToolError); + } + None +} + pub(super) fn with_request_id_header(mut response: Response, request_id: &str) -> Response { if let Ok(value) = HeaderValue::from_str(request_id) { response.headers_mut().insert(HEADER_X_REQUEST_ID, value); diff --git a/crates/crank-community-mcp/tests/integration/session.rs b/crates/crank-community-mcp/tests/integration/session.rs index 1d2e237..5dd7f4b 100644 --- a/crates/crank-community-mcp/tests/integration/session.rs +++ b/crates/crank-community-mcp/tests/integration/session.rs @@ -133,5 +133,9 @@ async fn postgres_transport_session_cleanup_removes_abandoned_expired_rows() { .unwrap(); assert_eq!(store.cleanup_expired(now).await.unwrap(), 1); + assert_eq!(store.active_count(now).await.unwrap(), 1); assert!(store.get(&active).await.unwrap().is_some()); + + assert!(store.delete(&active).await.unwrap()); + assert_eq!(store.active_count(now).await.unwrap(), 0); } diff --git a/crates/crank-community-mcp/tests/session_metrics.rs b/crates/crank-community-mcp/tests/session_metrics.rs new file mode 100644 index 0000000..ee03e15 --- /dev/null +++ b/crates/crank-community-mcp/tests/session_metrics.rs @@ -0,0 +1,76 @@ +use std::{sync::Arc, time::Duration}; + +use crank_community_mcp::session::{ + ActiveSessionMetrics, InMemorySessionStore, SharedSessionStore, TransportSessionStore, + spawn_session_cleanup, +}; +use metrics_util::debugging::{DebugValue, DebuggingRecorder, Snapshotter}; +use time::OffsetDateTime; + +#[tokio::test] +async fn active_session_sampler_tracks_create_delete_and_expiry() { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + recorder + .install() + .expect("isolated integration test recorder"); + + let store = Arc::new(InMemorySessionStore::default()); + let sessions: SharedSessionStore = store.clone(); + let sampler = ActiveSessionMetrics::start(Arc::clone(&sessions)); + assert_gauge(&sampler, &snapshotter, 0.0).await; + + let now = OffsetDateTime::now_utc(); + store + .create( + "2025-11-25", + "default", + "agent", + false, + now, + Some(now + time::Duration::milliseconds(100)), + ) + .await + .unwrap(); + assert_gauge(&sampler, &snapshotter, 1.0).await; + + spawn_session_cleanup(sessions, sampler, Duration::from_millis(10)); + assert_gauge_without_refresh(&snapshotter, 0.0).await; +} + +async fn assert_gauge_without_refresh(snapshotter: &Snapshotter, expected: f64) { + for _ in 0..100 { + tokio::time::sleep(Duration::from_millis(10)).await; + if gauge_value(snapshotter) == Some(expected) { + return; + } + } + panic!("active session gauge did not become {expected} through cleanup"); +} + +async fn assert_gauge(sampler: &ActiveSessionMetrics, snapshotter: &Snapshotter, expected: f64) { + for _ in 0..50 { + sampler.refresh(); + tokio::time::sleep(Duration::from_millis(10)).await; + if gauge_value(snapshotter) == Some(expected) { + return; + } + } + panic!("active session gauge did not become {expected}"); +} + +fn gauge_value(snapshotter: &Snapshotter) -> Option { + snapshotter + .snapshot() + .into_vec() + .into_iter() + .find_map(|(key, _, _, value)| { + if key.key().name() != "crank_mcp_active_sessions" { + return None; + } + match value { + DebugValue::Gauge(value) => Some(value.into_inner()), + _ => None, + } + }) +} diff --git a/crates/crank-community-mcp/tests/unit/session.rs b/crates/crank-community-mcp/tests/unit/session.rs index 1b57e69..ed2a9f3 100644 --- a/crates/crank-community-mcp/tests/unit/session.rs +++ b/crates/crank-community-mcp/tests/unit/session.rs @@ -93,8 +93,12 @@ async fn cleanup_removes_only_expired_sessions() { .unwrap(); assert_eq!(store.cleanup_expired(now).await.unwrap(), 1); + assert_eq!(store.active_count(now).await.unwrap(), 1); assert!(store.get(&expired).await.unwrap().is_none()); assert!(store.get(&active).await.unwrap().is_some()); + + assert!(store.delete(&active).await.unwrap()); + assert_eq!(store.active_count(now).await.unwrap(), 0); } #[test] diff --git a/crates/crank-metrics/Cargo.toml b/crates/crank-metrics/Cargo.toml new file mode 100644 index 0000000..23a8ece --- /dev/null +++ b/crates/crank-metrics/Cargo.toml @@ -0,0 +1,16 @@ +[package] +name = "crank-metrics" +edition.workspace = true +license.workspace = true +rust-version.workspace = true +publish.workspace = true +version.workspace = true + +[lib] +path = "src/lib.rs" + +[dependencies] +metrics.workspace = true + +[dev-dependencies] +metrics-util = "0.20.4" diff --git a/crates/crank-metrics/src/labels.rs b/crates/crank-metrics/src/labels.rs new file mode 100644 index 0000000..858a876 --- /dev/null +++ b/crates/crank-metrics/src/labels.rs @@ -0,0 +1,431 @@ +macro_rules! string_enum { + ( + $(#[$meta:meta])* + pub enum $name:ident { + $($variant:ident => $value:literal),+ $(,)? + } + ) => { + $(#[$meta])* + #[derive(Clone, Copy, Debug, Eq, PartialEq)] + pub enum $name { + $($variant),+ + } + + impl $name { + pub const fn as_str(self) -> &'static str { + match self { + $(Self::$variant => $value),+ + } + } + } + }; +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct HttpRoute(&'static str); + +impl HttpRoute { + pub const fn unmatched() -> Self { + Self("unmatched") + } + + pub fn from_matched_path(path: &str) -> Self { + match path { + "/health" => Self("/health"), + "/ready" => Self("/ready"), + "/api/auth/login" => Self("/api/auth/login"), + "/api/auth/logout" => Self("/api/auth/logout"), + "/api/auth/session" => Self("/api/auth/session"), + "/api/auth/profile" => Self("/api/auth/profile"), + "/api/auth/password" => Self("/api/auth/password"), + "/api/admin/capabilities" => Self("/api/admin/capabilities"), + "/api/admin/workspaces" => Self("/api/admin/workspaces"), + "/api/admin/workspaces/{workspace_id}" => Self("/api/admin/workspaces/{workspace_id}"), + "/api/admin/workspaces/{workspace_id}/operations" => { + Self("/api/admin/workspaces/{workspace_id}/operations") + } + "/api/admin/workspaces/{workspace_id}/imports/openapi/preview" => { + Self("/api/admin/workspaces/{workspace_id}/imports/openapi/preview") + } + "/api/admin/workspaces/{workspace_id}/imports/openapi/{job_id}/create" => { + Self("/api/admin/workspaces/{workspace_id}/imports/openapi/{job_id}/create") + } + "/api/admin/workspaces/{workspace_id}/operations/analyze-quality" => { + Self("/api/admin/workspaces/{workspace_id}/operations/analyze-quality") + } + "/api/admin/workspaces/{workspace_id}/operations/import" => { + Self("/api/admin/workspaces/{workspace_id}/operations/import") + } + "/api/admin/workspaces/{workspace_id}/operations/{operation_id}" => { + Self("/api/admin/workspaces/{workspace_id}/operations/{operation_id}") + } + "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/versions" => { + Self("/api/admin/workspaces/{workspace_id}/operations/{operation_id}/versions") + } + "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/versions/{version}" => { + Self( + "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/versions/{version}", + ) + } + "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/publish" => { + Self("/api/admin/workspaces/{workspace_id}/operations/{operation_id}/publish") + } + "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/archive" => { + Self("/api/admin/workspaces/{workspace_id}/operations/{operation_id}/archive") + } + "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/test-runs" => { + Self("/api/admin/workspaces/{workspace_id}/operations/{operation_id}/test-runs") + } + "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/samples/input-json" => { + Self( + "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/samples/input-json", + ) + } + "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/samples/output-json" => { + Self( + "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/samples/output-json", + ) + } + "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/drafts/generate" => { + Self( + "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/drafts/generate", + ) + } + "/api/admin/workspaces/{workspace_id}/operations/{operation_id}/export" => { + Self("/api/admin/workspaces/{workspace_id}/operations/{operation_id}/export") + } + "/api/admin/workspaces/{workspace_id}/agents" => { + Self("/api/admin/workspaces/{workspace_id}/agents") + } + "/api/admin/workspaces/{workspace_id}/agents/tool-search/preview" => { + Self("/api/admin/workspaces/{workspace_id}/agents/tool-search/preview") + } + "/api/admin/workspaces/{workspace_id}/agents/{agent_id}" => { + Self("/api/admin/workspaces/{workspace_id}/agents/{agent_id}") + } + "/api/admin/workspaces/{workspace_id}/agents/{agent_id}/versions/{version}" => { + Self("/api/admin/workspaces/{workspace_id}/agents/{agent_id}/versions/{version}") + } + "/api/admin/workspaces/{workspace_id}/agents/{agent_id}/bindings" => { + Self("/api/admin/workspaces/{workspace_id}/agents/{agent_id}/bindings") + } + "/api/admin/workspaces/{workspace_id}/agents/{agent_id}/publish" => { + Self("/api/admin/workspaces/{workspace_id}/agents/{agent_id}/publish") + } + "/api/admin/workspaces/{workspace_id}/agents/{agent_id}/unpublish" => { + Self("/api/admin/workspaces/{workspace_id}/agents/{agent_id}/unpublish") + } + "/api/admin/workspaces/{workspace_id}/agents/{agent_id}/archive" => { + Self("/api/admin/workspaces/{workspace_id}/agents/{agent_id}/archive") + } + "/api/admin/workspaces/{workspace_id}/agents/{agent_id}/platform-api-keys" => { + Self("/api/admin/workspaces/{workspace_id}/agents/{agent_id}/platform-api-keys") + } + "/api/admin/workspaces/{workspace_id}/agents/{agent_id}/platform-api-keys/{key_id}/revoke" => { + Self( + "/api/admin/workspaces/{workspace_id}/agents/{agent_id}/platform-api-keys/{key_id}/revoke", + ) + } + "/api/admin/workspaces/{workspace_id}/agents/{agent_id}/platform-api-keys/{key_id}" => { + Self( + "/api/admin/workspaces/{workspace_id}/agents/{agent_id}/platform-api-keys/{key_id}", + ) + } + "/api/admin/workspaces/{workspace_id}/auth-profiles" => { + Self("/api/admin/workspaces/{workspace_id}/auth-profiles") + } + "/api/admin/workspaces/{workspace_id}/auth-profiles/{auth_profile_id}" => { + Self("/api/admin/workspaces/{workspace_id}/auth-profiles/{auth_profile_id}") + } + "/api/admin/workspaces/{workspace_id}/upstreams" => { + Self("/api/admin/workspaces/{workspace_id}/upstreams") + } + "/api/admin/workspaces/{workspace_id}/upstreams/{upstream_id}" => { + Self("/api/admin/workspaces/{workspace_id}/upstreams/{upstream_id}") + } + "/api/admin/workspaces/{workspace_id}/secrets" => { + Self("/api/admin/workspaces/{workspace_id}/secrets") + } + "/api/admin/workspaces/{workspace_id}/secrets/{secret_id}" => { + Self("/api/admin/workspaces/{workspace_id}/secrets/{secret_id}") + } + "/api/admin/workspaces/{workspace_id}/secrets/{secret_id}/rotate" => { + Self("/api/admin/workspaces/{workspace_id}/secrets/{secret_id}/rotate") + } + "/api/admin/workspaces/{workspace_id}/export" => { + Self("/api/admin/workspaces/{workspace_id}/export") + } + "/api/admin/workspaces/{workspace_id}/logs" => { + Self("/api/admin/workspaces/{workspace_id}/logs") + } + "/api/admin/workspaces/{workspace_id}/logs/{log_id}" => { + Self("/api/admin/workspaces/{workspace_id}/logs/{log_id}") + } + "/api/admin/workspaces/{workspace_id}/approvals" => { + Self("/api/admin/workspaces/{workspace_id}/approvals") + } + "/api/admin/workspaces/{workspace_id}/approvals/{approval_id}" => { + Self("/api/admin/workspaces/{workspace_id}/approvals/{approval_id}") + } + "/api/admin/workspaces/{workspace_id}/usage" => { + Self("/api/admin/workspaces/{workspace_id}/usage") + } + "/api/admin/workspaces/{workspace_id}/usage/operations/{operation_id}" => { + Self("/api/admin/workspaces/{workspace_id}/usage/operations/{operation_id}") + } + "/api/admin/workspaces/{workspace_id}/usage/agents/{agent_id}" => { + Self("/api/admin/workspaces/{workspace_id}/usage/agents/{agent_id}") + } + "/v1/{workspace_slug}/{agent_slug}" => Self("/v1/{workspace_slug}/{agent_slug}"), + "/v1/{workspace_slug}/{agent_slug}/approvals" => { + Self("/v1/{workspace_slug}/{agent_slug}/approvals") + } + "/v1/{workspace_slug}/{agent_slug}/approvals/{approval_id}/approve" => { + Self("/v1/{workspace_slug}/{agent_slug}/approvals/{approval_id}/approve") + } + "/v1/{workspace_slug}/{agent_slug}/approvals/{approval_id}" => { + Self("/v1/{workspace_slug}/{agent_slug}/approvals/{approval_id}") + } + "/v1/{workspace_slug}/{agent_slug}/approvals/{approval_id}/deny" => { + Self("/v1/{workspace_slug}/{agent_slug}/approvals/{approval_id}/deny") + } + _ => Self::unmatched(), + } + } + + pub const fn as_str(self) -> &'static str { + self.0 + } +} + +string_enum! { + pub enum HttpMethod { + Get => "GET", + Post => "POST", + Put => "PUT", + Patch => "PATCH", + Delete => "DELETE", + Options => "OPTIONS", + Head => "HEAD", + Connect => "CONNECT", + Trace => "TRACE", + Other => "OTHER", + } +} + +impl HttpMethod { + pub fn classify(method: &str) -> Self { + match method { + "GET" => Self::Get, + "POST" => Self::Post, + "PUT" => Self::Put, + "PATCH" => Self::Patch, + "DELETE" => Self::Delete, + "OPTIONS" => Self::Options, + "HEAD" => Self::Head, + "CONNECT" => Self::Connect, + "TRACE" => Self::Trace, + _ => Self::Other, + } + } +} + +impl McpOutcome { + pub const fn from_http_status(status: u16) -> Self { + match status { + 200..=299 => Self::Success, + 400..=499 => Self::ClientError, + 500..=599 => Self::ServerError, + _ => Self::Other, + } + } +} + +string_enum! { + pub enum HttpStatusClass { + Informational => "1xx", + Success => "2xx", + Redirection => "3xx", + ClientError => "4xx", + ServerError => "5xx", + Other => "other", + } +} + +impl HttpStatusClass { + pub const fn from_status(status: u16) -> Self { + match status { + 100..=199 => Self::Informational, + 200..=299 => Self::Success, + 300..=399 => Self::Redirection, + 400..=499 => Self::ClientError, + 500..=599 => Self::ServerError, + _ => Self::Other, + } + } +} + +string_enum! { + pub enum McpMethod { + Initialize => "initialize", + Initialized => "initialized", + Ping => "ping", + ToolsList => "tools_list", + ToolsCall => "tools_call", + Notification => "notification", + Unsupported => "unsupported", + Response => "response", + Invalid => "invalid", + } +} + +string_enum! { + pub enum McpResponseMode { + Json => "json", + Sse => "sse", + Unknown => "unknown", + } +} + +string_enum! { + pub enum McpOutcome { + Success => "success", + ClientError => "client_error", + ServerError => "server_error", + JsonRpcError => "jsonrpc_error", + ToolError => "tool_error", + Aborted => "aborted", + Other => "other", + } +} + +string_enum! { + pub enum InvocationSource { + Internal => "internal", + AdminTestRun => "admin_test_run", + AgentToolCall => "agent_tool_call", + } +} + +string_enum! { + pub enum ToolOutcome { + Success => "success", + Error => "error", + Aborted => "aborted", + } +} + +string_enum! { + pub enum ToolErrorKind { + None => "none", + Schema => "schema", + Mapping => "mapping", + RestAdapter => "rest_adapter", + ProtocolAdapter => "protocol_adapter", + UnsupportedProtocol => "unsupported_protocol", + UnsupportedExecutionMode => "unsupported_execution_mode", + ConcurrencyLimit => "concurrency_limit", + InvalidPreparedRequest => "invalid_prepared_request", + ConfirmationRequired => "confirmation_required", + InvalidConfirmationToken => "invalid_confirmation_token", + ConfirmationStore => "confirmation_store", + IdempotencyStore => "idempotency_store", + IdempotencyInProgress => "idempotency_in_progress", + IdempotencyConflict => "idempotency_conflict", + IdempotencyOutcomeUnknown => "idempotency_outcome_unknown", + MissingAuthProfile => "missing_auth_profile", + MissingSecret => "missing_secret", + MissingSecretVersion => "missing_secret_version", + InvalidAuthSecret => "invalid_auth_secret", + SecretCrypto => "secret_crypto", + Aborted => "aborted", + } +} + +string_enum! { + pub enum UpstreamOperationKind { + Rest => "rest", + } +} + +string_enum! { + pub enum UpstreamOutcome { + Success => "success", + ClientError => "client_error", + ServerError => "server_error", + UnexpectedStatus => "unexpected_status", + Timeout => "timeout", + TransportError => "transport_error", + ResponseTooLarge => "response_too_large", + Rejected => "rejected", + WindowExpired => "window_expired", + InvalidResponse => "invalid_response", + InvalidRequest => "invalid_request", + Configuration => "configuration", + Aborted => "aborted", + } +} + +string_enum! { + pub enum LimitStage { + Concurrency => "concurrency", + RateLimit => "rate_limit", + McpStream => "mcp_stream", + } +} + +string_enum! { + pub enum CacheOutcome { + Hit => "hit", + Miss => "miss", + ReadError => "read_error", + DecodeError => "decode_error", + EvictError => "evict_error", + Stored => "stored", + WriteError => "write_error", + } +} + +string_enum! { + pub enum IdempotencyOutcome { + Execute => "execute", + Replay => "replay", + Completed => "completed", + Conflict => "conflict", + InProgress => "in_progress", + OutcomeUnknown => "outcome_unknown", + StoreUnavailable => "store_unavailable", + Error => "error", + } +} + +string_enum! { + pub enum ConfirmationOutcome { + Approved => "approved", + Required => "required", + InvalidToken => "invalid_token", + StoreUnavailable => "store_unavailable", + Error => "error", + } +} + +string_enum! { + pub enum DbPoolState { + Idle => "idle", + Used => "used", + } +} + +string_enum! { + pub enum SignalType { + Trace => "trace", + InvocationHistory => "invocation_history", + } +} + +string_enum! { + pub enum Exporter { + Otlp => "otlp", + Postgres => "postgres", + } +} diff --git a/crates/crank-metrics/src/lib.rs b/crates/crank-metrics/src/lib.rs new file mode 100644 index 0000000..ab14f87 --- /dev/null +++ b/crates/crank-metrics/src/lib.rs @@ -0,0 +1,26 @@ +//! Закрытый семантический контракт метрик Crank. +//! +//! Product crate передают только типизированные значения. Имена метрик, +//! ключи и значения labels сосредоточены здесь, поэтому пользовательские +//! идентификаторы и тексты нельзя случайно превратить во временные ряды. + +mod labels; +mod record; +mod schema; + +pub use labels::{ + CacheOutcome, ConfirmationOutcome, DbPoolState, Exporter, HttpMethod, HttpRoute, + HttpStatusClass, IdempotencyOutcome, InvocationSource, LimitStage, McpMethod, McpOutcome, + McpResponseMode, SignalType, ToolErrorKind, ToolOutcome, UpstreamOperationKind, + UpstreamOutcome, +}; +pub use record::{ + InFlightGuard, ToolInvocationMetrics, UpstreamRequestMetrics, initialize_gauges, + record_cache_outcome, record_confirmation_outcome, record_export_failure, record_http_request, + record_idempotency_outcome, record_invocation_history_lost, record_limit_rejection, + record_mcp_request, record_tool_invocation, record_upstream_request, set_catalog, + set_db_pool_connections, set_mcp_active_sessions, +}; +pub use schema::{ + DURATION_BUCKETS_SECONDS, MetricDefinition, MetricKind, MetricUnit, metric_schema, +}; diff --git a/crates/crank-metrics/src/record.rs b/crates/crank-metrics/src/record.rs new file mode 100644 index 0000000..78ea3be --- /dev/null +++ b/crates/crank-metrics/src/record.rs @@ -0,0 +1,243 @@ +use std::time::{Duration, Instant}; + +use metrics::Gauge; + +use crate::{ + CacheOutcome, ConfirmationOutcome, DbPoolState, Exporter, HttpMethod, HttpRoute, + HttpStatusClass, IdempotencyOutcome, InvocationSource, LimitStage, McpMethod, McpOutcome, + McpResponseMode, SignalType, ToolErrorKind, ToolOutcome, UpstreamOperationKind, + UpstreamOutcome, +}; + +pub fn record_http_request( + route: HttpRoute, + method: HttpMethod, + status: HttpStatusClass, + duration: Duration, +) { + metrics::counter!( + "crank_http_requests_total", + "route" => route.as_str(), + "method" => method.as_str(), + "status_class" => status.as_str() + ) + .increment(1); + metrics::histogram!( + "crank_http_request_duration_seconds", + "route" => route.as_str(), + "method" => method.as_str() + ) + .record(duration.as_secs_f64()); +} + +pub fn record_mcp_request(method: McpMethod, response_mode: McpResponseMode, outcome: McpOutcome) { + metrics::counter!( + "crank_mcp_requests_total", + "method" => method.as_str(), + "response_mode" => response_mode.as_str(), + "outcome" => outcome.as_str() + ) + .increment(1); +} + +pub fn set_mcp_active_sessions(count: u64) { + metrics::gauge!("crank_mcp_active_sessions").set(count as f64); +} + +pub fn record_tool_invocation( + source: InvocationSource, + outcome: ToolOutcome, + error_kind: ToolErrorKind, + duration: Duration, +) { + metrics::counter!( + "crank_tool_invocations_total", + "source" => source.as_str(), + "outcome" => outcome.as_str(), + "error_kind" => error_kind.as_str() + ) + .increment(1); + metrics::histogram!( + "crank_tool_invocation_duration_seconds", + "source" => source.as_str(), + "outcome" => outcome.as_str() + ) + .record(duration.as_secs_f64()); +} + +pub fn record_upstream_request( + operation_kind: UpstreamOperationKind, + outcome: UpstreamOutcome, + duration: Duration, +) { + metrics::counter!( + "crank_upstream_requests_total", + "operation_kind" => operation_kind.as_str(), + "outcome" => outcome.as_str() + ) + .increment(1); + metrics::histogram!( + "crank_upstream_request_duration_seconds", + "operation_kind" => operation_kind.as_str(), + "outcome" => outcome.as_str() + ) + .record(duration.as_secs_f64()); +} + +pub struct ToolInvocationMetrics { + source: InvocationSource, + started_at: Option, +} + +impl ToolInvocationMetrics { + pub fn start(source: InvocationSource) -> Self { + Self { + source, + started_at: Some(Instant::now()), + } + } + + pub fn complete(mut self, outcome: ToolOutcome, error_kind: ToolErrorKind) { + self.finish(outcome, error_kind); + } + + fn finish(&mut self, outcome: ToolOutcome, error_kind: ToolErrorKind) { + if let Some(started_at) = self.started_at.take() { + record_tool_invocation(self.source, outcome, error_kind, started_at.elapsed()); + } + } +} + +impl Drop for ToolInvocationMetrics { + fn drop(&mut self) { + self.finish(ToolOutcome::Aborted, ToolErrorKind::Aborted); + } +} + +pub struct UpstreamRequestMetrics { + operation_kind: UpstreamOperationKind, + started_at: Option, +} + +impl UpstreamRequestMetrics { + pub fn start(operation_kind: UpstreamOperationKind) -> Self { + Self { + operation_kind, + started_at: Some(Instant::now()), + } + } + + pub fn complete(mut self, outcome: UpstreamOutcome) { + self.finish(outcome); + } + + fn finish(&mut self, outcome: UpstreamOutcome) { + if let Some(started_at) = self.started_at.take() { + record_upstream_request(self.operation_kind, outcome, started_at.elapsed()); + } + } +} + +impl Drop for UpstreamRequestMetrics { + fn drop(&mut self) { + self.finish(UpstreamOutcome::Aborted); + } +} + +pub fn record_limit_rejection(stage: LimitStage) { + metrics::counter!( + "crank_runtime_limit_rejections_total", + "stage" => stage.as_str() + ) + .increment(1); +} + +pub fn record_cache_outcome(outcome: CacheOutcome) { + metrics::counter!( + "crank_runtime_cache_total", + "outcome" => outcome.as_str() + ) + .increment(1); +} + +pub fn record_idempotency_outcome(outcome: IdempotencyOutcome) { + metrics::counter!( + "crank_idempotency_total", + "outcome" => outcome.as_str() + ) + .increment(1); +} + +pub fn record_confirmation_outcome(outcome: ConfirmationOutcome) { + metrics::counter!( + "crank_confirmation_total", + "outcome" => outcome.as_str() + ) + .increment(1); +} + +pub fn set_db_pool_connections(state: DbPoolState, count: u32) { + metrics::gauge!( + "crank_db_pool_connections", + "state" => state.as_str() + ) + .set(f64::from(count)); +} + +pub fn set_catalog(tool_count: usize, estimated_context_tokens: usize, warnings: usize) { + metrics::gauge!("crank_catalog_tools").set(tool_count as f64); + metrics::gauge!("crank_catalog_estimated_context_tokens").set(estimated_context_tokens as f64); + metrics::gauge!("crank_catalog_warnings").set(warnings as f64); +} + +pub fn record_invocation_history_lost() { + metrics::counter!("crank_invocation_history_lost_total").increment(1); +} + +pub fn record_export_failure(signal: SignalType, exporter: Exporter) { + metrics::counter!( + "crank_telemetry_export_failures_total", + "signal_type" => signal.as_str(), + "exporter" => exporter.as_str() + ) + .increment(1); +} + +pub fn initialize_gauges() { + metrics::gauge!("crank_http_inflight").set(0.0); + metrics::gauge!("crank_mcp_active_sessions").set(0.0); + metrics::gauge!("crank_mcp_active_streams").set(0.0); + metrics::gauge!("crank_runtime_inflight").set(0.0); + set_db_pool_connections(DbPoolState::Idle, 0); + set_db_pool_connections(DbPoolState::Used, 0); + set_catalog(0, 0, 0); +} + +pub struct InFlightGuard { + gauge: Gauge, +} + +impl InFlightGuard { + pub fn http() -> Self { + Self::increment(metrics::gauge!("crank_http_inflight")) + } + + pub fn runtime() -> Self { + Self::increment(metrics::gauge!("crank_runtime_inflight")) + } + + pub fn mcp_stream() -> Self { + Self::increment(metrics::gauge!("crank_mcp_active_streams")) + } + + fn increment(gauge: Gauge) -> Self { + gauge.increment(1.0); + Self { gauge } + } +} + +impl Drop for InFlightGuard { + fn drop(&mut self) { + self.gauge.decrement(1.0); + } +} diff --git a/crates/crank-observability/src/metrics_schema.rs b/crates/crank-metrics/src/schema.rs similarity index 87% rename from crates/crank-observability/src/metrics_schema.rs rename to crates/crank-metrics/src/schema.rs index 505be01..6f3d2bb 100644 --- a/crates/crank-observability/src/metrics_schema.rs +++ b/crates/crank-metrics/src/schema.rs @@ -48,7 +48,12 @@ const METRIC_SCHEMA: &[MetricDefinition] = &[ gauge( "crank_mcp_active_sessions", &[], - "Active MCP transport sessions.", + "Active persisted MCP transport sessions.", + ), + gauge( + "crank_mcp_active_streams", + &[], + "MCP event streams currently being served.", ), counter( "crank_tool_invocations_total", @@ -80,6 +85,21 @@ const METRIC_SCHEMA: &[MetricDefinition] = &[ &["stage"], "Runtime executions rejected by a bounded limit.", ), + counter( + "crank_runtime_cache_total", + &["outcome"], + "Runtime response cache outcomes.", + ), + counter( + "crank_idempotency_total", + &["outcome"], + "Runtime idempotency outcomes.", + ), + counter( + "crank_confirmation_total", + &["outcome"], + "Runtime confirmation outcomes.", + ), gauge( "crank_db_pool_connections", &["state"], diff --git a/crates/crank-metrics/tests/contract.rs b/crates/crank-metrics/tests/contract.rs new file mode 100644 index 0000000..66bbe0a --- /dev/null +++ b/crates/crank-metrics/tests/contract.rs @@ -0,0 +1,235 @@ +use std::{collections::BTreeSet, time::Duration}; + +use crank_metrics::{ + CacheOutcome, ConfirmationOutcome, DbPoolState, Exporter, HttpMethod, HttpRoute, + HttpStatusClass, IdempotencyOutcome, InFlightGuard, InvocationSource, LimitStage, McpMethod, + McpOutcome, McpResponseMode, SignalType, ToolErrorKind, ToolInvocationMetrics, ToolOutcome, + UpstreamOperationKind, UpstreamOutcome, UpstreamRequestMetrics, initialize_gauges, + metric_schema, record_cache_outcome, record_confirmation_outcome, record_export_failure, + record_http_request, record_idempotency_outcome, record_invocation_history_lost, + record_limit_rejection, record_mcp_request, record_tool_invocation, record_upstream_request, + set_catalog, set_db_pool_connections, set_mcp_active_sessions, +}; +use metrics_util::debugging::DebuggingRecorder; + +#[test] +fn schema_names_are_unique_and_labels_are_closed() { + let schema = metric_schema(); + let names = schema + .iter() + .map(|metric| metric.name) + .collect::>(); + + assert_eq!(names.len(), schema.len()); + assert!(names.contains("crank_runtime_cache_total")); + assert!(names.contains("crank_idempotency_total")); + assert!(names.contains("crank_confirmation_total")); + assert!(names.contains("crank_mcp_active_streams")); + + let allowed_labels = [ + "route", + "method", + "status_class", + "response_mode", + "outcome", + "source", + "error_kind", + "operation_kind", + "stage", + "state", + "signal_type", + "exporter", + ]; + for metric in schema { + assert!( + metric + .labels + .iter() + .all(|label| allowed_labels.contains(label)), + "{} contains an unapproved label", + metric.name + ); + } +} + +#[test] +fn untrusted_http_values_collapse_to_closed_variants() { + assert_eq!( + HttpRoute::from_matched_path("/api/admin/workspaces/{workspace_id}/operations").as_str(), + "/api/admin/workspaces/{workspace_id}/operations" + ); + assert_eq!( + HttpRoute::from_matched_path("/customer-controlled"), + HttpRoute::unmatched() + ); + assert_eq!(HttpMethod::classify("CUSTOM"), HttpMethod::Other); + assert_eq!(HttpStatusClass::from_status(999), HttpStatusClass::Other); +} + +#[test] +fn application_outcomes_are_distinct_from_transport_success() { + assert_eq!(McpOutcome::from_http_status(200), McpOutcome::Success); + assert_ne!(McpOutcome::JsonRpcError, McpOutcome::Success); + assert_ne!(McpOutcome::ToolError, McpOutcome::Success); + + assert_eq!(CacheOutcome::Hit.as_str(), "hit"); + assert_eq!(CacheOutcome::Miss.as_str(), "miss"); + assert_eq!(IdempotencyOutcome::Replay.as_str(), "replay"); + assert_eq!(IdempotencyOutcome::Conflict.as_str(), "conflict"); + assert_eq!(ConfirmationOutcome::Approved.as_str(), "approved"); + assert_eq!(ConfirmationOutcome::Required.as_str(), "required"); +} + +#[test] +fn dropped_product_guards_record_aborted_outcomes() { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + + metrics::with_local_recorder(&recorder, || { + drop(ToolInvocationMetrics::start( + InvocationSource::AgentToolCall, + )); + drop(UpstreamRequestMetrics::start(UpstreamOperationKind::Rest)); + }); + + let snapshot = snapshotter.snapshot().into_vec(); + assert!(snapshot.iter().any(|(key, _, _, _)| { + key.key().name() == "crank_tool_invocations_total" + && has_label(key.key(), "outcome", "aborted") + && has_label(key.key(), "error_kind", "aborted") + })); + assert!(snapshot.iter().any(|(key, _, _, _)| { + key.key().name() == "crank_upstream_requests_total" + && has_label(key.key(), "outcome", "aborted") + })); +} + +fn has_label(key: &metrics::Key, name: &str, value: &str) -> bool { + key.labels() + .any(|label| label.key() == name && label.value() == value) +} + +#[test] +fn diverse_calls_cannot_create_unbounded_series() { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + + metrics::with_local_recorder(&recorder, || { + for status in 200..300 { + let route = format!("/customer-controlled/{status}"); + let method = format!("CUSTOM-{status}"); + record_http_request( + HttpRoute::from_matched_path(&route), + HttpMethod::classify(&method), + HttpStatusClass::from_status(status), + Duration::from_millis(u64::from(status)), + ); + record_mcp_request( + McpMethod::ToolsCall, + McpResponseMode::Json, + McpOutcome::ToolError, + ); + record_tool_invocation( + InvocationSource::AgentToolCall, + ToolOutcome::Error, + ToolErrorKind::Mapping, + Duration::from_millis(u64::from(status)), + ); + record_cache_outcome(CacheOutcome::Miss); + record_idempotency_outcome(IdempotencyOutcome::Replay); + record_confirmation_outcome(ConfirmationOutcome::Required); + } + }); + + let snapshot = snapshotter.snapshot().into_vec(); + let series = snapshot + .iter() + .map(|(key, _, _, _)| format!("{:?}", key.key())) + .collect::>(); + + assert_eq!(series.len(), 8); + assert!( + series + .iter() + .all(|series| !series.contains("customer-controlled")) + ); +} + +#[test] +fn every_recording_api_matches_the_declared_schema() { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + + metrics::with_local_recorder(&recorder, || { + initialize_gauges(); + record_http_request( + HttpRoute::from_matched_path("/api/auth/session"), + HttpMethod::Get, + HttpStatusClass::Success, + Duration::from_millis(1), + ); + record_mcp_request( + McpMethod::ToolsCall, + McpResponseMode::Json, + McpOutcome::Success, + ); + set_mcp_active_sessions(1); + let _stream = InFlightGuard::mcp_stream(); + let _runtime = InFlightGuard::runtime(); + record_tool_invocation( + InvocationSource::AgentToolCall, + ToolOutcome::Error, + ToolErrorKind::Mapping, + Duration::from_millis(1), + ); + record_upstream_request( + UpstreamOperationKind::Rest, + UpstreamOutcome::Timeout, + Duration::from_millis(1), + ); + record_limit_rejection(LimitStage::Concurrency); + record_cache_outcome(CacheOutcome::Hit); + record_idempotency_outcome(IdempotencyOutcome::Replay); + record_confirmation_outcome(ConfirmationOutcome::Required); + set_db_pool_connections(DbPoolState::Idle, 1); + set_catalog(1, 2, 3); + record_invocation_history_lost(); + record_export_failure(SignalType::Trace, Exporter::Otlp); + }); + + let snapshot = snapshotter.snapshot().into_vec(); + let actual_names = snapshot + .iter() + .map(|(key, _, _, _)| key.key().name().to_owned()) + .collect::>(); + let expected_names = metric_schema() + .iter() + .map(|definition| definition.name.to_owned()) + .collect::>(); + + assert_eq!(actual_names, expected_names); + for definition in metric_schema() { + let actual_label_sets = snapshot + .iter() + .filter(|(key, _, _, _)| key.key().name() == definition.name) + .map(|(key, _, _, _)| { + key.key() + .labels() + .map(|label| label.key().to_owned()) + .collect::>() + }) + .collect::>(); + let expected_labels = definition + .labels + .iter() + .map(|label| (*label).to_owned()) + .collect::>(); + + assert_eq!( + actual_label_sets, + BTreeSet::from([expected_labels]), + "recording API for {} diverges from the schema", + definition.name + ); + } +} diff --git a/crates/crank-observability/Cargo.toml b/crates/crank-observability/Cargo.toml index db55b0c..96eea7f 100644 --- a/crates/crank-observability/Cargo.toml +++ b/crates/crank-observability/Cargo.toml @@ -8,6 +8,7 @@ version.workspace = true [dependencies] axum.workspace = true +crank-metrics = { path = "../crank-metrics" } metrics.workspace = true metrics-exporter-prometheus.workspace = true opentelemetry.workspace = true @@ -21,7 +22,7 @@ sha2.workspace = true subtle.workspace = true thiserror.workspace = true time.workspace = true -tokio = { workspace = true, features = ["net"] } +tokio = { workspace = true, features = ["io-util", "net"] } tracing.workspace = true tracing-opentelemetry.workspace = true tracing-subscriber.workspace = true diff --git a/crates/crank-observability/src/incidents.rs b/crates/crank-observability/src/incidents.rs index 070782e..90417f7 100644 --- a/crates/crank-observability/src/incidents.rs +++ b/crates/crank-observability/src/incidents.rs @@ -14,13 +14,11 @@ pub fn record_operational_incident(incident: OperationalIncident) { }); match incident { OperationalIncident::InvocationHistoryLost => { - metrics::counter!("crank_invocation_history_lost_total").increment(1); - metrics::counter!( - "crank_telemetry_export_failures_total", - "signal_type" => "invocation_history", - "exporter" => "postgres" - ) - .increment(1); + crank_metrics::record_invocation_history_lost(); + crank_metrics::record_export_failure( + crank_metrics::SignalType::InvocationHistory, + crank_metrics::Exporter::Postgres, + ); } } } diff --git a/crates/crank-observability/src/instrumentation.rs b/crates/crank-observability/src/instrumentation.rs index 7901c20..eb831e0 100644 --- a/crates/crank-observability/src/instrumentation.rs +++ b/crates/crank-observability/src/instrumentation.rs @@ -5,7 +5,8 @@ use axum::{ middleware::Next, response::Response, }; -use metrics::{Gauge, Unit}; +use crank_metrics::{DbPoolState, HttpMethod, HttpRoute, HttpStatusClass, InFlightGuard}; +use metrics::Unit; use crate::{MetricKind, MetricUnit, metric_schema}; @@ -13,36 +14,28 @@ pub async fn record_http_request(request: Request, next: Next) -> Response { let route = request .extensions() .get::() - .map_or("unmatched", MatchedPath::as_str) - .to_owned(); - let method = normalized_http_method(request.method().as_str()); + .map_or(HttpRoute::unmatched(), |path| { + HttpRoute::from_matched_path(path.as_str()) + }); + let method = HttpMethod::classify(request.method().as_str()); let started_at = Instant::now(); - let _inflight = GaugeGuard::increment("crank_http_inflight"); + let _inflight = InFlightGuard::http(); let response = next.run(request).await; - let status_class = status_class(response.status().as_u16()); - - metrics::counter!( - "crank_http_requests_total", - "route" => route.clone(), - "method" => method, - "status_class" => status_class - ) - .increment(1); - metrics::histogram!( - "crank_http_request_duration_seconds", - "route" => route, - "method" => method - ) - .record(started_at.elapsed().as_secs_f64()); + crank_metrics::record_http_request( + route, + method, + HttpStatusClass::from_status(response.status().as_u16()), + started_at.elapsed(), + ); response } pub fn record_db_pool_connections(total: u32, idle: usize) { - let idle = idle.min(total as usize) as f64; - metrics::gauge!("crank_db_pool_connections", "state" => "idle").set(idle); - metrics::gauge!("crank_db_pool_connections", "state" => "used").set(f64::from(total) - idle); + let idle = u32::try_from(idle.min(total as usize)).unwrap_or(total); + crank_metrics::set_db_pool_connections(DbPoolState::Idle, idle); + crank_metrics::set_db_pool_connections(DbPoolState::Used, total.saturating_sub(idle)); } pub(crate) fn register_metric_schema() { @@ -64,70 +57,29 @@ pub(crate) fn register_metric_schema() { } } - metrics::gauge!("crank_http_inflight").set(0.0); - metrics::gauge!("crank_mcp_active_sessions").set(0.0); - metrics::gauge!("crank_runtime_inflight").set(0.0); - metrics::gauge!("crank_db_pool_connections", "state" => "idle").set(0.0); - metrics::gauge!("crank_db_pool_connections", "state" => "used").set(0.0); - metrics::gauge!("crank_catalog_tools").set(0.0); - metrics::gauge!("crank_catalog_estimated_context_tokens").set(0.0); - metrics::gauge!("crank_catalog_warnings").set(0.0); -} - -fn normalized_http_method(method: &str) -> &'static str { - match method { - "GET" => "GET", - "POST" => "POST", - "PUT" => "PUT", - "PATCH" => "PATCH", - "DELETE" => "DELETE", - "OPTIONS" => "OPTIONS", - "HEAD" => "HEAD", - "CONNECT" => "CONNECT", - "TRACE" => "TRACE", - _ => "OTHER", - } -} - -fn status_class(status: u16) -> &'static str { - match status { - 100..=199 => "1xx", - 200..=299 => "2xx", - 300..=399 => "3xx", - 400..=499 => "4xx", - 500..=599 => "5xx", - _ => "other", - } -} - -struct GaugeGuard { - gauge: Gauge, -} - -impl GaugeGuard { - fn increment(name: &'static str) -> Self { - let gauge = metrics::gauge!(name); - gauge.increment(1.0); - Self { gauge } - } -} - -impl Drop for GaugeGuard { - fn drop(&mut self) { - self.gauge.decrement(1.0); - } + crank_metrics::initialize_gauges(); } #[cfg(test)] mod tests { - use super::{normalized_http_method, status_class}; + use crank_metrics::{HttpMethod, HttpRoute, HttpStatusClass}; #[test] fn normalizes_unbounded_http_values() { - assert_eq!(normalized_http_method("GET"), "GET"); - assert_eq!(normalized_http_method("CUSTOM-user-controlled"), "OTHER"); - assert_eq!(status_class(204), "2xx"); - assert_eq!(status_class(429), "4xx"); - assert_eq!(status_class(999), "other"); + assert_eq!(HttpMethod::classify("GET"), HttpMethod::Get); + assert_eq!( + HttpMethod::classify("CUSTOM-user-controlled"), + HttpMethod::Other + ); + assert_eq!(HttpStatusClass::from_status(204), HttpStatusClass::Success); + assert_eq!( + HttpStatusClass::from_status(429), + HttpStatusClass::ClientError + ); + assert_eq!(HttpStatusClass::from_status(999), HttpStatusClass::Other); + assert_eq!( + HttpRoute::from_matched_path("/documents/{document_id}"), + HttpRoute::unmatched() + ); } } diff --git a/crates/crank-observability/src/lib.rs b/crates/crank-observability/src/lib.rs index d5f987f..2d0f17e 100644 --- a/crates/crank-observability/src/lib.rs +++ b/crates/crank-observability/src/lib.rs @@ -5,7 +5,6 @@ mod incidents; mod instrumentation; mod lifecycle; mod logging; -mod metrics_schema; mod otlp; mod prometheus; mod propagation; @@ -14,6 +13,9 @@ mod schema; pub use config::{ObservabilityConfig, ObservabilityConfigError, ServiceIdentity}; pub use correlation::RequestId; +pub use crank_metrics::{ + DURATION_BUCKETS_SECONDS, MetricDefinition, MetricKind, MetricUnit, metric_schema, +}; pub use error_reporting::{ CriticalErrorCategory, SentryConfig, SentryConfigError, capture_critical_error, with_request_correlation, @@ -22,9 +24,6 @@ pub use incidents::{OperationalIncident, operational_incident_total, record_oper pub use instrumentation::{record_db_pool_connections, record_http_request}; pub use lifecycle::{ObservabilityInitError, ObservabilityLifecycle, init}; pub use logging::build_subscriber; -pub use metrics_schema::{ - DURATION_BUCKETS_SECONDS, MetricDefinition, MetricKind, MetricUnit, metric_schema, -}; pub use otlp::{ OtlpBatchConfig, OtlpTraceConfig, OtlpTraceConfigError, OtlpTraceError, build_tracer_provider, }; diff --git a/crates/crank-observability/src/otlp.rs b/crates/crank-observability/src/otlp.rs index 845dedf..705bb83 100644 --- a/crates/crank-observability/src/otlp.rs +++ b/crates/crank-observability/src/otlp.rs @@ -340,12 +340,10 @@ impl SpanExporterTrait for ObservedSpanExporter { sanitize_trace_batch(&mut batch); let result = self.0.export(batch).await; if result.is_err() { - metrics::counter!( - "crank_telemetry_export_failures_total", - "signal_type" => "trace", - "exporter" => "otlp" - ) - .increment(1); + crank_metrics::record_export_failure( + crank_metrics::SignalType::Trace, + crank_metrics::Exporter::Otlp, + ); } result } diff --git a/crates/crank-observability/src/prometheus.rs b/crates/crank-observability/src/prometheus.rs index e9bd7fa..689c25b 100644 --- a/crates/crank-observability/src/prometheus.rs +++ b/crates/crank-observability/src/prometheus.rs @@ -155,7 +155,7 @@ impl MetricsSurface { Router::new() .route("/metrics", get(render_metrics)) .route("/health", get(metrics_health)) - .layer(middleware::from_fn_with_state( + .route_layer(middleware::from_fn_with_state( self.state.clone(), authorize_metrics, )) @@ -179,6 +179,12 @@ pub struct MetricsServer { } impl MetricsServer { + pub fn local_addr(&self) -> Result { + self.listener + .local_addr() + .map_err(|_| MetricsServeError::LocalAddress) + } + pub async fn serve(self) -> Result<(), MetricsServeError> { axum::serve(self.listener, self.router) .await @@ -198,6 +204,8 @@ pub enum MetricsServeError { Bind, #[error("metrics listener stopped unexpectedly")] Serve, + #[error("failed to read metrics listener address")] + LocalAddress, } pub(crate) fn install_prometheus_recorder( @@ -257,10 +265,15 @@ async fn authorize_metrics( } fn bearer_token(headers: &HeaderMap) -> Option<&[u8]> { - headers - .get(header::AUTHORIZATION)? - .as_bytes() - .strip_prefix(b"Bearer ") + let value = headers.get(header::AUTHORIZATION)?.as_bytes(); + let separator = value.iter().position(|byte| *byte == b' ')?; + let (scheme, token_with_spaces) = value.split_at(separator); + let token_start = token_with_spaces.iter().position(|byte| *byte != b' ')?; + let token = &token_with_spaces[token_start..]; + + scheme + .eq_ignore_ascii_case(b"bearer") + .then_some(token) .filter(|token| !token.is_empty()) } diff --git a/crates/crank-observability/tests/http_metrics.rs b/crates/crank-observability/tests/http_metrics.rs index 5f3eec6..c92856e 100644 --- a/crates/crank-observability/tests/http_metrics.rs +++ b/crates/crank-observability/tests/http_metrics.rs @@ -31,19 +31,21 @@ async fn http_metrics_use_matched_routes_and_closed_labels() { let metrics = lifecycle.metrics_surface(config).router(); let app = Router::new() .route( - "/documents/{document_id}", + "/api/admin/workspaces/{workspace_id}/operations/{operation_id}", get(|| async { StatusCode::NO_CONTENT }), ) .layer(middleware::from_fn(record_http_request)); - let sensitive_path_segment = "customer-secret-document-id"; + let sensitive_path_segment = "customer-secret-operation-id"; for index in 0..100 { let response = app .clone() .oneshot( - Request::get(format!("/documents/{sensitive_path_segment}-{index}")) - .body(Body::empty()) - .expect("request"), + Request::get(format!( + "/api/admin/workspaces/customer-secret-workspace-{index}/operations/{sensitive_path_segment}-{index}" + )) + .body(Body::empty()) + .expect("request"), ) .await .expect("response"); @@ -74,7 +76,9 @@ async fn http_metrics_use_matched_routes_and_closed_labels() { let body = String::from_utf8(body.to_vec()).expect("utf-8 metrics"); assert!(body.contains("crank_http_requests_total")); - assert!(body.contains("route=\"/documents/{document_id}\"")); + assert!( + body.contains("route=\"/api/admin/workspaces/{workspace_id}/operations/{operation_id}\"") + ); assert!(body.contains("method=\"GET\"")); assert!(body.contains("status_class=\"2xx\"")); assert!(body.contains("crank_http_request_duration_seconds_bucket")); @@ -83,7 +87,9 @@ async fn http_metrics_use_matched_routes_and_closed_labels() { body.lines() .filter(|line| { line.starts_with("crank_http_requests_total{") - && line.contains("route=\"/documents/{document_id}\"") + && line.contains( + "route=\"/api/admin/workspaces/{workspace_id}/operations/{operation_id}\"", + ) }) .count(), 1, diff --git a/crates/crank-observability/tests/metrics_signals.rs b/crates/crank-observability/tests/metrics_signals.rs new file mode 100644 index 0000000..d99f4da --- /dev/null +++ b/crates/crank-observability/tests/metrics_signals.rs @@ -0,0 +1,134 @@ +use std::{ + net::{IpAddr, Ipv4Addr, SocketAddr}, + time::Duration, +}; + +use crank_metrics::{ + CacheOutcome, ConfirmationOutcome, DbPoolState, Exporter, HttpMethod, HttpRoute, + HttpStatusClass, IdempotencyOutcome, InFlightGuard, InvocationSource, LimitStage, McpMethod, + McpOutcome, McpResponseMode, SignalType, ToolErrorKind, ToolOutcome, UpstreamOperationKind, + UpstreamOutcome, record_cache_outcome, record_confirmation_outcome, record_export_failure, + record_http_request, record_idempotency_outcome, record_invocation_history_lost, + record_limit_rejection, record_mcp_request, record_tool_invocation, record_upstream_request, + set_catalog, set_db_pool_connections, set_mcp_active_sessions, +}; +use crank_observability::{MetricsConfig, ObservabilityConfig, RedactionLimits, ServiceIdentity}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +#[tokio::test] +async fn typed_product_signals_are_exposed_by_a_real_scrape() { + let identity = + ServiceIdentity::try_new("metrics-signals", "0.3.1", "test").expect("valid identity"); + let lifecycle = crank_observability::init(ObservabilityConfig::new( + identity, + "off", + RedactionLimits::default(), + )) + .expect("observability lifecycle"); + let config = MetricsConfig::new( + true, + SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0), + None, + ) + .expect("loopback metrics config"); + let server = lifecycle + .metrics_surface(config) + .bind() + .await + .expect("metrics listener"); + let address = server.local_addr().expect("listener address"); + let task = tokio::spawn(server.serve()); + + record_http_request( + HttpRoute::from_matched_path("/api/auth/session"), + HttpMethod::Get, + HttpStatusClass::Success, + Duration::from_millis(7), + ); + record_mcp_request( + McpMethod::ToolsCall, + McpResponseMode::Json, + McpOutcome::ToolError, + ); + set_mcp_active_sessions(2); + let _stream = InFlightGuard::mcp_stream(); + let _runtime = InFlightGuard::runtime(); + record_tool_invocation( + InvocationSource::AgentToolCall, + ToolOutcome::Error, + ToolErrorKind::Mapping, + Duration::from_millis(11), + ); + record_upstream_request( + UpstreamOperationKind::Rest, + UpstreamOutcome::Timeout, + Duration::from_millis(13), + ); + record_limit_rejection(LimitStage::Concurrency); + record_cache_outcome(CacheOutcome::Hit); + record_idempotency_outcome(IdempotencyOutcome::Replay); + record_confirmation_outcome(ConfirmationOutcome::Required); + set_db_pool_connections(DbPoolState::Idle, 3); + set_db_pool_connections(DbPoolState::Used, 1); + set_catalog(5, 1200, 1); + record_invocation_history_lost(); + record_export_failure(SignalType::Trace, Exporter::Otlp); + + let response = raw_get(address, "/metrics").await; + task.abort(); + let _ = task.await; + + assert!(response.starts_with("HTTP/1.1 200 OK"), "{response}"); + for metric in [ + "crank_http_requests_total", + "crank_mcp_requests_total", + "crank_mcp_active_sessions", + "crank_mcp_active_streams", + "crank_tool_invocations_total", + "crank_upstream_requests_total", + "crank_runtime_inflight", + "crank_runtime_limit_rejections_total", + "crank_runtime_cache_total", + "crank_idempotency_total", + "crank_confirmation_total", + "crank_db_pool_connections", + "crank_catalog_tools", + "crank_invocation_history_lost_total", + "crank_telemetry_export_failures_total", + ] { + assert!(response.contains(metric), "missing metric {metric}"); + } + for expected_label in [ + "outcome=\"tool_error\"", + "outcome=\"timeout\"", + "outcome=\"hit\"", + "outcome=\"replay\"", + "outcome=\"required\"", + "error_kind=\"mapping\"", + ] { + assert!( + response.contains(expected_label), + "missing label {expected_label}" + ); + } + assert!(!response.contains("customer-secret")); +} + +async fn raw_get(address: SocketAddr, path: &str) -> String { + let mut stream = tokio::net::TcpStream::connect(address) + .await + .expect("metrics listener connection"); + stream + .write_all( + format!("GET {path} HTTP/1.1\r\nHost: {address}\r\nConnection: close\r\n\r\n") + .as_bytes(), + ) + .await + .expect("metrics request"); + let mut response = Vec::new(); + stream + .read_to_end(&mut response) + .await + .expect("metrics response"); + String::from_utf8(response).expect("utf-8 response") +} diff --git a/crates/crank-observability/tests/prometheus.rs b/crates/crank-observability/tests/prometheus.rs index 95768aa..2ebe971 100644 --- a/crates/crank-observability/tests/prometheus.rs +++ b/crates/crank-observability/tests/prometheus.rs @@ -8,6 +8,7 @@ use crank_observability::{ DURATION_BUCKETS_SECONDS, MetricsConfig, MetricsConfigError, MetricsSurface, ServiceIdentity, metric_schema, }; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tower::ServiceExt; fn identity() -> ServiceIdentity { @@ -27,6 +28,33 @@ fn loopback_is_allowed_without_a_token() { assert!(!config.requires_authentication()); } +#[tokio::test] +async fn default_service_ports_bind_and_serve_real_metrics_listeners() { + for (service, port) in [("admin-api", 9464), ("mcp-server", 9465)] { + let config = MetricsConfig::new( + true, + SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port), + None, + ) + .expect("default loopback metrics config"); + let identity = + ServiceIdentity::try_new(service, "0.3.1", "test").expect("valid service identity"); + let server = MetricsSurface::for_test(config, identity) + .expect("metrics surface") + .bind() + .await + .expect("default metrics port must bind"); + assert_eq!(server.local_addr().expect("local address").port(), port); + + let task = tokio::spawn(server.serve()); + let response = raw_get(port, "/metrics").await; + assert!(response.starts_with("HTTP/1.1 200 OK"), "{response}"); + assert!(response.contains("content-type: text/plain")); + task.abort(); + let _ = task.await; + } +} + #[test] fn non_loopback_without_a_token_is_rejected_without_secret_data() { let error = MetricsConfig::new( @@ -51,8 +79,13 @@ fn schema_is_closed_and_uses_fixed_duration_buckets() { assert!(names.contains(&"crank_http_requests_total")); assert!(names.contains(&"crank_http_request_duration_seconds")); assert!(names.contains(&"crank_mcp_requests_total")); + assert!(names.contains(&"crank_mcp_active_sessions")); + assert!(names.contains(&"crank_mcp_active_streams")); assert!(names.contains(&"crank_tool_invocations_total")); assert!(names.contains(&"crank_runtime_inflight")); + assert!(names.contains(&"crank_runtime_cache_total")); + assert!(names.contains(&"crank_idempotency_total")); + assert!(names.contains(&"crank_confirmation_total")); assert!(names.contains(&"crank_db_pool_connections")); assert!(names.contains(&"crank_catalog_tools")); assert!(names.contains(&"crank_invocation_history_lost_total")); @@ -108,7 +141,7 @@ async fn external_surface_protects_both_routes_and_exposes_nothing_else() { .clone() .oneshot( Request::get(path) - .header(header::AUTHORIZATION, format!("Bearer {token}")) + .header(header::AUTHORIZATION, format!("bEaReR {token}")) .body(Body::empty()) .expect("request"), ) @@ -121,14 +154,35 @@ async fn external_surface_protects_both_routes_and_exposes_nothing_else() { assert!(!String::from_utf8_lossy(&body).contains(token)); } - let absent = app - .oneshot( - Request::get("/api/operations") - .header(header::AUTHORIZATION, format!("Bearer {token}")) - .body(Body::empty()) - .expect("request"), + for authorization in [None, Some(format!("Bearer {token}"))] { + let mut request = Request::get("/api/operations"); + if let Some(authorization) = authorization { + request = request.header(header::AUTHORIZATION, authorization); + } + let absent = app + .clone() + .oneshot(request.body(Body::empty()).expect("request")) + .await + .expect("response"); + assert_eq!(absent.status(), StatusCode::NOT_FOUND); + } +} + +async fn raw_get(port: u16, path: &str) -> String { + let mut stream = tokio::net::TcpStream::connect((Ipv4Addr::LOCALHOST, port)) + .await + .expect("metrics listener connection"); + stream + .write_all( + format!("GET {path} HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\nConnection: close\r\n\r\n") + .as_bytes(), ) .await - .expect("response"); - assert_eq!(absent.status(), StatusCode::NOT_FOUND); + .expect("metrics request"); + let mut response = Vec::new(); + stream + .read_to_end(&mut response) + .await + .expect("metrics response"); + String::from_utf8(response).expect("utf-8 response") } diff --git a/crates/crank-runtime/Cargo.toml b/crates/crank-runtime/Cargo.toml index 39f7386..707a90b 100644 --- a/crates/crank-runtime/Cargo.toml +++ b/crates/crank-runtime/Cargo.toml @@ -19,10 +19,10 @@ base64.workspace = true crank-adapter-rest = { path = "../crank-adapter-rest" } crank-core = { path = "../crank-core" } crank-mapping = { path = "../crank-mapping" } +crank-metrics = { path = "../crank-metrics" } crank-schema = { path = "../crank-schema" } crank-trace = { path = "../crank-trace" } hkdf.workspace = true -metrics.workspace = true redis = { version = "0.29", features = ["tokio-comp", "connection-manager"] } serde.workspace = true serde_json.workspace = true @@ -36,6 +36,8 @@ uuid.workspace = true [dev-dependencies] axum.workspace = true futures-util = "0.3" +metrics.workspace = true +metrics-util = "0.20.4" testcontainers.workspace = true time.workspace = true tracing-subscriber.workspace = true diff --git a/crates/crank-runtime/src/confirmation.rs b/crates/crank-runtime/src/confirmation.rs index c4fc20d..f977bd0 100644 --- a/crates/crank-runtime/src/confirmation.rs +++ b/crates/crank-runtime/src/confirmation.rs @@ -38,7 +38,14 @@ pub async fn confirm_operation( .or_else(|| confirmation_token_from_input(input)); let Some(provided_token) = provided_token else { - let token = issue_confirmation_token(store, &scope, &input_hash, &safety).await?; + let token = issue_confirmation_token( + store, + operation.operation_id.as_str(), + &scope, + &input_hash, + &safety, + ) + .await?; return Err(RuntimeError::ConfirmationRequired { operation_id: operation.operation_id.as_str().to_owned(), safety_class: safety.class, @@ -47,7 +54,14 @@ pub async fn confirm_operation( }); }; - consume_confirmation_token(store, &scope, provided_token, &input_hash).await + consume_confirmation_token( + store, + operation.operation_id.as_str(), + &scope, + provided_token, + &input_hash, + ) + .await } pub(crate) fn is_applicable(operation: &RuntimeOperation) -> bool { @@ -108,6 +122,7 @@ fn confirmation_scope( async fn issue_confirmation_token( store: &dyn CoordinationStateStore, + operation_id: &str, scope: &str, input_hash: &str, safety: &OperationSafetyPolicy, @@ -128,15 +143,15 @@ async fn issue_confirmation_token( Duration::from_millis(ttl_ms), ) .await - .map_err(|error| RuntimeError::InvalidPreparedRequest { - field: "confirmation_token".to_owned(), - reason: error.to_string(), + .map_err(|_| RuntimeError::ConfirmationStoreUnavailable { + operation_id: operation_id.to_owned(), })?; Ok(token) } async fn consume_confirmation_token( store: &dyn CoordinationStateStore, + operation_id: &str, operation_scope: &str, token: &str, input_hash: &str, @@ -145,9 +160,8 @@ async fn consume_confirmation_token( let stored = store .take_value(CacheScope::Coordination, &key) .await - .map_err(|error| RuntimeError::InvalidPreparedRequest { - field: "confirmation_token".to_owned(), - reason: error.to_string(), + .map_err(|_| RuntimeError::ConfirmationStoreUnavailable { + operation_id: operation_id.to_owned(), })?; let Some(stored) = stored else { diff --git a/crates/crank-runtime/src/executor.rs b/crates/crank-runtime/src/executor.rs index bce29f2..5839e47 100644 --- a/crates/crank-runtime/src/executor.rs +++ b/crates/crank-runtime/src/executor.rs @@ -5,12 +5,17 @@ use crank_core::{ AdapterRegistry, CoordinationStateStore, ExecutionMode, InvocationSource, InvocationStatus, MeteringEvent, ResponseCacheStore, SharedMeteringSink, SharedProtocolAdapter, }; +use crank_metrics::{ + CacheOutcome, ConfirmationOutcome, IdempotencyOutcome, InFlightGuard, + InvocationSource as MetricInvocationSource, LimitStage, ToolErrorKind, ToolInvocationMetrics, + ToolOutcome, record_cache_outcome, record_confirmation_outcome, record_idempotency_outcome, + record_limit_rejection, +}; use crank_trace::{ErrorCategory, Stage, StageOutcome}; -use metrics::Gauge; use serde_json::{Map, Value, json}; use time::OffsetDateTime; use tokio::sync::{OwnedSemaphorePermit, Semaphore}; -use tracing::{Instrument, Span, debug}; +use tracing::{Instrument, Span, debug, warn}; use uuid::Uuid; use crate::{ @@ -42,6 +47,65 @@ pub struct RuntimeExecutionRequest<'a> { pub request_context: Option<&'a RuntimeRequestContext>, } +struct IdempotencyCancellationGuard { + cleanup: Option, + runtime: tokio::runtime::Handle, +} + +struct IdempotencyCancellationCleanup { + store: Arc, + operation: RuntimeOperation, + reservation: crate::idempotency::IdempotencyReservation, +} + +impl IdempotencyCancellationGuard { + fn new( + store: Arc, + operation: RuntimeOperation, + reservation: crate::idempotency::IdempotencyReservation, + ) -> Self { + Self { + cleanup: Some(IdempotencyCancellationCleanup { + store, + operation, + reservation, + }), + runtime: tokio::runtime::Handle::current(), + } + } + + fn disarm(&mut self) { + self.cleanup = None; + } +} + +impl Drop for IdempotencyCancellationGuard { + fn drop(&mut self) { + let Some(cleanup) = self.cleanup.take() else { + return; + }; + self.runtime.spawn(async move { + let result = crate::idempotency::mark_outcome_unknown( + cleanup.store.as_ref(), + &cleanup.operation, + &cleanup.reservation, + ) + .await; + record_idempotency_outcome(match &result { + Ok(()) => IdempotencyOutcome::OutcomeUnknown, + Err(error) => idempotency_error_outcome(error), + }); + if result.is_err() { + warn!( + name: "runtime.idempotency.cancellation_cleanup_failed", + error_category = "idempotency_store", + "failed to finalize cancelled idempotent execution" + ); + } + }); + } +} + impl<'a> RuntimeExecutionRequest<'a> { pub fn new(operation: &'a RuntimeOperation, input: &'a Value) -> Self { Self { @@ -180,10 +244,12 @@ impl RuntimeExecutor { ) -> Result { log_runtime_event("unary.execute", request.operation, request.request_context); let started_at = Instant::now(); + let invocation_metrics = + ToolInvocationMetrics::start(metric_invocation_source(request.request_context)); let runtime_span = Stage::RuntimeExecute.span(); let result = async { let _permit = self.acquire_unary_permit(request.operation)?; - let _inflight = RuntimeInFlightGuard::new(); + let _inflight = InFlightGuard::runtime(); let mapping_span = Stage::RuntimeArgumentsMap.span(); let prepared_request = mapping_span.in_scope(|| self.prepare_request(request.operation, request.input)); @@ -203,7 +269,11 @@ impl RuntimeExecutor { .await; record_runtime_result(&runtime_span, &result); drop(runtime_span); - record_execution_metrics(request.request_context, &result, started_at); + let (outcome, error_kind) = match &result { + Ok(_) => (ToolOutcome::Success, ToolErrorKind::None), + Err(error) => (ToolOutcome::Error, runtime_error_kind(error)), + }; + invocation_metrics.complete(outcome, error_kind); self.record_metering( request.operation, request.request_context, @@ -245,6 +315,7 @@ impl RuntimeExecutor { ) { Ok(key) => key, Err(error) if idempotency_applicable => { + record_idempotency_outcome(idempotency_error_outcome(&error)); let span = Stage::RuntimeIdempotency.span(); StageOutcome::Error.record(&span); ErrorCategory::Idempotency.record(&span); @@ -264,14 +335,19 @@ impl RuntimeExecutor { .instrument(approval_span.clone()) .await; match &approval_result { - Ok(()) => StageOutcome::Success.record(&approval_span), + Ok(()) => { + StageOutcome::Success.record(&approval_span); + record_confirmation_outcome(ConfirmationOutcome::Approved); + } Err(RuntimeError::ConfirmationRequired { .. }) => { StageOutcome::Required.record(&approval_span); ErrorCategory::Approval.record(&approval_span); + record_confirmation_outcome(ConfirmationOutcome::Required); } Err(error) => { StageOutcome::Error.record(&approval_span); runtime_error_category(error).record(&approval_span); + record_confirmation_outcome(confirmation_error_outcome(error)); } } drop(approval_span); @@ -292,9 +368,11 @@ impl RuntimeExecutor { match &result { Ok(crate::idempotency::IdempotencyAction::Execute(_)) => { StageOutcome::Execute.record(&idempotency_span); + record_idempotency_outcome(IdempotencyOutcome::Execute); } Ok(crate::idempotency::IdempotencyAction::Replay(_)) => { StageOutcome::Replay.record(&idempotency_span); + record_idempotency_outcome(IdempotencyOutcome::Replay); } Ok(crate::idempotency::IdempotencyAction::Disabled) => { StageOutcome::Skipped.record(&idempotency_span); @@ -302,6 +380,7 @@ impl RuntimeExecutor { Err(error) => { StageOutcome::Error.record(&idempotency_span); runtime_error_category(error).record(&idempotency_span); + record_idempotency_outcome(idempotency_error_outcome(error)); } } drop(idempotency_span); @@ -312,6 +391,18 @@ impl RuntimeExecutor { if let crate::idempotency::IdempotencyAction::Replay(response) = &idempotency { return transform_response(operation, response); } + let mut cancellation_guard = + if let crate::idempotency::IdempotencyAction::Execute(reservation) = &idempotency { + self.coordination_store.as_ref().map(|store| { + IdempotencyCancellationGuard::new( + Arc::clone(store), + operation.clone(), + reservation.clone(), + ) + }) + } else { + None + }; let adapter_result = match self .load_cached_adapter_response(operation, &prepared_request, request_context) @@ -346,6 +437,13 @@ impl RuntimeExecutor { .instrument(idempotency_span.clone()) .await; record_runtime_result(&idempotency_span, &cleanup_result); + record_idempotency_outcome(match &cleanup_result { + Ok(()) => IdempotencyOutcome::OutcomeUnknown, + Err(error) => idempotency_error_outcome(error), + }); + if let Some(guard) = &mut cancellation_guard { + guard.disarm(); + } } return Err(error); } @@ -359,6 +457,13 @@ impl RuntimeExecutor { .instrument(idempotency_span.clone()) .await; record_runtime_result(&idempotency_span, &completion_result); + record_idempotency_outcome(match &completion_result { + Ok(()) => IdempotencyOutcome::Completed, + Err(error) => idempotency_error_outcome(error), + }); + if let Some(guard) = &mut cancellation_guard { + guard.disarm(); + } drop(idempotency_span); completion_result?; } @@ -447,8 +552,13 @@ impl RuntimeExecutor { let response_cache = self.response_cache.as_ref()?; let cache_key = response_cache_key(operation, prepared_request, request_context)?; let cached = match response_cache.get(&cache_key).await { - Ok(cached) => cached?, + Ok(Some(cached)) => cached, + Ok(None) => { + record_cache_outcome(CacheOutcome::Miss); + return None; + } Err(_) => { + record_cache_outcome(CacheOutcome::ReadError); debug!( name: "runtime.response_cache.read_failed", operation_id = operation.operation_id.as_str(), @@ -460,15 +570,21 @@ impl RuntimeExecutor { }; match adapter_response_from_cached(cached) { - Ok(response) => Some(response), + Ok(response) => { + record_cache_outcome(CacheOutcome::Hit); + Some(response) + } Err(_) => { + record_cache_outcome(CacheOutcome::DecodeError); debug!( name: "runtime.response_cache.decode_failed", operation_id = operation.operation_id.as_str(), error_category = "cached_response", "cached response payload was invalid" ); - let _ = response_cache.delete(&cache_key).await; + if response_cache.delete(&cache_key).await.is_err() { + record_cache_outcome(CacheOutcome::EvictError); + } None } } @@ -505,12 +621,15 @@ impl RuntimeExecutor { .await .is_err() { + record_cache_outcome(CacheOutcome::WriteError); debug!( name: "runtime.response_cache.write_failed", operation_id = operation.operation_id.as_str(), error_category = "response_cache", "response cache write skipped" ); + } else { + record_cache_outcome(CacheOutcome::Stored); } } } @@ -634,86 +753,64 @@ fn try_acquire_limit( limit: usize, ) -> Result { limiter.try_acquire_owned().map_err(|_| { - metrics::counter!( - "crank_runtime_limit_rejections_total", - "stage" => "concurrency" - ) - .increment(1); + record_limit_rejection(LimitStage::Concurrency); RuntimeError::ConcurrencyLimitExceeded { kind, limit } }) } -fn record_execution_metrics( +fn metric_invocation_source( request_context: Option<&RuntimeRequestContext>, - result: &Result, - started_at: Instant, -) { - let source = request_context +) -> MetricInvocationSource { + request_context .and_then(RuntimeRequestContext::metering_context) - .map_or("internal", |context| match context.source { - InvocationSource::AdminTestRun => "admin_test_run", - InvocationSource::AgentToolCall => "agent_tool_call", - }); - let (outcome, error_kind) = match result { - Ok(_) => ("success", "none"), - Err(error) => ("error", runtime_error_kind(error)), - }; - - metrics::counter!( - "crank_tool_invocations_total", - "source" => source, - "outcome" => outcome, - "error_kind" => error_kind - ) - .increment(1); - metrics::histogram!( - "crank_tool_invocation_duration_seconds", - "source" => source, - "outcome" => outcome - ) - .record(started_at.elapsed().as_secs_f64()); + .map_or(MetricInvocationSource::Internal, |context| { + match context.source { + InvocationSource::AdminTestRun => MetricInvocationSource::AdminTestRun, + InvocationSource::AgentToolCall => MetricInvocationSource::AgentToolCall, + } + }) } -fn runtime_error_kind(error: &RuntimeError) -> &'static str { +fn runtime_error_kind(error: &RuntimeError) -> ToolErrorKind { match error { - RuntimeError::Schema(_) => "schema", - RuntimeError::Mapping(_) => "mapping", - RuntimeError::RestAdapter(_) => "rest_adapter", - RuntimeError::ProtocolAdapter(_) => "protocol_adapter", - RuntimeError::UnsupportedProtocol { .. } => "unsupported_protocol", - RuntimeError::UnsupportedExecutionMode { .. } => "unsupported_execution_mode", - RuntimeError::ConcurrencyLimitExceeded { .. } => "concurrency_limit", - RuntimeError::InvalidPreparedRequest { .. } => "invalid_prepared_request", - RuntimeError::ConfirmationRequired { .. } => "confirmation_required", - RuntimeError::InvalidConfirmationToken { .. } => "invalid_confirmation_token", - RuntimeError::ConfirmationStoreUnavailable { .. } => "confirmation_store", - RuntimeError::IdempotencyStoreUnavailable { .. } => "idempotency_store", - RuntimeError::IdempotencyInProgress { .. } => "idempotency_in_progress", - RuntimeError::IdempotencyConflict { .. } => "idempotency_conflict", - RuntimeError::IdempotencyOutcomeUnknown { .. } => "idempotency_outcome_unknown", - RuntimeError::MissingAuthProfile { .. } => "missing_auth_profile", - RuntimeError::MissingSecret { .. } => "missing_secret", - RuntimeError::MissingSecretVersion { .. } => "missing_secret_version", - RuntimeError::InvalidAuthSecretValue { .. } => "invalid_auth_secret", - RuntimeError::SecretCrypto { .. } => "secret_crypto", + RuntimeError::Schema(_) => ToolErrorKind::Schema, + RuntimeError::Mapping(_) => ToolErrorKind::Mapping, + RuntimeError::RestAdapter(_) => ToolErrorKind::RestAdapter, + RuntimeError::ProtocolAdapter(_) => ToolErrorKind::ProtocolAdapter, + RuntimeError::UnsupportedProtocol { .. } => ToolErrorKind::UnsupportedProtocol, + RuntimeError::UnsupportedExecutionMode { .. } => ToolErrorKind::UnsupportedExecutionMode, + RuntimeError::ConcurrencyLimitExceeded { .. } => ToolErrorKind::ConcurrencyLimit, + RuntimeError::InvalidPreparedRequest { .. } => ToolErrorKind::InvalidPreparedRequest, + RuntimeError::ConfirmationRequired { .. } => ToolErrorKind::ConfirmationRequired, + RuntimeError::InvalidConfirmationToken { .. } => ToolErrorKind::InvalidConfirmationToken, + RuntimeError::ConfirmationStoreUnavailable { .. } => ToolErrorKind::ConfirmationStore, + RuntimeError::IdempotencyStoreUnavailable { .. } => ToolErrorKind::IdempotencyStore, + RuntimeError::IdempotencyInProgress { .. } => ToolErrorKind::IdempotencyInProgress, + RuntimeError::IdempotencyConflict { .. } => ToolErrorKind::IdempotencyConflict, + RuntimeError::IdempotencyOutcomeUnknown { .. } => ToolErrorKind::IdempotencyOutcomeUnknown, + RuntimeError::MissingAuthProfile { .. } => ToolErrorKind::MissingAuthProfile, + RuntimeError::MissingSecret { .. } => ToolErrorKind::MissingSecret, + RuntimeError::MissingSecretVersion { .. } => ToolErrorKind::MissingSecretVersion, + RuntimeError::InvalidAuthSecretValue { .. } => ToolErrorKind::InvalidAuthSecret, + RuntimeError::SecretCrypto { .. } => ToolErrorKind::SecretCrypto, } } -struct RuntimeInFlightGuard { - gauge: Gauge, -} - -impl RuntimeInFlightGuard { - fn new() -> Self { - let gauge = metrics::gauge!("crank_runtime_inflight"); - gauge.increment(1.0); - Self { gauge } +fn idempotency_error_outcome(error: &RuntimeError) -> IdempotencyOutcome { + match error { + RuntimeError::IdempotencyConflict { .. } => IdempotencyOutcome::Conflict, + RuntimeError::IdempotencyInProgress { .. } => IdempotencyOutcome::InProgress, + RuntimeError::IdempotencyOutcomeUnknown { .. } => IdempotencyOutcome::OutcomeUnknown, + RuntimeError::IdempotencyStoreUnavailable { .. } => IdempotencyOutcome::StoreUnavailable, + _ => IdempotencyOutcome::Error, } } -impl Drop for RuntimeInFlightGuard { - fn drop(&mut self) { - self.gauge.decrement(1.0); +fn confirmation_error_outcome(error: &RuntimeError) -> ConfirmationOutcome { + match error { + RuntimeError::InvalidConfirmationToken { .. } => ConfirmationOutcome::InvalidToken, + RuntimeError::ConfirmationStoreUnavailable { .. } => ConfirmationOutcome::StoreUnavailable, + _ => ConfirmationOutcome::Error, } } diff --git a/crates/crank-runtime/src/idempotency.rs b/crates/crank-runtime/src/idempotency.rs index 998ad49..d909b76 100644 --- a/crates/crank-runtime/src/idempotency.rs +++ b/crates/crank-runtime/src/idempotency.rs @@ -21,6 +21,7 @@ pub(crate) enum IdempotencyAction { Replay(AdapterResponse), } +#[derive(Clone)] pub(crate) struct IdempotencyReservation { key: String, initial: CoordinationStateValue, diff --git a/crates/crank-runtime/src/rate_limit.rs b/crates/crank-runtime/src/rate_limit.rs index d8192c8..34776c6 100644 --- a/crates/crank-runtime/src/rate_limit.rs +++ b/crates/crank-runtime/src/rate_limit.rs @@ -5,6 +5,7 @@ use std::{ }; use crank_core::{RateLimitDecision, RateLimitStateStore}; +use crank_metrics::{LimitStage, record_limit_rejection}; use thiserror::Error; use tracing::warn; @@ -105,11 +106,7 @@ impl RequestRateLimiter { } }; if matches!(result, Err(RateLimitCheckError::Rejected(_))) { - metrics::counter!( - "crank_runtime_limit_rejections_total", - "stage" => "rate_limit" - ) - .increment(1); + record_limit_rejection(LimitStage::RateLimit); } result } diff --git a/crates/crank-runtime/tests/integration/confirmation.rs b/crates/crank-runtime/tests/integration/confirmation.rs index 993839e..5c8a48f 100644 --- a/crates/crank-runtime/tests/integration/confirmation.rs +++ b/crates/crank-runtime/tests/integration/confirmation.rs @@ -6,8 +6,9 @@ use std::sync::{ use async_trait::async_trait; use crank_core::{ - AdapterResponse, ConfirmationPolicy, ExecutionConfig, ExecutionMode, HttpMethod, Operation, - OperationId, OperationSafetyClass, OperationSafetyPolicy, OperationSecurityLevel, + AdapterResponse, CacheScope, CacheStoreError, ConfirmationPolicy, CoordinationStateReservation, + CoordinationStateStore, CoordinationStateValue, ExecutionConfig, ExecutionMode, HttpMethod, + Operation, OperationId, OperationSafetyClass, OperationSafetyPolicy, OperationSecurityLevel, OperationStatus, Protocol, ProtocolAdapter, ProtocolAdapterError, RestTarget, Target, ToolDescription, }; @@ -146,6 +147,111 @@ async fn confirmation_token_allows_only_one_concurrent_execution() { ); } +#[tokio::test] +async fn unavailable_confirmation_store_is_preserved_as_a_typed_error() { + let executor = RuntimeExecutorBuilder::new() + .register_adapter(Arc::new(CountingAdapter { + call_count: Arc::new(AtomicUsize::new(0)), + })) + .with_coordination_store(Arc::new(UnavailableCoordinationStore)) + .build(); + let operation: crank_runtime::RuntimeOperation = destructive_delete_operation().into(); + let context = RuntimeRequestContext::from_request_id("req_confirm_unavailable") + .with_response_cache_scope("workspace_1", "agent_1"); + + let issue_error = executor + .execute_with_context( + &operation, + &json!({ "order_id": "ord_123" }), + Some(&context), + ) + .await + .expect_err("unavailable store must prevent issuing a token"); + assert!(matches!( + issue_error, + RuntimeError::ConfirmationStoreUnavailable { ref operation_id } + if operation_id == "op_delete_order" + )); + + let consume_error = executor + .execute_with_context( + &operation, + &json!({ "order_id": "ord_123" }), + Some(&context.with_confirmation_token("ct_unavailable")), + ) + .await + .expect_err("unavailable store must not look like an invalid token"); + assert!(matches!( + consume_error, + RuntimeError::ConfirmationStoreUnavailable { ref operation_id } + if operation_id == "op_delete_order" + )); +} + +struct UnavailableCoordinationStore; + +impl UnavailableCoordinationStore { + fn error() -> CacheStoreError { + CacheStoreError::Unavailable { + message: "test backend unavailable".to_owned(), + } + } +} + +#[async_trait] +impl CoordinationStateStore for UnavailableCoordinationStore { + async fn get_value( + &self, + _scope: CacheScope, + _key: &str, + ) -> Result, CacheStoreError> { + Err(Self::error()) + } + + async fn put_value( + &self, + _scope: CacheScope, + _key: &str, + _value: CoordinationStateValue, + _ttl: std::time::Duration, + ) -> Result<(), CacheStoreError> { + Err(Self::error()) + } + + async fn delete_value(&self, _scope: CacheScope, _key: &str) -> Result<(), CacheStoreError> { + Err(Self::error()) + } + + async fn take_value( + &self, + _scope: CacheScope, + _key: &str, + ) -> Result, CacheStoreError> { + Err(Self::error()) + } + + async fn reserve_value( + &self, + _scope: CacheScope, + _key: &str, + _value: CoordinationStateValue, + _ttl: std::time::Duration, + ) -> Result { + Err(Self::error()) + } + + async fn compare_and_set_value( + &self, + _scope: CacheScope, + _key: &str, + _expected: &CoordinationStateValue, + _value: CoordinationStateValue, + _ttl: std::time::Duration, + ) -> Result { + Err(Self::error()) + } +} + struct CountingAdapter { call_count: Arc, } diff --git a/crates/crank-runtime/tests/product_metrics.rs b/crates/crank-runtime/tests/product_metrics.rs new file mode 100644 index 0000000..ce4dbee --- /dev/null +++ b/crates/crank-runtime/tests/product_metrics.rs @@ -0,0 +1,414 @@ +use std::{ + collections::BTreeMap, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, +}; + +use async_trait::async_trait; +use crank_core::{ + AdapterResponse, ConfirmationPolicy, ExecutionConfig, ExecutionMode, HttpMethod, + IdempotencyMode, IdempotencyPolicy, Operation, OperationId, OperationSafetyClass, + OperationSafetyPolicy, OperationSecurityLevel, OperationStatus, Protocol, ProtocolAdapter, + ProtocolAdapterError, ResponseCachePolicy, RestTarget, + RuntimeRequestContext as ProtocolRequestContext, Target, ToolDescription, +}; +use crank_mapping::{MappingRule, MappingSet}; +use crank_runtime::{ + InMemoryCoordinationStateStore, InMemoryResponseCacheStore, RuntimeError, + RuntimeExecutorBuilder, RuntimeRequestContext, +}; +use crank_schema::{Schema, SchemaKind}; +use metrics_util::debugging::DebuggingRecorder; +use serde_json::json; +use time::OffsetDateTime; + +#[tokio::test] +async fn real_runtime_paths_emit_cache_idempotency_and_confirmation_outcomes() { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + recorder + .install() + .expect("isolated integration test recorder"); + + exercise_response_cache().await; + exercise_idempotency().await; + exercise_cancelled_idempotency().await; + exercise_confirmation().await; + + let snapshot = snapshotter.snapshot().into_vec(); + for outcome in ["miss", "stored", "hit"] { + assert!(has_series( + &snapshot, + "crank_runtime_cache_total", + "outcome", + outcome + )); + } + for outcome in [ + "execute", + "completed", + "replay", + "conflict", + "outcome_unknown", + ] { + assert!(has_series( + &snapshot, + "crank_idempotency_total", + "outcome", + outcome + )); + } + for outcome in ["required", "approved", "invalid_token"] { + assert!( + has_series(&snapshot, "crank_confirmation_total", "outcome", outcome), + "missing confirmation outcome {outcome}: {snapshot:?}" + ); + } +} + +async fn exercise_cancelled_idempotency() { + let calls = Arc::new(AtomicUsize::new(0)); + let executor = RuntimeExecutorBuilder::new() + .register_adapter(Arc::new(BlockingAdapter { + calls: Arc::clone(&calls), + })) + .with_coordination_store(Arc::new(InMemoryCoordinationStateStore::default())) + .build(); + let operation: crank_runtime::RuntimeOperation = operation( + "cancelled_idempotent_write", + HttpMethod::Post, + None, + Some(IdempotencyPolicy { + mode: IdempotencyMode::Required, + ttl_ms: 60_000, + input_field: Some("key".to_owned()), + header_name: Some("Idempotency-Key".to_owned()), + }), + None, + ) + .into(); + let context = RuntimeRequestContext::from_request_id("req_cancelled_idempotency") + .with_response_cache_scope("workspace_cancelled", "agent_cancelled"); + let input = json!({"key": "cancelled-key"}); + + let running_executor = executor.clone(); + let running_operation = operation.clone(); + let running_context = context.clone(); + let running_input = input.clone(); + let task = tokio::spawn(async move { + running_executor + .execute_with_context(&running_operation, &running_input, Some(&running_context)) + .await + }); + wait_for_calls(&calls, 1).await; + task.abort(); + let _ = task.await; + + let retry = executor + .execute_with_context(&operation, &input, Some(&context)) + .await + .expect_err("cancelled external outcome must not be retried automatically"); + assert!(matches!( + retry, + RuntimeError::IdempotencyOutcomeUnknown { .. } + )); + assert_eq!(calls.load(Ordering::SeqCst), 1); +} + +async fn exercise_response_cache() { + let calls = Arc::new(AtomicUsize::new(0)); + let executor = executor(Arc::clone(&calls)); + let operation: crank_runtime::RuntimeOperation = operation( + "cached_lookup", + HttpMethod::Get, + Some(ResponseCachePolicy { ttl_ms: 60_000 }), + None, + None, + ) + .into(); + let context = RuntimeRequestContext::from_request_id("req_cache") + .with_response_cache_scope("workspace_cache", "agent_cache"); + + executor + .execute_with_context(&operation, &json!({"key": "cache-key"}), Some(&context)) + .await + .unwrap(); + executor + .execute_with_context(&operation, &json!({"key": "cache-key"}), Some(&context)) + .await + .unwrap(); + + assert_eq!(calls.load(Ordering::SeqCst), 1); +} + +async fn exercise_idempotency() { + let calls = Arc::new(AtomicUsize::new(0)); + let executor = executor(Arc::clone(&calls)); + let operation: crank_runtime::RuntimeOperation = operation( + "idempotent_write", + HttpMethod::Post, + None, + Some(IdempotencyPolicy { + mode: IdempotencyMode::Required, + ttl_ms: 60_000, + input_field: Some("key".to_owned()), + header_name: Some("Idempotency-Key".to_owned()), + }), + None, + ) + .into(); + let context = RuntimeRequestContext::from_request_id("req_idempotency") + .with_response_cache_scope("workspace_idempotency", "agent_idempotency"); + + let first_input = json!({"key": "stable-key", "variant": "first"}); + executor + .execute_with_context(&operation, &first_input, Some(&context)) + .await + .unwrap(); + executor + .execute_with_context(&operation, &first_input, Some(&context)) + .await + .unwrap(); + let conflict = executor + .execute_with_context( + &operation, + &json!({"key": "stable-key", "variant": "different"}), + Some(&context), + ) + .await + .expect_err("same idempotency key with a different input must conflict"); + + assert!(matches!(conflict, RuntimeError::IdempotencyConflict { .. })); + assert_eq!(calls.load(Ordering::SeqCst), 1); +} + +async fn exercise_confirmation() { + let calls = Arc::new(AtomicUsize::new(0)); + let executor = executor(Arc::clone(&calls)); + let operation: crank_runtime::RuntimeOperation = operation( + "destructive_write", + HttpMethod::Delete, + None, + None, + Some(OperationSafetyPolicy { + class: OperationSafetyClass::Destructive, + confirmation: Some(ConfirmationPolicy { ttl_ms: 60_000 }), + }), + ) + .into(); + let context = RuntimeRequestContext::from_request_id("req_confirmation") + .with_response_cache_scope("workspace_confirmation", "agent_confirmation"); + let input = json!({"key": "delete-key"}); + + let required = executor + .execute_with_context(&operation, &input, Some(&context)) + .await + .expect_err("destructive operation must require confirmation"); + let RuntimeError::ConfirmationRequired { + confirmation_token, .. + } = required + else { + panic!("expected confirmation token"); + }; + let confirmed = context + .clone() + .with_confirmation_token(confirmation_token.clone()); + executor + .execute_with_context(&operation, &input, Some(&confirmed)) + .await + .unwrap(); + let invalid = executor + .execute_with_context(&operation, &input, Some(&confirmed)) + .await + .expect_err("confirmation token must be single-use"); + + assert!(matches!( + invalid, + RuntimeError::InvalidConfirmationToken { .. } + )); + assert_eq!(calls.load(Ordering::SeqCst), 1); +} + +fn executor(calls: Arc) -> crank_runtime::RuntimeExecutor { + RuntimeExecutorBuilder::new() + .register_adapter(Arc::new(CountingAdapter { calls })) + .with_response_cache(Arc::new(InMemoryResponseCacheStore::default())) + .with_coordination_store(Arc::new(InMemoryCoordinationStateStore::default())) + .build() +} + +struct CountingAdapter { + calls: Arc, +} + +struct BlockingAdapter { + calls: Arc, +} + +#[async_trait] +impl ProtocolAdapter for BlockingAdapter { + fn protocol(&self) -> Protocol { + Protocol::Rest + } + + fn supports_mode(&self, mode: ExecutionMode) -> bool { + mode == ExecutionMode::Unary + } + + async fn invoke_unary( + &self, + _target: &Target, + _prepared: &crank_core::PreparedRequest, + _context: &ProtocolRequestContext, + ) -> Result { + self.calls.fetch_add(1, Ordering::SeqCst); + std::future::pending().await + } +} + +#[async_trait] +impl ProtocolAdapter for CountingAdapter { + fn protocol(&self) -> Protocol { + Protocol::Rest + } + + fn supports_mode(&self, mode: ExecutionMode) -> bool { + mode == ExecutionMode::Unary + } + + async fn invoke_unary( + &self, + _target: &Target, + _prepared: &crank_core::PreparedRequest, + _context: &ProtocolRequestContext, + ) -> Result { + self.calls.fetch_add(1, Ordering::SeqCst); + Ok(AdapterResponse { + status_code: 200, + headers: BTreeMap::new(), + body: json!({"result": "ok"}), + data: json!({"result": "ok"}), + }) + } +} + +fn operation( + name: &str, + method: HttpMethod, + response_cache: Option, + idempotency: Option, + safety: Option, +) -> Operation { + Operation { + id: OperationId::new(format!("op_{name}")), + name: name.to_owned(), + display_name: name.to_owned(), + category: "metrics".to_owned(), + protocol: Protocol::Rest, + security_level: OperationSecurityLevel::Standard, + status: OperationStatus::Published, + version: 1, + target: Target::Rest(RestTarget { + base_url: "https://metrics.example.invalid".to_owned(), + method, + path_template: "/resource".to_owned(), + static_headers: BTreeMap::new(), + }), + input_schema: object_schema("key"), + output_schema: object_schema("result"), + input_mapping: MappingSet { rules: Vec::new() }, + output_mapping: MappingSet { + rules: vec![MappingRule { + source: "$.response.body.result".to_owned(), + target: "$.output.result".to_owned(), + required: true, + default_value: None, + transform: None, + condition: None, + notes: None, + }], + }, + execution_config: ExecutionConfig { + timeout_ms: 1_000, + retry_policy: None, + response_cache, + idempotency, + safety, + approval_policy: None, + auth_profile_ref: None, + headers: BTreeMap::new(), + }, + tool_description: ToolDescription { + title: name.to_owned(), + description: "Exercises a real runtime metrics path.".to_owned(), + tags: Vec::new(), + examples: Vec::new(), + }, + samples: None, + generated_draft: None, + config_export: None, + wizard_state: None, + created_at: OffsetDateTime::UNIX_EPOCH, + updated_at: OffsetDateTime::UNIX_EPOCH, + published_at: None, + } +} + +fn object_schema(field: &str) -> Schema { + Schema { + kind: SchemaKind::Object, + description: None, + required: true, + nullable: false, + default_value: None, + fields: BTreeMap::from([( + field.to_owned(), + Schema { + kind: SchemaKind::String, + description: None, + required: true, + nullable: false, + default_value: None, + fields: BTreeMap::new(), + items: None, + enum_values: Vec::new(), + variants: Vec::new(), + }, + )]), + items: None, + enum_values: Vec::new(), + variants: Vec::new(), + } +} + +fn has_series( + snapshot: &[( + metrics_util::CompositeKey, + Option, + Option, + metrics_util::debugging::DebugValue, + )], + metric: &str, + label_name: &str, + label_value: &str, +) -> bool { + snapshot.iter().any(|(key, _, _, _)| { + key.key().name() == metric + && key + .key() + .labels() + .any(|label| label.key() == label_name && label.value() == label_value) + }) +} + +async fn wait_for_calls(calls: &AtomicUsize, expected: usize) { + for _ in 0..1_000 { + if calls.load(Ordering::SeqCst) >= expected { + return; + } + tokio::task::yield_now().await; + } + panic!("adapter did not receive {expected} call(s)"); +} diff --git a/docs/observability.md b/docs/observability.md index 8df0cd6..1ae16e1 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -165,11 +165,14 @@ scrape_configs: - `crank_http_requests_total`, `crank_http_request_duration_seconds`, `crank_http_inflight`; -- `crank_mcp_requests_total`, `crank_mcp_active_sessions`; +- `crank_mcp_requests_total`, `crank_mcp_active_sessions`, + `crank_mcp_active_streams`; - `crank_tool_invocations_total`, `crank_tool_invocation_duration_seconds`; - `crank_upstream_requests_total`, `crank_upstream_request_duration_seconds`; -- `crank_runtime_inflight`, `crank_runtime_limit_rejections_total`; +- `crank_runtime_inflight`, `crank_runtime_limit_rejections_total`, + `crank_runtime_cache_total`, `crank_idempotency_total`, + `crank_confirmation_total`; - `crank_db_pool_connections`; - `crank_catalog_tools`, `crank_catalog_estimated_context_tokens`, `crank_catalog_warnings`; @@ -178,11 +181,25 @@ scrape_configs: Recorder добавляет к рядам проверенные статические labels `service`, `version`, `environment`. Остальные labels имеют закрытый набор значений. +Имена, типы и допустимые значения labels определены в независимом нижнем +crate `crank-metrics`. Product-код вызывает только его типизированный фасад; +прямая production-зависимость от `metrics` вне `crank-metrics` и +`crank-observability` запрещена архитектурной проверкой Cargo metadata. Запрещено использовать workspace, идентификаторы агента, операции или запроса, фактический URL, текст ошибки, payload и пользовательский текст. HTTP route берётся только из шаблона Axum; неизвестные маршруты и методы сворачиваются в `unmatched` и `OTHER`. +`crank_mcp_active_sessions` отражает число ещё не истёкших транспортных +сессий в общем хранилище, а `crank_mcp_active_streams` — число открытых в +этом экземпляре SSE-потоков. Прикладные ошибки JSON-RPC и +`tools/call.result.isError=true` учитываются отдельно от успешного HTTP 200. +Счётчики cache, idempotency и confirmation описывают переходы +соответствующих стадий: hit/miss и ошибки хранилища, execute/replay/conflict +и required/approved/invalid token. Отменённый future runtime или upstream +завершает соответствующее измерение закрытым исходом `aborted`, поэтому +фактически начатый вызов не исчезает из показателей. + Buckets длительности фиксированы кодом: `0.005`, `0.01`, `0.025`, `0.05`, `0.1`, `0.25`, `0.5`, `1`, `2.5`, `5`, `10`, `30`, `60` секунд. Это пока техническая шкала, а не SLO. Фактические diff --git a/scripts/check-rust-boundaries.py b/scripts/check-rust-boundaries.py index d2fa1e0..89d1336 100755 --- a/scripts/check-rust-boundaries.py +++ b/scripts/check-rust-boundaries.py @@ -49,6 +49,8 @@ def package_category(name: str, manifest_path: Path, workspace_root: Path) -> st return "app" if name == "crank-core": return "core" + if name == "crank-metrics": + return "metrics" if name == "crank-observability": return "observability" if name == "crank-registry": @@ -92,6 +94,23 @@ def dependency_package_ids(raw_package: dict[str, Any], packages_by_name: dict[s return dependency_ids +def direct_dependency_violations(raw_package: dict[str, Any]) -> list[Violation]: + package_name = raw_package["name"] + if package_name in {"crank-metrics", "crank-observability"}: + return [] + + return [ + Violation( + source=package_name, + dependency="metrics", + reason="production crates must record metrics through crank-metrics", + ) + for dependency in raw_package.get("dependencies", []) + if dependency["name"] == "metrics" + and dependency.get("kind") != "dev" + ] + + def boundary_reason(source: Package, dependency: Package) -> str | None: if source.category == "app": if dependency.category == "app": @@ -101,7 +120,10 @@ def boundary_reason(source: Package, dependency: Package) -> str | None: if dependency.category == "app": return "workspace crates must not depend on apps" - if source.category == "observability": + if source.category == "metrics": + return "crank-metrics must not depend on other workspace crates" + + if source.category == "observability" and dependency.category != "metrics": return "crank-observability must not depend on other workspace crates" if ( @@ -133,6 +155,7 @@ def find_violations(metadata: dict[str, Any]) -> list[Violation]: for source in sorted(packages.values(), key=lambda package: package.name): raw_package = raw_packages_by_id[source.id] + violations.extend(direct_dependency_violations(raw_package)) for dependency_id in dependency_package_ids(raw_package, packages_by_name): dependency = packages[dependency_id] reason = boundary_reason(source, dependency) diff --git a/scripts/check-rust-module-boundaries.sh b/scripts/check-rust-module-boundaries.sh index cb01f22..91dcec6 100755 --- a/scripts/check-rust-module-boundaries.sh +++ b/scripts/check-rust-module-boundaries.sh @@ -47,6 +47,13 @@ check_no_match \ '^\s*use\s+(axum|sqlx)(::|[;\{])' \ "$ROOT_DIR/crates/crank-runtime/src" +check_no_match \ + "product modules must record metrics only through crank-metrics" \ + '(^|[^[:alnum:]_])(::)?metrics::(counter|gauge|histogram)!' \ + "$ROOT_DIR/apps" \ + "$ROOT_DIR/crates" \ + --glob '!**/crank-metrics/**' + if (( status != 0 )); then cat >&2 <<'EOF' @@ -58,6 +65,7 @@ Rules: - core remains framework/storage agnostic; - registry remains storage-only and HTTP-client agnostic; - runtime remains execution-only and storage/framework agnostic. +- names and labels of metrics remain inside the typed crank-metrics contract. EOF fi diff --git a/tests/unit/test_check_rust_boundaries.py b/tests/unit/test_check_rust_boundaries.py index a477617..503cfee 100644 --- a/tests/unit/test_check_rust_boundaries.py +++ b/tests/unit/test_check_rust_boundaries.py @@ -27,6 +27,10 @@ def package(root: Path, name: str, rel_dir: str, dependencies: list[str] | None } +def dependency(name: str, kind: str | None = None) -> dict: + return {"name": name, "kind": kind} + + def metadata(packages: list[dict], root: Path | None = None) -> dict: root = root or Path("/tmp/crank") return { @@ -124,6 +128,44 @@ class RustBoundaryCheckTests(unittest.TestCase): self.assertEqual(violations[0].source, "crank-observability") self.assertEqual(violations[0].dependency, dependency) + def test_allows_observability_and_product_crates_to_depend_on_metrics_contract(self) -> None: + packages = [ + package( + self.root, + "crank-observability", + "crates/crank-observability", + ["crank-metrics"], + ), + package( + self.root, + "crank-runtime", + "crates/crank-runtime", + ["crank-metrics"], + ), + package(self.root, "crank-metrics", "crates/crank-metrics"), + ] + + violations = self.checker.find_violations(metadata(packages, self.root)) + + self.assertEqual(violations, []) + + def test_rejects_metrics_contract_dependency_on_workspace_crates(self) -> None: + packages = [ + package( + self.root, + "crank-metrics", + "crates/crank-metrics", + ["crank-core"], + ), + package(self.root, "crank-core", "crates/crank-core"), + ] + + violations = self.checker.find_violations(metadata(packages, self.root)) + + self.assertEqual(len(violations), 1) + self.assertEqual(violations[0].source, "crank-metrics") + self.assertEqual(violations[0].dependency, "crank-core") + def test_rejects_domain_and_runtime_dependencies_on_observability(self) -> None: for source in ("crank-core", "crank-registry", "crank-runtime"): packages = [ @@ -146,6 +188,23 @@ class RustBoundaryCheckTests(unittest.TestCase): self.assertEqual(violations[0].source, source) self.assertEqual(violations[0].dependency, "crank-observability") + def test_rejects_direct_production_metrics_dependency_outside_contract(self) -> None: + app = package(self.root, "admin-api", "apps/admin-api") + app["dependencies"] = [dependency("metrics")] + + violations = self.checker.find_violations(metadata([app], self.root)) + + self.assertEqual(len(violations), 1) + self.assertEqual(violations[0].dependency, "metrics") + + def test_allows_metrics_as_test_only_dependency(self) -> None: + app = package(self.root, "admin-api", "apps/admin-api") + app["dependencies"] = [dependency("metrics", "dev")] + + violations = self.checker.find_violations(metadata([app], self.root)) + + self.assertEqual(violations, []) + if __name__ == "__main__": unittest.main()