Use bytes::Bytes as the HTTP request body in HttpClient. (#2485)

Co-authored-by: Zhongyang Wu <[email protected]>
Co-authored-by: Cijo Thomas <[email protected]>
Co-authored-by: Lalit Kumar Bhasin <[email protected]>

Author: 4 people

opentelemetry-http/CHANGELOG.md

@@ -4,6 +4,7 @@
 
 - Bump msrv to 1.75.0.
 - Add "internal-logs" feature flag (enabled by default), and emit internal logs.
+- Add `HttpClient::send_bytes` with `bytes::Bytes` request payload and deprecate old `HttpClient::send` function.
 
 ## 0.27.0
 

opentelemetry-http/src/lib.rs

@@ -55,13 +55,24 @@ pub type HttpError = Box<dyn std::error::Error + Send + Sync + 'static>;
 /// users to bring their choice of HTTP client.
 #[async_trait]
 pub trait HttpClient: Debug + Send + Sync {
-    /// Send the specified HTTP request
+    /// Send the specified HTTP request with `Vec<u8>` payload
     ///
     /// Returns the HTTP response including the status code and body.
     ///
     /// Returns an error if it can't connect to the server or the request could not be completed,
     /// e.g. because of a timeout, infinite redirects, or a loss of connection.
-    async fn send(&self, request: Request<Vec<u8>>) -> Result<Response<Bytes>, HttpError>;
+    #[deprecated(note = "Use `send_bytes` with `Bytes` payload instead.")]
+    async fn send(&self, request: Request<Vec<u8>>) -> Result<Response<Bytes>, HttpError> {
+        self.send_bytes(request.map(Into::into)).await
+    }
+
+    /// Send the specified HTTP request with `Bytes` payload.
+    ///
+    /// Returns the HTTP response including the status code and body.
+    ///
+    /// Returns an error if it can't connect to the server or the request could not be completed,
+    /// e.g. because of a timeout, infinite redirects, or a loss of connection.
+    async fn send_bytes(&self, request: Request<Bytes>) -> Result<Response<Bytes>, HttpError>;
 }
 
 #[cfg(feature = "reqwest")]
@@ -72,7 +83,7 @@ mod reqwest {
 
     #[async_trait]
     impl HttpClient for reqwest::Client {
-        async fn send(&self, request: Request<Vec<u8>>) -> Result<Response<Bytes>, HttpError> {
+        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()?;
@@ -89,7 +100,7 @@ mod reqwest {
     #[cfg(not(target_arch = "wasm32"))]
     #[async_trait]
     impl HttpClient for reqwest::blocking::Client {
-        async fn send(&self, request: Request<Vec<u8>>) -> Result<Response<Bytes>, HttpError> {
+        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()?;
@@ -159,7 +170,7 @@ pub mod hyper {
 
     #[async_trait]
     impl HttpClient for HyperClient {
-        async fn send(&self, request: Request<Vec<u8>>) -> Result<Response<Bytes>, HttpError> {
+        async fn send_bytes(&self, request: Request<Bytes>) -> Result<Response<Bytes>, HttpError> {
             otel_debug!(name: "HyperClient.Send");
             let (parts, body) = request.into_parts();
             let mut request = Request::from_parts(parts, Body(Full::from(body)));

opentelemetry-otlp/src/exporter/http/logs.rs

@@ -28,7 +28,7 @@ impl LogExporter for OtlpHttpClient {
                 .method(Method::POST)
                 .uri(&self.collector_endpoint)
                 .header(CONTENT_TYPE, content_type)
-                .body(body)
+                .body(body.into())
                 .map_err(|e| crate::Error::RequestFailed(Box::new(e)))?;
 
             for (k, v) in &self.headers {
@@ -37,7 +37,7 @@ impl LogExporter for OtlpHttpClient {
 
             let request_uri = request.uri().to_string();
             otel_debug!(name: "HttpLogsClient.CallingExport");
-            let response = client.send(request).await?;
+            let response = client.send_bytes(request).await?;
 
             if !response.status().is_success() {
                 let error = format!(

opentelemetry-otlp/src/exporter/http/metrics.rs

@@ -27,7 +27,7 @@ impl MetricsClient for OtlpHttpClient {
             .method(Method::POST)
             .uri(&self.collector_endpoint)
             .header(CONTENT_TYPE, content_type)
-            .body(body)
+            .body(body.into())
             .map_err(|e| crate::Error::RequestFailed(Box::new(e)))?;
 
         for (k, v) in &self.headers {
@@ -36,7 +36,7 @@ impl MetricsClient for OtlpHttpClient {
 
         otel_debug!(name: "HttpMetricsClient.CallingExport");
         client
-            .send(request)
+            .send_bytes(request)
             .await
             .map_err(|e| MetricError::ExportErr(Box::new(Error::RequestFailed(e))))?;
 

opentelemetry-otlp/src/exporter/http/trace.rs

@@ -30,7 +30,7 @@ impl SpanExporter for OtlpHttpClient {
             .method(Method::POST)
             .uri(&self.collector_endpoint)
             .header(CONTENT_TYPE, content_type)
-            .body(body)
+            .body(body.into())
         {
             Ok(req) => req,
             Err(e) => {
@@ -48,7 +48,7 @@ impl SpanExporter for OtlpHttpClient {
         Box::pin(async move {
             let request_uri = request.uri().to_string();
             otel_debug!(name: "HttpTracesClient.CallingExport");
-            let response = client.send(request).await?;
+            let response = client.send_bytes(request).await?;
 
             if !response.status().is_success() {
                 let error = format!(

opentelemetry-sdk/src/trace/sampler/jaeger_remote/sampler.rs

@@ -228,7 +228,7 @@ impl JaegerRemoteSampler {
     {
         let request = http::Request::get(endpoint)
             .header("Content-Type", "application/json")
-            .body(Vec::new())
+            .body(Default::default())
             .unwrap();
 
         let resp = client

opentelemetry-zipkin/src/exporter/uploader.rs

@@ -41,9 +41,9 @@ impl JsonV2Client {
             .method(Method::POST)
             .uri(self.collector_endpoint.clone())
             .header(CONTENT_TYPE, "application/json")
-            .body(serde_json::to_vec(&spans).unwrap_or_default())
+            .body(serde_json::to_vec(&spans).unwrap_or_default().into())
             .map_err::<Error, _>(Into::into)?;
-        let _ = self.client.send(req).await?.error_for_status()?;
+        let _ = self.client.send_bytes(req).await?.error_for_status()?;
         Ok(())
     }
 }

opentelemetry-zipkin/src/lib.rs

@@ -128,10 +128,10 @@
 //!
 //! #[async_trait]
 //! impl HttpClient for HyperClient {
-//!     async fn send(&self, req: Request<Vec<u8>>) -> Result<Response<Bytes>, HttpError> {
+//!     async fn send_bytes(&self, req: Request<Bytes>) -> Result<Response<Bytes>, HttpError> {
 //!         let resp = self
 //!             .0
-//!             .request(req.map(|v| Full::new(Bytes::from(v))))
+//!             .request(req.map(|v| Full::new(v)))
 //!             .await?;
 //!
 //!         let response = Response::builder()