From 89d4ac8883cb7ddebdc331ef272ec490e4d1f95a Mon Sep 17 00:00:00 2001 From: celia-oai Date: Mon, 17 Aug 2026 16:09:12 -0700 Subject: [PATCH] Add proxy-aware async OpenTelemetry HTTP transport --- codex-rs/Cargo.lock | 1 + codex-rs/Cargo.toml | 1 + codex-rs/http-client/Cargo.toml | 1 + codex-rs/http-client/src/lib.rs | 4 + codex-rs/http-client/src/telemetry_client.rs | 146 ++++++++++++++++++ .../http-client/src/telemetry_client_tests.rs | 140 +++++++++++++++++ 6 files changed, 293 insertions(+) create mode 100644 codex-rs/http-client/src/telemetry_client.rs create mode 100644 codex-rs/http-client/src/telemetry_client_tests.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index f90ba32e77..a27206300e 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3381,6 +3381,7 @@ dependencies = [ "http 1.4.0", "native-tls", "opentelemetry", + "opentelemetry-http", "opentelemetry_sdk", "pretty_assertions", "rcgen", diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index d796f62e05..57731319ea 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -371,6 +371,7 @@ once_cell = "1.20.2" openssl-sys = "*" opentelemetry = "0.31.0" opentelemetry-appender-tracing = "0.31.0" +opentelemetry-http = "0.31.0" opentelemetry-otlp = "0.31.0" opentelemetry-semantic-conventions = "0.31.0" opentelemetry_sdk = "0.31.0" diff --git a/codex-rs/http-client/Cargo.toml b/codex-rs/http-client/Cargo.toml index a77a1fecec..77cf8168a3 100644 --- a/codex-rs/http-client/Cargo.toml +++ b/codex-rs/http-client/Cargo.toml @@ -11,6 +11,7 @@ futures = { workspace = true } http = { workspace = true } native-tls = "0.2" opentelemetry = { workspace = true } +opentelemetry-http = { workspace = true, features = ["reqwest"] } reqwest = { workspace = true, features = ["blocking", "json", "rustls-tls-native-roots", "stream"] } rustls = { workspace = true } rustls-native-certs = { workspace = true } diff --git a/codex-rs/http-client/src/lib.rs b/codex-rs/http-client/src/lib.rs index cb7f1209e2..546a36d62b 100644 --- a/codex-rs/http-client/src/lib.rs +++ b/codex-rs/http-client/src/lib.rs @@ -8,6 +8,7 @@ mod outbound_proxy; mod request; mod route_aware_client_pool; mod route_aware_redirect; +mod telemetry_client; mod tls_backend_fallback; mod transport; @@ -53,7 +54,10 @@ pub use crate::route_aware_client_pool::RouteAwareClientPool; pub use crate::route_aware_client_pool::RouteAwareClientPoolError; pub use crate::route_aware_client_pool::RouteAwareRequestBuilder; pub use crate::route_aware_client_pool::RouteAwareRequestError; +pub use crate::telemetry_client::BuildTelemetryHttpClientError; +pub use crate::telemetry_client::TelemetryClientTlsConfig; pub use crate::transport::ByteStream; pub use crate::transport::HttpTransport; pub use crate::transport::ReqwestTransport; pub use crate::transport::StreamResponse; +pub use opentelemetry_http::HttpClient as TelemetryHttpClient; diff --git a/codex-rs/http-client/src/telemetry_client.rs b/codex-rs/http-client/src/telemetry_client.rs new file mode 100644 index 0000000000..7dedf620b3 --- /dev/null +++ b/codex-rs/http-client/src/telemetry_client.rs @@ -0,0 +1,146 @@ +//! Policy-aware HTTP transports for OpenTelemetry exporters. + +use crate::BuildRouteAwareHttpClientError; +use crate::ClientRouteClass; +use crate::HttpClientFactory; +use crate::OutboundProxyPolicy; +use crate::custom_ca::BuildCustomCaTransportError; +use opentelemetry_http::HttpClient; +use std::fs; +use std::io; +use std::path::Path; +use std::path::PathBuf; +use std::time::Duration; +use thiserror::Error; + +/// Optional collector-specific roots and client identity for telemetry export. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct TelemetryClientTlsConfig { + pub ca_certificate: Option, + pub client_certificate: Option, + pub client_private_key: Option, +} + +/// Failure while preparing a proxy-aware telemetry exporter transport. +#[derive(Debug, Error)] +pub enum BuildTelemetryHttpClientError { + #[error("failed to read {}: {source}", path.display())] + ReadTlsFile { + path: PathBuf, + #[source] + source: io::Error, + }, + + #[error("failed to parse certificate {}: {source}", path.display())] + InvalidCertificate { + path: PathBuf, + #[source] + source: reqwest::Error, + }, + + #[error( + "failed to parse client identity using {} and {}: {source}", + certificate_path.display(), + private_key_path.display() + )] + InvalidClientIdentity { + certificate_path: PathBuf, + private_key_path: PathBuf, + #[source] + source: reqwest::Error, + }, + + #[error("client_certificate and client_private_key must both be provided for mTLS")] + IncompleteClientIdentity, + + #[error(transparent)] + Route(#[from] BuildRouteAwareHttpClientError), + + #[error(transparent)] + CustomCa(#[from] BuildCustomCaTransportError), +} + +struct PreparedTlsConfig { + root_certificate: Option, + client_identity: Option, +} + +impl HttpClientFactory { + /// Builds an asynchronous exporter client for one fixed collector endpoint. + pub fn build_async_telemetry_client( + &self, + endpoint: &str, + timeout: Duration, + tls: &TelemetryClientTlsConfig, + ) -> Result, BuildTelemetryHttpClientError> { + let tls = prepare_tls_config(tls)?; + let mut builder = reqwest::Client::builder().timeout(timeout); + if let Some(certificate) = tls.root_certificate { + builder = builder + .tls_built_in_root_certs(false) + .add_root_certificate(certificate); + } + if let Some(identity) = tls.client_identity { + builder = builder.identity(identity).https_only(true); + } + if self.outbound_proxy_policy() == OutboundProxyPolicy::RespectSystemProxy { + builder = builder.redirect(reqwest::redirect::Policy::none()); + } + + self.build_reqwest_client(builder, endpoint, ClientRouteClass::Other) + .map_err(Into::into) + } + +} + +fn prepare_tls_config( + tls: &TelemetryClientTlsConfig, +) -> Result { + let root_certificate = tls + .ca_certificate + .as_ref() + .map(|path| { + let pem = read_tls_file(path)?; + reqwest::Certificate::from_pem(&pem).map_err(|source| { + BuildTelemetryHttpClientError::InvalidCertificate { + path: path.clone(), + source, + } + }) + }) + .transpose()?; + + let client_identity = match (&tls.client_certificate, &tls.client_private_key) { + (Some(certificate_path), Some(private_key_path)) => { + let mut pem = read_tls_file(certificate_path)?; + pem.extend_from_slice(&read_tls_file(private_key_path)?); + Some(reqwest::Identity::from_pem(&pem).map_err(|source| { + BuildTelemetryHttpClientError::InvalidClientIdentity { + certificate_path: certificate_path.clone(), + private_key_path: private_key_path.clone(), + source, + } + })?) + } + (Some(_), None) | (None, Some(_)) => { + return Err(BuildTelemetryHttpClientError::IncompleteClientIdentity); + } + (None, None) => None, + }; + + Ok(PreparedTlsConfig { + root_certificate, + client_identity, + }) +} + +fn read_tls_file(path: &Path) -> Result, BuildTelemetryHttpClientError> { + fs::read(path).map_err(|source| BuildTelemetryHttpClientError::ReadTlsFile { + path: path.to_path_buf(), + source, + }) +} + +#[cfg(test)] +#[path = "telemetry_client_tests.rs"] +mod tests; diff --git a/codex-rs/http-client/src/telemetry_client_tests.rs b/codex-rs/http-client/src/telemetry_client_tests.rs new file mode 100644 index 0000000000..eb501a560a --- /dev/null +++ b/codex-rs/http-client/src/telemetry_client_tests.rs @@ -0,0 +1,140 @@ +use super::BuildTelemetryHttpClientError; +use super::TelemetryClientTlsConfig; +use crate::HttpClientFactory; +use crate::OutboundProxyPolicy; +use crate::cache_system_proxy_route_for_test; +use bytes::Bytes; +use http::Request; +use opentelemetry_http::HttpClient as _; +use pretty_assertions::assert_eq; +use std::io::Read; +use std::io::Write; +use std::net::SocketAddr; +use std::net::TcpListener; +use std::path::PathBuf; +use std::thread; +use std::time::Duration; + +#[tokio::test] +async fn async_telemetry_client_uses_resolved_system_proxy() { + let (proxy_address, proxy) = spawn_proxy(); + let endpoint = "http://async-telemetry-proxy.test/v1/traces"; + cache_system_proxy_route_for_test(endpoint, format!("http://{proxy_address}")); + let client = HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy) + .build_async_telemetry_client( + endpoint, + Duration::from_secs(2), + &TelemetryClientTlsConfig::default(), + ) + .expect("async telemetry client should build"); + + let response = client + .send_bytes(telemetry_request(endpoint)) + .await + .expect("telemetry request should use proxy"); + let request = proxy.join().expect("proxy should complete"); + + assert_eq!(response.status(), http::StatusCode::OK); + assert_eq!( + request.lines().next(), + Some("POST http://async-telemetry-proxy.test/v1/traces HTTP/1.1") + ); +} + +#[tokio::test] +async fn async_system_proxy_telemetry_client_does_not_follow_redirects() { + let (proxy_address, proxy) = spawn_proxy_with_response( + "HTTP/1.1 307 Temporary Redirect\r\nLocation: http://different-route.test/v1/traces\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + ); + let endpoint = "http://async-telemetry-redirect.test/v1/traces"; + cache_system_proxy_route_for_test(endpoint, format!("http://{proxy_address}")); + let client = HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy) + .build_async_telemetry_client( + endpoint, + Duration::from_secs(2), + &TelemetryClientTlsConfig::default(), + ) + .expect("async telemetry client should build"); + + let response = client + .send_bytes(telemetry_request(endpoint)) + .await + .expect("redirect should be returned to the exporter"); + proxy.join().expect("proxy should complete"); + + assert_eq!(response.status(), http::StatusCode::TEMPORARY_REDIRECT); +} + +#[test] +fn telemetry_client_rejects_incomplete_client_identity() { + let result = HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault) + .build_async_telemetry_client( + "https://telemetry.example/v1/traces", + Duration::from_secs(2), + &TelemetryClientTlsConfig { + client_certificate: Some(PathBuf::from("client.pem")), + ..Default::default() + }, + ); + + assert!(matches!( + result, + Err(BuildTelemetryHttpClientError::IncompleteClientIdentity) + )); +} + +#[test] +fn telemetry_client_reports_invalid_collector_certificate() { + let temp = tempfile::tempdir().expect("temporary directory should exist"); + let certificate = temp.path().join("collector.pem"); + std::fs::write(&certificate, "not a certificate").expect("certificate fixture should write"); + + let result = HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault) + .build_async_telemetry_client( + "https://telemetry.example/v1/traces", + Duration::from_secs(2), + &TelemetryClientTlsConfig { + ca_certificate: Some(certificate.clone()), + ..Default::default() + }, + ); + + assert!(matches!( + result, + Err(BuildTelemetryHttpClientError::InvalidCertificate { path, .. }) if path == certificate + )); +} + +fn telemetry_request(endpoint: &str) -> Request { + Request::builder() + .method(http::Method::POST) + .uri(endpoint) + .body(Bytes::from_static(b"telemetry")) + .expect("telemetry request should build") +} + +fn spawn_proxy() -> (SocketAddr, thread::JoinHandle) { + spawn_proxy_with_response("HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") +} + +fn spawn_proxy_with_response(response: &str) -> (SocketAddr, thread::JoinHandle) { + let listener = TcpListener::bind("127.0.0.1:0").expect("proxy should bind"); + let address = listener.local_addr().expect("proxy should expose address"); + let response = response.to_owned(); + let handle = thread::spawn(move || { + let (mut stream, _) = listener.accept().expect("proxy should receive request"); + stream + .set_read_timeout(Some(Duration::from_secs(2))) + .expect("proxy should set timeout"); + let mut request = [0; 4096]; + let count = stream + .read(&mut request) + .expect("proxy should read request"); + stream + .write_all(response.as_bytes()) + .expect("proxy should write response"); + String::from_utf8_lossy(&request[..count]).into_owned() + }); + + (address, handle) +}