From 8b6f7077bef342c36baf6de16c6dbf7afccbc84b Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Wed, 1 Jul 2026 22:39:19 -0700 Subject: [PATCH] exec-server: log structured timing events --- codex-rs/exec-server/src/local_process.rs | 4 +- codex-rs/exec-server/src/server/processor.rs | 24 ++- codex-rs/exec-server/src/telemetry.rs | 73 +++++++- codex-rs/exec-server/src/telemetry_tests.rs | 170 +++++++++++++++++++ 4 files changed, 263 insertions(+), 8 deletions(-) create mode 100644 codex-rs/exec-server/src/telemetry_tests.rs diff --git a/codex-rs/exec-server/src/local_process.rs b/codex-rs/exec-server/src/local_process.rs index 3c488f0ad7..08d2d0319f 100644 --- a/codex-rs/exec-server/src/local_process.rs +++ b/codex-rs/exec-server/src/local_process.rs @@ -334,7 +334,7 @@ impl LocalProcess { output_notify: Arc::clone(&output_notify), open_streams: 2, closed: false, - metrics: Some(self.inner.telemetry.process_started()), + metrics: Some(self.inner.telemetry.process_started(process_id.as_ref())), termination_requested: false, sandbox: prepared.sandbox, sandbox_denied: false, @@ -1306,7 +1306,7 @@ mod tests { output_notify: Arc::clone(&output_notify), open_streams: 2, closed: false, - metrics: Some(backend.inner.telemetry.process_started()), + metrics: Some(backend.inner.telemetry.process_started(process_id.as_ref())), termination_requested: false, sandbox: SandboxType::None, sandbox_denied: false, diff --git a/codex-rs/exec-server/src/server/processor.rs b/codex-rs/exec-server/src/server/processor.rs index 63f6bd45d9..56690c65c5 100644 --- a/codex-rs/exec-server/src/server/processor.rs +++ b/codex-rs/exec-server/src/server/processor.rs @@ -128,6 +128,13 @@ async fn run_connection( JsonRpcConnectionEvent::Message(message) => match message { codex_exec_server_protocol::JSONRPCMessage::Request(request) => { let request_started_at = Instant::now(); + let request_log_context = telemetry.request_log_context( + &request.id, + request + .trace + .as_ref() + .and_then(|trace| trace.traceparent.as_deref()), + ); if let Some((method, route)) = router.request_route(request.method.as_str()) { let request_span = request_span(method, &request); let message = tokio::select! { @@ -135,6 +142,7 @@ async fn run_connection( _ = disconnected_rx.changed() => { request_span.record("result", "disconnected"); telemetry.request_completed( + request_log_context.as_ref(), method, "disconnected", request_started_at.elapsed(), @@ -149,6 +157,7 @@ async fn run_connection( { request_span.record("result", "disconnected"); telemetry.request_completed( + request_log_context.as_ref(), method, "disconnected", request_started_at.elapsed(), @@ -156,7 +165,12 @@ async fn run_connection( break; } request_span.record("result", result); - telemetry.request_completed(method, result, request_started_at.elapsed()); + telemetry.request_completed( + request_log_context.as_ref(), + method, + result, + request_started_at.elapsed(), + ); drop(request_span); } else { let method = "unknown"; @@ -174,6 +188,7 @@ async fn run_connection( { request_span.record("result", "disconnected"); telemetry.request_completed( + request_log_context.as_ref(), method, "disconnected", request_started_at.elapsed(), @@ -181,7 +196,12 @@ async fn run_connection( break; } request_span.record("result", "error"); - telemetry.request_completed(method, "error", request_started_at.elapsed()); + telemetry.request_completed( + request_log_context.as_ref(), + method, + "error", + request_started_at.elapsed(), + ); } } codex_exec_server_protocol::JSONRPCMessage::Notification(notification) => { diff --git a/codex-rs/exec-server/src/telemetry.rs b/codex-rs/exec-server/src/telemetry.rs index 650faacb2c..6191b55bf0 100644 --- a/codex-rs/exec-server/src/telemetry.rs +++ b/codex-rs/exec-server/src/telemetry.rs @@ -4,6 +4,7 @@ use std::time::Duration; use std::time::Instant; use codex_otel::MetricsClient; +use tracing::info; use tracing::warn; const CONNECTIONS_ACTIVE_METRIC: &str = "exec_server_connections_active"; @@ -96,10 +97,21 @@ pub(crate) struct ConnectionMetricGuard { pub(crate) struct ProcessMetricGuard { telemetry: ExecServerTelemetry, + log_context: Option, started_at: Instant, result: &'static str, } +struct ProcessLogContext { + process_id: String, + trace_id: String, +} + +pub(crate) struct RequestLogContext { + request_id: String, + traceparent: Option, +} + impl ExecServerTelemetry { pub fn new(metrics: MetricsClient) -> Self { let active = Arc::new(Mutex::new(ActiveCounts::default())); @@ -129,10 +141,22 @@ impl ExecServerTelemetry { pub(crate) fn request_completed( &self, + log_context: Option<&RequestLogContext>, method: &'static str, result: &'static str, duration: Duration, ) { + if let Some(log_context) = log_context { + info!( + event.name = "codex.exec_server_request", + request_id = %log_context.request_id, + method, + result, + duration_ms = u64::try_from(duration.as_millis()).unwrap_or(u64::MAX), + traceparent = log_context.traceparent.as_deref().unwrap_or(""), + "exec-server request completed" + ); + } self.with_inner(|inner| { let tags = [("method", method), ("result", result)]; inner.counter(REQUESTS_TOTAL_METRIC, REQUESTS_TOTAL_DESCRIPTION, &tags); @@ -145,6 +169,17 @@ impl ExecServerTelemetry { }); } + pub(crate) fn request_log_context( + &self, + request_id: &impl std::fmt::Display, + traceparent: Option<&str>, + ) -> Option { + self.info_events_enabled().then(|| RequestLogContext { + request_id: request_id.to_string(), + traceparent: traceparent.map(str::to_string), + }) + } + pub(crate) fn remote_registration_completed(&self, result: &'static str, duration: Duration) { self.record_operation(REMOTE_REGISTRATION_METRICS, result, duration); } @@ -163,18 +198,37 @@ impl ExecServerTelemetry { }); } - pub(crate) fn process_started(&self) -> ProcessMetricGuard { + pub(crate) fn process_started(&self, process_id: &str) -> ProcessMetricGuard { self.with_inner(|inner| { inner.adjust_process_count(/*delta*/ 1); }); ProcessMetricGuard { telemetry: self.clone(), + log_context: self.info_events_enabled().then(|| ProcessLogContext { + process_id: process_id.to_string(), + trace_id: codex_otel::current_span_trace_id().unwrap_or_default(), + }), started_at: Instant::now(), result: "unknown", } } - fn process_finished(&self, result: &'static str, duration: Duration) { + fn process_finished( + &self, + log_context: Option<&ProcessLogContext>, + result: &'static str, + duration: Duration, + ) { + if let Some(log_context) = log_context { + info!( + event.name = "codex.exec_server_process", + process_id = %log_context.process_id, + trace_id = %log_context.trace_id, + result, + duration_ms = u64::try_from(duration.as_millis()).unwrap_or(u64::MAX), + "exec-server process completed" + ); + } self.with_inner(|inner| { inner.adjust_process_count(/*delta*/ -1); inner.counter( @@ -191,6 +245,10 @@ impl ExecServerTelemetry { }); } + pub(crate) fn info_events_enabled(&self) -> bool { + tracing::enabled!(tracing::Level::INFO) + } + fn connection_finished(&self, transport: ConnectionTransport) { self.with_inner(|inner| { inner.adjust_connection_count(transport, /*delta*/ -1); @@ -236,8 +294,11 @@ impl ProcessMetricGuard { impl Drop for ProcessMetricGuard { fn drop(&mut self) { - self.telemetry - .process_finished(self.result, self.started_at.elapsed()); + self.telemetry.process_finished( + self.log_context.as_ref(), + self.result, + self.started_at.elapsed(), + ); } } @@ -338,3 +399,7 @@ fn register_active_gauge( warn!(metric = name, "failed to register exec-server gauge"); } } + +#[cfg(test)] +#[path = "telemetry_tests.rs"] +mod tests; diff --git a/codex-rs/exec-server/src/telemetry_tests.rs b/codex-rs/exec-server/src/telemetry_tests.rs new file mode 100644 index 0000000000..3caff2a93b --- /dev/null +++ b/codex-rs/exec-server/src/telemetry_tests.rs @@ -0,0 +1,170 @@ +use std::collections::BTreeMap; +use std::sync::Arc; +use std::sync::Mutex; +use std::time::Duration; + +use pretty_assertions::assert_eq; +use tracing::Event; +use tracing::Level; +use tracing::Subscriber; +use tracing::field::Field; +use tracing::field::Visit; +use tracing_subscriber::Layer; +use tracing_subscriber::layer::Context; +use tracing_subscriber::prelude::*; +use tracing_subscriber::registry::LookupSpan; + +use super::ExecServerTelemetry; + +#[derive(Clone, Debug, PartialEq, Eq)] +struct CapturedEvent { + level: Level, + target: String, + fields: BTreeMap, +} + +#[derive(Clone, Default)] +struct CaptureLayer { + events: Arc>>, +} + +impl CaptureLayer { + fn events(&self) -> Vec { + self.events + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() + } +} + +impl Layer for CaptureLayer +where + S: Subscriber + for<'lookup> LookupSpan<'lookup>, +{ + fn on_event(&self, event: &Event<'_>, _context: Context<'_, S>) { + let mut visitor = FieldVisitor::default(); + event.record(&mut visitor); + self.events + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .push(CapturedEvent { + level: *event.metadata().level(), + target: event.metadata().target().to_string(), + fields: visitor.fields, + }); + } +} + +#[derive(Default)] +struct FieldVisitor { + fields: BTreeMap, +} + +impl Visit for FieldVisitor { + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + self.fields + .insert(field.name().to_string(), format!("{value:?}")); + } + + fn record_str(&mut self, field: &Field, value: &str) { + self.fields + .insert(field.name().to_string(), value.to_string()); + } + + fn record_f64(&mut self, field: &Field, value: f64) { + self.fields + .insert(field.name().to_string(), value.to_string()); + } +} + +#[test] +fn exec_server_timing_events_are_structured_info_logs() { + let capture = CaptureLayer::default(); + let subscriber = tracing_subscriber::registry().with(capture.clone()); + + tracing::subscriber::with_default(subscriber, || { + tracing::callsite::rebuild_interest_cache(); + let telemetry = ExecServerTelemetry::default(); + let request_log_context = telemetry + .request_log_context(&"request-1", Some("00-trace-parent")) + .expect("INFO event should be enabled"); + telemetry.request_completed( + Some(&request_log_context), + "process/start", + "success", + Duration::from_millis(42), + ); + telemetry.process_started("process-1").finish("success"); + }); + + let events = capture.events(); + let request = events + .iter() + .find(|event| { + event + .fields + .get("event.name") + .is_some_and(|name| name == "codex.exec_server_request") + }) + .expect("request timing event"); + assert_eq!(request.level, Level::INFO); + assert_eq!(request.target, "codex_exec_server::telemetry"); + assert_eq!( + request.fields, + BTreeMap::from([ + ("duration_ms".to_string(), "42".to_string()), + ( + "event.name".to_string(), + "codex.exec_server_request".to_string(), + ), + ( + "message".to_string(), + "exec-server request completed".to_string(), + ), + ("method".to_string(), "process/start".to_string()), + ("request_id".to_string(), "request-1".to_string()), + ("result".to_string(), "success".to_string()), + ("traceparent".to_string(), "00-trace-parent".to_string()), + ]) + ); + + let process = events + .iter() + .find(|event| { + event + .fields + .get("event.name") + .is_some_and(|name| name == "codex.exec_server_process") + }) + .expect("process timing event"); + assert_eq!(process.level, Level::INFO); + assert_eq!(process.target, "codex_exec_server::telemetry"); + assert_eq!( + process.fields.get("message").map(String::as_str), + Some("exec-server process completed") + ); + assert_eq!( + process.fields.get("process_id").map(String::as_str), + Some("process-1") + ); + assert_eq!(process.fields.get("trace_id").map(String::as_str), Some("")); + assert_eq!( + process.fields.get("result").map(String::as_str), + Some("success") + ); + assert!(process.fields.contains_key("duration_ms")); +} + +#[test] +fn disabled_info_events_do_not_capture_process_log_values() { + let subscriber = tracing_subscriber::registry() + .with(tracing_subscriber::filter::filter_fn(|_metadata| false)); + + tracing::subscriber::with_default(subscriber, || { + tracing::callsite::rebuild_interest_cache(); + let telemetry = ExecServerTelemetry::default(); + assert!(!telemetry.info_events_enabled()); + let process = telemetry.process_started("process-1"); + assert!(process.log_context.is_none()); + }); +}