diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 3d8086c27b..a68293bdcc 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2323,6 +2323,7 @@ dependencies = [ "codex-memories-write", "codex-model-provider", "codex-models-manager", + "codex-otel", "codex-plugin", "codex-protocol", "codex-responses-api-proxy", @@ -3477,6 +3478,7 @@ dependencies = [ name = "codex-otel" version = "0.0.0" dependencies = [ + "async-trait", "chrono", "codex-api", "codex-app-server-protocol", @@ -3486,8 +3488,10 @@ dependencies = [ "eventsource-stream", "gethostname", "http 1.4.0", + "libc", "opentelemetry", "opentelemetry-appender-tracing", + "opentelemetry-http", "opentelemetry-otlp", "opentelemetry-semantic-conventions", "opentelemetry_sdk", @@ -3497,12 +3501,14 @@ dependencies = [ "serde", "serde_json", "strum_macros 0.28.0", + "tempfile", "thiserror 2.0.18", "tokio", "tokio-tungstenite", "tracing", "tracing-opentelemetry", "tracing-subscriber", + "wiremock", ] [[package]] diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index c72b7c7844..c7647fd053 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -328,6 +328,7 @@ once_cell = "1.20.2" openssl-sys = "*" opentelemetry = "0.31.0" opentelemetry-appender-tracing = "0.31.0" +opentelemetry-http = { version = "0.31.0", features = ["reqwest-blocking"] } opentelemetry-otlp = "0.31.0" opentelemetry-semantic-conventions = "0.31.0" opentelemetry_sdk = "0.31.0" diff --git a/codex-rs/cli/Cargo.toml b/codex-rs/cli/Cargo.toml index 6a4d78fbc7..a8f48b8e88 100644 --- a/codex-rs/cli/Cargo.toml +++ b/codex-rs/cli/Cargo.toml @@ -45,6 +45,7 @@ codex-mcp = { workspace = true } codex-mcp-server = { workspace = true } codex-model-provider = { workspace = true } codex-models-manager = { workspace = true } +codex-otel = { workspace = true } codex-plugin = { workspace = true } codex-protocol = { workspace = true } codex-responses-api-proxy = { workspace = true } diff --git a/codex-rs/cli/src/main.rs b/codex-rs/cli/src/main.rs index b194a5ab1b..6a51b68e68 100644 --- a/codex-rs/cli/src/main.rs +++ b/codex-rs/cli/src/main.rs @@ -919,6 +919,11 @@ fn stage_str(stage: Stage) -> &'static str { } fn main() -> anyhow::Result<()> { + if let Some(result) = codex_otel::run_process_exit_upload_if_requested() { + result?; + return Ok(()); + } + arg0_dispatch_or_else(|arg0_paths: Arg0DispatchPaths| async move { cli_main(arg0_paths).await?; Ok(()) diff --git a/codex-rs/otel/Cargo.toml b/codex-rs/otel/Cargo.toml index 9370119905..2225105fe3 100644 --- a/codex-rs/otel/Cargo.toml +++ b/codex-rs/otel/Cargo.toml @@ -13,6 +13,7 @@ path = "src/lib.rs" workspace = true [dependencies] +async-trait = { workspace = true } chrono = { workspace = true } codex-utils-absolute-path = { workspace = true } codex-utils-string = { workspace = true } @@ -21,8 +22,10 @@ codex-app-server-protocol = { workspace = true } codex-protocol = { workspace = true } eventsource-stream = { workspace = true } gethostname = { workspace = true } +libc = { workspace = true } opentelemetry = { workspace = true, features = ["logs", "metrics", "trace"] } opentelemetry-appender-tracing = { workspace = true } +opentelemetry-http = { workspace = true } opentelemetry-otlp = { workspace = true, features = [ "grpc-tonic", "http-proto", @@ -51,6 +54,7 @@ reqwest = { workspace = true, features = ["blocking", "rustls-tls"] } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } strum_macros = { workspace = true } +tempfile = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true } tokio-tungstenite = { workspace = true } @@ -64,3 +68,4 @@ opentelemetry_sdk = { workspace = true, features = [ "testing", ] } pretty_assertions = { workspace = true } +wiremock = { workspace = true } diff --git a/codex-rs/otel/src/lib.rs b/codex-rs/otel/src/lib.rs index 586dfa15a7..72649b9145 100644 --- a/codex-rs/otel/src/lib.rs +++ b/codex-rs/otel/src/lib.rs @@ -5,6 +5,7 @@ pub(crate) mod provider; pub(crate) mod trace_context; mod otlp; +mod process_exit_upload; mod targets; use crate::metrics::Result as MetricsResult; @@ -24,6 +25,8 @@ pub use crate::metrics::runtime_metrics::RuntimeMetricTotals; pub use crate::metrics::runtime_metrics::RuntimeMetricsSummary; pub use crate::metrics::timer::Timer; pub use crate::metrics::*; +pub use crate::process_exit_upload::ProcessExitUploadError; +pub use crate::process_exit_upload::run_process_exit_upload_if_requested; pub use crate::provider::OtelProvider; pub use crate::trace_context::context_from_w3c_trace_context; pub use crate::trace_context::current_span_trace_id; diff --git a/codex-rs/otel/src/metrics/client.rs b/codex-rs/otel/src/metrics/client.rs index 9d5ad5d166..bb0ca6b819 100644 --- a/codex-rs/otel/src/metrics/client.rs +++ b/codex-rs/otel/src/metrics/client.rs @@ -9,6 +9,8 @@ use crate::metrics::validation::validate_metric_name; use crate::metrics::validation::validate_tag_key; use crate::metrics::validation::validate_tag_value; use crate::metrics::validation::validate_tags; +use crate::process_exit_upload::StatsigUpload; +use crate::process_exit_upload::StatsigUploadClient; use codex_utils_string::sanitize_metric_tag_value; use opentelemetry::KeyValue; use opentelemetry::metrics::Counter; @@ -93,6 +95,7 @@ struct MetricsClientInner { duration_histograms: Mutex>>, runtime_reader: Option>, default_tags: BTreeMap, + process_exit_upload: Option, } impl MetricsClientInner { @@ -197,6 +200,13 @@ impl MetricsClientInner { .map_err(|source| MetricsError::ProviderShutdown { source })?; Ok(()) } + + fn shutdown_on_process_exit(&self) -> Result<()> { + debug!("handing off final OTEL metrics"); + self.meter_provider + .shutdown() + .map_err(|source| MetricsError::ProviderShutdown { source }) + } } /// OpenTelemetry metrics client used by Codex. @@ -239,13 +249,18 @@ impl MetricsClient { ) }); - let (meter_provider, meter) = match exporter { + let (meter_provider, meter, process_exit_upload) = match exporter { MetricsExporter::InMemory(exporter) => { - build_provider(resource, exporter, export_interval, runtime_reader.clone()) + let (meter_provider, meter) = + build_provider(resource, exporter, export_interval, runtime_reader.clone()); + (meter_provider, meter, None) } MetricsExporter::Otlp(exporter) => { - let exporter = build_otlp_metric_exporter(exporter, Temporality::Delta)?; - build_provider(resource, exporter, export_interval, runtime_reader.clone()) + let (exporter, process_exit_upload) = + build_otlp_metric_exporter(exporter, Temporality::Delta)?; + let (meter_provider, meter) = + build_provider(resource, exporter, export_interval, runtime_reader.clone()); + (meter_provider, meter, process_exit_upload) } }; @@ -257,6 +272,7 @@ impl MetricsClient { duration_histograms: Mutex::new(HashMap::new()), runtime_reader, default_tags, + process_exit_upload, }))) } @@ -319,6 +335,23 @@ impl MetricsClient { pub fn shutdown(&self) -> Result<()> { self.0.shutdown() } + + pub(crate) fn configure_statsig_metrics_uploader(&self, executable: std::path::PathBuf) { + if let Some(upload) = self.0.process_exit_upload.as_ref() { + upload.configure_executable(executable); + } + } + + pub(crate) fn prepare_process_exit_upload(&self) -> bool { + self.0 + .process_exit_upload + .as_ref() + .is_some_and(StatsigUpload::prepare) + } + + pub(crate) fn shutdown_on_process_exit(&self) -> Result<()> { + self.0.shutdown_on_process_exit() + } } fn os_resource_attributes() -> Vec { @@ -363,13 +396,28 @@ where fn build_otlp_metric_exporter( exporter: OtelExporter, temporality: Temporality, -) -> Result { +) -> Result<(opentelemetry_otlp::MetricExporter, Option)> { match exporter { OtelExporter::None => Err(MetricsError::ExporterDisabled), - OtelExporter::Statsig => build_otlp_metric_exporter( - crate::config::resolve_exporter(&OtelExporter::Statsig), - temporality, - ), + OtelExporter::Statsig => { + let OtelExporter::OtlpHttp { + endpoint, + headers, + protocol, + tls, + } = crate::config::resolve_exporter(&OtelExporter::Statsig) + else { + return Err(MetricsError::ExporterDisabled); + }; + build_http_metric_exporter( + endpoint, + headers, + protocol, + tls, + temporality, + /*process_exit_upload*/ true, + ) + } OtelExporter::OtlpGrpc { endpoint, headers, @@ -391,47 +439,75 @@ fn build_otlp_metric_exporter( None => base_tls_config, }; - opentelemetry_otlp::MetricExporter::builder() + let exporter = opentelemetry_otlp::MetricExporter::builder() .with_tonic() .with_endpoint(endpoint) .with_temporality(temporality) .with_metadata(MetadataMap::from_headers(header_map)) .with_tls_config(tls_config) .build() - .map_err(|source| MetricsError::ExporterBuild { source }) + .map_err(|source| MetricsError::ExporterBuild { source })?; + Ok((exporter, None)) } OtelExporter::OtlpHttp { endpoint, headers, protocol, tls, - } => { - debug!("Using OTLP Http exporter for metrics: {endpoint}"); - - let protocol = match protocol { - OtelHttpProtocol::Binary => Protocol::HttpBinary, - OtelHttpProtocol::Json => Protocol::HttpJson, - }; - - let mut exporter_builder = opentelemetry_otlp::MetricExporter::builder() - .with_http() - .with_endpoint(endpoint) - .with_temporality(temporality) - .with_protocol(protocol) - .with_headers(headers); - - if let Some(tls) = tls.as_ref() { - let client = - crate::otlp::build_http_client(tls, OTEL_EXPORTER_OTLP_METRICS_TIMEOUT) - .map_err(|err| MetricsError::InvalidConfig { - message: err.to_string(), - })?; - exporter_builder = exporter_builder.with_http_client(client); - } - - exporter_builder - .build() - .map_err(|source| MetricsError::ExporterBuild { source }) - } + } => build_http_metric_exporter( + endpoint, + headers, + protocol, + tls, + temporality, + /*process_exit_upload*/ false, + ), } } + +fn build_http_metric_exporter( + endpoint: String, + headers: HashMap, + protocol: OtelHttpProtocol, + tls: Option, + temporality: Temporality, + process_exit_upload: bool, +) -> Result<(opentelemetry_otlp::MetricExporter, Option)> { + debug!("Using OTLP Http exporter for metrics: {endpoint}"); + + let protocol = match protocol { + OtelHttpProtocol::Binary => Protocol::HttpBinary, + OtelHttpProtocol::Json => Protocol::HttpJson, + }; + + let mut exporter_builder = opentelemetry_otlp::MetricExporter::builder() + .with_http() + .with_endpoint(endpoint) + .with_temporality(temporality) + .with_protocol(protocol) + .with_headers(headers); + let upload = process_exit_upload.then(StatsigUpload::new); + + if tls.is_some() || upload.is_some() { + let tls = tls.unwrap_or_default(); + let client = crate::otlp::build_http_client(&tls, OTEL_EXPORTER_OTLP_METRICS_TIMEOUT) + .map_err(|err| MetricsError::InvalidConfig { + message: err.to_string(), + })?; + exporter_builder = match upload.as_ref() { + Some(upload) => { + exporter_builder.with_http_client(StatsigUploadClient::new(client, upload.clone())) + } + None => exporter_builder.with_http_client(client), + }; + } + + let exporter = exporter_builder + .build() + .map_err(|source| MetricsError::ExporterBuild { source })?; + Ok((exporter, upload)) +} + +#[cfg(test)] +#[path = "client_tests.rs"] +mod tests; diff --git a/codex-rs/otel/src/metrics/client_tests.rs b/codex-rs/otel/src/metrics/client_tests.rs new file mode 100644 index 0000000000..e8dae3c0da --- /dev/null +++ b/codex-rs/otel/src/metrics/client_tests.rs @@ -0,0 +1,55 @@ +#![cfg(unix)] + +use super::*; +use crate::config::STATSIG_OTLP_HTTP_ENDPOINT; +use std::fs; +use std::os::unix::fs::PermissionsExt; +use std::thread; + +#[test] +fn process_exit_upload_contains_final_metric() { + let temp_dir = tempfile::tempdir().expect("create temporary directory"); + let executable = temp_dir.path().join("upload-helper"); + let output = temp_dir.path().join("upload-helper.output"); + fs::write(&executable, "#!/bin/sh\ncat > \"$0.output\"\n").expect("write upload helper"); + let mut permissions = fs::metadata(&executable).unwrap().permissions(); + permissions.set_mode(0o700); + fs::set_permissions(&executable, permissions).expect("make upload helper executable"); + + let (exporter, process_exit_upload) = build_http_metric_exporter( + STATSIG_OTLP_HTTP_ENDPOINT.to_string(), + HashMap::new(), + OtelHttpProtocol::Json, + None, + Temporality::Delta, + /*process_exit_upload*/ true, + ) + .expect("build Statsig metric exporter"); + let upload = process_exit_upload.expect("process-exit upload state"); + upload.configure_executable(executable); + assert!(upload.prepare()); + + let (provider, meter) = build_provider( + Resource::builder_empty().build(), + exporter, + Some(Duration::from_secs(/*secs*/ 3600)), + None, + ); + meter + .u64_counter("codex.test.process_exit") + .build() + .add(1, &[]); + + provider.shutdown().expect("hand off final metric"); + + for _ in 0..200 { + if fs::read(&output).is_ok_and(|body| { + body.windows(b"codex.test.process_exit".len()) + .any(|window| window == b"codex.test.process_exit") + }) { + return; + } + thread::sleep(Duration::from_millis(10)); + } + panic!("detached upload payload did not contain the final metric"); +} diff --git a/codex-rs/otel/src/process_exit_upload.rs b/codex-rs/otel/src/process_exit_upload.rs new file mode 100644 index 0000000000..c6c45d51de --- /dev/null +++ b/codex-rs/otel/src/process_exit_upload.rs @@ -0,0 +1,365 @@ +use crate::config::STATSIG_API_KEY; +use crate::config::STATSIG_API_KEY_HEADER; +use crate::config::STATSIG_OTLP_HTTP_ENDPOINT; +use opentelemetry_http::Bytes; +use opentelemetry_http::HttpClient; +use opentelemetry_http::HttpError; +use opentelemetry_http::Request; +use opentelemetry_http::Response; +use opentelemetry_otlp::OTEL_EXPORTER_OTLP_METRICS_TIMEOUT; +use reqwest::header::CONTENT_TYPE; +use std::ffi::OsStr; +use std::fmt; +use std::fs::File; +use std::io; +use std::io::Read; +use std::io::Seek; +use std::io::SeekFrom; +use std::io::Write; +use std::path::PathBuf; +use std::process::Child; +use std::process::Command; +use std::process::Stdio; +use std::sync::Arc; +use std::sync::Condvar; +use std::sync::Mutex; +use std::sync::mpsc; +use thiserror::Error; + +const PROCESS_EXIT_UPLOAD_ARG: &str = "--codex-upload-statsig-metrics"; + +const MAX_UPLOAD_BYTES: usize = 1024 * 1024; +const MAX_PROCESS_EXIT_UPLOADS: usize = 2; + +#[derive(Clone)] +pub(crate) struct StatsigUpload { + inner: Arc, +} + +struct StatsigUploadInner { + state: Mutex, + normal_helper_finished: Condvar, +} + +#[derive(Debug, Default)] +struct StatsigUploadState { + executable: Option, + process_exit: bool, + normal_helper_active: bool, + active_process_exit_helpers: usize, + reaper: Option>, +} + +impl fmt::Debug for StatsigUpload { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("StatsigUpload") + .finish_non_exhaustive() + } +} + +impl StatsigUpload { + pub(crate) fn new() -> Self { + Self { + inner: Arc::new(StatsigUploadInner { + state: Mutex::new(StatsigUploadState::default()), + normal_helper_finished: Condvar::new(), + }), + } + } + + pub(crate) fn configure_executable(&self, executable: PathBuf) { + let mut state = self.state(); + if state.reaper.is_some() { + state.executable = Some(executable); + return; + } + let (sender, receiver) = mpsc::channel(); + match std::thread::Builder::new() + .name("codex-statsig-uploader-reaper".to_string()) + .spawn(move || reap_uploaders(receiver)) + { + Ok(_) => { + state.reaper = Some(sender); + state.executable = Some(executable); + } + Err(err) => { + tracing::warn!(%err, "Failed to start Statsig metrics uploader reaper"); + } + } + } + + pub(crate) fn prepare(&self) -> bool { + let mut state = self.state(); + if state.executable.is_none() { + return false; + } + state.process_exit = true; + self.inner.normal_helper_finished.notify_all(); + true + } + + fn action(&self) -> io::Result { + let mut state = self.state(); + let Some(executable) = state.executable.clone() else { + return Ok(StatsigUploadAction::SendDirectly); + }; + while state.normal_helper_active && !state.process_exit { + state = self + .inner + .normal_helper_finished + .wait(state) + .unwrap_or_else(std::sync::PoisonError::into_inner); + } + if state.process_exit { + if state.active_process_exit_helpers >= MAX_PROCESS_EXIT_UPLOADS { + return Err(io::Error::new( + io::ErrorKind::WouldBlock, + "Statsig metrics uploader has too many active process-exit helpers", + )); + } + state.active_process_exit_helpers += 1; + return Ok(StatsigUploadAction::Spawn( + executable, + HelperSlot::new(self.clone(), UploadKind::ProcessExit), + )); + } + state.normal_helper_active = true; + Ok(StatsigUploadAction::Spawn( + executable, + HelperSlot::new(self.clone(), UploadKind::Normal), + )) + } + + fn spawn(&self, executable: PathBuf, payload: Bytes, slot: HelperSlot) -> io::Result<()> { + let mut file = tempfile::tempfile()?; + file.write_all(&payload)?; + file.seek(SeekFrom::Start(0))?; + let child = spawn_uploader(executable, file)?; + self.reap(ReapRequest { child, _slot: slot }); + Ok(()) + } + + fn reap(&self, request: ReapRequest) { + let failed = match self.state().reaper.clone() { + Some(sender) => match sender.send(request) { + Ok(()) => return, + Err(err) => err.0, + }, + None => request, + }; + tracing::warn!("Statsig metrics uploader reaper is unavailable"); + let ReapRequest { child, _slot } = failed; + drop(child); + std::mem::forget(_slot); + } + + fn state(&self) -> std::sync::MutexGuard<'_, StatsigUploadState> { + self.inner + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } +} + +enum StatsigUploadAction { + SendDirectly, + Spawn(PathBuf, HelperSlot), +} + +#[derive(Clone, Copy, Eq, PartialEq)] +enum UploadKind { + Normal, + ProcessExit, +} + +struct HelperSlot { + upload: StatsigUpload, + kind: UploadKind, +} + +impl HelperSlot { + fn new(upload: StatsigUpload, kind: UploadKind) -> Self { + Self { upload, kind } + } +} + +impl Drop for HelperSlot { + fn drop(&mut self) { + let mut state = self.upload.state(); + match self.kind { + UploadKind::Normal => { + debug_assert!(state.normal_helper_active); + state.normal_helper_active = false; + self.upload.inner.normal_helper_finished.notify_all(); + } + UploadKind::ProcessExit => { + debug_assert!(state.active_process_exit_helpers > 0); + state.active_process_exit_helpers -= 1; + } + } + } +} + +struct ReapRequest { + child: Child, + _slot: HelperSlot, +} + +#[derive(Debug)] +pub(crate) struct StatsigUploadClient { + inner: reqwest::blocking::Client, + upload: StatsigUpload, +} + +impl StatsigUploadClient { + pub(crate) fn new(inner: reqwest::blocking::Client, upload: StatsigUpload) -> Self { + Self { inner, upload } + } + + fn validate(request: &Request) -> Result<(), HttpError> { + if request.body().len() > MAX_UPLOAD_BYTES { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "Statsig metrics upload exceeded the {MAX_UPLOAD_BYTES}-byte detached limit" + ), + ) + .into()); + } + + Ok(()) + } +} + +#[async_trait::async_trait] +impl HttpClient for StatsigUploadClient { + async fn send_bytes(&self, request: Request) -> Result, HttpError> { + // PeriodicReader serializes exports and shutdown on one worker. Once + // the TUI can re-exec Codex, keep network I/O out of that worker so a + // final flush cannot queue behind an in-flight HTTP request. + Self::validate(&request)?; + match self.upload.action()? { + StatsigUploadAction::SendDirectly => self.inner.send_bytes(request).await, + StatsigUploadAction::Spawn(executable, slot) => { + self.upload.spawn(executable, request.into_body(), slot)?; + synthetic_success() + } + } + } +} + +fn spawn_uploader(executable: PathBuf, payload: File) -> io::Result { + let mut command = Command::new(executable); + command + .arg(PROCESS_EXIT_UPLOAD_ARG) + .stdin(Stdio::from(payload)) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + + #[cfg(unix)] + { + use std::os::unix::process::CommandExt; + + // SAFETY: setsid is async-signal-safe and does not access parent memory. + unsafe { + command.pre_exec(|| { + if libc::setsid() == -1 { + return Err(io::Error::last_os_error()); + } + Ok(()) + }); + } + } + + #[cfg(windows)] + { + use std::os::windows::process::CommandExt; + + const DETACHED_PROCESS: u32 = 0x0000_0008; + const CREATE_NEW_PROCESS_GROUP: u32 = 0x0000_0200; + const CREATE_NO_WINDOW: u32 = 0x0800_0000; + command.creation_flags(DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW); + } + + command.spawn() +} + +fn synthetic_success() -> Result, HttpError> { + Ok(Response::builder() + .status(http::StatusCode::OK) + .body(Bytes::new())?) +} + +fn reap_uploaders(receiver: mpsc::Receiver) { + while let Ok(mut request) = receiver.recv() { + log_uploader_status(request.child.wait()); + } +} + +fn log_uploader_status(status: io::Result) { + match status { + Ok(status) if status.success() => {} + Ok(status) => { + tracing::warn!(?status, "Statsig metrics uploader exited unsuccessfully"); + } + Err(err) => { + tracing::warn!(%err, "Failed to reap Statsig metrics uploader"); + } + } +} + +#[derive(Debug, Error)] +pub enum ProcessExitUploadError { + #[error("failed to read process-exit metrics payload")] + Read(#[source] io::Error), + + #[error("process-exit metrics payload exceeded the {MAX_UPLOAD_BYTES}-byte limit")] + TooLarge, + + #[error("failed to build Statsig process-exit upload client")] + BuildClient(#[source] reqwest::Error), + + #[error("failed to upload process-exit metrics")] + Upload(#[source] reqwest::Error), +} + +pub fn run_process_exit_upload_if_requested() -> Option> { + (std::env::args_os().nth(1).as_deref() == Some(OsStr::new(PROCESS_EXIT_UPLOAD_ARG))) + .then(upload_statsig_metrics_from_stdin) +} + +fn upload_statsig_metrics_from_stdin() -> Result<(), ProcessExitUploadError> { + upload_statsig_metrics(io::stdin().lock(), STATSIG_OTLP_HTTP_ENDPOINT) +} + +fn upload_statsig_metrics(reader: impl Read, endpoint: &str) -> Result<(), ProcessExitUploadError> { + let mut body = Vec::new(); + reader + .take((MAX_UPLOAD_BYTES + 1) as u64) + .read_to_end(&mut body) + .map_err(ProcessExitUploadError::Read)?; + if body.len() > MAX_UPLOAD_BYTES { + return Err(ProcessExitUploadError::TooLarge); + } + + let client = reqwest::blocking::Client::builder() + .timeout(crate::otlp::resolve_otlp_timeout( + OTEL_EXPORTER_OTLP_METRICS_TIMEOUT, + )) + .build() + .map_err(ProcessExitUploadError::BuildClient)?; + client + .post(endpoint) + .header(CONTENT_TYPE, "application/json") + .header(STATSIG_API_KEY_HEADER, STATSIG_API_KEY) + .body(body) + .send() + .and_then(reqwest::blocking::Response::error_for_status) + .map_err(ProcessExitUploadError::Upload)?; + Ok(()) +} + +#[cfg(test)] +#[path = "process_exit_upload_tests.rs"] +mod tests; diff --git a/codex-rs/otel/src/process_exit_upload_tests.rs b/codex-rs/otel/src/process_exit_upload_tests.rs new file mode 100644 index 0000000000..179c3f3628 --- /dev/null +++ b/codex-rs/otel/src/process_exit_upload_tests.rs @@ -0,0 +1,167 @@ +use super::*; +use pretty_assertions::assert_eq; +#[cfg(unix)] +use std::fs; +use std::io::Cursor; +#[cfg(unix)] +use std::os::unix::fs::PermissionsExt; +#[cfg(unix)] +use std::sync::Arc; +#[cfg(unix)] +use std::sync::mpsc; +use std::thread; +#[cfg(unix)] +use std::time::Duration; +use wiremock::Mock; +use wiremock::MockServer; +use wiremock::ResponseTemplate; +use wiremock::matchers::body_json; +use wiremock::matchers::header; +use wiremock::matchers::method; +use wiremock::matchers::path; + +#[tokio::test] +async fn upload_statsig_metrics_sends_exact_body_and_headers() { + let server = MockServer::start().await; + let body = serde_json::json!({"resourceMetrics": []}); + Mock::given(method("POST")) + .and(path("/v1/metrics")) + .and(header("content-type", "application/json")) + .and(header(STATSIG_API_KEY_HEADER, STATSIG_API_KEY)) + .and(body_json(&body)) + .respond_with(ResponseTemplate::new(200)) + .expect(1) + .mount(&server) + .await; + let endpoint = format!("{}/v1/metrics", server.uri()); + let body = body.to_string(); + + tokio::task::spawn_blocking(move || upload_statsig_metrics(Cursor::new(body), &endpoint)) + .await + .expect("join upload task") + .expect("upload metrics"); +} + +#[cfg(unix)] +#[test] +fn detached_uploader_outlives_parent_process() { + const CHILD_ENV: &str = "CODEX_TEST_DETACHED_STATSIG_UPLOAD"; + if let Some(executable) = std::env::var_os(CHILD_ENV) { + let mut payload = tempfile::tempfile().expect("create upload payload"); + payload.write_all(b"detached payload").unwrap(); + payload.seek(SeekFrom::Start(0)).unwrap(); + spawn_uploader(executable.into(), payload).expect("spawn detached uploader"); + std::process::exit(0); + } + + let temp_dir = tempfile::tempdir().expect("create temporary directory"); + let executable = temp_dir.path().join("upload-helper"); + let output = temp_dir.path().join("upload-helper.output"); + fs::write(&executable, "#!/bin/sh\nsleep 0.2\ncat > \"$0.output\"\n") + .expect("write upload helper"); + let mut permissions = fs::metadata(&executable).unwrap().permissions(); + permissions.set_mode(0o700); + fs::set_permissions(&executable, permissions).expect("make upload helper executable"); + + let status = Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "process_exit_upload::tests::detached_uploader_outlives_parent_process", + ]) + .env(CHILD_ENV, &executable) + .status() + .expect("run uploader parent process"); + assert!(status.success()); + + for _ in 0..200 { + if fs::read(&output).is_ok_and(|body| body == b"detached payload") { + return; + } + thread::sleep(Duration::from_millis(10)); + } + panic!("detached uploader did not finish after its parent exited"); +} + +#[test] +fn process_exit_client_does_not_fall_back_when_helper_spawn_fails() { + let upload = StatsigUpload::new(); + upload.configure_executable(PathBuf::from("/path/that/does/not/exist/codex")); + let client = StatsigUploadClient::new(reqwest::blocking::Client::new(), upload); + let runtime = tokio::runtime::Builder::new_current_thread() + .build() + .expect("build current-thread runtime"); + + let result = runtime.block_on(client.send_bytes(statsig_request())); + let retry = runtime.block_on(client.send_bytes(statsig_request())); + + assert!(result.is_err()); + assert!(retry.is_err()); +} + +#[cfg(unix)] +#[test] +fn statsig_uploader_backpressures_normal_uploads_and_reserves_exit_helpers() { + let temp_dir = tempfile::tempdir().expect("create temporary directory"); + let executable = temp_dir.path().join("upload-helper"); + fs::write(&executable, "#!/bin/sh\ncat >/dev/null\nsleep 1\n").expect("write upload helper"); + let mut permissions = fs::metadata(&executable).unwrap().permissions(); + permissions.set_mode(0o700); + fs::set_permissions(&executable, permissions).expect("make upload helper executable"); + + let upload = StatsigUpload::new(); + upload.configure_executable(executable); + let client = Arc::new(StatsigUploadClient::new( + reqwest::blocking::Client::new(), + upload.clone(), + )); + let runtime = tokio::runtime::Builder::new_current_thread() + .build() + .expect("build current-thread runtime"); + + let first = runtime.block_on(client.send_bytes(statsig_request())); + assert!(first.is_ok()); + + let (sender, receiver) = mpsc::channel(); + let waiting_client = Arc::clone(&client); + thread::spawn(move || { + let runtime = tokio::runtime::Builder::new_current_thread() + .build() + .expect("build current-thread runtime"); + sender + .send(runtime.block_on(waiting_client.send_bytes(statsig_request()))) + .expect("send waiting upload result"); + }); + assert!(receiver.recv_timeout(Duration::from_millis(100)).is_err()); + + assert!(upload.prepare()); + let waiting_upload = receiver + .recv_timeout(Duration::from_secs(1)) + .expect("waiting upload should be handed off"); + let final_upload = runtime.block_on(client.send_bytes(statsig_request())); + let over_limit = runtime.block_on(client.send_bytes(statsig_request())); + assert!(waiting_upload.is_ok()); + assert!(final_upload.is_ok()); + assert!(over_limit.is_err()); +} + +#[test] +fn upload_statsig_metrics_rejects_oversized_body() { + let result = upload_statsig_metrics( + Cursor::new(vec![0; MAX_UPLOAD_BYTES + 1]), + "http://127.0.0.1:1/v1/metrics", + ); + + assert_eq!( + result.unwrap_err().to_string(), + format!("process-exit metrics payload exceeded the {MAX_UPLOAD_BYTES}-byte limit") + ); +} + +fn statsig_request() -> Request { + Request::builder() + .method(http::Method::POST) + .uri(STATSIG_OTLP_HTTP_ENDPOINT) + .header(CONTENT_TYPE, "application/json") + .body(Bytes::from_static(br#"{"resourceMetrics":[]}"#)) + .expect("build Statsig request") +} diff --git a/codex-rs/otel/src/provider.rs b/codex-rs/otel/src/provider.rs index a1804b3e20..e70e23732f 100644 --- a/codex-rs/otel/src/provider.rs +++ b/codex-rs/otel/src/provider.rs @@ -40,6 +40,7 @@ use opentelemetry_semantic_conventions as semconv; use std::collections::BTreeMap; use std::error::Error; use std::mem::ManuallyDrop; +use std::path::PathBuf; use std::time::Duration; use tracing::debug; use tracing_subscriber::Layer; @@ -93,9 +94,17 @@ impl OtelProvider { } fn shutdown_on_process_exit_with( - self, + mut self, spawn: impl FnOnce(Box) -> std::io::Result<()>, ) { + if let Some(metrics) = self.metrics.take() { + if metrics.prepare_process_exit_upload() { + let _ = metrics.shutdown_on_process_exit(); + } else { + self.metrics = Some(metrics); + } + } + let provider = ManuallyDrop::new(self); if let Err(err) = spawn(Box::new(move || { drop(ManuallyDrop::into_inner(provider)); @@ -134,7 +143,7 @@ impl OtelProvider { settings.environment.clone(), settings.service_name.clone(), settings.service_version.clone(), - metric_exporter, + settings.metrics_exporter.clone(), ); if settings.runtime_metrics { config = config.with_runtime_reader(); @@ -222,6 +231,13 @@ impl OtelProvider { pub fn metrics(&self) -> Option<&MetricsClient> { self.metrics.as_ref() } + + /// Route built-in Statsig metric exports through detached Codex helpers. + pub fn configure_statsig_metrics_uploader(&self, executable: PathBuf) { + if let Some(metrics) = self.metrics.as_ref() { + metrics.configure_statsig_metrics_uploader(executable); + } + } } impl Drop for OtelProvider { diff --git a/codex-rs/tui/src/lib.rs b/codex-rs/tui/src/lib.rs index 1659886da7..fe73821964 100644 --- a/codex-rs/tui/src/lib.rs +++ b/codex-rs/tui/src/lib.rs @@ -1144,6 +1144,11 @@ pub async fn run_main( None } }; + if let Some(otel) = otel.as_ref() + && let Some(executable) = arg0_paths.codex_self_exe.clone() + { + otel.configure_statsig_metrics_uploader(executable); + } crate::legacy_core::otel_init::record_process_start(otel.as_ref(), otel_originator.as_str()); crate::legacy_core::otel_init::install_sqlite_telemetry( otel.as_ref(), diff --git a/codex-rs/tui/src/main.rs b/codex-rs/tui/src/main.rs index 9c69c7c81a..75552ee601 100644 --- a/codex-rs/tui/src/main.rs +++ b/codex-rs/tui/src/main.rs @@ -45,6 +45,11 @@ struct TopCli { } fn main() -> anyhow::Result<()> { + if let Some(result) = codex_otel::run_process_exit_upload_if_requested() { + result?; + return Ok(()); + } + arg0_dispatch_or_else(|arg0_paths: Arg0DispatchPaths| async move { let top_cli = TopCli::parse(); let mut inner = top_cli.inner;