Upgrade to hyper/http v1.0 (#1674)

Co-authored-by: Lalit Kumar Bhasin <labhas@microsoft.com>
open-telemetry · Jul 12, 2024 · 621a5a9 · 621a5a9
1 parent 166a127
commit 621a5a9
Show file tree

Hide file tree

Showing 26 changed files with 187 additions and 137 deletions.
diff --git a/Cargo.toml b/Cargo.toml
@@ -23,23 +23,25 @@ criterion = "0.5"
 futures-core = "0.3"
 futures-executor = "0.3"
 futures-util = { version = "0.3", default-features = false }
-hyper = { version = "0.14", default-features = false }
-http = { version = "0.2", default-features = false }
+http = { version = "1.1", default-features = false, features = ["std"] }
+http-body-util = "0.1"
+hyper = { version = "1.3", default-features = false }
+hyper-util = "0.1"
 log = "0.4.21"
 once_cell = "1.13"
 ordered-float = "4.0"
 pin-project-lite = "0.2"
-prost = "0.12"
-prost-build = "0.12"
-prost-types = "0.12"
+prost = "0.13"
+prost-build = "0.13"
+prost-types = "0.13"
 rand = { version = "0.8", default-features = false }
-reqwest = { version = "0.11", default-features = false }
+reqwest = { version = "0.12", default-features = false }
 serde = { version = "1.0", default-features = false }
 serde_json = "1.0"
 temp-env = "0.3.6"
 thiserror = { version = "1", default-features = false }
-tonic = { version = "0.11", default-features = false }
-tonic-build = "0.11"
+tonic = { version = "0.12", default-features = false }
+tonic-build = "0.12"
 tokio = { version = "1", default-features = false }
 tokio-stream = "0.1.1"
 tracing = { version = "0.1", default-features = false }

diff --git a/examples/tracing-http-propagator/Cargo.toml b/examples/tracing-http-propagator/Cargo.toml
@@ -16,7 +16,9 @@ path = "src/client.rs"
 doc = false
 
 [dependencies]
+http-body-util = { workspace = true }
 hyper = { workspace = true, features = ["full"] }
+hyper-util = { workspace = true, features = ["full"] }
 tokio = { workspace = true, features = ["full"] }
 opentelemetry = { path = "../../opentelemetry" }
 opentelemetry_sdk = { path = "../../opentelemetry-sdk" }

diff --git a/examples/tracing-http-propagator/src/client.rs b/examples/tracing-http-propagator/src/client.rs
@@ -1,10 +1,11 @@
-use hyper::{body::Body, Client};
+use http_body_util::Full;
+use hyper_util::{client::legacy::Client, rt::TokioExecutor};
 use opentelemetry::{
     global,
     trace::{SpanKind, TraceContextExt, Tracer},
     Context, KeyValue,
 };
-use opentelemetry_http::HeaderInjector;
+use opentelemetry_http::{Bytes, HeaderInjector};
 use opentelemetry_sdk::{propagation::TraceContextPropagator, trace::TracerProvider};
 use opentelemetry_stdout::SpanExporter;
 
@@ -24,7 +25,7 @@ async fn send_request(
     body_content: &str,
     span_name: &str,
 ) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync + 'static>> {
-    let client = Client::new();
+    let client = Client::builder(TokioExecutor::new()).build_http();
     let tracer = global::tracer("example/client");
     let span = tracer
         .span_builder(String::from(span_name))
@@ -37,7 +38,7 @@ async fn send_request(
         propagator.inject_context(&cx, &mut HeaderInjector(req.headers_mut().unwrap()))
     });
     let res = client
