zzos/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-29 16:57:06 +08:00
Code Issues Packages Projects Releases Wiki Activity

Measure total exec-server request duration including queueing (#44207)

Browse Source 
## 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
This commit is contained in:
bkotsopoulos
2026-09-09 17:03:57 +00:00
committed by copyberry
parent 8ff4aa8ee4
commit 4f2449b4b2
3 changed files with 254 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/exec-server/src/server/request_dispatcher.rs
+25 -7
View File
@@ -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
};
codex-rs/exec-server/src/server/request_dispatcher_tests.rs
+220 -1
View File
@@ -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<RpcServerOutboundMessage>,
disconnected_tx: watch::Sender<bool>,
}
fn request_telemetry_fixture(router: RpcRouter<ExecServerHandler>) -> 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<ExecServerHandler>, _params: ()| async {
Ok::<_, codex_exec_server_protocol::JSONRPCErrorError>(())
},
);
router.request(
"test/error",
|_handler: Arc<ExecServerHandler>, _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<ExecServerHandler>, _params: ()| {
let execution_started = Arc::clone(&notify_execution_started);
async move {
execution_started.notify_one();
std::future::pending::<Result<(), codex_exec_server_protocol::JSONRPCErrorError>>()
.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");
}
codex-rs/exec-server/src/telemetry.rs
+9
View File
@@ -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,
);
});
}