Add proxy-aware async OpenTelemetry HTTP transport

codex-rs/Cargo.lock

@@ -3381,6 +3381,7 @@ dependencies = [
  "http 1.4.0",
  "native-tls",
  "opentelemetry",
+ "opentelemetry-http",
  "opentelemetry_sdk",
  "pretty_assertions",
  "rcgen",

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"

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 }

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;

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<PathBuf>,
+    pub client_certificate: Option<PathBuf>,
+    pub client_private_key: Option<PathBuf>,
+}
+
+/// 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<reqwest::Certificate>,
+    client_identity: Option<reqwest::Identity>,
+}
+
+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<impl HttpClient + 'static + use<>, 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<PreparedTlsConfig, BuildTelemetryHttpClientError> {
+    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<Vec<u8>, BuildTelemetryHttpClientError> {
+    fs::read(path).map_err(|source| BuildTelemetryHttpClientError::ReadTlsFile {
+        path: path.to_path_buf(),
+        source,
+    })
+}
+
+#[cfg(test)]
+#[path = "telemetry_client_tests.rs"]
+mod tests;

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<Bytes> {
+    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<String>) {
+    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<String>) {
+    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)
+}