chore(proxy/http): address hyper::client::conn::SendRequest deprecation

cratelyn · cratelyn · commit 1e519082e038 · 2024-12-03T15:37:56.000-05:00
this commit updates code in `linkerd-proxy-http`'s HTTP/2 client code, and the `linkerd-app-test` crate's `TestServer`, to use the new `hyper::client::conn::http2::SendRequest` backported from the 1.x major release. see <hyperium/hyper#2960> for more information. this commit refrains from updating the broader client connection system, and addresses the breaking changes to `SendRequest` made in the 1.0 major release, namely: * send request is no longer a tower service: * <https://docs.rs/hyper/0.14.31/hyper/client/conn/struct.SendRequest.html#impl-Service%3CRequest%3CB%3E%3E-for-SendRequest%3CB%3E> * <https://docs.rs/hyper/1.5.1/hyper/client/conn/http2/struct.SendRequest.html#trait-implementations> * `send_request()` now returns an anonymous `impl Future` and not a named `ResponseFuture`, as in `0.14`. * <https://docs.rs/hyper/0.14.31/hyper/client/conn/struct.ResponseFuture.html> * <https://docs.rs/hyper/1.5.1/hyper/client/conn/http2/struct.SendRequest.html#method.send_request> NB: this change depends on <hyperium/hyper#3798>.
diff --git a/linkerd/app/test/src/http_util.rs b/linkerd/app/test/src/http_util.rs
@@ -3,15 +3,15 @@ use crate::{
     io, ContextError,
 };
 use futures::FutureExt;
-use hyper::{body::HttpBody, Body, Request, Response};
+use hyper::{body::HttpBody, client::conn::http2::SendRequest, Body, Request, Response};
 use parking_lot::Mutex;
 use std::{future::Future, sync::Arc};
 use tokio::task::JoinHandle;
 use tower::{util::ServiceExt, Service};
 use tracing::Instrument;
 
 #[allow(deprecated)] // linkerd/linkerd2#8733
-use hyper::client::conn::{Builder as ClientBuilder, SendRequest};
+use hyper::client::conn::Builder as ClientBuilder;
 
 pub struct Server {
     #[allow(deprecated)] // linkerd/linkerd2#8733
@@ -72,7 +72,7 @@ pub async fn connect_client(
             res.map_err(Into::into)
         })
         .instrument(tracing::info_span!("client_bg"));
-    (client, tokio::spawn(client_bg))
+    (client.into(), tokio::spawn(client_bg))
 }
 
 #[allow(deprecated)] // linkerd/linkerd2#8733
@@ -98,19 +98,21 @@ pub async fn connect_and_accept(
 }
 
 #[tracing::instrument(skip(client))]
-#[allow(deprecated)] // linkerd/linkerd2#8733
 pub async fn http_request(
     client: &mut SendRequest<Body>,
     request: Request<Body>,
 ) -> Result<Response<Body>, Error> {
-    let rsp = client
+    // Wait for the dispatcher to be ready for a request.
+    client
         .ready()
         .await
-        .map_err(ContextError::ctx("HTTP client poll_ready failed"))?
-        .call(request)
+        .map_err(ContextError::ctx("HTTP client poll_ready failed"))?;
+
+    // Send the request, awaiting a response.
+    let rsp = client
+        .send_request(request)
         .await
         .map_err(ContextError::ctx("HTTP client request failed"))?;
-
     tracing::info!(?rsp);
 
     Ok(rsp)
diff --git a/linkerd/proxy/http/Cargo.toml b/linkerd/proxy/http/Cargo.toml
@@ -21,6 +21,7 @@ http = "0.2"
 http-body = "0.4"
 httparse = "1"
 hyper = { version = "0.14", features = [
+    "backports",
     "client",
     "deprecated",
     "http1",
diff --git a/linkerd/proxy/http/src/h2.rs b/linkerd/proxy/http/src/h2.rs
@@ -23,8 +23,7 @@ pub struct Connect<C, B> {
 
 #[derive(Debug)]
 pub struct Connection<B> {
-    #[allow(deprecated)] // linkerd/linkerd2#8733
-    tx: hyper::client::conn::SendRequest<B>,
+    tx: hyper::client::conn::http2::SendRequest<B>,
 }
 
 // === impl Connect ===
@@ -136,7 +135,7 @@ where
                         .instrument(trace_span!("conn").or_current()),
                 );
 
-                Ok(Connection { tx })
+                Ok(Connection { tx: tx.into() })
             }
             .instrument(debug_span!("h2").or_current()),
         )
@@ -153,7 +152,7 @@ where
 {
     type Response = http::Response<hyper::Body>;
     type Error = hyper::Error;
-    type Future = conn::ResponseFuture;
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
 
     #[inline]
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
@@ -175,6 +174,6 @@ where
             *req.version_mut() = http::Version::HTTP_11;
         }
 
-        self.tx.send_request(req)
+        self.tx.send_request(req).boxed()
     }
 }
