From 4f2449b4b21988d5015ce6edf755fbd6a37a4908 Mon Sep 17 00:00:00 2001 From: bkotsopoulos Date: Wed, 9 Sep 2026 16:54:18 +0000 Subject: [PATCH] Measure total exec-server request duration including queueing (#44207) ## Why The existing request duration metric starts at dispatch, leaving out time spent waiting before dispatch. ## What changed Add `exec_server_request_total_duration_seconds` to measure time from decoded receipt until response enqueue or disconnection, including queueing. Record it alongside the existing duration metric with the same `method` and `result` labels for success, error, and disconnection paths. Preserve existing dispatch and queue timing semantics. ## Testing Add coverage for successful requests, route errors, unknown methods, response delivery failures, and disconnection during execution. Extend admission-wait coverage to verify total timing includes queueing and route setup without double-counting, and assert each completion and duration is recorded once. GitOrigin-RevId: ed0f719751b821b8f9f9556914b0f67dc6c5ad8f --- .../src/server/request_dispatcher.rs | 32 ++- .../src/server/request_dispatcher_tests.rs | 221 +++++++++++++++++- codex-rs/exec-server/src/telemetry.rs | 9 + 3 files changed, 254 insertions(+), 8 deletions(-) diff --git a/codex-rs/exec-server/src/server/request_dispatcher.rs b/codex-rs/exec-server/src/server/request_dispatcher.rs index 5460921e68..c1f16b64fe 100644 --- a/codex-rs/exec-server/src/server/request_dispatcher.rs +++ b/codex-rs/exec-server/src/server/request_dispatcher.rs @@ -195,13 +195,21 @@ impl RequestDispatcher { .is_err() { request_span.record("result", "disconnected"); - self.telemetry - .request_completed(method, "disconnected", started_at.elapsed()); + self.telemetry.request_completed( + method, + "disconnected", + started_at.elapsed(), + queued_at.elapsed(), + ); return RequestTaskResult::ConnectionClosed; } request_span.record("result", "error"); - self.telemetry - .request_completed(method, "error", started_at.elapsed()); + self.telemetry.request_completed( + method, + "error", + started_at.elapsed(), + queued_at.elapsed(), + ); return RequestTaskResult::Completed; }; @@ -221,7 +229,12 @@ impl RequestDispatcher { message = route.instrument(request_span.clone()) => message, _ = disconnected_rx.changed() => { request_span.record("result", "disconnected"); - telemetry.request_completed(method, "disconnected", started_at.elapsed()); + telemetry.request_completed( + method, + "disconnected", + started_at.elapsed(), + queued_at.elapsed(), + ); return RequestTaskResult::ConnectionClosed; } }; @@ -235,11 +248,16 @@ impl RequestDispatcher { }; if !response_sent { request_span.record("result", "disconnected"); - telemetry.request_completed(method, "disconnected", started_at.elapsed()); + telemetry.request_completed( + method, + "disconnected", + started_at.elapsed(), + queued_at.elapsed(), + ); return RequestTaskResult::ConnectionClosed; } request_span.record("result", result); - telemetry.request_completed(method, result, started_at.elapsed()); + telemetry.request_completed(method, result, started_at.elapsed(), queued_at.elapsed()); RequestTaskResult::Completed }; diff --git a/codex-rs/exec-server/src/server/request_dispatcher_tests.rs b/codex-rs/exec-server/src/server/request_dispatcher_tests.rs index 3772e621ae..ead6dd07c8 100644 --- a/codex-rs/exec-server/src/server/request_dispatcher_tests.rs +++ b/codex-rs/exec-server/src/server/request_dispatcher_tests.rs @@ -1,5 +1,7 @@ +use std::collections::BTreeMap; use std::sync::Arc; use std::time::Duration; +use std::time::Instant; use codex_exec_server_protocol::JSONRPCMessage; use codex_exec_server_protocol::JSONRPCRequest; @@ -34,6 +36,7 @@ use crate::connection::JsonRpcConnectionEvent; use crate::rpc::RpcNotificationSender; use crate::rpc::RpcRouter; use crate::rpc::RpcServerOutboundMessage; +use crate::rpc::invalid_request; use crate::server::ExecServerHandler; use crate::server::session_registry::SessionRegistry; use crate::telemetry::ExecServerTelemetry; @@ -143,7 +146,7 @@ fn request_span_uses_bounded_name_wire_method_and_inbound_trace_parent() { assert_eq!(request_span.parent_span_id, parent_span_id); } -/// One end-to-end request span covers queueing and execution; the metric isolates admission wait. +/// Total request timing includes queueing without changing dispatch or queue timing. #[tokio::test] async fn request_queue_waits_for_dispatcher_admission_before_recording_telemetry() { let span_exporter = InMemorySpanExporter::default(); @@ -343,5 +346,221 @@ async fn request_queue_waits_for_dispatcher_admission_before_recording_telemetry "queue latency must exclude synchronous request decoding and route setup" ); + let [dispatch_duration, total_duration] = + assert_request_completion(&metrics, "test/queued", "success"); + assert!(dispatch_duration >= route_setup_duration + Duration::from_millis(25)); + assert!(total_duration >= dispatch_duration); + assert!( + total_duration <= request_duration + Duration::from_millis(5), + "total request timing must not add admission wait twice" + ); + assert!( + total_duration >= queue_duration + route_setup_duration - Duration::from_millis(5), + "total request timing must include admission and route setup exactly once" + ); + metrics.shutdown().expect("shutdown metrics"); } + +struct RequestTelemetryFixture { + metrics: MetricsClient, + dispatcher: RequestDispatcher, + outgoing_rx: mpsc::Receiver, + disconnected_tx: watch::Sender, +} + +fn request_telemetry_fixture(router: RpcRouter) -> RequestTelemetryFixture { + let metrics = MetricsClient::new( + MetricsConfig::in_memory( + "test", + "codex-exec-server", + env!("CARGO_PKG_VERSION"), + InMemoryMetricExporter::default(), + ) + .with_runtime_reader(), + ) + .expect("metrics client"); + let telemetry = ExecServerTelemetry::new(metrics.clone()); + let (outgoing_tx, outgoing_rx) = mpsc::channel(/*buffer*/ 1); + let notifications = RpcNotificationSender::new(outgoing_tx.clone()); + let requests = notifications.request_sender(); + let handler = Arc::new(ExecServerHandler::new( + SessionRegistry::new(telemetry.clone()), + notifications, + ExecServerRuntimePaths::new( + std::env::current_exe().expect("current executable"), + /*codex_linux_sandbox_exe*/ None, + ) + .expect("runtime paths"), + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + )); + let (disconnected_tx, disconnected_rx) = watch::channel(/*init*/ false); + let dispatcher = RequestDispatcher::new( + Arc::new(router), + handler, + outgoing_tx, + disconnected_rx, + requests, + telemetry, + RequestDispatchMode::Inline, + ); + RequestTelemetryFixture { + metrics, + dispatcher, + outgoing_rx, + disconnected_tx, + } +} + +fn assert_request_completion(metrics: &MetricsClient, method: &str, result: &str) -> [Duration; 2] { + let snapshot = metrics.snapshot().expect("completed request metrics"); + let metric = snapshot + .scope_metrics() + .flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics) + .find(|metric| metric.name() == "exec_server_requests_total") + .expect("request counter"); + let AggregatedMetrics::U64(MetricData::Sum(sum)) = metric.data() else { + panic!("request counter should be a u64 sum"); + }; + let points: Vec<_> = sum.data_points().collect(); + assert_eq!(points.len(), 1); + assert_eq!(points[0].value(), 1, "record the completion only once"); + + [ + "exec_server_request_duration_seconds", + "exec_server_request_total_duration_seconds", + ] + .map(|name| { + let metric = snapshot + .scope_metrics() + .flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics) + .find(|metric| metric.name() == name) + .expect("request duration histogram"); + let AggregatedMetrics::F64(MetricData::Histogram(histogram)) = metric.data() else { + panic!("request duration should be an f64 histogram"); + }; + let points: Vec<_> = histogram.data_points().collect(); + assert_eq!(points.len(), 1); + let point = points[0]; + let attributes: BTreeMap<_, _> = point + .attributes() + .map(|attribute| { + ( + attribute.key.as_str().to_string(), + attribute.value.as_str().into_owned(), + ) + }) + .collect(); + assert_eq!( + attributes, + BTreeMap::from([ + ("method".to_string(), method.to_string()), + ("result".to_string(), result.to_string()), + ]) + ); + assert_eq!(point.count(), 1, "record each duration only once"); + Duration::from_secs_f64(point.sum()) + }) +} + +/// A synthetic receipt timestamp makes the pre-dispatch boundary deterministic for every result. +#[tokio::test] +async fn total_duration_preserves_dispatch_duration_and_completion_results() { + for (method, expected_method, expected_result, close_response) in [ + ("test/success", "test/success", "success", false), + ("test/error", "test/error", "error", false), + ("test/unknown", "unknown", "error", false), + ("test/success", "test/success", "disconnected", true), + ("test/unknown", "unknown", "disconnected", true), + ] { + let mut router = RpcRouter::new(); + router.request( + "test/success", + |_handler: Arc, _params: ()| async { + Ok::<_, codex_exec_server_protocol::JSONRPCErrorError>(()) + }, + ); + router.request( + "test/error", + |_handler: Arc, _params: ()| async { + Err::<(), _>(invalid_request("synthetic route error".to_string())) + }, + ); + let mut fixture = request_telemetry_fixture(router); + if close_response { + fixture.outgoing_rx.close(); + } + let pre_dispatch_wait = Duration::from_secs(5); + let received_at = Instant::now() - pre_dispatch_wait; + let result = fixture + .dispatcher + .dispatch_request( + JSONRPCRequest { + id: RequestId::Integer(1), + method: method.to_string(), + params: None, + trace: None, + }, + tracing::Span::none(), + received_at, + ) + .await; + assert_eq!( + matches!(result, RequestTaskResult::ConnectionClosed), + close_response + ); + let [dispatch_duration, total_duration] = + assert_request_completion(&fixture.metrics, expected_method, expected_result); + assert!(total_duration >= dispatch_duration + pre_dispatch_wait); + fixture.metrics.shutdown().expect("shutdown metrics"); + } +} + +/// Disconnecting a running request emits the same result and count for both duration definitions. +#[tokio::test] +async fn total_duration_records_disconnection_during_execution() { + let execution_started = Arc::new(Notify::new()); + let notify_execution_started = Arc::clone(&execution_started); + let mut router = RpcRouter::new(); + router.request( + "test/pending", + move |_handler: Arc, _params: ()| { + let execution_started = Arc::clone(¬ify_execution_started); + async move { + execution_started.notify_one(); + std::future::pending::>() + .await + } + }, + ); + let mut fixture = request_telemetry_fixture(router); + let pre_dispatch_wait = Duration::from_secs(5); + let received_at = Instant::now() - pre_dispatch_wait; + let dispatch = fixture.dispatcher.dispatch_request( + JSONRPCRequest { + id: RequestId::Integer(1), + method: "test/pending".to_string(), + params: None, + trace: None, + }, + tracing::Span::none(), + received_at, + ); + let disconnect = async { + execution_started.notified().await; + fixture + .disconnected_tx + .send(/*value*/ true) + .expect("disconnect request"); + }; + let (result, ()) = timeout(Duration::from_secs(1), async { + tokio::join!(dispatch, disconnect) + }) + .await + .expect("disconnected request should finish"); + assert!(matches!(result, RequestTaskResult::ConnectionClosed)); + let [dispatch_duration, total_duration] = + assert_request_completion(&fixture.metrics, "test/pending", "disconnected"); + assert!(total_duration >= dispatch_duration + pre_dispatch_wait); + fixture.metrics.shutdown().expect("shutdown metrics"); +} diff --git a/codex-rs/exec-server/src/telemetry.rs b/codex-rs/exec-server/src/telemetry.rs index 96c9a908d2..16fe4d138a 100644 --- a/codex-rs/exec-server/src/telemetry.rs +++ b/codex-rs/exec-server/src/telemetry.rs @@ -35,6 +35,8 @@ const REQUESTS_TOTAL_METRIC: &str = "exec_server_requests_total"; const REQUESTS_TOTAL_DESCRIPTION: &str = "Total number of exec-server requests."; const REQUEST_DURATION_METRIC: &str = "exec_server_request_duration_seconds"; const REQUEST_DURATION_DESCRIPTION: &str = "Duration of exec-server requests in seconds."; +const REQUEST_TOTAL_DURATION_METRIC: &str = "exec_server_request_total_duration_seconds"; +const REQUEST_TOTAL_DURATION_DESCRIPTION: &str = "Total exec-server request duration in seconds, including queueing, from decoded receipt until response enqueue or disconnection."; const REQUEST_QUEUE_DURATION_METRIC: &str = "exec_server_request_queue_duration_seconds"; const REQUEST_QUEUE_DURATION_DESCRIPTION: &str = "Time exec-server requests spend queued before execution in seconds."; @@ -157,6 +159,7 @@ impl ExecServerTelemetry { method: &'static str, result: &'static str, duration: Duration, + total_duration: Duration, ) { self.with_inner(|inner| { let tags = [("method", method), ("result", result)]; @@ -167,6 +170,12 @@ impl ExecServerTelemetry { duration, &tags, ); + inner.duration( + REQUEST_TOTAL_DURATION_METRIC, + REQUEST_TOTAL_DURATION_DESCRIPTION, + total_duration, + &tags, + ); }); }