From c0cc9fc820f62e4dbdde68214dd6d792d6c8a7ee Mon Sep 17 00:00:00 2001 From: Abhinav Date: Thu, 12 Nov 2026 05:08:00 +0000 Subject: [PATCH] Add refreshable headers for OTLP HTTP exporters (#61028) ## Why Allow collector credentials to refresh without rebuilding OpenTelemetry providers. ## What changed Add `otel.headers_helper`, a command run from `CODEX_HOME` that returns a JSON object of header strings. Share its results across explicitly configured HTTP log, trace, and metric exporters. Refresh every 29 minutes by default, configurable through `otel.headers_helper_refresh_interval_ms`. Helper headers override matching static headers. Failed refreshes block affected exports until the helper recovers, and HTTP redirects are disabled when helper headers are used. Repository-local configuration cannot set the helper; exporters set by admin requirements, gRPC exporters, and built-in analytics do not use it. Helper execution respects network policy. ## Testing Add tests for shared execution, credential refresh and recovery, blocked exports after failure, network restrictions, and configuration validation. Collector tests cover header precedence across all three signals using JSON and binary HTTP encodings, including exclusion of an admin-managed exporter. GitOrigin-RevId: 89018ac09c2c99523d3f92bf77048e7ed9dae5f2 --- codex-rs/Cargo.lock | 2 + .../src/config_manager_service_tests.rs | 6 + codex-rs/code-mode-host/src/main.rs | 1 + codex-rs/config/src/loader/tests.rs | 32 +++ codex-rs/config/src/types.rs | 10 + codex-rs/core/config.schema.json | 10 + codex-rs/core/src/config/config_tests.rs | 16 ++ codex-rs/core/src/config/otel.rs | 2 + codex-rs/core/src/otel_init.rs | 27 ++ codex-rs/core/tests/suite/skills_extension.rs | 1 + .../src/server/process_otel_tests.rs | 1 + codex-rs/otel/Cargo.toml | 2 + codex-rs/otel/README.md | 10 +- codex-rs/otel/src/config.rs | 26 ++ codex-rs/otel/src/headers_helper.rs | 156 +++++++++++ codex-rs/otel/src/headers_helper_tests.rs | 257 ++++++++++++++++++ codex-rs/otel/src/lib.rs | 2 + codex-rs/otel/src/metrics/client.rs | 8 +- codex-rs/otel/src/metrics/config.rs | 5 + codex-rs/otel/src/network_policy.rs | 11 +- codex-rs/otel/src/otlp.rs | 27 +- codex-rs/otel/src/provider.rs | 53 +++- codex-rs/otel/src/provider_shutdown_tests.rs | 1 + .../tests/suite/buffered_operations_tests.rs | 1 + .../otel/tests/suite/otel_headers_helper.rs | 125 +++++++++ .../otel/tests/suite/otlp_http_loopback.rs | 9 + codex-rs/windows-sandbox-rs/src/wfp_setup.rs | 1 + 27 files changed, 789 insertions(+), 13 deletions(-) create mode 100644 codex-rs/otel/src/headers_helper.rs create mode 100644 codex-rs/otel/src/headers_helper_tests.rs create mode 100644 codex-rs/otel/tests/suite/otel_headers_helper.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 13762059f2..836156e7d8 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -4565,8 +4565,10 @@ dependencies = [ "chrono", "codex-api", "codex-http-client", + "codex-http-headers-helper", "codex-protocol", "codex-utils-absolute-path", + "codex-utils-process", "codex-utils-stream-parser", "codex-utils-string", "eventsource-stream", diff --git a/codex-rs/app-server/src/config_manager_service_tests.rs b/codex-rs/app-server/src/config_manager_service_tests.rs index ea1ce912bb..e663d2ef64 100644 --- a/codex-rs/app-server/src/config_manager_service_tests.rs +++ b/codex-rs/app-server/src/config_manager_service_tests.rs @@ -2011,6 +2011,8 @@ async fn read_projects_managed_otel_values_and_preserves_unmanaged_settings() { std::fs::write( tmp.path().join(CONFIG_TOML_FILE), r#"[otel] +headers_helper = "get-otel-headers" +headers_helper_refresh_interval_ms = 60000 tool_result = { max_bytes = 1234 } log_user_prompt = true environment = "user" @@ -2046,6 +2048,8 @@ tracestate = { managed = { tenant = "enterprise" }, empty = {} }"#, assert_eq!( otel, &serde_json::json!({ + "headers_helper": "get-otel-headers", + "headers_helper_refresh_interval_ms": 60000, "tool_result": { "max_bytes": 1234 }, "log_sse_deltas": null, "log_user_prompt": false, @@ -2073,6 +2077,8 @@ tracestate = { managed = { tenant = "enterprise" }, empty = {} }"#, ("otel.tracestate.managed.tenant", false), ("otel.tracestate.managed.region", true), ("otel.environment", true), + ("otel.headers_helper", true), + ("otel.headers_helper_refresh_interval_ms", true), ("otel.tool_result.max_bytes", true), ("otel.span_attributes.service.version", true), ("otel.tracestate.local.tenant", true), diff --git a/codex-rs/code-mode-host/src/main.rs b/codex-rs/code-mode-host/src/main.rs index 8f823c6624..a4c9187462 100644 --- a/codex-rs/code-mode-host/src/main.rs +++ b/codex-rs/code-mode-host/src/main.rs @@ -104,6 +104,7 @@ fn main() -> anyhow::Result<()> { fn build_trace_provider(endpoint: &str) -> anyhow::Result { OtelProvider::try_new(&OtelSettings { + headers_helper: None, http_client_factory: codex_http_client::HttpClientFactory::new( codex_http_client::OutboundProxyPolicy::ReqwestDefault, ), diff --git a/codex-rs/config/src/loader/tests.rs b/codex-rs/config/src/loader/tests.rs index 840882ac43..b6cf45848f 100644 --- a/codex-rs/config/src/loader/tests.rs +++ b/codex-rs/config/src/loader/tests.rs @@ -1273,3 +1273,35 @@ fn project_config_cannot_change_system_proxy_routing_or_credential_masking() { } } } + +#[test] +fn otel_headers_helpers_cannot_come_from_repository_config() { + let mut config: TomlValue = toml::from_str( + "[otel]\nheaders_helper = 'untrusted-command'\nheaders_helper_refresh_interval_ms = 1", + ) + .unwrap(); + let ignored = + sanitize_project_config(&mut config, CredentialBrokerProjectState::Unconfigured, &[]); + assert_eq!(ignored, ["otel"]); + assert!(config.get("otel").is_none()); +} + +#[test] +fn otel_headers_helper_interval_must_be_positive() { + assert!( + toml::from_str::( + "[otel]\nheaders_helper = 'helper'\nheaders_helper_refresh_interval_ms = 0" + ) + .is_err() + ); + let config: ConfigToml = toml::from_str( + "[otel]\nheaders_helper = 'helper'\nheaders_helper_refresh_interval_ms = 900000", + ) + .unwrap(); + let otel = config.otel.unwrap(); + assert_eq!(otel.headers_helper.as_deref(), Some("helper")); + assert_eq!( + otel.headers_helper_refresh_interval_ms.unwrap().get(), + 900000 + ); +} diff --git a/codex-rs/config/src/types.rs b/codex-rs/config/src/types.rs index 177faa086c..e23e9764f8 100644 --- a/codex-rs/config/src/types.rs +++ b/codex-rs/config/src/types.rs @@ -29,6 +29,7 @@ use codex_utils_absolute_path::AbsolutePathBuf; use std::collections::BTreeMap; use std::collections::HashMap; use std::fmt; +use std::num::NonZeroU64; use codex_utils_redacted_string::RedactedString; use schemars::JsonSchema; @@ -637,6 +638,11 @@ pub enum OtelExporterKind { #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default, JsonSchema)] #[schemars(deny_unknown_fields)] pub struct OtelConfigToml { + /// Command that prints a JSON object of HTTP headers for explicit OTLP HTTP exporters. + /// Runs from CODEX_HOME. Repository-local configuration cannot set OTel options. + pub headers_helper: Option, + /// Interval between helper executions in milliseconds. Defaults to 29 minutes. + pub headers_helper_refresh_interval_ms: Option, /// Byte limit for tool-result log output; independent of model-visible output. pub tool_result: Option, /// Log successful SSE events whose kind ends in `.delta`. Defaults to true. @@ -674,6 +680,8 @@ pub struct OtelConfigToml { /// Effective OTEL settings after defaults are applied. #[derive(Debug, Clone, PartialEq)] pub struct OtelConfig { + pub headers_helper: Option, + pub headers_helper_refresh_interval_ms: Option, pub tool_result: codex_protocol::config_types::ToolResultLogConfig, pub log_sse_deltas: bool, pub log_user_prompt: bool, @@ -690,6 +698,8 @@ pub struct OtelConfig { impl Default for OtelConfig { fn default() -> Self { OtelConfig { + headers_helper: None, + headers_helper_refresh_interval_ms: None, tool_result: Default::default(), log_sse_deltas: true, log_user_prompt: false, diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index b2de022d80..bfbd2b7300 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -3936,6 +3936,16 @@ ], "description": "Optional log exporter" }, + "headers_helper": { + "description": "Command that prints a JSON object of HTTP headers for explicit OTLP HTTP exporters. Runs from CODEX_HOME. Repository-local configuration cannot set OTel options.", + "type": "string" + }, + "headers_helper_refresh_interval_ms": { + "description": "Interval between helper executions in milliseconds. Defaults to 29 minutes.", + "format": "uint64", + "minimum": 1.0, + "type": "integer" + }, "log_agent_responses": { "description": "Opt in to logging final main-agent and spawned-subagent responses to an OTLP log exporter. Defaults to false. Response text can be sensitive and is capped at 64 KiB.", "type": "boolean" diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index 961ccd19db..2d41d3fcb4 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -10542,6 +10542,10 @@ async fn load_config_applies_otel_trace_metadata() -> std::io::Result<()> { let mut fixture = create_test_fixture()?; fixture.cfg = toml::from_str( r#" +[otel] +headers_helper = "company-auth telemetry-headers" +headers_helper_refresh_interval_ms = 900000 + [otel.span_attributes] "example.trace_attr" = "enabled" @@ -10562,6 +10566,18 @@ beta = "two" ) .await?; + assert_eq!( + config.otel.headers_helper.as_deref(), + Some("company-auth telemetry-headers") + ); + assert_eq!( + config + .otel + .headers_helper_refresh_interval_ms + .unwrap() + .get(), + 900000 + ); assert_eq!( config.otel.span_attributes, HashMap::from([("example.trace_attr".to_string(), "enabled".to_string())]) diff --git a/codex-rs/core/src/config/otel.rs b/codex-rs/core/src/config/otel.rs index 9195e8c066..36a075a650 100644 --- a/codex-rs/core/src/config/otel.rs +++ b/codex-rs/core/src/config/otel.rs @@ -27,6 +27,8 @@ pub(crate) fn resolve_config( let tracestate = resolve_tracestate(config.tracestate, startup_warnings); OtelConfig { + headers_helper: config.headers_helper, + headers_helper_refresh_interval_ms: config.headers_helper_refresh_interval_ms, tool_result: config.tool_result.unwrap_or_default(), log_sse_deltas: config.log_sse_deltas.unwrap_or(true), log_user_prompt, diff --git a/codex-rs/core/src/otel_init.rs b/codex-rs/core/src/otel_init.rs index 1fc9a05abe..b187bbd891 100644 --- a/codex-rs/core/src/otel_init.rs +++ b/codex-rs/core/src/otel_init.rs @@ -4,11 +4,16 @@ use codex_config::types::OtelHttpProtocol as Protocol; use codex_features::Feature; use codex_login::default_client::originator; use codex_otel::OtelExporter; +use codex_otel::OtelHeadersHelper; use codex_otel::OtelHttpProtocol; use codex_otel::OtelProvider; use codex_otel::OtelSettings; use codex_otel::OtelTlsConfig as OtelTlsSettings; use std::error::Error; +use std::num::NonZeroU64; +use std::time::Duration; + +const DEFAULT_HEADERS_HELPER_REFRESH_INTERVAL_MS: u64 = 29 * 60 * 1_000; /// Build an OpenTelemetry provider from the app Config. /// @@ -82,7 +87,29 @@ pub fn build_provider( let service_name = service_name_override.unwrap_or(originator.value.as_str()); let runtime_metrics = config.features.enabled(Feature::RuntimeMetrics); + let managed = config + .config_layer_stack + .requirements_toml() + .otel + .as_deref(); + let headers_helper = config + .otel + .headers_helper + .as_ref() + .map(|command| OtelHeadersHelper { + command: command.clone(), + apply_to_logs: managed.is_none_or(|otel| otel.exporter.is_none()), + apply_to_traces: managed.is_none_or(|otel| otel.trace_exporter.is_none()), + apply_to_metrics: managed.is_none_or(|otel| otel.metrics_exporter.is_none()), + refresh_interval: Duration::from_millis( + config + .otel + .headers_helper_refresh_interval_ms + .map_or(DEFAULT_HEADERS_HELPER_REFRESH_INTERVAL_MS, NonZeroU64::get), + ), + }); OtelProvider::try_new(&OtelSettings { + headers_helper, http_client_factory: config.http_client_factory(), service_name: service_name.to_string(), service_version: service_version.to_string(), diff --git a/codex-rs/core/tests/suite/skills_extension.rs b/codex-rs/core/tests/suite/skills_extension.rs index 004a9abc05..5a61c4ccbe 100644 --- a/codex-rs/core/tests/suite/skills_extension.rs +++ b/codex-rs/core/tests/suite/skills_extension.rs @@ -2734,6 +2734,7 @@ async fn assert_catalog_model_switch(max_context_tokens: Option) -> Resul let codex_home = Arc::new(TempDir::new()?); // Use the normal metrics sink to verify core's model attribution. let telemetry = OtelProvider::try_new(&OtelSettings { + headers_helper: None, http_client_factory: codex_core::test_support::default_http_client_factory(), environment: "test".to_string(), service_name: "skills-model-switch".to_string(), diff --git a/codex-rs/exec-server/src/server/process_otel_tests.rs b/codex-rs/exec-server/src/server/process_otel_tests.rs index 6b1d0b3421..b1e398985b 100644 --- a/codex-rs/exec-server/src/server/process_otel_tests.rs +++ b/codex-rs/exec-server/src/server/process_otel_tests.rs @@ -624,6 +624,7 @@ fn exported_logs( tls: None, }; let otel = OtelProvider::try_new(&OtelSettings { + headers_helper: None, http_client_factory: codex_http_client::HttpClientFactory::new( codex_http_client::OutboundProxyPolicy::ReqwestDefault, ), diff --git a/codex-rs/otel/Cargo.toml b/codex-rs/otel/Cargo.toml index 5d60385356..68c5dce934 100644 --- a/codex-rs/otel/Cargo.toml +++ b/codex-rs/otel/Cargo.toml @@ -20,6 +20,8 @@ chrono = { workspace = true } codex-utils-absolute-path = { workspace = true } codex-utils-stream-parser = { workspace = true } codex-utils-string = { workspace = true } +codex-utils-process = { workspace = true } +codex-http-headers-helper = { workspace = true } codex-api = { workspace = true } codex-protocol = { workspace = true } eventsource-stream = { workspace = true } diff --git a/codex-rs/otel/README.md b/codex-rs/otel/README.md index 43287ecb45..277e172f2d 100644 --- a/codex-rs/otel/README.md +++ b/codex-rs/otel/README.md @@ -15,13 +15,16 @@ metrics (when enabled), then attach its layers to your `tracing_subscriber` registry: ```rust -use codex_otel::config::OtelExporter; -use codex_otel::config::OtelHttpProtocol; -use codex_otel::config::OtelSettings; +use codex_http_client::{HttpClientFactory, OutboundProxyPolicy}; +use codex_otel::OtelExporter; +use codex_otel::OtelHttpProtocol; +use codex_otel::OtelSettings; use codex_otel::OtelProvider; use tracing_subscriber::prelude::*; let settings = OtelSettings { + headers_helper: None, + http_client_factory: HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), environment: "dev".to_string(), service_name: "codex-cli".to_string(), service_version: env!("CARGO_PKG_VERSION").to_string(), @@ -39,6 +42,7 @@ let settings = OtelSettings { tls: None, }, metrics_exporter: OtelExporter::None, + runtime_metrics: false, span_attributes: std::collections::HashMap::new(), tracestate: std::collections::BTreeMap::new(), }; diff --git a/codex-rs/otel/src/config.rs b/codex-rs/otel/src/config.rs index 761b913286..0828b6085e 100644 --- a/codex-rs/otel/src/config.rs +++ b/codex-rs/otel/src/config.rs @@ -1,6 +1,8 @@ use std::collections::BTreeMap; use std::collections::HashMap; +use std::fmt; use std::path::PathBuf; +use std::time::Duration; use codex_utils_absolute_path::AbsolutePathBuf; use serde::Deserialize; @@ -49,6 +51,8 @@ pub fn validate_span_attributes(attributes: &HashMap) -> std::io #[derive(Clone, Debug)] pub struct OtelSettings { + /// Optional credential helper shared by explicitly configured OTLP HTTP exporters. + pub headers_helper: Option, pub http_client_factory: codex_http_client::HttpClientFactory, pub environment: String, pub service_name: String, @@ -62,6 +66,28 @@ pub struct OtelSettings { pub tracestate: BTreeMap>, } +/// A shared command for explicit OTLP HTTP exporters; never used by built-in analytics. +/// The command runs from `OtelSettings::codex_home` and prints a JSON object of strings. +/// Its headers override static headers. A failed run suspends exports until recovery. +#[derive(Clone)] +pub struct OtelHeadersHelper { + pub command: String, + pub refresh_interval: Duration, + /// Exporters set by admin requirements ignore the headers helper. + pub apply_to_logs: bool, + pub apply_to_traces: bool, + pub apply_to_metrics: bool, +} + +impl fmt::Debug for OtelHeadersHelper { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("OtelHeadersHelper") + .field("command", &"[REDACTED]") + .field("refresh_interval", &self.refresh_interval) + .finish() + } +} + /// Resolved Statsig metrics settings that another process can use to recreate /// the built-in metrics exporter configuration without receiving generic /// exporter credentials in-process. diff --git a/codex-rs/otel/src/headers_helper.rs b/codex-rs/otel/src/headers_helper.rs new file mode 100644 index 0000000000..a2036bc0e7 --- /dev/null +++ b/codex-rs/otel/src/headers_helper.rs @@ -0,0 +1,156 @@ +//! Periodically refresh collector credentials without rebuilding OTel providers. +//! One worker serves all explicit HTTP signals. Failed refreshes replace the cache +//! with an error so exports cannot silently fall back to stale or static credentials. + +use crate::config::OtelHeadersHelper; +use codex_http_client::NetworkPolicy; +use http::HeaderMap; +use std::fmt; +use std::io; +use std::io::Write; +use std::path::Path; +use std::sync::Arc; +use std::sync::Mutex; +use tokio::sync::oneshot; +use tokio::sync::watch; + +type HeaderResult = Result; + +pub(crate) struct HeadersHelper { + headers: watch::Receiver>, + stop: Mutex>>, + pub(crate) runtime: tokio::runtime::Handle, +} + +impl fmt::Debug for HeadersHelper { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("HeadersHelper").finish_non_exhaustive() + } +} + +impl HeadersHelper { + pub(crate) fn start( + config: &OtelHeadersHelper, + cwd: &Path, + policy: NetworkPolicy, + ) -> io::Result> { + if config.command.trim().is_empty() || config.refresh_interval.is_zero() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "OTel headers helper requires a command and a positive refresh interval", + )); + } + // SDK exports can run on OS threads without a Tokio runtime. Keep helper + // execution and its HTTP clients on a runtime with the same lifetime. + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()?; + let handle = runtime.handle().clone(); + let (tx, rx) = watch::channel(None); + let (stop, stopped) = oneshot::channel(); + let config = config.clone(); + let cwd = cwd.to_path_buf(); + std::thread::Builder::new() + .name("otel-headers-helper".to_string()) + .spawn(move || { + runtime.block_on(async move { + tokio::select! { + _ = stopped => {} + _ = refresh_headers(&config, &cwd, &policy, tx) => {} + } + }); + })?; + Ok(Arc::new(Self { + headers: rx, + stop: Mutex::new(Some(stop)), + runtime: handle, + })) + } + + pub(crate) async fn headers(&self) -> HeaderResult { + let mut headers = self.headers.clone(); + let value = headers + .wait_for(Option::is_some) + .await + .map_err(|_| "OTel headers helper stopped")?; + value.clone().ok_or("OTel headers helper has no result")? + } + + pub(crate) fn shutdown(&self) { + if let Some(stop) = self + .stop + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .take() + { + let _ = stop.send(()); + } + } +} + +impl Drop for HeadersHelper { + fn drop(&mut self) { + self.shutdown(); + } +} + +async fn refresh_headers( + config: &OtelHeadersHelper, + cwd: &Path, + policy: &NetworkPolicy, + updates: watch::Sender>, +) { + let mut warned = false; + loop { + // The script can open its own sockets, so it needs the same revocable + // permit as an SDK that cannot enforce individual destinations. + let result = match policy.acquire_for_unsupported_sdk() { + Ok(permit) => permit + .run(run_helper(&config.command, cwd)) + .await + .unwrap_or_else(|error| Err(error.to_string())), + Err(error) => Err(error.to_string()), + }; + let error = result.as_ref().err().cloned(); + if updates.send(Some(result)).is_err() { + return; + } + if let Some(error) = error { + tracing::warn!( + error, + "OTel headers helper failed; telemetry is not being exported" + ); + if !warned { + let _ = writeln!( + io::stderr().lock(), + "WARNING: OTel headers helper failed; telemetry is not being exported: {error}" + ); + warned = true; + } + } + tokio::time::sleep(config.refresh_interval).await; + } +} + +async fn run_helper(command: &str, cwd: &Path) -> HeaderResult { + #[cfg(windows)] + let mut process = codex_http_headers_helper::shell_command(command); + #[cfg(not(windows))] + let mut process = { + // Executable paths containing spaces need no shell quoting. + let executable = cwd.join(command); + if executable.is_file() { + codex_utils_process::background_command(executable) + } else { + codex_http_headers_helper::shell_command(command) + } + }; + process.current_dir(cwd); + codex_http_headers_helper::run(process) + .await + .map_err(|err| err.to_string()) +} + +#[cfg(test)] +#[path = "headers_helper_tests.rs"] +mod tests; diff --git a/codex-rs/otel/src/headers_helper_tests.rs b/codex-rs/otel/src/headers_helper_tests.rs new file mode 100644 index 0000000000..e2a6d59b11 --- /dev/null +++ b/codex-rs/otel/src/headers_helper_tests.rs @@ -0,0 +1,257 @@ +#![cfg(test)] + +use super::*; +use pretty_assertions::assert_eq; +use std::collections::BTreeMap; +use std::time::Duration; + +fn read_command() -> &'static str { + if cfg!(windows) { + "type headers.json" + } else { + "cat headers.json" + } +} + +#[tokio::test] +async fn restricted_policy_skips_command_until_a_refresh_is_permitted() { + use codex_http_client::DestinationPolicy; + use codex_http_client::NetworkPolicyController; + + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("headers.json"), "{}").unwrap(); + let controller = NetworkPolicyController::default(); + let policy = controller.policy(); + let revision = policy.revision(); + controller.publish( + revision, + DestinationPolicy::Restricted { + allowed_hosts: Default::default(), + }, + ); + let helper = HeadersHelper::start( + &OtelHeadersHelper { + command: format!("echo invocation >> calls && {}", read_command()), + refresh_interval: Duration::from_millis(20), + apply_to_logs: true, + apply_to_traces: true, + apply_to_metrics: true, + }, + dir.path(), + policy, + ) + .unwrap(); + assert!(helper.headers().await.is_err()); + assert!(!dir.path().join("calls").exists()); + let mut updates = helper.headers.clone(); + controller.publish(revision, DestinationPolicy::Unrestricted); + tokio::time::timeout( + Duration::from_secs(5), + updates.wait_for(|v| matches!(v, Some(Ok(_)))), + ) + .await + .unwrap() + .unwrap(); + assert!(dir.path().join("calls").exists()); + helper.shutdown(); +} + +#[tokio::test] +async fn concurrent_exports_share_one_helper_execution() { + let dir = tempfile::tempdir().unwrap(); + std::fs::write( + dir.path().join("headers.json"), + r#"{"Authorization":"first"}"#, + ) + .unwrap(); + let helper = HeadersHelper::start( + &OtelHeadersHelper { + command: format!("echo invocation >> calls && {}", read_command()), + refresh_interval: Duration::from_secs(3600), + apply_to_logs: true, + apply_to_traces: true, + apply_to_metrics: true, + }, + dir.path(), + NetworkPolicy::unmanaged(), + ) + .unwrap(); + let mut tasks = tokio::task::JoinSet::new(); + for _ in 0..20 { + let helper = helper.clone(); + tasks.spawn(async move { helper.headers().await.unwrap() }); + } + while let Some(result) = tasks.join_next().await { + assert_eq!(result.unwrap()["authorization"], "first"); + } + assert_eq!( + std::fs::read_to_string(dir.path().join("calls")) + .unwrap() + .lines() + .count(), + 1 + ); + helper.shutdown(); +} + +#[tokio::test] +async fn periodic_refresh_replaces_credentials_and_recovers_after_failure() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("headers.json"); + std::fs::write(&path, r#"{"Authorization":"first","X-Old":"old"}"#).unwrap(); + let helper = HeadersHelper::start( + &OtelHeadersHelper { + command: read_command().to_string(), + refresh_interval: Duration::from_millis(20), + apply_to_logs: true, + apply_to_traces: true, + apply_to_metrics: true, + }, + dir.path(), + NetworkPolicy::unmanaged(), + ) + .unwrap(); + assert_eq!(helper.headers().await.unwrap()["authorization"], "first"); + let mut updates = helper.headers.clone(); + std::fs::write(&path, "invalid output containing a secret").unwrap(); + tokio::time::timeout( + Duration::from_secs(5), + updates.wait_for(|value| matches!(value, Some(Err(_)))), + ) + .await + .unwrap() + .unwrap(); + assert!(helper.headers().await.is_err()); + std::fs::write(&path, r#"{"Authorization":"second"}"#).unwrap(); + tokio::time::timeout( + Duration::from_secs(5), + updates.wait_for( + |value| matches!(value, Some(Ok(headers)) if headers["authorization"] == "second"), + ), + ) + .await + .unwrap() + .unwrap(); + let headers = helper.headers().await.unwrap(); + assert_eq!(headers.len(), 1); + assert_eq!(headers["authorization"], "second"); + helper.shutdown(); +} + +#[tokio::test] +async fn helper_failure_is_an_export_error_and_output_is_bounded() { + let dir = tempfile::tempdir().unwrap(); + assert!(run_helper("exit 1", dir.path()).await.is_err()); + std::fs::write(dir.path().join("headers.json"), vec![b'x'; 64 * 1024 + 1]).unwrap(); + assert_eq!( + run_helper(read_command(), dir.path()).await, + Err("headers helper output exceeds 64 KiB".to_string()) + ); +} + +#[tokio::test] +async fn failed_helper_prevents_http_export_even_with_static_credentials() { + use opentelemetry_http::HttpClient as _; + let dir = tempfile::tempdir().unwrap(); + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + let helper = HeadersHelper::start( + &OtelHeadersHelper { + command: "exit 1".to_string(), + refresh_interval: Duration::from_secs(3600), + apply_to_logs: true, + apply_to_traces: true, + apply_to_metrics: true, + }, + dir.path(), + NetworkPolicy::unmanaged(), + ) + .unwrap(); + let factory = codex_http_client::HttpClientFactory::new( + codex_http_client::OutboundProxyPolicy::ReqwestDefault, + ); + let client = crate::otlp::build_async_http_client( + &factory, + /*tls*/ None, + opentelemetry_otlp::OTEL_EXPORTER_OTLP_LOGS_TIMEOUT, + Some(helper.clone()), + ) + .unwrap(); + let request = http::Request::post(format!("http://{}/v1/logs", listener.local_addr().unwrap())) + .header("authorization", "Bearer static-credential") + .body(bytes::Bytes::new()) + .unwrap(); + assert!(client.send_bytes(request).await.is_err()); + assert_eq!( + listener.accept().unwrap_err().kind(), + std::io::ErrorKind::WouldBlock + ); + helper.shutdown(); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn grpc_and_builtin_metrics_do_not_start_the_helper() { + use crate::config::OtelExporter; + let helper = OtelHeadersHelper { + command: String::new(), + refresh_interval: Duration::ZERO, + apply_to_logs: true, + apply_to_traces: true, + apply_to_metrics: true, + }; + for exporter in [ + OtelExporter::None, + OtelExporter::Statsig, + OtelExporter::OtlpGrpc { + endpoint: "http://127.0.0.1:4317".to_string(), + headers: Default::default(), + tls: None, + }, + ] { + let settings = crate::config::OtelSettings { + headers_helper: Some(helper.clone()), + http_client_factory: codex_http_client::HttpClientFactory::new( + codex_http_client::OutboundProxyPolicy::ReqwestDefault, + ), + environment: "test".to_string(), + service_name: "test".to_string(), + service_version: "test".to_string(), + codex_home: std::path::PathBuf::from("."), + exporter: OtelExporter::None, + trace_exporter: OtelExporter::None, + metrics_exporter: exporter, + runtime_metrics: false, + span_attributes: Default::default(), + tracestate: BTreeMap::new(), + }; + // Invalid helper settings would fail provider construction if the helper started. + let provider = crate::OtelProvider::try_new(&settings).unwrap(); + if let Some(provider) = provider { + provider.shutdown(); + } + } +} + +#[cfg(unix)] +#[tokio::test] +async fn executable_paths_with_spaces_and_timeouts_are_supported() { + use std::os::unix::fs::PermissionsExt; + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("helper with spaces"); + std::fs::write( + &path, + "#!/bin/sh\nprintf '%s' '{\"Authorization\":\"token\"}'", + ) + .unwrap(); + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o700)).unwrap(); + assert_eq!( + run_helper(path.to_str().unwrap(), dir.path()) + .await + .unwrap()["authorization"], + "token" + ); + assert_eq!( + run_helper("sleep 60", dir.path()).await, + Err("headers helper timed out after 10 seconds".to_string()) + ); +} diff --git a/codex-rs/otel/src/lib.rs b/codex-rs/otel/src/lib.rs index 91e060ace4..ecbc444355 100644 --- a/codex-rs/otel/src/lib.rs +++ b/codex-rs/otel/src/lib.rs @@ -7,6 +7,7 @@ pub(crate) mod trace_context; mod agent_response; mod guardian_assessment; +mod headers_helper; mod network_policy; mod otlp; mod skill_invocation; @@ -21,6 +22,7 @@ use strum_macros::Display; pub use crate::agent_response::AgentResponseContext; pub use crate::agent_response::AgentResponseLogger; pub use crate::config::OtelExporter; +pub use crate::config::OtelHeadersHelper; pub use crate::config::OtelHttpProtocol; pub use crate::config::OtelSettings; pub use crate::config::OtelTlsConfig; diff --git a/codex-rs/otel/src/metrics/client.rs b/codex-rs/otel/src/metrics/client.rs index 0dee169517..e5fc11ab79 100644 --- a/codex-rs/otel/src/metrics/client.rs +++ b/codex-rs/otel/src/metrics/client.rs @@ -1,5 +1,6 @@ use crate::config::OtelExporter; use crate::config::OtelHttpProtocol; +use crate::headers_helper::HeadersHelper; use crate::metrics::MetricsError; use crate::metrics::Result; use crate::metrics::config::MetricsConfig; @@ -345,6 +346,7 @@ impl MetricsClient { /// Build a metrics client from configuration and validate defaults. pub fn new(config: MetricsConfig) -> Result { let MetricsConfig { + headers_helper, http_client_factory, environment, service_name, @@ -397,6 +399,7 @@ impl MetricsClient { exporter, temporality, &http_client_factory, + headers_helper, )?, policy: http_client_factory.network_policy().clone(), }; @@ -657,6 +660,7 @@ fn build_otlp_metric_exporter( exporter: OtelExporter, temporality: Temporality, factory: &codex_http_client::HttpClientFactory, + headers_helper: Option>, ) -> Result { match exporter { OtelExporter::None => Err(MetricsError::ExporterDisabled), @@ -664,6 +668,7 @@ fn build_otlp_metric_exporter( crate::config::resolve_exporter(&OtelExporter::Statsig), temporality, factory, + /*headers_helper*/ None, ), OtelExporter::OtlpGrpc { endpoint, @@ -715,11 +720,12 @@ fn build_otlp_metric_exporter( .with_protocol(protocol) .with_headers(headers); - if factory.network_policy().is_managed() { + if headers_helper.is_some() || factory.network_policy().is_managed() { let client = crate::otlp::build_async_http_client( factory, tls.as_ref(), OTEL_EXPORTER_OTLP_METRICS_TIMEOUT, + headers_helper, ) .map_err(|err| MetricsError::InvalidConfig { message: err.to_string(), diff --git a/codex-rs/otel/src/metrics/config.rs b/codex-rs/otel/src/metrics/config.rs index df2203a213..933092da00 100644 --- a/codex-rs/otel/src/metrics/config.rs +++ b/codex-rs/otel/src/metrics/config.rs @@ -1,4 +1,5 @@ use crate::config::OtelExporter; +use crate::headers_helper::HeadersHelper; use crate::metrics::Result; use crate::metrics::names::API_CALL_COUNT_METRIC; use crate::metrics::names::API_CALL_DURATION_METRIC; @@ -14,6 +15,7 @@ use crate::metrics::validation::validate_tag_key; use crate::metrics::validation::validate_tag_value; use opentelemetry_sdk::metrics::InMemoryMetricExporter; use std::collections::BTreeMap; +use std::sync::Arc; use std::time::Duration; const CONVERSATION_TURN_COUNT_METRIC: &str = "codex.conversation.turn.count"; @@ -43,6 +45,7 @@ pub enum MetricsExporter { #[derive(Clone, Debug)] pub struct MetricsConfig { + pub(crate) headers_helper: Option>, pub(crate) http_client_factory: codex_http_client::HttpClientFactory, pub(crate) environment: String, pub(crate) service_name: String, @@ -75,6 +78,7 @@ impl MetricsConfig { &[] }; Self { + headers_helper: None, http_client_factory: codex_http_client::HttpClientFactory::new( codex_http_client::OutboundProxyPolicy::ReqwestDefault, ), @@ -97,6 +101,7 @@ impl MetricsConfig { exporter: InMemoryMetricExporter, ) -> Self { Self { + headers_helper: None, http_client_factory: codex_http_client::HttpClientFactory::new( codex_http_client::OutboundProxyPolicy::ReqwestDefault, ), diff --git a/codex-rs/otel/src/network_policy.rs b/codex-rs/otel/src/network_policy.rs index 9b9455a1a5..2311dd975d 100644 --- a/codex-rs/otel/src/network_policy.rs +++ b/codex-rs/otel/src/network_policy.rs @@ -1,5 +1,7 @@ //! OTLP SDK transports remain disabled under application destination restrictions. //! Export futures are revocable, including HTTP exporters driven from SDK worker threads. +use crate::headers_helper::HeadersHelper; +use std::sync::Arc; use bytes::Bytes; use codex_http_client::NetworkPolicy; @@ -79,6 +81,7 @@ impl PushMetricExporter for PolicyExporter { /// Keeps blocking SDK worker threads from performing uncancellable HTTP I/O. #[derive(Debug)] pub(crate) struct RuntimeHttpClient { + pub(crate) headers_helper: Option>, pub(crate) client: codex_http_client::HttpClient, pub(crate) timeout: Duration, pub(crate) runtime: tokio::runtime::Handle, @@ -96,10 +99,16 @@ impl opentelemetry_http::HttpClient for RuntimeHttpClient { Box::pin(async move { let client = self.client.clone(); let timeout = self.timeout; + let headers_helper = self.headers_helper.clone(); let mut requests = tokio::task::JoinSet::new(); requests.spawn_on( async move { - let (parts, body) = request.into_parts(); + let (mut parts, body) = request.into_parts(); + if let Some(helper) = &headers_helper { + // HeaderMap replaces names case-insensitively. Each export starts + // from the static headers, so removed helper keys do not linger. + parts.headers.extend(helper.headers().await?); + } let response = client .request(parts.method, parts.uri.to_string()) .version(parts.version) diff --git a/codex-rs/otel/src/otlp.rs b/codex-rs/otel/src/otlp.rs index a20ad5d65b..53ba300f23 100644 --- a/codex-rs/otel/src/otlp.rs +++ b/codex-rs/otel/src/otlp.rs @@ -1,4 +1,5 @@ use crate::config::OtelTlsConfig; +use crate::headers_helper::HeadersHelper; use codex_utils_absolute_path::AbsolutePathBuf; use http::Uri; use opentelemetry_otlp::OTEL_EXPORTER_OTLP_TIMEOUT; @@ -17,6 +18,7 @@ use std::fs; use std::io; use std::io::ErrorKind; use std::path::PathBuf; +use std::sync::Arc; use std::time::Duration; pub(crate) fn build_header_map(headers: &std::collections::HashMap) -> HeaderMap { @@ -149,6 +151,7 @@ pub(crate) fn build_async_http_client( factory: &codex_http_client::HttpClientFactory, tls: Option<&OtelTlsConfig>, timeout_var: &str, + headers_helper: Option>, ) -> Result> { let mut config = codex_http_client::HttpClientTlsConfig::default(); @@ -189,12 +192,19 @@ pub(crate) fn build_async_http_client( } } + let mut builder = codex_http_client::HttpClientBuilder::new().without_request_logging(); + if headers_helper.is_some() { + // Custom credential headers must never follow a collector redirect. + builder = builder.without_redirects(); + } Ok(crate::network_policy::RuntimeHttpClient { - client: codex_http_client::HttpClientBuilder::new() - .without_request_logging() - .build_with_tls(factory, codex_http_client::ClientRouteClass::Other, config), + client: builder.build_with_tls(factory, codex_http_client::ClientRouteClass::Other, config), timeout: resolve_otlp_timeout(timeout_var), - runtime: tokio::runtime::Handle::try_current()?, + runtime: match &headers_helper { + Some(helper) => helper.runtime.clone(), + None => tokio::runtime::Handle::try_current()?, + }, + headers_helper, }) } @@ -294,8 +304,13 @@ mod tests { let (_temp_dir, tls) = write_test_mtls_config(); let factory = HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault); - build_async_http_client(&factory, Some(&tls), OTEL_EXPORTER_OTLP_TIMEOUT) - .expect("async HTTP client should accept PEM client identity"); + build_async_http_client( + &factory, + Some(&tls), + OTEL_EXPORTER_OTLP_TIMEOUT, + /*headers_helper*/ None, + ) + .expect("async HTTP client should accept PEM client identity"); } fn write_test_mtls_config() -> (TempDir, OtelTlsConfig) { diff --git a/codex-rs/otel/src/provider.rs b/codex-rs/otel/src/provider.rs index d04d107a0e..1b4ecffdc7 100644 --- a/codex-rs/otel/src/provider.rs +++ b/codex-rs/otel/src/provider.rs @@ -2,6 +2,7 @@ use crate::config::OtelExporter; use crate::config::OtelHttpProtocol; use crate::config::OtelSettings; use crate::config::StatsigMetricsSettings; +use crate::headers_helper::HeadersHelper; use crate::metrics::MetricsClient; use crate::metrics::MetricsConfig; use crate::targets::is_log_export_target; @@ -44,6 +45,7 @@ use std::collections::HashMap; use std::error::Error; use std::io; use std::mem::ManuallyDrop; +use std::sync::Arc; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; use std::sync::mpsc; @@ -66,6 +68,7 @@ pub struct OtelProvider { pub tracer_provider: Option, pub tracer: Option, pub metrics: Option, + headers_helper: Option>, shutdown_started: AtomicBool, shutdown_worker: Option>, } @@ -109,6 +112,9 @@ impl OtelProvider { if let Some(logger) = &self.logger { let _ = logger.shutdown(); } + if let Some(helper) = &self.headers_helper { + helper.shutdown(); + } } /// Starts the detached shutdown worker before shutdown-time resource pressure. @@ -218,6 +224,33 @@ impl OtelProvider { } crate::trace_context::validate_tracestate_entries(&settings.tracestate)?; + let (helper_for_logs, helper_for_traces, helper_for_metrics) = settings + .headers_helper + .as_ref() + .map(|helper| { + ( + helper.apply_to_logs + && matches!(settings.exporter, OtelExporter::OtlpHttp { .. }), + helper.apply_to_traces + && matches!(settings.trace_exporter, OtelExporter::OtlpHttp { .. }), + helper.apply_to_metrics + && matches!(settings.metrics_exporter, OtelExporter::OtlpHttp { .. }), + ) + }) + .unwrap_or_default(); + let headers_helper = settings + .headers_helper + .as_ref() + .filter(|_| helper_for_logs || helper_for_traces || helper_for_metrics) + .map(|config| { + HeadersHelper::start( + config, + &settings.codex_home, + settings.http_client_factory.network_policy().clone(), + ) + }) + .transpose()?; + let metrics = if matches!(metric_exporter, OtelExporter::None) { None } else { @@ -228,6 +261,9 @@ impl OtelProvider { settings.metrics_exporter.clone(), ) .with_http_client_factory(settings.http_client_factory.clone()); + if helper_for_metrics { + config.headers_helper = headers_helper.clone(); + } if settings.runtime_metrics { config = config.with_runtime_reader(); } @@ -242,6 +278,7 @@ impl OtelProvider { &log_resource, &settings.exporter, &settings.http_client_factory, + headers_helper.clone().filter(|_| helper_for_logs), ) }) .transpose()?; @@ -253,6 +290,7 @@ impl OtelProvider { &settings.trace_exporter, settings.span_attributes.clone(), &settings.http_client_factory, + headers_helper.clone().filter(|_| helper_for_traces), ) }) .transpose()?; @@ -266,6 +304,7 @@ impl OtelProvider { tracer_provider, tracer, metrics, + headers_helper, shutdown_started: AtomicBool::default(), shutdown_worker: None, }; @@ -455,7 +494,10 @@ fn build_logger( resource: &Resource, exporter: &OtelExporter, factory: &codex_http_client::HttpClientFactory, + headers_helper: Option>, ) -> Result> { + let headers_helper = + headers_helper.filter(|_| matches!(exporter, OtelExporter::OtlpHttp { .. })); let mut builder = SdkLoggerProvider::builder().with_resource(resource.clone()); match crate::config::resolve_exporter(exporter) { @@ -510,11 +552,12 @@ fn build_logger( .with_protocol(protocol) .with_headers(headers); - if factory.network_policy().is_managed() { + if headers_helper.is_some() || factory.network_policy().is_managed() { let client = crate::otlp::build_async_http_client( factory, tls.as_ref(), OTEL_EXPORTER_OTLP_LOGS_TIMEOUT, + headers_helper, )?; exporter_builder = exporter_builder.with_http_client(client); } else if let Some(tls) = tls.as_ref() { @@ -539,7 +582,10 @@ fn build_tracer_provider( exporter: &OtelExporter, span_attributes: HashMap, factory: &codex_http_client::HttpClientFactory, + headers_helper: Option>, ) -> Result> { + let headers_helper = + headers_helper.filter(|_| matches!(exporter, OtelExporter::OtlpHttp { .. })); let span_exporter = match crate::config::resolve_exporter(exporter) { OtelExporter::None => return Ok(tracer_provider_builder(resource, span_attributes).build()), OtelExporter::Statsig => unreachable!("statsig exporter should be resolved"), @@ -592,6 +638,7 @@ fn build_tracer_provider( factory, tls.as_ref(), OTEL_EXPORTER_OTLP_TRACES_TIMEOUT, + headers_helper, )?; exporter_builder = exporter_builder.with_http_client(client); @@ -620,11 +667,12 @@ fn build_tracer_provider( .with_protocol(protocol) .with_headers(headers); - if factory.network_policy().is_managed() { + if headers_helper.is_some() || factory.network_policy().is_managed() { let client = crate::otlp::build_async_http_client( factory, tls.as_ref(), OTEL_EXPORTER_OTLP_TRACES_TIMEOUT, + headers_helper, )?; exporter_builder = exporter_builder.with_http_client(client); } else if let Some(tls) = tls.as_ref() { @@ -868,6 +916,7 @@ mod tests { fn test_otel_settings() -> OtelSettings { OtelSettings { + headers_helper: None, http_client_factory: HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), environment: "test".to_string(), service_name: "codex-test".to_string(), diff --git a/codex-rs/otel/src/provider_shutdown_tests.rs b/codex-rs/otel/src/provider_shutdown_tests.rs index 768aa0f473..1187b127f7 100644 --- a/codex-rs/otel/src/provider_shutdown_tests.rs +++ b/codex-rs/otel/src/provider_shutdown_tests.rs @@ -155,6 +155,7 @@ fn test_provider(behavior: ShutdownBehavior) -> TestProvider { TestProvider { provider: OtelProvider { + headers_helper: None, logger: None, tracer_provider: Some(tracer_provider), tracer: None, diff --git a/codex-rs/otel/tests/suite/buffered_operations_tests.rs b/codex-rs/otel/tests/suite/buffered_operations_tests.rs index 63b3dd8eda..d9dcf9ce3b 100644 --- a/codex-rs/otel/tests/suite/buffered_operations_tests.rs +++ b/codex-rs/otel/tests/suite/buffered_operations_tests.rs @@ -81,6 +81,7 @@ fn shutdown_buffers_until_replacement_and_preserves_newer_installations() { fn rejected_opt_out_preserves_recording_and_accepted_opt_out_survives_shutdown() { let metrics = install(); let mut settings = OtelSettings { + headers_helper: None, http_client_factory: HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), environment: "test".to_string(), service_name: "test".to_string(), diff --git a/codex-rs/otel/tests/suite/otel_headers_helper.rs b/codex-rs/otel/tests/suite/otel_headers_helper.rs new file mode 100644 index 0000000000..625b0ff224 --- /dev/null +++ b/codex-rs/otel/tests/suite/otel_headers_helper.rs @@ -0,0 +1,125 @@ +//! Collector-level coverage for dynamic headers on every HTTP signal and encoding. + +use super::*; +use codex_otel::OtelHeadersHelper; +use pretty_assertions::assert_eq; + +#[test] +fn http_headers_helper_authenticates_all_signals() { + for protocol in [OtelHttpProtocol::Json, OtelHttpProtocol::Binary] { + check_all_signals(protocol.clone(), /*managed_logs*/ false); + tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + .expect("create test runtime") + .block_on(async { + check_all_signals(protocol, /*managed_logs*/ true) + }); + } +} + +fn check_all_signals(protocol: OtelHttpProtocol, managed_logs: bool) { + let dir = tempfile::tempdir().expect("create helper directory"); + std::fs::write( + dir.path().join("headers.json"), + r#"{"Authorization":"Bearer refreshed","X-Helper":"present"}"#, + ) + .expect("write helper headers"); + let listener = TcpListener::bind("127.0.0.1:0").expect("bind collector"); + let address = listener.local_addr().expect("collector address"); + listener + .set_nonblocking(true) + .expect("nonblocking collector"); + let collector = thread::spawn(move || { + let deadline = Instant::now() + Duration::from_secs(10); + let mut received = BTreeMap::new(); + while Instant::now() < deadline && received.len() < 3 { + match listener.accept() { + Ok((mut stream, _)) => { + let (path, headers, _) = read_http_request(&mut stream).expect("read export"); + write_http_response(&mut stream, "200 OK").expect("acknowledge export"); + received.insert(path, headers); + } + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + thread::sleep(Duration::from_millis(5)); + } + Err(error) => panic!("collector failed: {error}"), + } + } + received + }); + let exporter = |signal| OtelExporter::OtlpHttp { + endpoint: format!("http://{address}/v1/{signal}"), + headers: HashMap::from([ + ("authorization".to_string(), "Bearer stale".to_string()), + ("X-Static".to_string(), "retained".to_string()), + ]), + protocol: protocol.clone(), + tls: None, + }; + let command = if cfg!(windows) { + "type headers.json" + } else { + "cat headers.json" + }; + let helper = OtelHeadersHelper { + command: format!("echo invocation >> calls && {command}"), + refresh_interval: Duration::from_secs(3600), + apply_to_logs: !managed_logs, + apply_to_traces: true, + apply_to_metrics: true, + }; + let provider = OtelProvider::try_new(&OtelSettings { + headers_helper: Some(helper), + http_client_factory: HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + environment: "test".to_string(), + service_name: "otel-helper-test".to_string(), + service_version: "test".to_string(), + codex_home: dir.path().to_path_buf(), + exporter: exporter("logs"), + trace_exporter: exporter("traces"), + metrics_exporter: exporter("metrics"), + runtime_metrics: false, + span_attributes: HashMap::new(), + tracestate: BTreeMap::new(), + }) + .expect("initialize exporters") + .expect("exporters are enabled"); + let subscriber = tracing_subscriber::registry() + .with(provider.logger_layer().expect("logs enabled")) + .with(provider.tracing_layer().expect("traces enabled")); + tracing::subscriber::with_default(subscriber, || { + let span = tracing::info_span!("helper-test"); + let _entered = span.enter(); + tracing::event!(target: "codex_otel.log_only", tracing::Level::INFO, event.name = "codex.test.helper", "test"); + }); + provider + .metrics() + .expect("metrics enabled") + .counter("codex.test.helper", /*inc*/ 1, &[]) + .expect("record counter"); + provider.shutdown(); + let received = collector.join().expect("collector completed"); + assert_eq!( + received.keys().map(String::as_str).collect::>(), + ["/v1/logs", "/v1/metrics", "/v1/traces"] + ); + for (path, headers) in &received { + if managed_logs && path == "/v1/logs" { + assert_eq!(headers["authorization"], "Bearer stale"); + assert!(!headers.contains_key("x-helper")); + } else { + assert_eq!(headers["authorization"], "Bearer refreshed"); + assert_eq!(headers["x-helper"], "present"); + } + assert_eq!(headers["x-static"], "retained"); + } + assert_eq!( + std::fs::read_to_string(dir.path().join("calls")) + .expect("read helper invocation count") + .lines() + .count(), + 1 + ); +} diff --git a/codex-rs/otel/tests/suite/otlp_http_loopback.rs b/codex-rs/otel/tests/suite/otlp_http_loopback.rs index 44b73bd898..2600201780 100644 --- a/codex-rs/otel/tests/suite/otlp_http_loopback.rs +++ b/codex-rs/otel/tests/suite/otlp_http_loopback.rs @@ -28,6 +28,9 @@ use tracing_subscriber::layer::SubscriberExt; static TRACE_CONTEXT_CONFIG_LOCK: Mutex<()> = Mutex::new(()); +#[path = "otel_headers_helper.rs"] +mod otel_headers_helper; + struct CapturedRequest { path: String, content_type: Option, @@ -397,6 +400,7 @@ fn otlp_http_exporter_sends_logs_to_collector() }); let otel = OtelProvider::try_new(&OtelSettings { + headers_helper: None, http_client_factory: HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), environment: "test".to_string(), service_name: "codex-cli".to_string(), @@ -457,6 +461,7 @@ fn otlp_http_exporter_sends_logs_to_collector() #[test] fn otel_provider_rejects_header_unsafe_configured_tracestate() { let result = OtelProvider::try_new(&OtelSettings { + headers_helper: None, http_client_factory: HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), environment: "test".to_string(), service_name: "codex-cli".to_string(), @@ -523,6 +528,7 @@ fn otlp_http_exporter_sends_traces_to_collector() }); let otel = OtelProvider::try_new(&OtelSettings { + headers_helper: None, http_client_factory: HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), environment: "test".to_string(), service_name: "codex-cli".to_string(), @@ -641,6 +647,7 @@ async fn otlp_http_exporter_sends_traces_to_collector_with_bounded_shutdown_in_t listener.set_nonblocking(true).expect("set_nonblocking"); let otel = OtelProvider::try_new(&OtelSettings { + headers_helper: None, http_client_factory: HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), environment: "test".to_string(), service_name: "codex-cli".to_string(), @@ -770,6 +777,7 @@ fn otlp_http_exporter_times_out_when_collector_stalls_during_bounded_shutdown() .expect("build tokio runtime"); let (result, elapsed) = runtime.block_on(async move { let otel = OtelProvider::try_new(&OtelSettings { + headers_helper: None, http_client_factory: HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), environment: "test".to_string(), service_name: "codex-cli".to_string(), @@ -870,6 +878,7 @@ fn otlp_http_exporter_sends_traces_to_collector_in_current_thread_tokio_runtime( let result = runtime.block_on(async move { let otel = OtelProvider::try_new(&OtelSettings { + headers_helper: None, http_client_factory: HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), environment: "test".to_string(), service_name: "codex-cli".to_string(), diff --git a/codex-rs/windows-sandbox-rs/src/wfp_setup.rs b/codex-rs/windows-sandbox-rs/src/wfp_setup.rs index 3cdc7d75d1..790e966a3d 100644 --- a/codex-rs/windows-sandbox-rs/src/wfp_setup.rs +++ b/codex-rs/windows-sandbox-rs/src/wfp_setup.rs @@ -48,6 +48,7 @@ fn build_wfp_metrics_provider( // Statsig environment in the elevation payload. Other exporters are // intentionally omitted from this helper path. OtelProvider::try_new(&OtelSettings { + headers_helper: None, http_client_factory: codex_http_client::HttpClientFactory::new( codex_http_client::OutboundProxyPolicy::ReqwestDefault, ),