diff --git a/linkerd/proxy/http/src/server/tests.rs b/linkerd/proxy/http/src/server/tests.rs
@@ -2,6 +2,7 @@ use std::vec;
 
 use super::*;
 use bytes::Bytes;
+use futures::FutureExt;
 use http_body::Body;
 use linkerd_stack::CloneParam;
 use tokio::time;
@@ -144,8 +145,7 @@ async fn h2_stream_window_exhaustion() {
 const LOG_LEVEL: &str = "h2::proto=trace,hyper=trace,linkerd=trace,info";
 
 struct TestServer {
-    #[allow(deprecated)] // linkerd/linkerd2#8733
-    client: hyper::client::conn::SendRequest<BoxBody>,
+    client: hyper::client::conn::http2::SendRequest<BoxBody>,
     server: Handle,
 }
 
@@ -202,7 +202,10 @@ impl TestServer {
             .expect("client connect");
         tokio::spawn(task.instrument(info_span!("client")));
 
-        Self { client, server }
+        Self {
+            client: client.into(),
+            server,
+        }
     }
 
     #[allow(deprecated)] // linkerd/linkerd2#8733
@@ -228,7 +231,8 @@ impl TestServer {
         self.server.allow(1);
         let mut call0 = self
             .client
-            .send_request(http::Request::new(BoxBody::default()));
+            .send_request(http::Request::new(BoxBody::default()))
+            .boxed();
         let (_req, next) = tokio::select! {
             _ = (&mut call0) => unreachable!("client cannot receive a response"),
             next = self.server.next_request() => next.expect("server not dropped"),

Original file line number	Diff line number	Diff line change
`@@ -23,8 +23,7 @@ pub struct Connect<C, B> {`
`23`	`23`
`24`	`24`	`#[derive(Debug)]`
`25`	`25`	`pub struct Connection<B> {`
`26`		`- #[allow(deprecated)] // linkerd/linkerd2#8733`
`27`		`- tx: hyper::client::conn::SendRequest<B>,`
	`26`	`+ tx: hyper::client::conn::http2::SendRequest<B>,`
`28`	`27`	`}`
`29`	`28`
`30`	`29`	`// === impl Connect ===`
`@@ -136,7 +135,7 @@ where`
`136`	`135`	`.instrument(trace_span!("conn").or_current()),`
`137`	`136`	`);`
`138`	`137`
`139`		`- Ok(Connection { tx })`
	`138`	`+ Ok(Connection { tx: tx.into() })`
`140`	`139`	`}`
`141`	`140`	`.instrument(debug_span!("h2").or_current()),`
`142`	`141`	`)`
`@@ -153,7 +152,7 @@ where`
`153`	`152`	`{`
`154`	`153`	`type Response = http::Response<hyper::Body>;`
`155`	`154`	`type Error = hyper::Error;`
`156`		`- type Future = conn::ResponseFuture;`
	`155`	`+ type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;`
`157`	`156`
`158`	`157`	`#[inline]`
`159`	`158`	`fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {`
`@@ -175,6 +174,6 @@ where`
`175`	`174`	`*req.version_mut() = http::Version::HTTP_11;`
`176`	`175`	`}`
`177`	`176`
`178`		`- self.tx.send_request(req)`
	`177`	`+ self.tx.send_request(req).boxed()`
`179`	`178`	`}`
`180`	`179`	`}`