openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-09 15:58:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

exec-server: log structured timing events

Browse Source
This commit is contained in:
Michael Bolin
2026-07-01 22:39:19 -07:00
parent c899466b7d
commit 8b6f7077be
4 changed files with 263 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
codex-rs/exec-server/src/local_process.rs
View File

@@ -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,

24
codex-rs/exec-server/src/server/processor.rs
View File

@@ -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) => {

73
codex-rs/exec-server/src/telemetry.rs
View File

@@ -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<ProcessLogContext>,
started_at: Instant,
result: &'static str,
}
struct ProcessLogContext {
process_id: String,
trace_id: String,
}
pub(crate) struct RequestLogContext {
request_id: String,
traceparent: Option<String>,
}
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<RequestLogContext> {
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;

170
codex-rs/exec-server/src/telemetry_tests.rs Normal file
View File

@@ -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<String, String>,
}
#[derive(Clone, Default)]
struct CaptureLayer {
events: Arc<Mutex<Vec<CapturedEvent>>>,
}
impl CaptureLayer {
fn events(&self) -> Vec<CapturedEvent> {
self.events
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone()
}
}
impl<S> Layer<S> 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<String, String>,
}
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());
});
}