tracing: Avoid high-cardinality client in INFO spans (#947)

Our default TCP server sets an info-level tracing span that includes the
client IP and port. This client-level metadata is potentially
high-cardinality, it's not helpful on the outbound side, and it is only
rendered in rare situations.

This change modifies the server to set its `accept` span on the debug
level. The inbound and outbound proxies handle setting info-level spans
with the relevant metadata -- the inbound proxy includes only a server
port, and the outbound includes the original dst address. The error
responder has been modified to include the client address in warning
messages so that it need not be derived from span metadata.

Author: olix0r

linkerd/app/core/src/errors.rs

@@ -4,7 +4,7 @@ use linkerd_error::Error;
 use linkerd_error_metrics as metrics;
 use linkerd_error_respond as respond;
 pub use linkerd_error_respond::RespondLayer;
-use linkerd_proxy_http::{client_handle::Close, ClientHandle, HasH2Reason};
+use linkerd_proxy_http::{ClientHandle, HasH2Reason};
 use linkerd_timeout::{error::ResponseTimeout, FailFastError};
 use linkerd_tls as tls;
 use pin_project::pin_project;
@@ -55,7 +55,7 @@ pub struct NewRespond(());
 pub struct Respond {
     version: http::Version,
     is_grpc: bool,
-    close: Option<Close>,
+    client: Option<ClientHandle>,
 }
 
 #[pin_project(project = ResponseBodyProj)]
@@ -142,10 +142,9 @@ 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());
+        let client = req.extensions().get::<ClientHandle>().cloned();
+        debug_assert!(client.is_some(), "Missing client handle");
+
         match req.version() {
             http::Version::HTTP_2 => {
                 let is_grpc = req
@@ -155,13 +154,13 @@ impl<ReqB, RspB: Default + hyper::body::HttpBody>
                     .unwrap_or(false);
                 Respond {
                     is_grpc,
-                    close,
+                    client,
                     version: http::Version::HTTP_2,
                 }
             }
             version => Respond {
                 version,
-                close,
+                client,
                 is_grpc: false,
             },
         }
