outbound: Fix incorrect l5d-proxy-connection logs (#2344)

We've received reports of proxies inexplicibly emiting log lines
including "Received unmeshed response with l5d-proxy-connection set".
These messages may arise when the endpoint stack returns a synthesized
response.

Furthermore, we have a note in the code explaining that this connection
closure logic should not apply to load balanced requests, though it
currently is.

This change fixes both of these issues:

* The proxy_connection_close module is renamed to
  handle_proxy_error_headers.
* It is now only applied in the endpoint stack. It doesn't make any
  sense for it to be applied in the server stack, since we'll already
  have cleared any headers set by peers. Removing this module prevents
  the application of teardown logic on synthetic responses.
* `NewHandleProxyErrorHeaders` is now configured by a `Closable`
  parameter that determines whether teardown logic should be applied.
  This parameter is only enabled when forwarding to a single endpoint.
  No teardown logic is applied when load balancing.
* In a future change, we should stop emitting l5d-proxy-connection when
  synthesizing outbound responses.

Author: olix0r

linkerd/app/outbound/src/http.rs

@@ -1,7 +1,4 @@
-use self::{
-    proxy_connection_close::ProxyConnectionClose, require_id_header::NewRequireIdentity,
-    strip_proxy_error::NewStripProxyError,
-};
+use self::require_id_header::NewRequireIdentity;
 use crate::Outbound;
 use linkerd_app_core::{
     proxy::{
@@ -17,12 +14,11 @@ use tokio::sync::watch;
 
 pub mod concrete;
 mod endpoint;
+mod handle_proxy_error_headers;
 pub mod logical;
-mod proxy_connection_close;
 mod require_id_header;
 mod retry;
 mod server;
-mod strip_proxy_error;
 
 pub use self::logical::{policy, profile, LogicalAddr, Routes};
 pub(crate) use self::require_id_header::IdentityRequired;

linkerd/app/outbound/src/http/concrete.rs

@@ -1,7 +1,7 @@
 //! A stack that (optionally) resolves a service to a set of endpoint replicas
 //! and distributes HTTP requests among them.
 
-use super::{balance, client};
+use super::{balance, client, handle_proxy_error_headers};
 use crate::{http, stack_labels, Outbound};
 use linkerd_app_core::{
     metrics, profiles,
@@ -41,6 +41,7 @@ pub struct Endpoint<T> {
     is_local: bool,
     metadata: Metadata,
     parent: T,
+    close_server_connection_on_remote_proxy_error: bool,
 }
 
 /// A target configuring a load balancer stack.
@@ -129,6 +130,10 @@ impl<N> Outbound<N> {
                             metadata,
                             is_local,
                             parent: target.parent,
+                            // We don't close server-side connections when we
+                            // get `l5d-proxy-connection: close` response headers
+                            // going through the balancer.
+                            close_server_connection_on_remote_proxy_error: false,
                         }
                     }
                 })
@@ -158,6 +163,7 @@ impl<N> Outbound<N> {
                                     addr,
                                     metadata,
                                     parent,
+                                    close_server_connection_on_remote_proxy_error: true,
                                 }
                             }),
                         })
@@ -220,6 +226,14 @@ impl<T> svc::Param<Option<http::AuthorityOverride>> for Endpoint<T> {
     }
 }
 
+impl<T> svc::Param<handle_proxy_error_headers::CloseServerConnection> for Endpoint<T> {
+    fn param(&self) -> handle_proxy_error_headers::CloseServerConnection {
+        handle_proxy_error_headers::CloseServerConnection(
+            self.close_server_connection_on_remote_proxy_error,
+        )
+    }
+}
+
 impl<T> svc::Param<transport::labels::Key> for Endpoint<T>
 where
     T: svc::Param<Option<http::uri::Authority>>,

linkerd/app/outbound/src/http/endpoint.rs

@@ -1,6 +1,9 @@
 //! A stack that sends requests to an HTTP endpoint.
 
