diff --git a/codex-rs/cli/tests/exec_server.rs b/codex-rs/cli/tests/exec_server.rs index 051a829113..4f60d2b1b8 100644 --- a/codex-rs/cli/tests/exec_server.rs +++ b/codex-rs/cli/tests/exec_server.rs @@ -2,6 +2,7 @@ use std::io::Read as _; use std::io::Write as _; use std::net::TcpListener; use std::path::Path; +use std::sync::mpsc; use std::thread; use std::time::Duration; use std::time::Instant; @@ -55,6 +56,53 @@ fn local_exec_server_ignores_invalid_config_without_strict_config() -> Result<() Ok(()) } +#[test] +fn local_exec_server_exports_real_otel_metrics() -> Result<()> { + let collector = TestCollector::start()?; + let codex_home = TempDir::new()?; + let base_url = &collector.base_url; + std::fs::write( + codex_home.path().join("config.toml"), + format!( + r#" +[analytics] +enabled = true + +[otel] +environment = "test" +metrics_exporter = {{ otlp-http = {{ endpoint = "{base_url}/v1/metrics", protocol = "json" }} }} +"# + ), + )?; + + let mut cmd = codex_command(codex_home.path())?; + cmd.args(["exec-server", "--listen", "stdio"]) + .write_stdin( + r#"{"id":1,"method":"initialize","params":{"clientName":"otel-test","resumeSessionId":null}}"#, + ) + .assert() + .success(); + + let requests = collector.finish()?; + let metrics = requests + .iter() + .filter(|request| request.path == "/v1/metrics") + .map(|request| request.body.as_str()) + .collect::>() + .join("\n"); + assert!( + metrics.contains("exec_server_connections_active"), + "{metrics}" + ); + assert!(metrics.contains("exec_server_requests_total"), "{metrics}"); + assert!(metrics.contains("initialize"), "{metrics}"); + assert!( + metrics.contains("success") || metrics.contains("disconnected"), + "{metrics}" + ); + Ok(()) +} + #[test] fn remote_exec_server_preserves_websocket_error_in_stderr() -> Result<()> { let failed_websocket_listener = TcpListener::bind("127.0.0.1:0")?; @@ -109,6 +157,60 @@ struct CapturedRequest { body: String, } +struct TestCollector { + base_url: String, + requests: mpsc::Receiver>, + stop: mpsc::Sender<()>, + server: thread::JoinHandle<()>, +} + +impl TestCollector { + fn start() -> Result { + let listener = TcpListener::bind("127.0.0.1:0")?; + let addr = listener.local_addr()?; + listener.set_nonblocking(true)?; + let (tx, requests) = mpsc::channel(); + let (stop, stop_rx) = mpsc::channel(); + let server = thread::spawn(move || { + let mut captured = Vec::new(); + loop { + match listener.accept() { + Ok((mut stream, _)) => { + if let Ok(request) = read_http_request(&mut stream) { + captured.push(request); + } + let _ = stream.write_all( + b"HTTP/1.1 202 Accepted\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + ); + } + Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => { + if stop_rx.try_recv().is_ok() { + break; + } + thread::sleep(Duration::from_millis(10)); + } + Err(_) => break, + } + } + let _ = tx.send(captured); + }); + Ok(Self { + base_url: format!("http://{addr}"), + requests, + stop, + server, + }) + } + + fn finish(self) -> Result> { + let _ = self.stop.send(()); + self.server + .join() + .map_err(|_| anyhow::anyhow!("collector thread panicked"))?; + Ok(self.requests.recv_timeout(Duration::from_secs(1))?) + } +} + struct TestRegistry { base_url: String, server: thread::JoinHandle>, diff --git a/codex-rs/exec-server/src/server/transport_tests.rs b/codex-rs/exec-server/src/server/transport_tests.rs index 98c93c977a..51c2fff96e 100644 --- a/codex-rs/exec-server/src/server/transport_tests.rs +++ b/codex-rs/exec-server/src/server/transport_tests.rs @@ -6,6 +6,9 @@ use codex_app_server_protocol::JSONRPCNotification; use codex_app_server_protocol::JSONRPCRequest; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::RequestId; +use codex_otel::MetricsConfig; +use opentelemetry_sdk::metrics::InMemoryMetricExporter; +use opentelemetry_sdk::metrics::data::ScopeMetrics; use pretty_assertions::assert_eq; use tokio::io::AsyncBufReadExt; use tokio::io::AsyncWriteExt; @@ -112,6 +115,76 @@ async fn stdio_listen_transport_serves_initialize() { .expect("stdio transport should not fail"); } +#[tokio::test] +async fn stdio_listen_transport_emits_connection_and_request_metrics() { + let exporter = InMemoryMetricExporter::default(); + let metrics = codex_otel::MetricsClient::new(MetricsConfig::in_memory( + "test", + "codex-exec-server", + env!("CARGO_PKG_VERSION"), + exporter.clone(), + )) + .expect("metrics"); + let telemetry = crate::ExecServerTelemetry::new(Some(metrics.clone())); + let (mut client_writer, server_reader) = duplex(1 << 20); + let (server_writer, client_reader) = duplex(1 << 20); + let server_task = tokio::spawn(run_stdio_connection_with_io( + server_reader, + server_writer, + test_runtime_paths(), + telemetry, + )); + let mut client_lines = BufReader::new(client_reader).lines(); + + write_jsonrpc_line( + &mut client_writer, + &JSONRPCMessage::Request(JSONRPCRequest { + id: RequestId::Integer(1), + method: INITIALIZE_METHOD.to_string(), + params: Some( + serde_json::to_value(InitializeParams { + client_name: "exec-server-metrics-test".to_string(), + resume_session_id: None, + }) + .expect("initialize params should serialize"), + ), + trace: None, + }), + ) + .await; + let _response = timeout(Duration::from_secs(1), client_lines.next_line()) + .await + .expect("initialize response should arrive") + .expect("initialize response read should succeed") + .expect("initialize response should be present"); + + drop(client_writer); + drop(client_lines); + timeout(Duration::from_secs(1), server_task) + .await + .expect("stdio transport should finish after client disconnect") + .expect("stdio transport task should join") + .expect("stdio transport should not fail"); + metrics.shutdown().expect("shutdown metrics"); + + let metric_names = exporter + .get_finished_metrics() + .expect("finished metrics") + .into_iter() + .flat_map(|metrics| { + metrics + .scope_metrics() + .flat_map(ScopeMetrics::metrics) + .map(|metric| metric.name().to_string()) + .collect::>() + }) + .collect::>(); + assert!(metric_names.contains(&"exec_server_connections_active".to_string())); + assert!(metric_names.contains(&"exec_server_connections_total".to_string())); + assert!(metric_names.contains(&"exec_server_requests_total".to_string())); + assert!(metric_names.contains(&"exec_server_request_duration_seconds".to_string())); +} + #[test] fn parse_listen_url_accepts_websocket_url() { let transport =