Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Dec 17, 2024
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions Cargo.lock
Original file line number Diff line number Diff line change
@@ -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",
1 change: 1 addition & 0 deletions hyper-balance/Cargo.toml
Original file line number Diff line number Diff line change
@@ -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"] }
22 changes: 11 additions & 11 deletions hyper-balance/src/lib.rs
Original file line number Diff line number Diff line change
@@ -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;

1 change: 1 addition & 0 deletions linkerd/app/admin/Cargo.toml
Original file line number Diff line number Diff line change
@@ -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 }
52 changes: 25 additions & 27 deletions linkerd/app/admin/src/server.rs
Original file line number Diff line number Diff line change
@@ -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::from_static("ready\n"))
.expect("builder with known status code must not fail")
} else {
Response::builder()
.status(StatusCode::SERVICE_UNAVAILABLE)
.body("not ready\n".into())
.body(BoxBody::from_static("not ready\n"))
.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::from_static("live\n"))
.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,58 @@ 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::from_static("shutdown endpoint is not enabled\n"))
.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::from_static("shutdown\n"))
.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::from_static("shutdown listener dropped\n"))
.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 +213,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 +329,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")
32 changes: 22 additions & 10 deletions linkerd/app/admin/src/server/json.rs
Original file line number Diff line number Diff line change
@@ -1,14 +1,17 @@
static JSON_MIME: &str = "application/json";
pub(in crate::server) static JSON_HEADER_VAL: HeaderValue = HeaderValue::from_static(JSON_MIME);

use bytes::Bytes;
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 +21,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 +45,34 @@ 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::from_static(JSON_MIME))
.expect("builder with known status code must not fail"));
}
}

Ok(())
}

fn mk_rsp(status: StatusCode, val: &impl serde::Serialize) -> http::Response<Body> {
match serde_json::to_vec(val) {
Ok(json) => http::Response::builder()
fn mk_rsp(status: StatusCode, val: &impl serde::Serialize) -> http::Response<BoxBody> {
// Serialize the value into JSON, and then place the bytes in a boxed response body.
let json = serde_json::to_vec(val)
.map(Bytes::from)
.map(http_body::Full::new)
.map(BoxBody::new);

match json {
Ok(body) => http::Response::builder()
.status(status)
.header(header::CONTENT_TYPE, JSON_HEADER_VAL.clone())
.body(json.into())
.body(body)
.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(format!(
"failed to serialize JSON value: {error}"
)))
.expect("builder with known status code must not fail")
}
}
Loading