Skip to content
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.

Sign up for GitHub

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

http: Teardown serverside connections on error #747

Merged
merged 8 commits into from
Nov 18, 2020
Merged

http: Teardown serverside connections on error #747

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
af6d446
outbound: add test for fwding conn errors to HTTP clients (#745)
hawkw Nov 17, 2020
29e754f
http: Add a means to gracefully teardown servers
olix0r Nov 18, 2020
ffdb86b
Modify the HTTP error handler to teardown connections on error
olix0r Nov 18, 2020
d667f24
outbound: Remove the FailOnError layer
olix0r Nov 18, 2020
da3471d
http: Teardown serverside connections on error
olix0r Nov 18, 2020
94c4d7f
fixup Cargo.toml
olix0r Nov 18, 2020
c0dde92
serverside => server-side
olix0r Nov 18, 2020
dd078ac
Use a Pin<Box<..>> instead of a Closed future
olix0r Nov 18, 2020
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
56 changes: 12 additions & 44 deletions Cargo.lock
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -732,9 +732,9 @@ dependencies = [

[[package]]
name = "lazy_static"
version = "1.3.0"
version = "1.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bc5729f27f159ddd61f4df6228e827e86643d4d3e7c32183cb30a1c08f604a14"
checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646"

[[package]]
name = "libc"
Expand Down Expand Up @@ -916,6 +916,7 @@ dependencies = [
"bytes 0.5.4",
"futures 0.3.5",
"http 0.2.1",
"hyper",
"indexmap",
"ipnet 1.0.0",
"linkerd2-app-core",
Expand Down Expand Up @@ -979,7 +980,7 @@ dependencies = [
"futures 0.3.5",
"linkerd2-error",
"linkerd2-stack",
"parking_lot 0.11.0",
"parking_lot",
"tokio",
"tower",
"tracing",
Expand Down Expand Up @@ -1548,15 +1549,6 @@ dependencies = [
"tracing-subscriber",
]

[[package]]
name = "lock_api"
version = "0.3.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c4da24a77a3d8a6d4862d95f72e6fdb9c09a643ecdb402d754004a557f2bec75"
dependencies = [
"scopeguard",
]

[[package]]
name = "lock_api"
version = "0.4.1"
Expand Down Expand Up @@ -1808,39 +1800,15 @@ dependencies = [
"tonic-build",
]

[[package]]
name = "parking_lot"
version = "0.10.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d3a704eb390aafdc107b0e392f56a82b668e3a71366993b5340f5833fd62505e"
dependencies = [
"lock_api 0.3.4",
"parking_lot_core 0.7.2",
]

[[package]]
name = "parking_lot"
version = "0.11.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a4893845fa2ca272e647da5d0e46660a314ead9c2fdd9a883aabc32e481a8733"
dependencies = [
"instant",
"lock_api 0.4.1",
"parking_lot_core 0.8.0",
]

[[package]]
name = "parking_lot_core"
version = "0.7.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d58c7c768d4ba344e3e8d72518ac13e259d7c7ade24167003b8488e10b6740a3"
dependencies = [
"cfg-if 0.1.10",
"cloudabi 0.0.3",
"libc",
"redox_syscall",
"smallvec",
"winapi 0.3.8",
"lock_api",
"parking_lot_core",
]

[[package]]
Expand Down Expand Up @@ -2522,9 +2490,9 @@ dependencies = [

[[package]]
name = "tokio"
version = "0.2.22"
version = "0.2.23"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5d34ca54d84bf2b5b4d7d31e901a8464f7b60ac145a284fba25ceb801f2ddccd"
checksum = "a6d7ad61edd59bfcc7e80dababf0f4aed2e6d5e0ba1659356ae889752dfc12ff"
dependencies = [
"bytes 0.5.4",
"fnv",
Expand All @@ -2537,7 +2505,7 @@ dependencies = [
"mio-named-pipes",
"mio-uds",
"num_cpus",
"parking_lot 0.10.2",
"parking_lot",
"pin-project-lite",
"signal-hook-registry",
"slab",
Expand Down Expand Up @@ -2578,9 +2546,9 @@ dependencies = [

[[package]]
name = "tokio-macros"
version = "0.2.5"
version = "0.2.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f0c3acc6aa564495a0f2e1d59fab677cd7f81a19994cfc7f3ad0e64301560389"
checksum = "e44da00bfc73a25f814cd8d7e57a68a5c31b74b3152a0a1d1f590c97ed06265a"
dependencies = [
"proc-macro2",
"quote",
Expand Down Expand Up @@ -2825,7 +2793,7 @@ dependencies = [
"chrono",
"lazy_static",
"matchers",
"parking_lot 0.11.0",
"parking_lot",
"regex 1.3.9",
"serde",
"serde_json",
Expand Down
23 changes: 17 additions & 6 deletions linkerd/app/core/src/admin/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,11 @@
//! * `/ready` -- returns 200 when the proxy is ready to participate in meshed traffic.

use crate::{
proxy::http::{ClientAddr, SetClientAddr},
proxy::http::{ClientHandle, SetClientHandle},
svc, trace,
transport::{io, tls},
};
use futures::{future, TryFutureExt};
use futures::future;
use http::StatusCode;
use hyper::{Body, Request, Response};
use linkerd2_error::{Error, Never};
Expand Down Expand Up @@ -98,8 +98,8 @@ impl<M> Admin<M> {

fn client_is_localhost<B>(req: &Request<B>) -> bool {
req.extensions()
.get::<ClientAddr>()
.map(|a| a.as_ref().ip().is_loopback())
.get::<ClientHandle>()
.map(|a| a.addr.ip().is_loopback())
.unwrap_or(false)
}
}
Expand Down Expand Up @@ -178,9 +178,20 @@ impl<M: FmtMetrics + Clone + Send + 'static> svc::Service<io::BoxedIo> for Serve
// Since the `/proxy-log-level` controls access based on the
// client's IP address, we wrap the service with a new service
// that adds the remote IP as a request extension.
let svc = SetClientAddr::new(meta.addrs.peer(), svc.clone());
let (svc, closed) = SetClientHandle::new(meta.addrs.peer(), svc.clone());
let mut conn = server.serve_connection(io, svc);

Box::pin(async move {
tokio::select! {
res = &mut conn => res?,
() = closed.closed() => {
Pin::new(&mut conn).graceful_shutdown();
conn.await?;
}
}

Box::pin(server.serve_connection(io, svc).map_err(Into::into))
Ok(())
})
}
}

Expand Down
83 changes: 54 additions & 29 deletions linkerd/app/core/src/errors.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ use linkerd2_error::Error;
use linkerd2_error_metrics as metrics;
use linkerd2_error_respond as respond;
pub use linkerd2_error_respond::RespondLayer;
use linkerd2_proxy_http::HasH2Reason;
use linkerd2_proxy_http::{client_handle::Close, ClientHandle, HasH2Reason};
use linkerd2_timeout::{error::ResponseTimeout, FailFastError};
use pin_project::pin_project;
use std::pin::Pin;
Expand Down Expand Up @@ -51,10 +51,11 @@ pub enum Reason {
#[derive(Copy, Clone, Debug)]
pub struct NewRespond(());

#[derive(Copy, Clone, Debug)]
pub enum Respond {
Http1(http::Version),
Http2 { is_grpc: bool },
#[derive(Clone, Debug)]
pub struct Respond {
version: http::Version,
is_grpc: bool,
close: Option<Close>,
}

#[pin_project(project = ResponseBodyProj)]
Expand Down Expand Up @@ -139,30 +140,39 @@ impl<ReqB, RspB: Default + hyper::body::HttpBody>
type Respond = Respond;

fn new_respond(&self, req: &http::Request<ReqB>) -> Self::Respond {
let close = req
.extensions()
.get::<ClientHandle>()
.map(|h| h.close.clone());
match req.version() {
http::Version::HTTP_2 => {
let is_grpc = req
.headers()
.get(http::header::CONTENT_TYPE)
.and_then(|v| v.to_str().ok().map(|s| s.starts_with("application/grpc")))
.unwrap_or(false);
Respond::Http2 { is_grpc }
Respond {
is_grpc,
close,
version: http::Version::HTTP_2,
}
}
version => Respond::Http1(version),
version => Respond {
version,
close,
is_grpc: false,
},
}
}
}

impl<RspB: Default + hyper::body::HttpBody> respond::Respond<http::Response<RspB>> for Respond {
type Response = http::Response<ResponseBody<RspB>>;

fn respond(
&self,
reseponse: Result<http::Response<RspB>, Error>,
) -> Result<Self::Response, Error> {
match reseponse {
fn respond(&self, res: Result<http::Response<RspB>, Error>) -> Result<Self::Response, Error> {
match res {
Ok(response) => Ok(response.map(|b| match *self {
Respond::Http2 { is_grpc } if is_grpc == true => ResponseBody::Grpc {
Respond { is_grpc: true, .. } => ResponseBody::Grpc {
inner: b,
trailers: None,
},
Expand All @@ -171,33 +181,36 @@ impl<RspB: Default + hyper::body::HttpBody> respond::Respond<http::Response<RspB
Err(error) => {
warn!("Failed to proxy request: {}", error);

if let Respond::Http2 { is_grpc } = self {
if self.version == http::Version::HTTP_2 {
if let Some(reset) = error.h2_reason() {
debug!(%reset, "Propagating HTTP2 reset");
return Err(error);
}
}

if *is_grpc {
let mut rsp = http::Response::builder()
.version(http::Version::HTTP_2)
.header(http::header::CONTENT_LENGTH, "0")
.body(ResponseBody::default())
.expect("app::errors response is valid");
let code = set_grpc_status(&*error, rsp.headers_mut());
debug!(?code, "Handling error with gRPC status");
return Ok(rsp);
// Gracefully teardown the serverside connection.
if should_teardown_connection(&*error) {
if let Some(c) = self.close.as_ref() {
debug!("Closing serverside connection");
c.close();
}
}

let version = match self {
Respond::Http1(ref version) => version.clone(),
Respond::Http2 { .. } => http::Version::HTTP_2,
};
if self.is_grpc {
let mut rsp = http::Response::builder()
.version(http::Version::HTTP_2)
.header(http::header::CONTENT_LENGTH, "0")
.body(ResponseBody::default())
.expect("app::errors response is valid");
let code = set_grpc_status(&*error, rsp.headers_mut());
debug!(?code, "Handling error with gRPC status");
return Ok(rsp);
}

let status = http_status(&*error);
debug!(%status, ?version, "Handling error with HTTP response");
debug!(%status, version = ?self.version, "Handling error with HTTP response");
Ok(http::Response::builder()
.version(version)
.version(self.version)
.status(status)
.header(http::header::CONTENT_LENGTH, "0")
.body(ResponseBody::default())
Expand All @@ -207,6 +220,18 @@ impl<RspB: Default + hyper::body::HttpBody> respond::Respond<http::Response<RspB
}
}

fn should_teardown_connection(error: &(dyn std::error::Error + 'static)) -> bool {
if error.is::<ResponseTimeout>() {
false
} else if error.is::<tower::timeout::error::Elapsed>() {
false
} else if let Some(e) = error.source() {
should_teardown_connection(e)
Comment on lines +228 to +229
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it occurs to me that it might be more efficient to do source-chain traversal with a loop rather than recursively in these functions. but, i don't think we ever have a lot of errors that are more than one or two sources deep, so i don't think this actually matters.

} else {
true
}
}

fn http_status(error: &(dyn std::error::Error + 'static)) -> StatusCode {
if let Some(HttpError { http, .. }) = error.downcast_ref::<HttpError>() {
*http
Expand Down
3 changes: 2 additions & 1 deletion linkerd/app/outbound/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,8 +30,9 @@ features = [
]

[dev-dependencies]
hyper = "0.13"
ipnet = "1.0"
linkerd2-io = { path = "../../io", features = ["tokio-test"] }
linkerd2-app-test = { path = "../test" }
linkerd2-io = { path = "../../io", features = ["tokio-test"] }
tokio = { version = "0.2", features = ["full", "macros"]}
tracing-futures = "0.2"
8 changes: 1 addition & 7 deletions linkerd/app/outbound/src/http/logical.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -123,13 +123,7 @@ where
.push_switch(
Logical::should_resolve,
svc::stack(endpoint)
.push_on_response(
svc::layers()
.push(svc::layer::mk(
svc::stack::FailOnError::<std::io::Error, S>::new,
))
.box_http_request(),
)
.push_on_response(svc::layers().box_http_request())
.push_map_target(Endpoint::from_logical(
ReasonForNoPeerName::NotProvidedByServiceDiscovery,
))
Expand Down
9 changes: 5 additions & 4 deletions linkerd/app/outbound/src/http/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,14 +2,17 @@ pub mod endpoint;
pub mod logical;
mod require_identity_on_endpoint;

#[cfg(test)]
mod tests;

use crate::tcp;
use indexmap::IndexMap;
pub use linkerd2_app_core::proxy::http::*;
use linkerd2_app_core::{
dst, profiles,
proxy::{
api_resolve::ProtocolHint,
http::{self, CanOverrideAuthority, ClientAddr},
http::{self, CanOverrideAuthority, ClientHandle},
identity, tap,
},
transport::tls,
Expand Down Expand Up @@ -71,9 +74,7 @@ impl CanOverrideAuthority for Endpoint {

impl tap::Inspect for Endpoint {
fn src_addr<B>(&self, req: &http::Request<B>) -> Option<SocketAddr> {
req.extensions()
.get::<ClientAddr>()
.map(|c| c.as_ref().clone())
req.extensions().get::<ClientHandle>().map(|c| c.addr)
}

fn src_tls<'a, B>(
Expand Down
Loading