proxy http work in progress

Author: cratelyn

Cargo.lock

@@ -1995,6 +1995,7 @@ dependencies = [
  "async-trait",
  "bytes",
  "futures",
+ "hyper-util",
  "linkerd-errno",
  "pin-project",
  "tokio",
@@ -2327,9 +2328,11 @@ dependencies = [
  "h2",
  "http 1.2.0",
  "http-body",
+ "http-body-util",
  "httparse",
  "hyper",
  "hyper-balance",
+ "hyper-util",
  "linkerd-detect",
  "linkerd-duplex",
  "linkerd-error",

linkerd/io/Cargo.toml

@@ -16,6 +16,7 @@ default = []
 async-trait = "0.1"
 futures = { version = "0.3", default-features = false }
 bytes = { workspace = true }
+hyper-util = { workspace = true, features = ["tokio"] }
 linkerd-errno = { path = "../errno" }
 tokio = { version = "1", features = ["io-util", "net"] }
 tokio-test = { version = "0.4", optional = true }

linkerd/io/src/lib.rs

@@ -76,3 +76,10 @@ impl PeerAddr for tokio::io::DuplexStream {
         Ok(([0, 0, 0, 0], 0).into())
     }
 }
+
+impl<T: PeerAddr> PeerAddr for hyper_util::rt::tokio::TokioIo<T> {
+    #[inline]
+    fn peer_addr(&self) -> Result<SocketAddr> {
+        self.inner().peer_addr()
+    }
+}

linkerd/proxy/http/Cargo.toml

@@ -27,6 +27,12 @@ hyper = { workspace = true, features = [
     "server",
 ] }
 hyper-balance = { path = "../../../hyper-balance" }
+hyper-util = { workspace = true, default-features = false, features = [
+    "client",
+    "client-legacy",
+    "http1",
+    "service",
+] }
 parking_lot = "0.12"
 pin-project = "1"
 rand = "0.9"
@@ -54,6 +60,7 @@ linkerd-proxy-balance = { path = "../balance" }
 linkerd-stack = { path = "../../stack" }
 
 [dev-dependencies]
+http-body-util = { workspace = true, features = ["channel"] }
 tokio-test = "0.4"
 tower-test = "0.4"
 linkerd-tracing = { path = "../../tracing", features = ["ansi"] }

linkerd/proxy/http/src/client.rs

@@ -63,7 +63,7 @@ where
     C::Connection: Unpin + Send,
     C::Metadata: Send,
     C::Future: Unpin + Send + 'static,
-    B: crate::Body + Send + 'static,
+    B: crate::Body + Send + Unpin + 'static,
     B::Data: Send,
     B::Error: Into<Error> + Send + Sync,
 {
@@ -123,7 +123,7 @@ where
     C::Connection: Unpin + Send,
     C::Future: Unpin + Send + 'static,
     C::Error: Into<Error>,
-    B: crate::Body + Send + 'static,
+    B: crate::Body + Send + Unpin + 'static,
     B::Data: Send,
     B::Error: Into<Error> + Send + Sync,
 {

linkerd/proxy/http/src/h1.rs

@@ -33,8 +33,8 @@ pub struct PoolSettings {
 pub struct Client<C, T, B> {
     connect: C,
     target: T,
-    absolute_form: Option<hyper::Client<HyperConnect<C, T>, B>>,
-    origin_form: Option<hyper::Client<HyperConnect<C, T>, B>>,
+    absolute_form: Option<hyper_util::client::legacy::Client<HyperConnect<C, T>, B>>,
+    origin_form: Option<hyper_util::client::legacy::Client<HyperConnect<C, T>, B>>,
     pool: PoolSettings,
 }
 
@@ -68,9 +68,9 @@ impl<C, T, B> Client<C, T, B>
 where
     T: Clone + Send + Sync + 'static,
     C: MakeConnection<(crate::Variant, T)> + Clone + Send + Sync + 'static,
-    C::Connection: Unpin + Send,
+    C::Connection: Unpin + Send + hyper::rt::Read + hyper::rt::Write,
     C::Future: Unpin + Send + 'static,
-    B: crate::Body + Send + 'static,
+    B: crate::Body + Send + Unpin + 'static,
     B::Data: Send,
     B::Error: Into<Error> + Send + Sync,
 {
@@ -94,10 +94,9 @@ where
             // ish, so we just build a one-off client for the connection.
             // There's no real reason to hold the client for re-use.
             debug!(use_absolute_form, is_missing_host, "Using one-off client");
-            hyper::Client::builder()
+            hyper_util::client::legacy::Client::builder(TracingExecutor)
                 .pool_max_idle_per_host(0)
                 .set_host(use_absolute_form)
-                .executor(TracingExecutor)
                 .build(HyperConnect::new(
                     self.connect.clone(),
                     self.target.clone(),
@@ -120,11 +119,10 @@ where
             if client.is_none() {
                 debug!(use_absolute_form, "Caching new client");
                 *client = Some(
-                    hyper::Client::builder()
+                    hyper_util::client::legacy::Client::builder(TracingExecutor)
                         .pool_max_idle_per_host(self.pool.max_idle)
                         .pool_idle_timeout(self.pool.idle_timeout)
                         .set_host(use_absolute_form)
-                        .executor(TracingExecutor)
                         .build(HyperConnect::new(
                             self.connect.clone(),
                             self.target.clone(),

linkerd/proxy/http/src/h2.rs

@@ -55,7 +55,7 @@ where
     C::Connection: Send + Unpin + 'static,
     C::Metadata: Send,
     C::Future: Send + 'static,
-    B: Body + Send + 'static,
+    B: Body + Send + Unpin + 'static,
     B::Data: Send,
     B::Error: Into<Error> + Send + Sync,
 {
@@ -147,7 +147,7 @@ where
     B::Data: Send,
     B::Error: Into<Error> + Send + Sync,
 {
-    type Response = http::Response<hyper::Body>;
+    type Response = http::Response<hyper::body::Incoming>;
     type Error = hyper::Error;
     type Future = Pin<Box<dyn Send + Future<Output = Result<Self::Response, Self::Error>>>>;
 

linkerd/proxy/http/src/orig_proto.rs

@@ -1,6 +1,7 @@
 use super::{h1, h2, Body};
 use futures::prelude::*;
 use http::header::{HeaderValue, TRANSFER_ENCODING};
+use http_body::Frame;
 use linkerd_error::{Error, Result};
 use linkerd_http_box::BoxBody;
 use linkerd_stack::{layer, MakeConnection, Service};
@@ -56,7 +57,7 @@ where
     C: MakeConnection<(crate::Variant, T)> + Clone + Send + Sync + 'static,
     C::Connection: Unpin + Send,
     C::Future: Unpin + Send + 'static,
-    B: crate::Body + Send + 'static,
+    B: crate::Body + Send + Unpin + 'static,
     B::Data: Send,
     B::Error: Into<Error> + Send + Sync,
 {
@@ -211,23 +212,13 @@ where
         self.inner.is_end_stream()
     }
 
-    fn poll_data(
+    fn poll_frame(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
+    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
         self.project()
             .inner
-            .poll_data(cx)
-            .map_err(downgrade_h2_error)
-    }
-
-    fn poll_trailers(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
-        self.project()
-            .inner
-            .poll_trailers(cx)
+            .poll_frame(cx)
             .map_err(downgrade_h2_error)
     }
 

linkerd/proxy/http/src/server.rs

@@ -1,8 +1,7 @@
-use crate::{
-    client_handle::SetClientHandle, h2, BoxBody, BoxRequest, ClientHandle, TracingExecutor, Variant,
-};
+use crate::{client_handle::SetClientHandle, h2, BoxBody, ClientHandle, TracingExecutor, Variant};
 use linkerd_error::Error;
-use linkerd_io::{self as io, PeerAddr};
+use linkerd_http_upgrade::glue::UpgradeBody;
+use linkerd_io::PeerAddr;
 use linkerd_stack::{layer, ExtractParam, NewService};
 use std::{
     future::Future,
@@ -126,13 +125,22 @@ where
 
 impl<I, N, S> Service<I> for ServeHttp<N>
 where
-    I: io::AsyncRead + io::AsyncWrite + PeerAddr + Send + Unpin + 'static,
+    I: hyper::rt::Read + hyper::rt::Write + PeerAddr + Send + Unpin + 'static,
     N: NewService<ClientHandle, Service = S> + Send + 'static,
-    S: Service<http::Request<BoxBody>, Response = http::Response<BoxBody>, Error = Error>
-        + Unpin
+    S: Service<
+            http::Request<hyper::body::Incoming>,
+            Response = http::Response<BoxBody>,
+            Error = Error,
+        > + Service<
+            http::Request<UpgradeBody<hyper::body::Incoming>>,
+            Response = http::Response<BoxBody>,
+            Error = Error,
+        > + Clone
         + Send
+        + Unpin
         + 'static,
-    S::Future: Send + 'static,
+    <S as Service<http::Request<hyper::body::Incoming>>>::Future: Send + 'static,
+    <S as Service<http::Request<UpgradeBody<hyper::body::Incoming>>>>::Future: Send + 'static,
 {
     type Response = ();
     type Error = Error;
@@ -162,10 +170,8 @@ where
                 match version {
                     Variant::Http1 => {
                         // Enable support for HTTP upgrades (CONNECT and websockets).
-                        let svc = linkerd_http_upgrade::upgrade::Service::new(
-                            BoxRequest::new(svc),
-                            drain.clone(),
-                        );
+                        let svc = linkerd_http_upgrade::upgrade::Service::new(svc, drain.clone());
+                        let svc = hyper_util::service::TowerToHyperService::new(svc);
                         let mut conn = http1.serve_connection(io, svc).with_upgrades();
 
                         tokio::select! {
@@ -187,7 +193,8 @@ where
                     }
 
                     Variant::H2 => {
-                        let mut conn = http2.serve_connection(io, BoxRequest::new(svc));
+                        let svc = hyper_util::service::TowerToHyperService::new(svc);
+                        let mut conn = http2.serve_connection(io, svc);
 
                         tokio::select! {
                             res = &mut conn => {