-use super::{NewRequireIdentity, NewStripProxyError, ProxyConnectionClose};
+use super::{
+    handle_proxy_error_headers::{self, NewHandleProxyErrorHeaders},
+    NewRequireIdentity,
+};
 use crate::{tcp::tagged_transport, Outbound};
 use linkerd_app_core::{
     classify, config, errors, http_tracing, metrics,
@@ -87,9 +90,10 @@ impl<N> Outbound<N> {
     pub fn push_http_endpoint<T, B, NSvc>(self) -> Outbound<svc::ArcNewHttp<T, B>>
     where
         // Http endpoint target.
-        T: svc::Param<http::client::Settings>,
         T: svc::Param<Remote<ServerAddr>>,
+        T: svc::Param<http::client::Settings>,
         T: svc::Param<Option<http::AuthorityOverride>>,
+        T: svc::Param<handle_proxy_error_headers::CloseServerConnection>,
         T: svc::Param<metrics::EndpointLabels>,
         T: svc::Param<tls::ConditionalClientTls>,
         T: tap::Inspect,
@@ -119,17 +123,13 @@ impl<N> Outbound<N> {
                 // actively polled.
                 .push_on_service(svc::layer::mk(svc::SpawnReady::new))
                 .push_new_reconnect(backoff)
-                // Set the TLS status on responses so that the stack can detect whether the request
-                // was sent over a meshed connection.
-                .push_http_response_insert_target::<tls::ConditionalClientTls>()
                 .push(svc::NewMapErr::layer_from_target::<EndpointError, _>())
-                // If the outbound proxy is not configured to emit headers, then strip the
-                // `l5d-proxy-errors` header if set by the peer.
-                .push(NewStripProxyError::layer(config.emit_headers))
-                // Tear down server connections when a peer proxy generates an error.
-                // TODO(ver) this should only be honored when forwarding and not when the connection
-                // is part of a balancer.
-                .push(ProxyConnectionClose::layer())
+                .push_on_service(svc::MapErr::layer_boxed())
+                // Tear down server connections when a peer proxy generates a
+                // response with the `l5d-proxy-connection: close` header. This
+                // is only done when the `Closable` parameter is set to true.
+                // This module always strips error headers from responses.
+                .push(NewHandleProxyErrorHeaders::layer())
                 // Handle connection-level errors eagerly so that we can report 5XX failures in tap
                 // and metrics. HTTP error metrics are not incremented here so that errors are not
                 // double-counted--i.e., endpoint metrics track these responses and error metrics

linkerd/app/outbound/src/http/endpoint/tests.rs

@@ -256,6 +256,12 @@ impl svc::Param<Remote<ServerAddr>> for Endpoint {
     }
 }
 
+impl svc::Param<http::handle_proxy_error_headers::CloseServerConnection> for Endpoint {
+    fn param(&self) -> http::handle_proxy_error_headers::CloseServerConnection {
+        http::handle_proxy_error_headers::CloseServerConnection(true)
+    }
+}
+
 impl svc::Param<tls::ConditionalClientTls> for Endpoint {
     fn param(&self) -> tls::ConditionalClientTls {
         tls::ConditionalClientTls::None(tls::NoClientTls::Disabled)

linkerd/app/outbound/src/http/handle_proxy_error_headers.rs

@@ -0,0 +1,248 @@
+use futures::prelude::*;
+use linkerd_app_core::{
+    errors::respond::{L5D_PROXY_CONNECTION, L5D_PROXY_ERROR},
+    proxy::http::ClientHandle,
+    svc, tls,
+};
+use std::{
+    future::Future,
+    pin::Pin,
+    task::{Context, Poll},
+};
+use tracing::debug;
+
+#[derive(Copy, Clone, Debug)]
+pub struct CloseServerConnection(pub bool);
+
+/// Close the accepted connection if the response from a peer proxy has the
+/// `l5d-proxy-connection: close` header. This means the peer proxy encountered
+/// an inbound connection error with its application and therefore the accepted
+/// connection should be torn down.
+#[derive(Clone, Debug)]
+pub struct NewHandleProxyErrorHeaders<X, N> {
+    inner: N,
+    extract: X,
+}
+
+#[derive(Clone, Debug)]
+pub struct HandleProxyErrorHeaders<S> {
+    inner: S,
+    closable: bool,
+}
+
+#[pin_project::pin_project]
+#[derive(Debug)]
+pub struct ResponseFuture<F> {
+    #[pin]
+    inner: F,
+    client: ClientHandle,
+    closable: bool,
+}
+
+impl<X: Clone, N> NewHandleProxyErrorHeaders<X, N> {
+    fn new(extract: X, inner: N) -> Self {
+        Self { extract, inner }
+    }
+
+    pub fn layer_via(extract: X) -> impl svc::layer::Layer<N, Service = Self> + Clone {
+        svc::layer::mk(move |inner| Self::new(extract.clone(), inner))
+    }
+}
+
+impl<N> NewHandleProxyErrorHeaders<(), N> {
+    pub fn layer() -> impl svc::layer::Layer<N, Service = Self> + Clone {
+        Self::layer_via(())
+    }
+}
+
+impl<T, X, N> svc::NewService<T> for NewHandleProxyErrorHeaders<X, N>
+where
+    X: svc::ExtractParam<CloseServerConnection, T>,
+    X: svc::ExtractParam<tls::ConditionalClientTls, T>,
+    N: svc::NewService<T>,
+{
+    type Service = HandleProxyErrorHeaders<N::Service>;
+
+    #[inline]
+    fn new_service(&self, target: T) -> Self::Service {
+        let CloseServerConnection(closable) = self.extract.extract_param(&target);
+        let is_meshed = matches!(
+            self.extract.extract_param(&target),
+            tls::ConditionalClientTls::Some(_)
+        );
+        let inner = self.inner.new_service(target);
+        HandleProxyErrorHeaders {
+            inner,
+            closable: closable && is_meshed,
+        }
+    }
+}
+
+impl<S, A, B> svc::Service<http::Request<A>> for HandleProxyErrorHeaders<S>
+where
+    S: svc::Service<http::Request<A>, Response = http::Response<B>>,
+    S::Response: Send,
+    S::Future: Send + 'static,
+    B: Send + 'static,
+{
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = ResponseFuture<S::Future>;
+
+    #[inline]
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    #[inline]
+    fn call(&mut self, req: http::Request<A>) -> Self::Future {
+        let client = req
+            .extensions()
+            .get::<ClientHandle>()
+            .cloned()
+            .expect("missing client handle");
+        let inner = self.inner.call(req);
+        ResponseFuture {
+            inner,
+            client,
+            closable: self.closable,
+        }
+    }
+}
+
+impl<B, F: TryFuture<Ok = http::Response<B>>> Future for ResponseFuture<F> {
+    type Output = Result<http::Response<B>, F::Error>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.project();
+        let mut rsp = futures::ready!(this.inner.try_poll(cx))?;
+        if update_response(&mut rsp, *this.closable) {
+            // Signal that the proxy's server-side connection should be
+            // terminated. This handles the remote error as if the local proxy
+            // encountered an error.
+            debug!("Closing server connection");
+            this.client.close.close();
+        }
+        Poll::Ready(Ok(rsp))
+    }
+}
+
+fn update_response<B>(rsp: &mut http::Response<B>, closable: bool) -> bool {
+    // Clear the headers.
+    let hdr = rsp.headers_mut().remove(L5D_PROXY_CONNECTION);
+    let err = rsp.headers_mut().get(L5D_PROXY_ERROR);
+    debug!(
+        error = err
+            .and_then(|v| v.to_str().ok())
+            .map(tracing::field::display),
+        "Remote proxy error"
+    );
+
+    if !closable {
+        return false;
+    }
+
+    static CLOSE: http::HeaderValue = http::HeaderValue::from_static("close");
+    if hdr.as_ref() != Some(&CLOSE) {
+        return false;
+    }
+
+    if rsp.version() == http::Version::HTTP_11 {
+        // If the response is HTTP/1.1, we need to send a Connection: close
+        // header to tell the application this connection is being closed.
+        rsp.headers_mut().insert(
+            http::header::CONNECTION,
+            http::HeaderValue::from_static("close"),
+        );
+    }
+    true
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use futures::future;
+    use linkerd_app_core::{
+        svc::{self, ServiceExt},
+        Infallible,
+    };
+    use linkerd_tracing::test;
+    use tokio::time;
+
+    impl<S> HandleProxyErrorHeaders<S> {
+        fn for_test(closable: bool, inner: S) -> Self {
+            Self { closable, inner }
+        }
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn connection_closes_after_meshed_response_header() {
+        let _trace = test::trace_init();
+
+        // Build the client that should be closed after receiving a response
+        // with the l5d-proxy-error header.
+        let mut req = http::Request::builder()
+            .uri("http://foo.example.com")
+            .body(hyper::Body::default())
+            .unwrap();
+        let (handle, closed) = ClientHandle::new(([192, 0, 2, 3], 50000).into());
+        req.extensions_mut().insert(handle);
+
+        let svc = HandleProxyErrorHeaders::for_test(
+            true,
+            svc::mk(|_: http::Request<hyper::Body>| {
+                future::ok::<_, Infallible>(
+                    http::Response::builder()
+                        .status(http::StatusCode::BAD_GATEWAY)
+                        .header(L5D_PROXY_CONNECTION, "close")
+                        .body(hyper::Body::default())
+                        .unwrap(),
+                )
+            }),
+        );
+
+        let rsp = svc.oneshot(req).await.expect("request must succeed");
+        assert_eq!(rsp.status(), http::StatusCode::BAD_GATEWAY);
+
+        // The client handle close future should fire.
+        time::timeout(time::Duration::from_secs(10), closed)
+            .await
+            .expect("client handle must close");
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn header_ignored_in_unmeshed_response_header() {
+        let _trace = test::trace_init();
+
+        // Build the client that should be closed after receiving a response
+        // with the l5d-proxy-error header.
+        let mut req = http::Request::builder()
+            .uri("http://foo.example.com")
+            .body(hyper::Body::default())
+            .unwrap();
+        let (handle, closed) = ClientHandle::new(([192, 0, 2, 3], 50000).into());
+        req.extensions_mut().insert(handle);
+
+        let svc = HandleProxyErrorHeaders::for_test(
+            false,
+            svc::mk(|_: http::Request<hyper::Body>| {
+                future::ok::<_, Infallible>(
+                    http::Response::builder()
+                        .status(http::StatusCode::BAD_GATEWAY)
+                        .header(L5D_PROXY_CONNECTION, "close")
+                        .body(hyper::Body::default())
+                        .unwrap(),
+                )
+            }),
+        );
+
+        let rsp = svc.oneshot(req).await.expect("request must succeed");
+        assert_eq!(rsp.status(), http::StatusCode::BAD_GATEWAY);
+
+        // The client handle close future should fire.
+        tokio::select! {
+            _ = time::sleep(time::Duration::from_secs(10)) => {},
+            _ = closed => panic!("connection shouldn't close"),
+        }
+    }
+}