refactor: move away from legacy hyper::body::HttpBody

this is an incremental step away from hyper 0.14's request and response
body interfaces, and towards the 1.0 body types. see
<linkerd/linkerd2#8733> for more information
about upgrading to hyper 1.0.

hyper 0.14 provides a `hyper::body::Body` that is removed in the 1.0
interface. `hyper-util` now provides a workable default body type. hyper
0.14 reëxports `http_body::Body` as `HttpBody`. hyper 1.0 reëxports this
trait as `hyper::body::Body` without any renaming.

this commit moves application code away from hyper's legacy `Body` type
and the `HttpBody` trait alias. this commit moves assorted interfaces
towards the boxed `BoxBody` type instead. when possible, code is tweaked
such that it refers to the reëxport in `linkerd-proxy-http`, rather than
directly through `hyper`.

NB: this commit is based upon #3466.

Signed-off-by: katelyn martin <[email protected]>

Cargo.lock

@@ -969,6 +969,7 @@ version = "0.1.0"
 dependencies = [
  "futures",
  "http",
+ "http-body",
  "hyper",
  "pin-project",
  "tokio",
@@ -1312,6 +1313,7 @@ dependencies = [
 name = "linkerd-app-admin"
 version = "0.1.0"
 dependencies = [
+ "bytes",
  "deflate",
  "futures",
  "http",
@@ -1999,7 +2001,9 @@ version = "0.1.0"
 dependencies = [
  "deflate",
  "http",
+ "http-body",
  "hyper",
+ "linkerd-http-box",
  "linkerd-stack",
  "linkerd-system",
  "parking_lot",
@@ -2335,8 +2339,10 @@ dependencies = [
 name = "linkerd-proxy-tap"
 version = "0.1.0"
 dependencies = [
+ "bytes",
  "futures",
  "http",
+ "http-body",
  "hyper",
  "ipnet",
  "linkerd-conditional",

hyper-balance/Cargo.toml

@@ -9,6 +9,7 @@ publish = false
 [dependencies]
 futures = { version = "0.3", default-features = false }
 http = { workspace = true }
+http-body = { workspace = true }
 hyper = { workspace = true, features = ["deprecated"] }
 pin-project = "1"
 tower = { version = "0.4", default-features = false, features = ["load"] }

hyper-balance/src/lib.rs

@@ -1,7 +1,7 @@
 #![deny(rust_2018_idioms, clippy::disallowed_methods, clippy::disallowed_types)]
 #![forbid(unsafe_code)]
 
-use hyper::body::HttpBody;
+use http_body::Body;
 use pin_project::pin_project;
 use std::pin::Pin;
 use std::task::{Context, Poll};
@@ -38,7 +38,7 @@ pub struct PendingUntilEosBody<T, B> {
 
 impl<T, B> TrackCompletion<T, http::Response<B>> for PendingUntilFirstData
 where
-    B: HttpBody,
+    B: Body,
 {
     type Output = http::Response<PendingUntilFirstDataBody<T, B>>;
 
@@ -59,7 +59,7 @@ where
 
 impl<T, B> TrackCompletion<T, http::Response<B>> for PendingUntilEos
 where
-    B: HttpBody,
+    B: Body,
 {
     type Output = http::Response<PendingUntilEosBody<T, B>>;
 
@@ -80,7 +80,7 @@ where
 
 impl<T, B> Default for PendingUntilFirstDataBody<T, B>
 where
-    B: HttpBody + Default,
+    B: Body + Default,
 {
     fn default() -> Self {
         Self {
@@ -90,9 +90,9 @@ where
     }
 }
 
-impl<T, B> HttpBody for PendingUntilFirstDataBody<T, B>
+impl<T, B> Body for PendingUntilFirstDataBody<T, B>
 where
-    B: HttpBody,
+    B: Body,
     T: Send + 'static,
 {
     type Data = B::Data;
@@ -138,7 +138,7 @@ where
 
 impl<T, B> Default for PendingUntilEosBody<T, B>
 where
-    B: HttpBody + Default,
+    B: Body + Default,
 {
     fn default() -> Self {
         Self {
@@ -148,7 +148,7 @@ where
     }
 }
 
-impl<T: Send + 'static, B: HttpBody> HttpBody for PendingUntilEosBody<T, B> {
+impl<T: Send + 'static, B: Body> Body for PendingUntilEosBody<T, B> {
     type Data = B::Data;
     type Error = B::Error;
 
@@ -198,7 +198,7 @@ impl<T: Send + 'static, B: HttpBody> HttpBody for PendingUntilEosBody<T, B> {
 mod tests {
     use super::{PendingUntilEos, PendingUntilFirstData};
     use futures::future::poll_fn;
-    use hyper::body::HttpBody;
+    use http_body::Body;
     use std::collections::VecDeque;
     use std::io::Cursor;
     use std::pin::Pin;
@@ -429,7 +429,7 @@ mod tests {
 
     #[derive(Default)]
     struct TestBody(VecDeque<&'static str>, Option<http::HeaderMap>);
-    impl HttpBody for TestBody {
+    impl Body for TestBody {
         type Data = Cursor<&'static str>;
         type Error = &'static str;
 
@@ -456,7 +456,7 @@ mod tests {
 
     #[derive(Default)]
     struct ErrBody(Option<&'static str>);
-    impl HttpBody for ErrBody {
+    impl Body for ErrBody {
         type Data = Cursor<&'static str>;
         type Error = &'static str;
 

linkerd/app/admin/Cargo.toml

@@ -15,6 +15,7 @@ pprof = ["deflate", "dep:pprof"]
 log-streaming = ["linkerd-tracing/stream"]
 
 [dependencies]
+bytes = "1"
 deflate = { version = "1", optional = true, features = ["gzip"] }
 http = { workspace = true }
 http-body = { workspace = true }

linkerd/app/admin/src/server.rs

@@ -12,13 +12,9 @@
 
 use futures::future::{self, TryFutureExt};
 use http::StatusCode;
-use hyper::{
-    body::{Body, HttpBody},
-    Request, Response,
-};
 use linkerd_app_core::{
     metrics::{self as metrics, FmtMetrics},
-    proxy::http::ClientHandle,
+    proxy::http::{Body, BoxBody, ClientHandle, Request, Response},
     trace, Error, Result,
 };
 use std::{
@@ -45,7 +41,7 @@ pub struct Admin<M> {
     pprof: Option<crate::pprof::Pprof>,
 }
 
-pub type ResponseFuture = Pin<Box<dyn Future<Output = Result<Response<Body>>> + Send + 'static>>;
+pub type ResponseFuture = Pin<Box<dyn Future<Output = Result<Response<BoxBody>>> + Send + 'static>>;
 
 impl<M> Admin<M> {
     pub fn new(
@@ -73,30 +69,30 @@ impl<M> Admin<M> {
         self
     }
 
-    fn ready_rsp(&self) -> Response<Body> {
+    fn ready_rsp(&self) -> Response<BoxBody> {
         if self.ready.is_ready() {
             Response::builder()
                 .status(StatusCode::OK)
                 .header(http::header::CONTENT_TYPE, "text/plain")
-                .body("ready\n".into())
+                .body(BoxBody::new::<String>("ready\n".into()))
                 .expect("builder with known status code must not fail")
         } else {
             Response::builder()
                 .status(StatusCode::SERVICE_UNAVAILABLE)
-                .body("not ready\n".into())
+                .body(BoxBody::new::<String>("not ready\n".into()))
                 .expect("builder with known status code must not fail")
         }
     }
 
-    fn live_rsp() -> Response<Body> {
+    fn live_rsp() -> Response<BoxBody> {
         Response::builder()
             .status(StatusCode::OK)
             .header(http::header::CONTENT_TYPE, "text/plain")
-            .body("live\n".into())
+            .body(BoxBody::new::<String>("live\n".into()))
             .expect("builder with known status code must not fail")
     }
 
-    fn env_rsp<B>(req: Request<B>) -> Response<Body> {
+    fn env_rsp<B>(req: Request<B>) -> Response<BoxBody> {
         use std::{collections::HashMap, env, ffi::OsString};
 
         if req.method() != http::Method::GET {
@@ -142,56 +138,60 @@ impl<M> Admin<M> {
         json::json_rsp(&env)
     }
 
-    fn shutdown(&self) -> Response<Body> {
+    fn shutdown(&self) -> Response<BoxBody> {
         if !self.enable_shutdown {
             return Response::builder()
                 .status(StatusCode::NOT_FOUND)
                 .header(http::header::CONTENT_TYPE, "text/plain")
-                .body("shutdown endpoint is not enabled\n".into())
+                .body(BoxBody::new::<String>(
+                    "shutdown endpoint is not enabled\n".into(),
+                ))
                 .expect("builder with known status code must not fail");
         }
         if self.shutdown_tx.send(()).is_ok() {
             Response::builder()
                 .status(StatusCode::OK)
                 .header(http::header::CONTENT_TYPE, "text/plain")
-                .body("shutdown\n".into())
+                .body(BoxBody::new::<String>("shutdown\n".into()))
                 .expect("builder with known status code must not fail")
         } else {
             Response::builder()
                 .status(StatusCode::INTERNAL_SERVER_ERROR)
                 .header(http::header::CONTENT_TYPE, "text/plain")
-                .body("shutdown listener dropped\n".into())
+                .body(BoxBody::new::<String>("shutdown listener dropped\n".into()))
                 .expect("builder with known status code must not fail")
         }
     }
 
-    fn internal_error_rsp(error: impl ToString) -> http::Response<Body> {
+    fn internal_error_rsp(error: impl ToString) -> http::Response<BoxBody> {
         http::Response::builder()
             .status(http::StatusCode::INTERNAL_SERVER_ERROR)
             .header(http::header::CONTENT_TYPE, "text/plain")
-            .body(error.to_string().into())
+            .body(BoxBody::new(error.to_string()))
             .expect("builder with known status code should not fail")
     }
 
-    fn not_found() -> Response<Body> {
+    fn not_found() -> Response<BoxBody> {
         Response::builder()
             .status(http::StatusCode::NOT_FOUND)
-            .body(Body::empty())
+            .body(BoxBody::new(hyper::Body::empty()))
             .expect("builder with known status code must not fail")
     }
 
-    fn method_not_allowed() -> Response<Body> {
+    fn method_not_allowed() -> Response<BoxBody> {
         Response::builder()
             .status(http::StatusCode::METHOD_NOT_ALLOWED)
-            .body(Body::empty())
+            .body(BoxBody::new(hyper::Body::empty()))
             .expect("builder with known status code must not fail")
     }
 
-    fn forbidden_not_localhost() -> Response<Body> {
+    fn forbidden_not_localhost() -> Response<BoxBody> {
         Response::builder()
             .status(http::StatusCode::FORBIDDEN)
             .header(http::header::CONTENT_TYPE, "text/plain")
-            .body("Requests are only permitted from localhost.".into())
+            .body(BoxBody::new::<String>(
+                "Requests are only permitted from localhost.".into(),
+            ))
             .expect("builder with known status code must not fail")
     }
 
@@ -215,11 +215,11 @@ impl<M> Admin<M> {
 impl<M, B> tower::Service<http::Request<B>> for Admin<M>
 where
     M: FmtMetrics,
-    B: HttpBody + Send + 'static,
+    B: Body + Send + 'static,
     B::Error: Into<Error>,
     B::Data: Send,
 {
-    type Response = http::Response<Body>;
+    type Response = http::Response<BoxBody>;
     type Error = Error;
     type Future = ResponseFuture;
 
@@ -331,7 +331,7 @@ mod tests {
                 let r = Request::builder()
                     .method(Method::GET)
                     .uri("http://0.0.0.0/ready")
-                    .body(Body::empty())
+                    .body(hyper::Body::empty())
                     .unwrap();
                 let f = admin.clone().oneshot(r);
                 timeout(TIMEOUT, f).await.expect("timeout").expect("call")

linkerd/app/admin/src/server/json.rs

@@ -3,12 +3,14 @@ pub(in crate::server) static JSON_HEADER_VAL: HeaderValue = HeaderValue::from_st
 
 use hyper::{
     header::{self, HeaderValue},
-    Body, StatusCode,
+    StatusCode,
 };
+use linkerd_app_core::proxy::http::BoxBody;
+
 pub(crate) fn json_error_rsp(
     error: impl ToString,
     status: http::StatusCode,
-) -> http::Response<Body> {
+) -> http::Response<BoxBody> {
     mk_rsp(
         status,
         &serde_json::json!({
@@ -18,11 +20,12 @@ pub(crate) fn json_error_rsp(
     )
 }
 
-pub(crate) fn json_rsp(val: &impl serde::Serialize) -> http::Response<Body> {
+pub(crate) fn json_rsp(val: &impl serde::Serialize) -> http::Response<BoxBody> {
     mk_rsp(StatusCode::OK, val)
 }
 
-pub(crate) fn accepts_json<B>(req: &http::Request<B>) -> Result<(), http::Response<Body>> {
+#[allow(clippy::result_large_err)]
+pub(crate) fn accepts_json<B>(req: &http::Request<B>) -> Result<(), http::Response<BoxBody>> {
     if let Some(accept) = req.headers().get(header::ACCEPT) {
         let accept = match std::str::from_utf8(accept.as_bytes()) {
             Ok(accept) => accept,
@@ -41,26 +44,28 @@ pub(crate) fn accepts_json<B>(req: &http::Request<B>) -> Result<(), http::Respon
             tracing::warn!(?accept, "Accept header will not accept 'application/json'");
             return Err(http::Response::builder()
                 .status(StatusCode::NOT_ACCEPTABLE)
-                .body(JSON_MIME.into())
+                .body(BoxBody::new::<String>(JSON_MIME.into()))
                 .expect("builder with known status code must not fail"));
         }
     }
 
     Ok(())
 }
 
-fn mk_rsp(status: StatusCode, val: &impl serde::Serialize) -> http::Response<Body> {
+fn mk_rsp(status: StatusCode, val: &impl serde::Serialize) -> http::Response<BoxBody> {
     match serde_json::to_vec(val) {
         Ok(json) => http::Response::builder()
             .status(status)
             .header(header::CONTENT_TYPE, JSON_HEADER_VAL.clone())
-            .body(json.into())
+            .body(BoxBody::new(http_body::Full::new(bytes::Bytes::from(json))))
             .expect("builder with known status code must not fail"),
         Err(error) => {
             tracing::warn!(?error, "failed to serialize JSON value");
             http::Response::builder()
                 .status(StatusCode::INTERNAL_SERVER_ERROR)
-                .body(format!("failed to serialize JSON value: {error}").into())
+                .body(BoxBody::new::<String>(format!(
+                    "failed to serialize JSON value: {error}"
+                )))
                 .expect("builder with known status code must not fail")
         }
     }