fix(http): preserve error responses for retry classification (#3627)

Copilot-Session: 0bc45e17-c983-47d7-9a97-f6b1909ceefb

Author: cijothomas

opentelemetry-http/CHANGELOG.md

@@ -2,6 +2,11 @@
 
 ## vNext
 
+- Return HTTP error responses from the built-in reqwest and hyper clients instead
+  of converting 4xx and 5xx statuses into transport errors. This preserves the
+  response status and headers for exporter retry classification.
+  If your code relied on `send_bytes` returning `Err` for non-success statuses,
+  call `ResponseExt::error_for_status()` on the response instead.
 - **Breaking** Removed `reqwest-rustls-webpki-roots` feature. The `webpki-roots` cargo feature was
   removed from `reqwest` in v0.13.0. Use `reqwest-rustls` instead, which now correctly enables
   `reqwest/rustls` (platform native trust roots). To use Mozilla's embedded CA bundle, construct a

opentelemetry-http/Cargo.toml

@@ -29,6 +29,10 @@ opentelemetry = { workspace = true, features = ["trace"] }
 reqwest = { workspace = true, optional = true }
 tokio = { workspace = true, features = ["time"], optional = true }
 
+[dev-dependencies]
+futures-executor = { workspace = true }
+tokio = { workspace = true, features = ["net", "rt", "time"] }
+
 [lints]
 workspace = true
 

opentelemetry-http/src/lib.rs

@@ -101,7 +101,7 @@ mod reqwest {
         async fn send_bytes(&self, request: Request<Bytes>) -> Result<Response<Bytes>, HttpError> {
             otel_debug!(name: "ReqwestClient.Send");
             let request = request.try_into()?;
-            let mut response = self.execute(request).await?.error_for_status()?;
+            let mut response = self.execute(request).await?;
             let headers = std::mem::take(response.headers_mut());
             let mut http_response = Response::builder()
                 .status(response.status())
@@ -119,7 +119,7 @@ mod reqwest {
         async fn send_bytes(&self, request: Request<Bytes>) -> Result<Response<Bytes>, HttpError> {
             otel_debug!(name: "ReqwestBlockingClient.Send");
             let request = request.try_into()?;
-            let mut response = self.execute(request)?.error_for_status()?;
+            let mut response = self.execute(request)?;
             let headers = std::mem::take(response.headers_mut());
             let mut http_response = Response::builder()
                 .status(response.status())
@@ -134,7 +134,6 @@ mod reqwest {
 #[cfg(feature = "hyper")]
 pub mod hyper {
     use super::{async_trait, Bytes, HttpClient, HttpError, Request, Response};
-    use crate::ResponseExt;
     use http::HeaderValue;
     use http_body_util::{BodyExt, Full};
     use hyper::body::{Body as HttpBody, Frame};
@@ -207,7 +206,7 @@ pub mod hyper {
                 .body(response.into_body().collect().await?.to_bytes())?;
             *http_response.headers_mut() = headers;
 
-            Ok(http_response.error_for_status()?)
+            Ok(http_response)
         }
     }
 
@@ -258,6 +257,21 @@ impl<T> ResponseExt for Response<T> {
 mod tests {
     use super::*;
     use http::HeaderValue;
+    #[cfg(all(
+        any(feature = "hyper", feature = "reqwest", feature = "reqwest-blocking"),
+        not(target_arch = "wasm32")
+    ))]
+    use std::io::{Read, Write};
+    #[cfg(all(
+        any(feature = "hyper", feature = "reqwest", feature = "reqwest-blocking"),
+        not(target_arch = "wasm32")
+    ))]
+    use std::net::{SocketAddr, TcpListener};
+    #[cfg(all(
+        any(feature = "hyper", feature = "reqwest", feature = "reqwest-blocking"),
+        not(target_arch = "wasm32")
+    ))]
+    use std::thread::JoinHandle;
 
     #[test]
     fn http_headers_get() {
@@ -309,6 +323,85 @@ mod tests {
         assert!(got.contains(&"headername2"));
     }
 
+    #[cfg(all(
+        any(feature = "hyper", feature = "reqwest", feature = "reqwest-blocking"),
+        not(target_arch = "wasm32")
+    ))]
+    fn spawn_error_response_server() -> (SocketAddr, JoinHandle<()>) {
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let address = listener.local_addr().unwrap();
+        let server = std::thread::spawn(move || {
+            let (mut stream, _) = listener.accept().unwrap();
+            let mut request = [0; 1024];
+            let _ = stream.read(&mut request).unwrap();
+            stream
+                .write_all(
+                    b"HTTP/1.1 429 Too Many Requests\r\n\
+Retry-After: 7\r\n\
+Content-Length: 0\r\n\
+Connection: close\r\n\r\n",
+                )
+                .unwrap();
+        });
+        (address, server)
+    }
+
+    #[cfg(all(feature = "reqwest-blocking", not(target_arch = "wasm32")))]
+    #[test]
+    fn reqwest_blocking_preserves_error_response_status_and_headers() {
+        let (address, server) = spawn_error_response_server();
+        let client = ::reqwest::blocking::Client::new();
+        let request = Request::post(format!("http://{address}/v1/traces"))
+            .body(Bytes::new())
+            .unwrap();
+        let response = futures_executor::block_on(client.send_bytes(request)).unwrap();
+
+        server.join().unwrap();
+        assert_eq!(response.status(), http::StatusCode::TOO_MANY_REQUESTS);
+        assert_eq!(response.headers().get("retry-after").unwrap(), "7");
+    }
+
+    #[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))]
+    #[test]
+    fn reqwest_async_preserves_error_response_status_and_headers() {
+        let (address, server) = spawn_error_response_server();
+        let client = ::reqwest::Client::new();
+        let request = Request::post(format!("http://{address}/v1/traces"))
+            .body(Bytes::new())
+            .unwrap();
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        let response = runtime.block_on(client.send_bytes(request)).unwrap();
+
+        server.join().unwrap();
+        assert_eq!(response.status(), http::StatusCode::TOO_MANY_REQUESTS);
+        assert_eq!(response.headers().get("retry-after").unwrap(), "7");
+    }
+
+    #[cfg(all(feature = "hyper", not(target_arch = "wasm32")))]
+    #[test]
+    fn hyper_preserves_error_response_status_and_headers() {
+        let (address, server) = spawn_error_response_server();
+        let client = crate::hyper::HyperClient::with_default_connector(
+            std::time::Duration::from_secs(2),
+            None,
+        );
+        let request = Request::post(format!("http://{address}/v1/traces"))
+            .body(Bytes::new())
+            .unwrap();
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        let response = runtime.block_on(client.send_bytes(request)).unwrap();
+
+        server.join().unwrap();
+        assert_eq!(response.status(), http::StatusCode::TOO_MANY_REQUESTS);
+        assert_eq!(response.headers().get("retry-after").unwrap(), "7");
+    }
+
     #[test]
     fn http_headers_reserve() {
         let mut carrier = http::HeaderMap::new();