openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-31 19:28:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

otel: hand off Statsig metrics to detached uploader

Browse Source
This commit is contained in:
Zanie Blue
2026-06-09 16:28:31 -05:00
parent facb81e58d
commit 058195b9e4
13 changed files with 751 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+6
View File
@@ -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]]
codex-rs/Cargo.toml
+1
View File
@@ -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"
codex-rs/cli/Cargo.toml
+1
View File
@@ -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 }
codex-rs/cli/src/main.rs
+5
View File
@@ -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(())
codex-rs/otel/Cargo.toml
+5
View File
@@ -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 }
codex-rs/otel/src/lib.rs
+3
View File
@@ -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;
codex-rs/otel/src/metrics/client.rs
+115 -39
View File
@@ -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<HashMap<String, Histogram<f64>>>,
runtime_reader: Option<Arc<ManualReader>>,
default_tags: BTreeMap<String, String>,
process_exit_upload: Option<StatsigUpload>,
}
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<KeyValue> {
@@ -363,13 +396,28 @@ where
fn build_otlp_metric_exporter(
exporter: OtelExporter,
temporality: Temporality,
) -> Result<opentelemetry_otlp::MetricExporter> {
) -> Result<(opentelemetry_otlp::MetricExporter, Option<StatsigUpload>)> {
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<String, String>,
protocol: OtelHttpProtocol,
tls: Option<crate::config::OtelTlsConfig>,
temporality: Temporality,
process_exit_upload: bool,
) -> Result<(opentelemetry_otlp::MetricExporter, Option<StatsigUpload>)> {
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;
codex-rs/otel/src/metrics/client_tests.rs
+55
View File
@@ -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");
}
codex-rs/otel/src/process_exit_upload.rs
+365
View File
@@ -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<StatsigUploadInner>,
}
struct StatsigUploadInner {
state: Mutex<StatsigUploadState>,
normal_helper_finished: Condvar,
}
#[derive(Debug, Default)]
struct StatsigUploadState {
executable: Option<PathBuf>,
process_exit: bool,
normal_helper_active: bool,
active_process_exit_helpers: usize,
reaper: Option<mpsc::Sender<ReapRequest>>,
}
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<StatsigUploadAction> {
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<Bytes>) -> 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<Bytes>) -> Result<Response<Bytes>, 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<Child> {
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<Response<Bytes>, HttpError> {
Ok(Response::builder()
.status(http::StatusCode::OK)
.body(Bytes::new())?)
}
fn reap_uploaders(receiver: mpsc::Receiver<ReapRequest>) {
while let Ok(mut request) = receiver.recv() {
log_uploader_status(request.child.wait());
}
}
fn log_uploader_status(status: io::Result<std::process::ExitStatus>) {
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<Result<(), ProcessExitUploadError>> {
(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;
codex-rs/otel/src/process_exit_upload_tests.rs
+167
View File
@@ -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<Bytes> {
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")
}
codex-rs/otel/src/provider.rs
+18 -2
View File
@@ -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<dyn FnOnce() + Send + 'static>) -> 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 {
codex-rs/tui/src/lib.rs
+5
View File
@@ -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(),
codex-rs/tui/src/main.rs
+5
View File
@@ -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;