diff --git a/codex-rs/app-server/src/otel_reloader.rs b/codex-rs/app-server/src/otel_reloader.rs index 9f46caa23f..4ad302fc24 100644 --- a/codex-rs/app-server/src/otel_reloader.rs +++ b/codex-rs/app-server/src/otel_reloader.rs @@ -28,16 +28,17 @@ where .map(Layer::boxed); let (logger_layer, logger_handle) = reload::Layer::new(logger_export_layer); - let mut layers: Vec + Send + Sync + 'static>> = vec![ + let layers: Vec + Send + Sync + 'static>> = vec![ logger_layer .with_filter(tracing_subscriber::filter::filter_fn( OtelProvider::log_export_filter, )) .boxed(), + // The subscriber graph is fixed after startup. Keep the global tracer + // bridge installed even when tracing starts disabled so an account + // switch can install a trace exporter later. + OtelProvider::reloadable_tracing_layer(OTEL_SERVICE_NAME).boxed(), ]; - if provider.is_some_and(|provider| provider.tracer.is_some()) { - layers.push(OtelProvider::reloadable_tracing_layer(OTEL_SERVICE_NAME).boxed()); - } (layers, logger_handle) } diff --git a/codex-rs/app-server/tests/suite/v2/otel.rs b/codex-rs/app-server/tests/suite/v2/otel.rs index a5ec203eb4..302e749dd1 100644 --- a/codex-rs/app-server/tests/suite/v2/otel.rs +++ b/codex-rs/app-server/tests/suite/v2/otel.rs @@ -27,6 +27,11 @@ const NEXT_EMAIL: &str = "next-account@example.com"; const PARENT_TRACE_ID: &str = "4bf92f3577b34da6a3ce929d0e0e4736"; const PARENT_TRACEPARENT: &str = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"; +enum TraceExporterSetting { + Disabled, + Collector, +} + #[tokio::test] async fn account_switch_reloads_telemetry_collectors_and_preserves_trace_context() -> Result<()> { let collector = MockServer::start().await; @@ -37,7 +42,11 @@ async fn account_switch_reloads_telemetry_collectors_and_preserves_trace_context let codex_home = TempDir::new()?; let initial_endpoint = format!("{}/initial", collector.uri()); - write_otel_config(codex_home.path(), &initial_endpoint)?; + write_otel_config( + codex_home.path(), + &initial_endpoint, + TraceExporterSetting::Collector, + )?; write_chatgpt_auth( codex_home.path(), ChatGptAuthFixture::new("initial-access-token") @@ -81,7 +90,11 @@ async fn account_switch_reloads_telemetry_collectors_and_preserves_trace_context .context("initial account telemetry must be exported before its permission is revoked")?; let next_endpoint = format!("{}/next", collector.uri()); - write_otel_config(codex_home.path(), &next_endpoint)?; + write_otel_config( + codex_home.path(), + &next_endpoint, + TraceExporterSetting::Collector, + )?; let access_token = encode_id_token( &ChatGptIdTokenClaims::new() .email(NEXT_EMAIL) @@ -184,7 +197,124 @@ async fn account_switch_reloads_telemetry_collectors_and_preserves_trace_context Ok(()) } -fn write_otel_config(codex_home: &Path, collector_endpoint: &str) -> Result<()> { +#[tokio::test] +async fn account_switch_enables_trace_exporter_after_startup_without_tracing() -> Result<()> { + let collector = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200)) + .mount(&collector) + .await; + + let codex_home = TempDir::new()?; + let initial_endpoint = format!("{}/initial", collector.uri()); + write_otel_config( + codex_home.path(), + &initial_endpoint, + TraceExporterSetting::Disabled, + )?; + write_trace_requirements( + codex_home.path(), + &initial_endpoint, + TraceExporterSetting::Disabled, + )?; + write_chatgpt_auth( + codex_home.path(), + ChatGptAuthFixture::new("initial-access-token") + .account_id(INITIAL_ACCOUNT_ID) + .chatgpt_account_id(INITIAL_ACCOUNT_ID) + .chatgpt_user_id("initial-user") + .plan_type("pro") + .email(INITIAL_EMAIL), + AuthCredentialsStoreMode::File, + )?; + write_models_cache(codex_home.path()).await?; + + let mut app_server = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .with_env_overrides(&[("TRACEPARENT", Some(PARENT_TRACEPARENT))]) + .with_json_logging("codex_app_server::otel_reloader=info") + .build_initialized_with_timeout(TEST_TIMEOUT) + .await?; + app_server + .start_thread(ThreadStartParams::default()) + .await?; + + let next_endpoint = format!("{}/next", collector.uri()); + write_trace_requirements( + codex_home.path(), + &next_endpoint, + TraceExporterSetting::Collector, + )?; + let access_token = encode_id_token( + &ChatGptIdTokenClaims::new() + .email(NEXT_EMAIL) + .plan_type("pro") + .chatgpt_account_id(NEXT_ACCOUNT_ID) + .chatgpt_user_id("next-user"), + )?; + let request_id = app_server + .send_chatgpt_auth_tokens_login_request( + access_token, + NEXT_ACCOUNT_ID.to_string(), + Some("pro".to_string()), + ) + .await?; + let response: LoginAccountResponse = + timeout(TEST_TIMEOUT, app_server.read_response(request_id)).await??; + assert_eq!(response, LoginAccountResponse::ChatgptAuthTokens {}); + timeout( + TEST_TIMEOUT, + app_server.wait_for_json_log_event( + "codex.app_server.otel_reloaded", + Duration::from_secs(/*secs*/ 10), + ), + ) + .await??; + app_server + .start_thread(ThreadStartParams::default()) + .await?; + + let status = timeout(TEST_TIMEOUT, app_server.shutdown_gracefully()).await??; + assert!(status.success(), "app-server did not shut down cleanly"); + + let requests = collector + .received_requests() + .await + .context("collector did not record requests")?; + let exported = |path: &str| { + requests + .iter() + .filter(|request| request.url.path() == path) + .map(|request| String::from_utf8_lossy(&request.body).into_owned()) + .collect::>() + .join("\n") + }; + let initial_traces = exported("/initial/traces"); + let next_traces = exported("/next/traces"); + + assert!( + initial_traces.is_empty(), + "traces were exported before tracing was configured: {initial_traces}" + ); + assert!( + next_traces.contains(PARENT_TRACE_ID), + "the requirements-managed trace exporter did not start after account switch: {next_traces}" + ); + + Ok(()) +} + +fn write_otel_config( + codex_home: &Path, + collector_endpoint: &str, + trace_exporter_setting: TraceExporterSetting, +) -> Result<()> { + let trace_exporter = match trace_exporter_setting { + TraceExporterSetting::Disabled => String::new(), + TraceExporterSetting::Collector => format!( + r#"trace_exporter = {{ otlp-http = {{ endpoint = "{collector_endpoint}/traces", protocol = "json" }} }}"# + ), + }; std::fs::write( codex_home.join("config.toml"), format!( @@ -202,10 +332,32 @@ enabled = false [otel] environment = "test" exporter = {{ otlp-http = {{ endpoint = "{collector_endpoint}/logs", protocol = "json", headers = {{ x-collector-key = "test-collector-key" }} }} }} -trace_exporter = {{ otlp-http = {{ endpoint = "{collector_endpoint}/traces", protocol = "json" }} }} +{trace_exporter} metrics_exporter = {{ otlp-http = {{ endpoint = "{collector_endpoint}/metrics", protocol = "json" }} }} "# ), )?; Ok(()) } + +fn write_trace_requirements( + codex_home: &Path, + collector_endpoint: &str, + trace_exporter_setting: TraceExporterSetting, +) -> Result<()> { + let trace_exporter = match trace_exporter_setting { + TraceExporterSetting::Disabled => "trace_exporter = \"none\"".to_string(), + TraceExporterSetting::Collector => format!( + r#"trace_exporter = {{ otlp-http = {{ endpoint = "{collector_endpoint}/traces", protocol = "json" }} }}"# + ), + }; + std::fs::write( + codex_home.join("requirements.toml"), + format!( + r#"[otel] +{trace_exporter} +"# + ), + )?; + Ok(()) +} diff --git a/codex-rs/otel/src/provider.rs b/codex-rs/otel/src/provider.rs index 24b118a8a9..d012b49d91 100644 --- a/codex-rs/otel/src/provider.rs +++ b/codex-rs/otel/src/provider.rs @@ -13,6 +13,7 @@ use opentelemetry::global; use opentelemetry::trace::Span as _; use opentelemetry::trace::SpanBuilder; use opentelemetry::trace::TracerProvider as _; +use opentelemetry::trace::noop::NoopTracerProvider; use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge; use opentelemetry_otlp::LogExporter; use opentelemetry_otlp::OTEL_EXPORTER_OTLP_LOGS_TIMEOUT; @@ -202,6 +203,7 @@ impl OtelProvider { // settings do not install an active provider. crate::trace_context::set_tracestate_entries(BTreeMap::new())?; crate::metrics::buffered::GLOBAL.disable(); + install_global_tracer_provider(/*tracer_provider*/ None); debug!("No OTEL exporter enabled in settings."); return Ok(None); } @@ -269,10 +271,7 @@ impl OtelProvider { provider.prepare_shutdown_worker()?; crate::trace_context::set_tracestate_entries(settings.tracestate.clone())?; - if let Some(tracer_provider) = provider.tracer_provider.clone() { - global::set_tracer_provider(tracer_provider); - global::set_text_map_propagator(TraceContextPropagator::new()); - } + install_global_tracer_provider(provider.tracer_provider.clone()); if let Some(metrics) = provider.metrics.as_mut() { *metrics = crate::metrics::install_global(metrics.clone()); if matches!(settings.metrics_exporter, OtelExporter::Statsig) { @@ -354,6 +353,15 @@ impl OtelProvider { } } +fn install_global_tracer_provider(tracer_provider: Option) { + if let Some(tracer_provider) = tracer_provider { + global::set_tracer_provider(tracer_provider); + global::set_text_map_propagator(TraceContextPropagator::new()); + } else { + global::set_tracer_provider(NoopTracerProvider::new()); + } +} + impl Drop for OtelProvider { fn drop(&mut self) { self.shutdown(); @@ -659,6 +667,7 @@ mod tests { use crate::metrics::TURN_TOKEN_USAGE_METRIC; use codex_http_client::HttpClientFactory; use codex_http_client::OutboundProxyPolicy; + use opentelemetry::trace::Tracer as _; use opentelemetry_sdk::metrics::InMemoryMetricExporter; use pretty_assertions::assert_eq; use std::path::PathBuf; @@ -726,6 +735,26 @@ mod tests { assert!(!is_trace_safe_target("codex_otel.network_proxy")); } + #[test] + fn disabled_trace_settings_replace_global_tracer_with_noop() -> Result<(), Box> { + global::set_tracer_provider(SdkTracerProvider::builder().build()); + let before = global::tracer("codex-otel-test").start("before-disable"); + assert!( + before.is_recording(), + "the test provider should record before tracing is disabled" + ); + + let provider = OtelProvider::try_new(&test_otel_settings())?; + assert!(provider.is_none()); + + let after = global::tracer("codex-otel-test").start("after-disable"); + assert!( + !after.is_recording(), + "new spans should not record after tracing is disabled" + ); + Ok(()) + } + #[test] fn cached_global_metrics_follow_reinstalled_provider() -> Result<(), Box> { let initial =