diff --git a/codex-rs/exec-server/src/client_telemetry.rs b/codex-rs/exec-server/src/client_telemetry.rs new file mode 100644 index 0000000000..f35b17af38 --- /dev/null +++ b/codex-rs/exec-server/src/client_telemetry.rs @@ -0,0 +1,23 @@ +//! Caller-side RPC attempt counts using the client's protocol method names, independent of tracing. + +use codex_otel::EXEC_SERVER_CLIENT_REQUEST_COUNT_METRIC; +use codex_otel::MetricsClient; + +pub(crate) fn record_client_request(metrics: Option<&MetricsClient>, method: &str) { + let Some(metrics) = metrics else { + return; + }; + // Record before local admission so failures and cancelled calls still count + // as attempts. Notifications and responses never enter these call paths. + if metrics + .counter_with_description( + EXEC_SERVER_CLIENT_REQUEST_COUNT_METRIC, + "Total number of client-side exec-server RPC attempts, including local failures.", + /*inc*/ 1, + &[("method", method)], + ) + .is_err() + { + tracing::warn!("failed to emit exec-server client request counter"); + } +} diff --git a/codex-rs/exec-server/src/lib.rs b/codex-rs/exec-server/src/lib.rs index 633b7b0848..141c7b7eda 100644 --- a/codex-rs/exec-server/src/lib.rs +++ b/codex-rs/exec-server/src/lib.rs @@ -3,6 +3,7 @@ mod capability_discovery; mod capability_discovery_cache; mod client; mod client_api; +mod client_telemetry; mod client_transport; mod connection; mod environment; diff --git a/codex-rs/exec-server/src/rpc.rs b/codex-rs/exec-server/src/rpc.rs index 998413756a..5047b06562 100644 --- a/codex-rs/exec-server/src/rpc.rs +++ b/codex-rs/exec-server/src/rpc.rs @@ -16,6 +16,7 @@ use codex_exec_server_protocol::JSONRPCNotification; use codex_exec_server_protocol::JSONRPCRequest; use codex_exec_server_protocol::JSONRPCResponse; use codex_exec_server_protocol::RequestId; +use codex_otel::MetricsClient; use codex_protocol::protocol::W3cTraceContext; use serde::Serialize; use serde::de::DeserializeOwned; @@ -30,11 +31,16 @@ use tokio::sync::watch; use tokio::task::JoinHandle; use tokio::time::timeout; +use crate::client_telemetry::record_client_request; use crate::connection::JsonRpcConnection; use crate::connection::JsonRpcConnectionEvent; use crate::connection::JsonRpcTransport; use crate::rpc_server_requests::RpcServerRequestSender; +#[cfg(test)] +#[path = "rpc_client_metrics_tests.rs"] +mod client_metrics_tests; + pub(crate) const SESSION_ALREADY_ATTACHED_ERROR_CODE: i64 = -32010; const MAX_IN_FLIGHT_REGULAR_CALLS: usize = 1024; const RESERVED_CLEANUP_CALLS: usize = 1; @@ -313,6 +319,7 @@ where } pub(crate) struct RpcClient { + metrics: Option, write_tx: mpsc::Sender, pending: Arc>>, inbound_request_ids: Arc>>, @@ -397,6 +404,7 @@ impl RpcClient { ( Self { + metrics: codex_otel::global(), write_tx, pending, inbound_request_ids: Arc::new(StdMutex::new(HashSet::new())), @@ -559,6 +567,7 @@ impl RpcClient { P: Serialize, T: DeserializeOwned, { + record_client_request(self.metrics.as_ref(), method); let _call_slot = self.acquire_regular_call_slot()?; self.call_inner(method, params, RpcCallTimeout::None).await } @@ -573,6 +582,7 @@ impl RpcClient { P: Serialize, T: DeserializeOwned, { + record_client_request(self.metrics.as_ref(), method); let _call_slot = self.acquire_regular_call_slot()?; self.call_inner(method, params, RpcCallTimeout::After(call_timeout)) .await @@ -597,6 +607,7 @@ impl RpcClient { P: Serialize, T: DeserializeOwned, { + record_client_request(self.metrics.as_ref(), method); let _call_slot = match self.shared_call_slots.try_acquire() { Ok(call_slot) => call_slot, Err(_) => match self.cleanup_call_slots.try_acquire() { diff --git a/codex-rs/exec-server/src/rpc_client_metrics_tests.rs b/codex-rs/exec-server/src/rpc_client_metrics_tests.rs new file mode 100644 index 0000000000..e58d504c43 --- /dev/null +++ b/codex-rs/exec-server/src/rpc_client_metrics_tests.rs @@ -0,0 +1,252 @@ +//! Exercise caller-side counters through the RPC transport, including failures and cancellation. + +use std::collections::BTreeMap; +use std::time::Duration; + +use codex_otel::MetricsClient; +use codex_otel::MetricsConfig; +use opentelemetry_sdk::metrics::InMemoryMetricExporter; +use opentelemetry_sdk::metrics::data::AggregatedMetrics; +use opentelemetry_sdk::metrics::data::MetricData; +use pretty_assertions::assert_eq; +use serde_json::Value; +use tokio::sync::mpsc; +use tokio::sync::watch; + +use super::RpcCallError; +use super::RpcClient; +use super::RpcClientEvent; +use crate::connection::JsonRpcConnection; +use crate::connection::JsonRpcConnectionEvent; +use crate::connection::JsonRpcTransport; +use crate::protocol::FS_READ_FILE_METHOD; +use crate::protocol::INITIALIZED_METHOD; +use crate::protocol::JSONRPCMessage; +use crate::protocol::JSONRPCResponse; +use crate::protocol::RequestId; + +struct Harness { + client: RpcClient, + metrics: MetricsClient, + outgoing: mpsc::Receiver, + incoming: mpsc::Sender, + _events: mpsc::Receiver, + _disconnected: watch::Sender, +} + +impl Harness { + fn new() -> Self { + let metrics = MetricsClient::new( + MetricsConfig::in_memory( + "test", + "exec-server-client-test", + env!("CARGO_PKG_VERSION"), + InMemoryMetricExporter::default(), + ) + .with_runtime_reader(), + ) + .expect("metrics client"); + let (outgoing_tx, outgoing) = mpsc::channel(/*buffer*/ 8); + let (incoming, incoming_rx) = mpsc::channel(/*buffer*/ 8); + let (disconnected, disconnected_rx) = watch::channel(/*init*/ false); + let (mut client, events) = RpcClient::new(JsonRpcConnection { + outgoing_tx, + incoming_rx, + disconnected_rx, + task_handles: Vec::new(), + transport: JsonRpcTransport::Plain, + }); + client.metrics = Some(metrics.clone()); + Self { + client, + metrics, + outgoing, + incoming, + _events: events, + _disconnected: disconnected, + } + } + + fn counts(&self) -> BTreeMap { + let snapshot = self.metrics.snapshot().expect("metrics snapshot"); + let mut counts = BTreeMap::new(); + for metric in snapshot + .scope_metrics() + .flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics) + .filter(|metric| metric.name() == "exec_server_client_requests_total") + { + let AggregatedMetrics::U64(MetricData::Sum(sum)) = metric.data() else { + panic!("client request count should be a u64 sum"); + }; + for point in sum.data_points() { + let attributes = point + .attributes() + .map(|attribute| { + ( + attribute.key.as_str(), + attribute.value.as_str().into_owned(), + ) + }) + .collect::>(); + assert_eq!(attributes.len(), 1, "only the method is labeled"); + assert_eq!(attributes[0].0, "method"); + *counts.entry(attributes[0].1.clone()).or_default() += point.value(); + } + } + counts + } +} + +#[derive(Clone, Copy)] +enum CallKind { + Regular, + Untraced, + WithTimeout, + Cleanup, +} + +#[tokio::test] +async fn each_request_entry_point_counts_once_and_preserves_response() { + for kind in [ + CallKind::Regular, + CallKind::Untraced, + CallKind::WithTimeout, + CallKind::Cleanup, + ] { + let mut harness = Harness::new(); + let params = serde_json::json!({"path": "/sensitive-test-path"}); + let request = async { + match kind { + CallKind::Regular => { + harness + .client + .call::<_, Value>(FS_READ_FILE_METHOD, ¶ms) + .await + } + CallKind::Untraced => { + harness + .client + .call_untraced::<_, Value>(FS_READ_FILE_METHOD, ¶ms) + .await + } + CallKind::WithTimeout => { + harness + .client + .call_with_timeout::<_, Value>( + FS_READ_FILE_METHOD, + ¶ms, + Duration::from_secs(1), + ) + .await + } + CallKind::Cleanup => { + harness + .client + .call_for_cleanup::<_, Value>(FS_READ_FILE_METHOD, ¶ms) + .await + } + } + }; + let server = async { + let Some(JSONRPCMessage::Request(request)) = harness.outgoing.recv().await else { + panic!("expected request"); + }; + assert_eq!(request.method, FS_READ_FILE_METHOD); + assert_eq!(request.params, Some(params.clone())); + harness + .incoming + .send(JsonRpcConnectionEvent::Message(JSONRPCMessage::Response( + JSONRPCResponse { + id: request.id, + result: serde_json::json!({"ok": true}), + }, + ))) + .await + .expect("response accepted"); + }; + let (response, ()) = tokio::join!(request, server); + assert_eq!( + response.expect("RPC response"), + serde_json::json!({"ok": true}) + ); + assert_eq!( + harness.counts(), + BTreeMap::from([(FS_READ_FILE_METHOD.to_string(), 1)]) + ); + } +} + +#[tokio::test] +async fn local_rejections_and_closed_transport_count_as_attempts() { + let harness = Harness::new(); + let slots = harness + .client + .shared_call_slots + .acquire_many(super::MAX_IN_FLIGHT_REGULAR_CALLS as u32) + .await + .expect("occupy slots"); + let rejected = harness + .client + .call::<_, Value>(FS_READ_FILE_METHOD, &()) + .await; + assert!(matches!( + rejected, + Err(RpcCallError::PendingRequestLimitExceeded { .. }) + )); + drop(slots); + harness.client.close_transport().await; + let closed = harness + .client + .call::<_, Value>(FS_READ_FILE_METHOD, &()) + .await; + assert!(matches!(closed, Err(RpcCallError::Closed))); + assert_eq!( + harness.counts(), + BTreeMap::from([(FS_READ_FILE_METHOD.to_string(), 2)]) + ); +} + +#[tokio::test(start_paused = true)] +async fn timeout_and_cancellation_count_as_attempts() { + let mut harness = Harness::new(); + let timed_out = harness + .client + .call_with_timeout::<_, Value>(FS_READ_FILE_METHOD, &(), Duration::from_secs(1)) + .await; + assert!(matches!(timed_out, Err(RpcCallError::TimedOut { .. }))); + harness.outgoing.recv().await.expect("timed out request"); + let mut cancelled = Box::pin(harness.client.call::<_, Value>(FS_READ_FILE_METHOD, &())); + assert!(futures::poll!(cancelled.as_mut()).is_pending()); + harness.outgoing.recv().await.expect("cancelled request"); + drop(cancelled); + assert_eq!( + harness.counts(), + BTreeMap::from([(FS_READ_FILE_METHOD.to_string(), 2)]) + ); +} + +#[tokio::test] +async fn notifications_responses_and_disabled_metrics_do_not_record_attempts() { + let mut harness = Harness::new(); + harness + .client + .notify(INITIALIZED_METHOD, &()) + .await + .expect("notification"); + harness + .client + .respond(RequestId::Integer(1), &()) + .await + .expect("response"); + assert_eq!(harness.counts(), BTreeMap::new()); + harness.client.metrics = None; + harness.client.close_transport().await; + assert!(matches!( + harness + .client + .call::<_, Value>(FS_READ_FILE_METHOD, &()) + .await, + Err(RpcCallError::Closed) + )); + assert_eq!(harness.counts(), BTreeMap::new()); +} diff --git a/codex-rs/otel/src/metrics/config.rs b/codex-rs/otel/src/metrics/config.rs index 22d8d001b2..212b63d346 100644 --- a/codex-rs/otel/src/metrics/config.rs +++ b/codex-rs/otel/src/metrics/config.rs @@ -2,6 +2,7 @@ use crate::config::OtelExporter; use crate::metrics::Result; use crate::metrics::names::API_CALL_COUNT_METRIC; use crate::metrics::names::API_CALL_DURATION_METRIC; +use crate::metrics::names::EXEC_SERVER_CLIENT_REQUEST_COUNT_METRIC; use crate::metrics::names::RESPONSES_API_ENGINE_IAPI_TTFT_DURATION_METRIC; use crate::metrics::names::RESPONSES_API_ENGINE_SERVICE_TBT_DURATION_METRIC; use crate::metrics::names::RESPONSES_API_ENGINE_SERVICE_TTFT_DURATION_METRIC; @@ -23,6 +24,8 @@ const STATSIG_DISABLED_METRICS: &[&str] = &[ API_CALL_COUNT_METRIC, API_CALL_DURATION_METRIC, CONVERSATION_TURN_COUNT_METRIC, + // Caller-side executor volume belongs in configured observability collectors. + EXEC_SERVER_CLIENT_REQUEST_COUNT_METRIC, RESPONSES_API_ENGINE_IAPI_TTFT_DURATION_METRIC, RESPONSES_API_ENGINE_SERVICE_TBT_DURATION_METRIC, RESPONSES_API_ENGINE_SERVICE_TTFT_DURATION_METRIC, diff --git a/codex-rs/otel/src/metrics/names.rs b/codex-rs/otel/src/metrics/names.rs index 7398554d31..1d4994e918 100644 --- a/codex-rs/otel/src/metrics/names.rs +++ b/codex-rs/otel/src/metrics/names.rs @@ -5,6 +5,8 @@ pub const ARTIFACT_OPERATION_STARTED_METRIC: &str = "codex.artifact.operation.st pub const ARTIFACT_OPERATION_EXPECTED_OUTPUT_COUNT_METRIC: &str = "codex.artifact.operation.expected_output_count"; pub const PROCESS_START_METRIC: &str = "codex.process.start"; +/// Caller-side exec-server RPC attempts, including local admission and transport failures. +pub const EXEC_SERVER_CLIENT_REQUEST_COUNT_METRIC: &str = "exec_server_client_requests_total"; pub const API_CALL_COUNT_METRIC: &str = "codex.api_request"; pub const API_CALL_DURATION_METRIC: &str = "codex.api_request.duration_ms"; pub const SSE_EVENT_COUNT_METRIC: &str = "codex.sse_event"; diff --git a/codex-rs/otel/src/provider.rs b/codex-rs/otel/src/provider.rs index e46f58ac30..514453d32b 100644 --- a/codex-rs/otel/src/provider.rs +++ b/codex-rs/otel/src/provider.rs @@ -605,6 +605,7 @@ mod tests { use super::*; use crate::metrics::API_CALL_COUNT_METRIC; use crate::metrics::API_CALL_DURATION_METRIC; + use crate::metrics::EXEC_SERVER_CLIENT_REQUEST_COUNT_METRIC; use crate::metrics::MetricsExporter; use crate::metrics::RESPONSES_API_ENGINE_IAPI_TTFT_DURATION_METRIC; use crate::metrics::RESPONSES_API_ENGINE_SERVICE_TBT_DURATION_METRIC; @@ -731,6 +732,12 @@ mod tests { metrics.counter(API_CALL_COUNT_METRIC, /*inc*/ 1, &[])?; metrics.record_duration(API_CALL_DURATION_METRIC, Duration::from_millis(100), &[])?; + metrics.counter_with_description( + EXEC_SERVER_CLIENT_REQUEST_COUNT_METRIC, + "Client-side exec-server RPC attempts.", + /*inc*/ 1, + &[("method", "fs/readFile")], + )?; metrics.counter("codex.conversation.turn.count", /*inc*/ 1, &[])?; metrics.record_duration( RESPONSES_API_ENGINE_IAPI_TTFT_DURATION_METRIC, diff --git a/codex-rs/otel/tests/suite/otlp_http_loopback.rs b/codex-rs/otel/tests/suite/otlp_http_loopback.rs index bcfe02b47f..7bd209b138 100644 --- a/codex-rs/otel/tests/suite/otlp_http_loopback.rs +++ b/codex-rs/otel/tests/suite/otlp_http_loopback.rs @@ -1,3 +1,4 @@ +use codex_otel::EXEC_SERVER_CLIENT_REQUEST_COUNT_METRIC; use codex_otel::MetricsClient; use codex_otel::MetricsConfig; use codex_otel::OtelExporter; @@ -187,6 +188,12 @@ fn otlp_http_exporter_sends_metrics_to_collector() -> Result<()> { metrics.counter("codex.turns", /*inc*/ 1, &[("source", "test")])?; metrics.counter("codex.api_request", /*inc*/ 1, &[("status", "200")])?; + metrics.counter_with_description( + EXEC_SERVER_CLIENT_REQUEST_COUNT_METRIC, + "Client-side exec-server RPC attempts.", + /*inc*/ 1, + &[("method", "fs/readFile")], + )?; metrics.record_duration( "codex.api_request.duration_ms", Duration::from_millis(100), @@ -259,6 +266,10 @@ fn otlp_http_exporter_sends_metrics_to_collector() -> Result<()> { "expected API-request counter not found; body prefix: {}", &body.chars().take(2000).collect::() ); + assert!( + body.contains(EXEC_SERVER_CLIENT_REQUEST_COUNT_METRIC), + "custom OTLP must retain the exec-server client counter excluded from built-in Statsig" + ); assert!( body.contains("\"codex.api_request.duration_ms\""), "expected API-request duration not found; body prefix: {}",