-        .request(req.body(Body::from(String::from(body_content)))?)
+        .request(req.body(Full::new(Bytes::from(body_content.to_string())))?)
         .await?;
 
     cx.span().add_event(

diff --git a/examples/tracing-http-propagator/src/server.rs b/examples/tracing-http-propagator/src/server.rs
@@ -1,50 +1,64 @@
-use hyper::{
-    service::{make_service_fn, service_fn},
-    Body, Request, Response, Server, StatusCode,
-};
+use http_body_util::{combinators::BoxBody, BodyExt, Full};
+use hyper::{body::Incoming, service::service_fn, Request, Response, StatusCode};
+use hyper_util::rt::{TokioExecutor, TokioIo};
 use opentelemetry::{
     global,
     trace::{FutureExt, Span, SpanKind, TraceContextExt, Tracer},
     Context, KeyValue,
 };
-use opentelemetry_http::HeaderExtractor;
+use opentelemetry_http::{Bytes, HeaderExtractor};
 use opentelemetry_sdk::{propagation::TraceContextPropagator, trace::TracerProvider};
 use opentelemetry_semantic_conventions::trace;
 use opentelemetry_stdout::SpanExporter;
 use std::{convert::Infallible, net::SocketAddr};
+use tokio::net::TcpListener;
 
 // Utility function to extract the context from the incoming request headers
-fn extract_context_from_request(req: &Request<Body>) -> Context {
+fn extract_context_from_request(req: &Request<Incoming>) -> Context {
     global::get_text_map_propagator(|propagator| {
         propagator.extract(&HeaderExtractor(req.headers()))
     })
 }
 
 // Separate async function for the handle endpoint
-async fn handle_health_check(_req: Request<Body>) -> Result<Response<Body>, Infallible> {
+async fn handle_health_check(
+    _req: Request<Incoming>,
+) -> Result<Response<BoxBody<Bytes, hyper::Error>>, Infallible> {
     let tracer = global::tracer("example/server");
     let mut span = tracer
         .span_builder("health_check")
         .with_kind(SpanKind::Internal)
         .start(&tracer);
     span.add_event("Health check accessed", vec![]);
-    let res = Response::new(Body::from("Server is up and running!"));
+
+    let res = Response::new(
+        Full::new(Bytes::from_static(b"Server is up and running!"))
+            .map_err(|err| match err {})
+            .boxed(),
+    );
+
     Ok(res)
 }
 
 // Separate async function for the echo endpoint
-async fn handle_echo(req: Request<Body>) -> Result<Response<Body>, Infallible> {
+async fn handle_echo(
+    req: Request<Incoming>,
+) -> Result<Response<BoxBody<Bytes, hyper::Error>>, Infallible> {
     let tracer = global::tracer("example/server");
     let mut span = tracer
         .span_builder("echo")
         .with_kind(SpanKind::Internal)
         .start(&tracer);
     span.add_event("Echoing back the request", vec![]);
-    let res = Response::new(req.into_body());
+
+    let res = Response::new(req.into_body().boxed());
+
     Ok(res)
 }
 
-async fn router(req: Request<Body>) -> Result<Response<Body>, Infallible> {
+async fn router(
+    req: Request<Incoming>,
+) -> Result<Response<BoxBody<Bytes, hyper::Error>>, Infallible> {
     // Extract the context from the incoming request headers
     let parent_cx = extract_context_from_request(&req);
     let response = {
@@ -64,12 +78,13 @@ async fn router(req: Request<Body>) -> Result<Response<Body>, Infallible> {
             _ => {
                 cx.span()
                     .set_attribute(KeyValue::new(trace::HTTP_RESPONSE_STATUS_CODE, 404));
-                let mut not_found = Response::default();
+                let mut not_found = Response::new(BoxBody::default());
                 *not_found.status_mut() = StatusCode::NOT_FOUND;
                 Ok(not_found)
             }
         }
     };
+
     response
 }
 
@@ -87,15 +102,18 @@ fn init_tracer() {
 
 #[tokio::main]
 async fn main() {
+    use hyper_util::server::conn::auto::Builder;
+
     init_tracer();
     let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
+    let listener = TcpListener::bind(addr).await.unwrap();
 
-    let make_svc = make_service_fn(|_conn| async { Ok::<_, Infallible>(service_fn(router)) });
-
-    let server = Server::bind(&addr).serve(make_svc);
-
-    println!("Listening on {addr}");
-    if let Err(e) = server.await {
-        eprintln!("server error: {e}");
+    while let Ok((stream, _addr)) = listener.accept().await {
+        if let Err(err) = Builder::new(TokioExecutor::new())
+            .serve_connection(TokioIo::new(stream), service_fn(router))
+            .await
+        {
+            eprintln!("{err}");
+        }
     }
 }
diff --git a/opentelemetry-appender-tracing/Cargo.toml b/opentelemetry-appender-tracing/Cargo.toml
@@ -12,7 +12,6 @@ rust-version = "1.65"
 
 [dependencies]
 log = { workspace = true, optional = true }
-once_cell = { workspace = true }
 opentelemetry = { version = "0.23", path = "../opentelemetry", features = ["logs"] }
 tracing = { workspace = true, features = ["std"]}
 tracing-core = { workspace = true }

diff --git a/opentelemetry-http/CHANGELOG.md b/opentelemetry-http/CHANGELOG.md
@@ -4,6 +4,7 @@
 
 - **Breaking** Correct the misspelling of "webkpi" to "webpki" in features [#1842](https://github.com/open-telemetry/opentelemetry-rust/pull/1842)
 - **Breaking** Remove support for the `isahc` HTTP client [#1924](https://github.com/open-telemetry/opentelemetry-rust/pull/1924)
+- Update to `http` v1 [#1674](https://github.com/open-telemetry/opentelemetry-rust/pull/1674)
 
 ## v0.12.0
 

diff --git a/opentelemetry-http/Cargo.toml b/opentelemetry-http/Cargo.toml
@@ -10,14 +10,17 @@ edition = "2021"
 rust-version = "1.65"
 
 [features]
+hyper = ["dep:http-body-util", "dep:hyper", "dep:hyper-util", "dep:tokio"]
 reqwest-rustls = ["reqwest", "reqwest/rustls-tls-native-roots"]
 reqwest-rustls-webpki-roots = ["reqwest", "reqwest/rustls-tls-webpki-roots"]
 
 [dependencies]
 async-trait = { workspace = true }
 bytes = { workspace = true }
 http = { workspace = true }
-hyper = { workspace = true, features = ["http2", "client", "tcp"], optional = true }
+http-body-util = { workspace = true, optional = true }
+hyper = { workspace = true, optional = true }
+hyper-util = { workspace = true, features = ["client-legacy", "http2"], optional = true }
 opentelemetry = { version = "0.23", path = "../opentelemetry", features = ["trace"] }
 reqwest = { workspace = true, features = ["blocking"], optional = true }
 tokio = { workspace = true, features = ["time"], optional = true }
diff --git a/opentelemetry-http/src/lib.rs b/opentelemetry-http/src/lib.rs
@@ -105,21 +105,24 @@ pub mod hyper {
 
     use super::{async_trait, Bytes, HttpClient, HttpError, Request, Response};
     use http::HeaderValue;
-    use hyper::client::connect::Connect;
-    use hyper::Client;
+    use http_body_util::{BodyExt, Full};
+    use hyper::body::{Body as HttpBody, Frame};
+    use hyper_util::client::legacy::{connect::Connect, Client};
     use std::fmt::Debug;
+    use std::pin::Pin;
+    use std::task::{self, Poll};
     use std::time::Duration;
     use tokio::time;
 
     #[derive(Debug, Clone)]
     pub struct HyperClient<C> {
-        inner: Client<C>,
+        inner: Client<C, Body>,
         timeout: Duration,
         authorization: Option<HeaderValue>,
     }
 
     impl<C> HyperClient<C> {
-        pub fn new_with_timeout(inner: Client<C>, timeout: Duration) -> Self {
+        pub fn new_with_timeout(inner: Client<C, Body>, timeout: Duration) -> Self {
             Self {
                 inner,
                 timeout,
@@ -128,7 +131,7 @@ pub mod hyper {
         }
 
         pub fn new_with_timeout_and_authorization_header(
-            inner: Client<C>,
+            inner: Client<C, Body>,
             timeout: Duration,
             authorization: HeaderValue,
         ) -> Self {
@@ -147,22 +150,49 @@ pub mod hyper {
     {
         async fn send(&self, request: Request<Vec<u8>>) -> Result<Response<Bytes>, HttpError> {
             let (parts, body) = request.into_parts();
-            let mut request = Request::from_parts(parts, body.into());
+            let mut request = Request::from_parts(parts, Body(Full::from(body)));
             if let Some(ref authorization) = self.authorization {
                 request
                     .headers_mut()
                     .insert(http::header::AUTHORIZATION, authorization.clone());
             }
             let mut response = time::timeout(self.timeout, self.inner.request(request)).await??;
             let headers = std::mem::take(response.headers_mut());
+
             let mut http_response = Response::builder()
                 .status(response.status())
-                .body(hyper::body::to_bytes(response.into_body()).await?)?;
+                .body(response.into_body().collect().await?.to_bytes())?;
             *http_response.headers_mut() = headers;
 
             Ok(http_response.error_for_status()?)
         }
     }
+
+    pub struct Body(Full<Bytes>);
+
+    impl HttpBody for Body {
+        type Data = Bytes;
+        type Error = Box<dyn std::error::Error + Send + Sync + 'static>;
+
+        #[inline]
+        fn poll_frame(
+            self: Pin<&mut Self>,
+            cx: &mut task::Context<'_>,
+        ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
+            let inner_body = unsafe { self.map_unchecked_mut(|b| &mut b.0) };
+            inner_body.poll_frame(cx).map_err(Into::into)
+        }
+
+        #[inline]
+        fn is_end_stream(&self) -> bool {
+            self.0.is_end_stream()
+        }
+
+        #[inline]
+        fn size_hint(&self) -> hyper::body::SizeHint {
+            self.0.size_hint()
+        }
+    }
 }
 
 /// Methods to make working with responses from the [`HttpClient`] trait easier.

diff --git a/opentelemetry-otlp/CHANGELOG.md b/opentelemetry-otlp/CHANGELOG.md
@@ -20,7 +20,7 @@ now use `.with_resource(RESOURCE::default())` to configure Resource when using
   previous release.
 - **Breaking** [1869](https://github.com/open-telemetry/opentelemetry-rust/pull/1869) The OTLP logs exporter now overrides the [InstrumentationScope::name](https://github.com/open-telemetry/opentelemetry-proto/blob/b3060d2104df364136d75a35779e6bd48bac449a/opentelemetry/proto/common/v1/common.proto#L73) field with the `target` from `LogRecord`, if target is populated.
 - Groups batch of `LogRecord` and `Span` by their resource and instrumentation scope before exporting, for better efficiency [#1873](https://github.com/open-telemetry/opentelemetry-rust/pull/1873).
-
+- **Breaking** Update to `http` v1 and `tonic` v0.12 [#1674](https://github.com/open-telemetry/opentelemetry-rust/pull/1674)
 
 ## v0.16.0
 

diff --git a/opentelemetry-otlp/examples/basic-otlp-http/Cargo.toml b/opentelemetry-otlp/examples/basic-otlp-http/Cargo.toml
@@ -8,7 +8,7 @@ publish = false
 [features]
 default = ["reqwest"]
 reqwest = ["opentelemetry-otlp/reqwest-client"]
-hyper = ["dep:async-trait", "dep:http", "dep:hyper", "dep:opentelemetry-http", "dep:bytes"]
+hyper = ["dep:async-trait", "dep:http", "dep:http-body-util", "dep:hyper", "dep:hyper-util", "dep:opentelemetry-http", "dep:bytes"]
 
 
 [dependencies]
@@ -23,7 +23,9 @@ opentelemetry-semantic-conventions = { path = "../../../opentelemetry-semantic-c
 async-trait = { workspace = true, optional = true }
 bytes = { workspace = true, optional = true }
 http = { workspace = true, optional = true }
+http-body-util = { workspace = true, optional = true  }
 hyper = { workspace = true, features = ["client"], optional = true }
+hyper-util = { workspace = true, features = ["client-legacy"], optional = true }
 tokio = { workspace = true, features = ["full"] }
 tracing = { workspace = true, features = ["std"]}
 tracing-core = { workspace = true }

diff --git a/opentelemetry-otlp/examples/basic-otlp-http/src/hyper.rs b/opentelemetry-otlp/examples/basic-otlp-http/src/hyper.rs
@@ -1,20 +1,24 @@
 use async_trait::async_trait;
 use bytes::Bytes;
 use http::{Request, Response};
-use hyper::{
-    client::{connect::Connect, HttpConnector},
-    Body, Client,
+use http_body_util::{BodyExt, Full};
+use hyper_util::{
+    client::legacy::{
+        connect::{Connect, HttpConnector},
+        Client,
+    },
+    rt::TokioExecutor,
 };
 use opentelemetry_http::{HttpClient, HttpError, ResponseExt};
 
 pub struct HyperClient<C> {
-    inner: hyper::Client<C>,
+    inner: hyper_util::client::legacy::Client<C, Full<Bytes>>,
 }
 
 impl Default for HyperClient<HttpConnector> {
     fn default() -> Self {
         Self {
-            inner: Client::new(),
+            inner: Client::builder(TokioExecutor::new()).build_http(),
         }
     }
 }
@@ -30,15 +34,15 @@ impl<C> std::fmt::Debug for HyperClient<C> {
 #[async_trait]
 impl<C: Connect + Clone + Send + Sync + 'static> HttpClient for HyperClient<C> {
     async fn send(&self, request: Request<Vec<u8>>) -> Result<Response<Bytes>, HttpError> {
-        let request = request.map(Body::from);
+        let request = request.map(|body| Full::new(Bytes::from(body)));
 
         let (parts, body) = self
             .inner
             .request(request)
             .await?
             .error_for_status()?
             .into_parts();
-        let body = hyper::body::to_bytes(body).await?;
+        let body = body.collect().await?.to_bytes();
 
         Ok(Response::from_parts(parts, body))
     }