From 59cd46774ea6526d144a304141ba7d8510793910 Mon Sep 17 00:00:00 2001 From: Sama Setty Date: Mon, 1 Jun 2026 18:30:18 -0700 Subject: [PATCH] Add streamable HTTP MCP failure metric --- codex-rs/Cargo.lock | 1 + codex-rs/rmcp-client/Cargo.toml | 1 + .../rmcp-client/src/http_client_adapter.rs | 188 +++++++++++++++++- codex-rs/rmcp-client/src/rmcp_client.rs | 9 +- 4 files changed, 196 insertions(+), 3 deletions(-) diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 27b66293e1..dfdff5ecc4 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3559,6 +3559,7 @@ dependencies = [ "codex-config", "codex-exec-server", "codex-keyring-store", + "codex-otel", "codex-protocol", "codex-utils-cargo-bin", "codex-utils-home-dir", diff --git a/codex-rs/rmcp-client/Cargo.toml b/codex-rs/rmcp-client/Cargo.toml index 63c6d74e4a..0cf9b51af8 100644 --- a/codex-rs/rmcp-client/Cargo.toml +++ b/codex-rs/rmcp-client/Cargo.toml @@ -20,6 +20,7 @@ codex-client = { workspace = true } codex-config = { workspace = true } codex-exec-server = { workspace = true } codex-keyring-store = { workspace = true } +codex-otel = { workspace = true } codex-protocol = { workspace = true } codex-utils-pty = { workspace = true } codex-utils-home-dir = { workspace = true } diff --git a/codex-rs/rmcp-client/src/http_client_adapter.rs b/codex-rs/rmcp-client/src/http_client_adapter.rs index 6f98f78920..3ace0b4685 100644 --- a/codex-rs/rmcp-client/src/http_client_adapter.rs +++ b/codex-rs/rmcp-client/src/http_client_adapter.rs @@ -46,9 +46,15 @@ const EVENT_STREAM_MIME_TYPE: &str = "text/event-stream"; const JSON_MIME_TYPE: &str = "application/json"; const HEADER_SESSION_ID: &str = "Mcp-Session-Id"; const NON_JSON_RESPONSE_BODY_PREVIEW_BYTES: usize = 8_192; +const STREAMABLE_HTTP_POST_MESSAGE_FAILURE_METRIC: &str = + "codex.mcp.streamable_http.post_message.failure"; +const CODEX_APPS_MCP_SERVER_NAME: &str = "codex_apps"; +const CHATGPT_WHAM_APPS_ENDPOINT_LABEL: &str = "chatgpt_wham_apps"; +const OTHER_METRIC_LABEL: &str = "other"; #[derive(Clone)] pub(crate) struct StreamableHttpClientAdapter { + server_name: String, http_client: Arc, default_headers: HeaderMap, auth_provider: Option, @@ -66,11 +72,13 @@ pub(crate) enum StreamableHttpClientAdapterError { impl StreamableHttpClientAdapter { pub(crate) fn new( + server_name: &str, http_client: Arc, default_headers: HeaderMap, auth_provider: Option, ) -> Self { Self { + server_name: server_name.to_string(), http_client, default_headers, auth_provider, @@ -140,12 +148,14 @@ impl StreamableHttpClient for StreamableHttpClientAdapter { let (response, mut body_stream) = match response { Ok(response) => response, Err(error) => { - log_post_message_http_error( + record_post_message_http_error( + &self.server_name, &uri, mcp_method.as_deref(), mcp_request_id.as_deref(), has_session_id, has_authorization_header, + &error, ); return Err(StreamableHttpError::Client( StreamableHttpClientAdapterError::from(error), @@ -385,14 +395,24 @@ fn client_jsonrpc_message_fields( } } -fn log_post_message_http_error( +fn record_post_message_http_error( + server_name: &str, uri: &str, mcp_method: Option<&str>, mcp_request_id: Option<&str>, has_session_id: bool, has_authorization_header: bool, + error: &ExecServerError, ) { let parsed_url = reqwest::Url::parse(uri).ok(); + emit_post_message_failure_metric(PostMessageFailureMetricFields { + server_name, + parsed_url: parsed_url.as_ref(), + mcp_method, + has_session_id, + has_authorization_header, + error, + }); tracing::warn!( endpoint_scheme = parsed_url .as_ref() @@ -415,6 +435,98 @@ fn log_post_message_http_error( ); } +struct PostMessageFailureMetricFields<'a> { + server_name: &'a str, + parsed_url: Option<&'a reqwest::Url>, + mcp_method: Option<&'a str>, + has_session_id: bool, + has_authorization_header: bool, + error: &'a ExecServerError, +} + +fn emit_post_message_failure_metric(fields: PostMessageFailureMetricFields<'_>) { + let Some(metrics) = codex_otel::global() else { + return; + }; + let tags = post_message_failure_metric_tags(fields); + let tag_refs: Vec<(&str, &str)> = tags + .iter() + .map(|(key, value)| (*key, value.as_str())) + .collect(); + let _ = metrics.counter( + STREAMABLE_HTTP_POST_MESSAGE_FAILURE_METRIC, + /*inc*/ 1, + &tag_refs, + ); +} + +fn post_message_failure_metric_tags( + fields: PostMessageFailureMetricFields<'_>, +) -> Vec<(&'static str, String)> { + vec![ + ( + "server_name", + mcp_server_metric_label(fields.server_name).to_string(), + ), + ( + "endpoint", + endpoint_metric_label(fields.parsed_url).to_string(), + ), + ( + "mcp_method", + codex_otel::sanitize_metric_tag_value(fields.mcp_method.unwrap_or("none")), + ), + ("has_session_id", fields.has_session_id.to_string()), + ( + "has_authorization_header", + fields.has_authorization_header.to_string(), + ), + ( + "error_type", + exec_server_error_type(fields.error).to_string(), + ), + ] +} + +fn mcp_server_metric_label(server_name: &str) -> &'static str { + if server_name == CODEX_APPS_MCP_SERVER_NAME { + CODEX_APPS_MCP_SERVER_NAME + } else { + OTHER_METRIC_LABEL + } +} + +fn endpoint_metric_label(parsed_url: Option<&reqwest::Url>) -> &'static str { + if parsed_url.is_some_and(|url| { + url.scheme() == "https" + && url.host_str() == Some("chatgpt.com") + && url.path() == "/backend-api/wham/apps" + }) { + CHATGPT_WHAM_APPS_ENDPOINT_LABEL + } else { + OTHER_METRIC_LABEL + } +} + +fn exec_server_error_type(error: &ExecServerError) -> &'static str { + match error { + ExecServerError::Spawn(_) => "spawn", + ExecServerError::WebSocketConnectTimeout { .. } => "websocket_connect_timeout", + ExecServerError::WebSocketConnect { .. } => "websocket_connect", + ExecServerError::InitializeTimedOut { .. } => "initialize_timed_out", + ExecServerError::Closed => "closed", + ExecServerError::Disconnected(_) => "disconnected", + ExecServerError::Json(_) => "json", + ExecServerError::HttpRequest(_) => "http_request", + ExecServerError::Protocol(_) => "protocol", + ExecServerError::Server { .. } => "server", + ExecServerError::EnvironmentRegistryHttp { .. } => "environment_registry_http", + ExecServerError::EnvironmentRegistryConfig(_) => "environment_registry_config", + ExecServerError::EnvironmentRegistryAuth(_) => "environment_registry_auth", + ExecServerError::EnvironmentRegistryRequest(_) => "environment_registry_request", + } +} + fn insert_header( headers: &mut HeaderMap, name: HeaderName, @@ -490,3 +602,75 @@ fn sse_stream_from_body( })) .boxed() } + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use pretty_assertions::assert_eq; + + use super::*; + + #[test] + fn post_message_failure_metric_tags_include_initialize_endpoint() { + let url = reqwest::Url::parse("https://chatgpt.com/backend-api/wham/apps").unwrap(); + let error = ExecServerError::HttpRequest("error sending request".to_string()); + + let tags = post_message_failure_metric_tags(PostMessageFailureMetricFields { + server_name: "codex_apps", + parsed_url: Some(&url), + mcp_method: Some("initialize"), + has_session_id: false, + has_authorization_header: true, + error: &error, + }); + + assert_eq!( + tags, + vec![ + ("server_name", "codex_apps".to_string()), + ("endpoint", "chatgpt_wham_apps".to_string()), + ("mcp_method", "initialize".to_string()), + ("has_session_id", "false".to_string()), + ("has_authorization_header", "true".to_string()), + ("error_type", "http_request".to_string()), + ] + ); + } + + #[test] + fn exec_server_error_type_covers_timeout_variant() { + let error = ExecServerError::InitializeTimedOut { + timeout: Duration::from_secs(5), + }; + + assert_eq!(exec_server_error_type(&error), "initialize_timed_out"); + } + + #[test] + fn post_message_failure_metric_tags_bucket_unknown_server_and_endpoint() { + let url = reqwest::Url::parse("https://example.com/custom/mcp").unwrap(); + let error = ExecServerError::HttpRequest("error sending request".to_string()); + + let tags = post_message_failure_metric_tags(PostMessageFailureMetricFields { + server_name: "user_configured_server", + parsed_url: Some(&url), + mcp_method: Some("tools/list"), + has_session_id: true, + has_authorization_header: false, + error: &error, + }); + + assert_eq!( + tags, + vec![ + ("server_name", "other".to_string()), + ("endpoint", "other".to_string()), + ("mcp_method", "tools/list".to_string()), + ("has_session_id", "true".to_string()), + ("has_authorization_header", "false".to_string()), + ("error_type", "http_request".to_string()), + ] + ); + } +} diff --git a/codex-rs/rmcp-client/src/rmcp_client.rs b/codex-rs/rmcp-client/src/rmcp_client.rs index 90b09d724c..dd8e42c747 100644 --- a/codex-rs/rmcp-client/src/rmcp_client.rs +++ b/codex-rs/rmcp-client/src/rmcp_client.rs @@ -784,6 +784,7 @@ impl RmcpClient { .auth_header(access_token); let transport = StreamableHttpClientTransport::with_client( StreamableHttpClientAdapter::new( + server_name, Arc::clone(http_client), default_headers, /*auth_provider*/ None, @@ -803,6 +804,7 @@ impl RmcpClient { let transport = StreamableHttpClientTransport::with_client( StreamableHttpClientAdapter::new( + server_name, Arc::clone(http_client), default_headers, auth_provider.clone(), @@ -1040,7 +1042,12 @@ async fn create_oauth_transport_and_runtime( }; let auth_client = AuthClient::new( - StreamableHttpClientAdapter::new(http_client, default_headers, /*auth_provider*/ None), + StreamableHttpClientAdapter::new( + server_name, + http_client, + default_headers, + /*auth_provider*/ None, + ), manager, ); let auth_manager = auth_client.auth_manager.clone();