@@ -181,7 +180,15 @@ impl<RspB: Default + hyper::body::HttpBody> respond::Respond<http::Response<RspB
                 _ => ResponseBody::NonGrpc(b),
             })),
             Err(error) => {
-                warn!("Failed to proxy request: {}", error);
+                let addr = self
+                    .client
+                    .as_ref()
+                    .map(|ClientHandle { ref addr, .. }| *addr)
+                    .unwrap_or_else(|| {
+                        debug!("Missing client address");
+                        ([0, 0, 0, 0], 0).into()
+                    });
+                warn!(client.addr = %addr, "Failed to proxy request: {}", error);
 
                 if self.version == http::Version::HTTP_2 {
                     if let Some(reset) = error.h2_reason() {
@@ -191,9 +198,9 @@ impl<RspB: Default + hyper::body::HttpBody> respond::Respond<http::Response<RspB
                 }
 
                 // Gracefully teardown the server-side connection.
-                if let Some(c) = self.close.as_ref() {
+                if let Some(ClientHandle { ref close, .. }) = self.client.as_ref() {
                     debug!("Closing server-side connection");
-                    c.close();
+                    close.close();
                 }
 
                 if self.is_grpc {

linkerd/app/core/src/serve.rs

@@ -5,7 +5,7 @@ use linkerd_error::Error;
 use linkerd_proxy_transport::listen::Addrs;
 use tower::util::ServiceExt;
 use tracing::instrument::Instrument;
-use tracing::{debug, info, info_span, warn};
+use tracing::{debug, debug_span, info, warn};
 
 /// Spawns a task that binds an `L`-typed listener with an `A`-typed
 /// connection-accepting service.
@@ -38,13 +38,9 @@ pub async fn serve<M, A, I>(
                     };
 
                     // The local addr should be instrumented from the listener's context.
-                    let span = info_span!(
-                        "accept",
-                        client.addr = %addrs.client(),
-                        target.addr = %addrs.target_addr(),
-                    );
+                    let span = debug_span!("accept", client.addr = %addrs.client());
 
-                    let accept = new_accept.new_service(addrs);
+                    let accept = span.in_scope(|| new_accept.new_service(addrs));
 
                     // Dispatch all of the work for a given connection onto a connection-specific task.
                     tokio::spawn(

linkerd/app/inbound/src/direct.rs

@@ -9,7 +9,7 @@ use linkerd_app_core::{
     Conditional, Error, NameAddr, Never,
 };
 use std::{convert::TryFrom, fmt::Debug, net::SocketAddr};
-use tracing::debug_span;
+use tracing::{debug_span, info_span};
 
 #[derive(Clone, Debug)]
 struct WithTransportHeaderAlpn(LocalCrtKey);
@@ -95,7 +95,6 @@ impl<T> Inbound<T> {
         let detect_timeout = config.proxy.detect_protocol_timeout;
 
         let stack = tcp
-            .check_new_service::<TcpEndpoint, FwdIo<I>>()
             .instrument(|_: &TcpEndpoint| debug_span!("opaque"))
             // When the transport header is present, it may be used for either local
             // TCP forwarding, or we may be processing an HTTP gateway connection.
@@ -112,13 +111,11 @@ impl<T> Inbound<T> {
                         port,
                         name: Some(name),
                         protocol,
-                    } => Ok(svc::Either::B(GatewayConnection::TransportHeader(
-                        GatewayTransportHeader {
-                            target: NameAddr::from((name, port)),
-                            protocol,
-                            client,
-                        },
-                    ))),
+                    } => Ok(svc::Either::B(GatewayTransportHeader {
+                        target: NameAddr::from((name, port)),
+                        protocol,
+                        client,
+                    })),
                     TransportHeader {
                         name: None,
                         protocol: Some(_),
@@ -129,8 +126,8 @@ impl<T> Inbound<T> {
                 // header indicates the connection's HTTP version.
                 svc::stack(gateway.clone())
                     .push_on_response(svc::MapTargetLayer::new(io::EitherIo::Left))
-                    .check_new_service::<GatewayConnection, FwdIo<I>>()
-                    .instrument(|_: &GatewayConnection| debug_span!("gateway"))
+                    .push_map_target(GatewayConnection::TransportHeader)
+                    .instrument(|g: &GatewayTransportHeader| info_span!("gateway", dst = %g.target))
                     .into_inner(),
             )
             // Use ALPN to determine whether a transport header should be read.
@@ -150,13 +147,10 @@ impl<T> Inbound<T> {
                 // with transport header support.
                 svc::stack(gateway)
                     .push_on_response(svc::MapTargetLayer::new(io::EitherIo::Right))
-                    .check_new_service::<GatewayConnection, SensorIo<tls::server::Io<I>>>()
-                    .instrument(|_: &GatewayConnection| debug_span!("legacy"))
+                    .instrument(|_: &GatewayConnection| info_span!("gateway", legacy = true))
                     .into_inner(),
             )
-            .check_new_service::<ClientInfo, SensorIo<tls::server::Io<I>>>()
             .push(rt.metrics.transport.layer_accept())
-            .instrument(|_: &ClientInfo| debug_span!("direct"))
             // Build a ClientInfo target for each accepted connection. Refuse the
             // connection if it doesn't include an mTLS identity.
             .push_request_filter(ClientInfo::try_from)

linkerd/app/inbound/src/lib.rs

@@ -31,7 +31,7 @@ use linkerd_app_core::{
     Error, NameMatch, ProxyRuntime,
 };
 use std::{fmt::Debug, future::Future, net::SocketAddr, time::Duration};
-use tracing::debug_span;
+use tracing::{debug_span, info_span};
 
 #[derive(Clone, Debug)]
 pub struct Config {
@@ -258,7 +258,7 @@ where
                 self.runtime.identity.clone(),
                 config.detect_protocol_timeout,
             ))
-            .check_new_service::<listen::Addrs, I>()
+            .instrument(|_: &_| debug_span!("proxy"))
             .push_switch(
                 disable_detect,
                 self.clone()
@@ -267,7 +267,7 @@ where
                     .push_map_target(TcpEndpoint::from)
                     .push(self.runtime.metrics.transport.layer_accept())
                     .push_map_target(TcpAccept::port_skipped)
-                    .check_new_service::<listen::Addrs, I>()
+                    .instrument(|_: &_| debug_span!("forward"))
                     .into_inner(),
             )
             .check_new_service::<listen::Addrs, I>()
@@ -276,9 +276,10 @@ where
                 self.push_tcp_forward(server_port)
                     .push_direct(gateway)
                     .stack
-                    .check_new_service::<listen::Addrs, I>()
+                    .instrument(|_: &_| debug_span!("direct"))
                     .into_inner(),
             )
+            .instrument(|a: &listen::Addrs| info_span!("server", port = %a.target_addr().port()))
             .check_new_service::<listen::Addrs, I>()
             .into_inner()
     }

linkerd/app/inbound/src/prevent_loop.rs

@@ -50,11 +50,7 @@ impl Predicate<Addrs> for PreventLoop {
 
 impl std::fmt::Display for LoopPrevented {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(
-            f,
-            "inbound requests must not target localhost:{}",
-            self.port
-        )
+        write!(f, "inbound connection must not target port {}", self.port)
     }
 }
 

linkerd/app/outbound/src/discover.rs

@@ -5,6 +5,7 @@ use linkerd_app_core::{
     Error, IpMatch,
 };
 use std::convert::TryFrom;
+use tracing::info_span;
 
 impl<N> Outbound<N> {
     /// Discovers the profile for a TCP endpoint.
@@ -36,11 +37,8 @@ impl<N> Outbound<N> {
         let allow = AllowProfile(config.allow_discovery.clone().into());
 
         let stack = accept
-            .check_new::<tcp::Logical>()
-            .check_new_service::<tcp::Logical, SensorIo<I>>()
             .push_map_target(tcp::Logical::from)
             .push(profiles::discover::layer(profiles, allow))
-            .check_new_service::<tcp::Accept, SensorIo<I>>()
             .push_on_response(
                 svc::layers()
                     // If the traffic split is empty/unavailable, eagerly fail
@@ -55,10 +53,9 @@ impl<N> Outbound<N> {
                     ))
                     .push_spawn_buffer(config.proxy.buffer_capacity),
             )
-            .check_new_service::<tcp::Accept, SensorIo<I>>()
             .push(rt.metrics.transport.layer_accept())
             .push_cache(config.proxy.cache_max_idle_age)
-            .check_new_service::<tcp::Accept, I>()
+            .instrument(|a: &tcp::Accept| info_span!("server", orig_dst = %a.orig_dst))
             .push_request_filter(tcp::Accept::try_from)
             .check_new_service::<listen::Addrs, I>();
 

linkerd/app/outbound/src/ingress.rs

@@ -118,9 +118,7 @@ impl Outbound<()> {
                     ))
                     .push(http::BoxResponse::layer()),
             )
-            .check_new_service::<http::Accept, http::Request<_>>()
             .instrument(|a: &http::Accept| debug_span!("http", v = %a.protocol))
-            .check_new_service::<http::Accept, http::Request<_>>()
             .push(http::NewServeHttp::layer(
                 h2_settings,
                 self.runtime.drain.clone(),
@@ -133,13 +131,13 @@ impl Outbound<()> {
                 detect_protocol_timeout,
                 http::DetectHttp::default(),
             ))
-            .check_new_service::<tcp::Accept, transport::metrics::SensorIo<I>>()
             .push(self.runtime.metrics.transport.layer_accept())
+            .instrument(|a: &tcp::Accept| info_span!("ingress", orig_dst = %a.orig_dst))
             .push_request_filter(tcp::Accept::try_from)
-            .check_new_service::<listen::Addrs, I>()
             // Boxing is necessary purely to limit the link-time overhead of
             // having enormous types.
             .push(svc::BoxNewService::layer())
+            .check_new_service::<listen::Addrs, I>()
             .into_inner()
     }
 }