Include server address in server error logs

When the proxy's TCP server encounters an error (usually due to one of the connections failing, we log the error and the client's address. The server's address was omitted because it varies based on context that is not known in this module: in some cases it's the actual server address on the socket, but when proxying a connection it may be determined by the value retrieved from the SO_ORIGINAL_DST socket option. To fix this, the server now requires that connection metadata be able to materialize an 'AddrPair' parameter that describes a client-server connection. The TCP listener impls are updated to satisfy this based on the appropriate metadata; and the TCP server consumes this type to include both client and server addresses in the relevant logs/contexts.
linkerd · Nov 3, 2023 · a2db0ac · a2db0ac
1 parent cbf226e
commit a2db0ac
Show file tree

Hide file tree

Showing 10 changed files with 75 additions and 17 deletions.
diff --git a/linkerd/app/admin/src/stack.rs b/linkerd/app/admin/src/stack.rs
@@ -7,7 +7,9 @@ use linkerd_app_core::{
     serve,
     svc::{self, ExtractParam, InsertParam, Param},
     tls, trace,
-    transport::{self, listen::Bind, ClientAddr, Local, OrigDstAddr, Remote, ServerAddr},
+    transport::{
+        self, addrs::AddrPair, listen::Bind, ClientAddr, Local, OrigDstAddr, Remote, ServerAddr,
+    },
     Error, Result,
 };
 use linkerd_app_inbound as inbound;
@@ -84,7 +86,9 @@ impl Config {
     where
         R: FmtMetrics + Clone + Send + Sync + Unpin + 'static,
         B: Bind<ServerConfig>,
-        B::Addrs: svc::Param<Remote<ClientAddr>> + svc::Param<Local<ServerAddr>>,
+        B::Addrs: svc::Param<Remote<ClientAddr>>,
+        B::Addrs: svc::Param<Local<ServerAddr>>,
+        B::Addrs: svc::Param<AddrPair>,
     {
         let (listen_addr, listen) = bind.bind(&self.server)?;
 
@@ -202,6 +206,14 @@ impl Param<Remote<ClientAddr>> for Http {
     }
 }
 
+impl Param<AddrPair> for Http {
+    fn param(&self) -> AddrPair {
+        let Remote(client) = self.tcp.client;
+        let Local(server) = self.tcp.addr;
+        AddrPair(client, server)
+    }
+}
+
 impl Param<tls::ConditionalServerTls> for Http {
     fn param(&self) -> tls::ConditionalServerTls {
         self.tcp.tls.clone()

diff --git a/linkerd/app/core/src/serve.rs b/linkerd/app/core/src/serve.rs
@@ -1,11 +1,11 @@
 use crate::{
     io, is_caused_by,
     svc::{self, Param},
-    transport::{ClientAddr, Remote},
     Result,
 };
 use futures::prelude::*;
 use linkerd_error::Error;
+use linkerd_proxy_transport::AddrPair;
 use tower::util::ServiceExt;
 use tracing::{debug, debug_span, info, instrument::Instrument, warn};
 
@@ -18,7 +18,7 @@ pub async fn serve<M, S, I, A>(
     shutdown: impl Future,
 ) where
     I: Send + 'static,
-    A: Param<Remote<ClientAddr>>,
+    A: Param<AddrPair>,
     M: svc::NewService<A, Service = S>,
     S: tower::Service<io::ScopedIo<I>, Response = ()> + Send + 'static,
     S::Error: Into<Error>,
@@ -40,8 +40,8 @@ pub async fn serve<M, S, I, A>(
                     };
 
                     // The local addr should be instrumented from the listener's context.
-                    let Remote(ClientAddr(client_addr)) = addrs.param();
-                    let span = debug_span!("accept", client.addr = %client_addr).entered();
+                    let AddrPair(client_addr, server_addr) = addrs.param();
+                    let span = debug_span!("accept", client.addr = %client_addr, server.addr = %server_addr).entered();
                     let accept = new_accept.new_service(addrs);
 
                     // Dispatch all of the work for a given connection onto a
@@ -57,10 +57,20 @@ pub async fn serve<M, S, I, A>(
                                     {
                                         Ok(()) => debug!("Connection closed"),
                                         Err(reason) if is_caused_by::<std::io::Error>(&*reason) => {
-                                            debug!(%reason, "Connection closed")
+                                            debug!(
+                                                reason,
+                                                client.addr = %client_addr,
+                                                server.addr = %server_addr,
+                                                "Connection closed"
+                                            );
                                         }
                                         Err(error) => {
-                                            info!(error, client.addr = %client_addr, "Connection closed")
+                                            info!(
+                                                error,
+                                                client.addr = %client_addr,
+                                                server.addr = %server_addr,
+                                                "Connection closed"
+                                            );
                                         }
                                     }
                                     // Hold the service until the connection is complete. This

diff --git a/linkerd/app/inbound/src/server.rs b/linkerd/app/inbound/src/server.rs
@@ -5,7 +5,7 @@ use linkerd_app_core::{
     io, profiles,
     proxy::http,
     serve, svc,
-    transport::{self, ClientAddr, Local, OrigDstAddr, Remote, ServerAddr},
+    transport::{self, addrs::*},
     Error, Result,
 };
 use std::{fmt::Debug, sync::Arc};
@@ -43,7 +43,10 @@ impl Inbound<()> {
         profiles: P,
         gateway: G,
     ) where
-        A: svc::Param<Remote<ClientAddr>> + svc::Param<OrigDstAddr> + Clone + Send + Sync + 'static,
+        A: svc::Param<Remote<ClientAddr>>,
+        A: svc::Param<OrigDstAddr>,
+        A: svc::Param<AddrPair>,
+        A: Clone + Send + Sync + 'static,
         I: io::AsyncRead + io::AsyncWrite + io::Peek + io::PeerAddr,
         I: Debug + Unpin + Send + Sync + 'static,
         G: svc::NewService<direct::GatewayTransportHeader, Service = GSvc>,

diff --git a/linkerd/app/outbound/src/lib.rs b/linkerd/app/outbound/src/lib.rs
@@ -230,7 +230,7 @@ impl Outbound<()> {
         resolve: R,
     ) where
         // Target describing a server-side connection.
-        T: svc::Param<Remote<ClientAddr>>,
+        T: svc::Param<AddrPair>,
         T: svc::Param<OrigDstAddr>,
         T: Clone + Send + Sync + 'static,
         // Server-side socket.

diff --git a/linkerd/app/src/lib.rs b/linkerd/app/src/lib.rs
@@ -22,7 +22,7 @@ use linkerd_app_core::{
     metrics::FmtMetrics,
     svc::Param,
     telemetry,
-    transport::{listen::Bind, ClientAddr, Local, OrigDstAddr, Remote, ServerAddr},
+    transport::{addrs::*, listen::Bind},
     Error, ProxyRuntime,
 };
 use linkerd_app_gateway as gateway;
@@ -102,11 +102,17 @@ impl Config {
     ) -> Result<App, Error>
     where
         BIn: Bind<ServerConfig> + 'static,
-        BIn::Addrs: Param<Remote<ClientAddr>> + Param<Local<ServerAddr>> + Param<OrigDstAddr>,
+        BIn::Addrs: Param<Remote<ClientAddr>>
+            + Param<Local<ServerAddr>>
+            + Param<OrigDstAddr>
+            + Param<AddrPair>,
         BOut: Bind<ServerConfig> + 'static,
-        BOut::Addrs: Param<Remote<ClientAddr>> + Param<Local<ServerAddr>> + Param<OrigDstAddr>,
+        BOut::Addrs: Param<Remote<ClientAddr>>
+            + Param<Local<ServerAddr>>
+            + Param<OrigDstAddr>
+            + Param<AddrPair>,
         BAdmin: Bind<ServerConfig> + Clone + 'static,
-        BAdmin::Addrs: Param<Remote<ClientAddr>> + Param<Local<ServerAddr>>,
+        BAdmin::Addrs: Param<Remote<ClientAddr>> + Param<Local<ServerAddr>> + Param<AddrPair>,
     {
         let Config {
             admin,

diff --git a/linkerd/app/src/tap.rs b/linkerd/app/src/tap.rs
@@ -6,7 +6,7 @@ use linkerd_app_core::{
     serve,
     svc::{self, ExtractParam, InsertParam, Param},
     tls,
-    transport::{listen::Bind, ClientAddr, Local, Remote, ServerAddr},
+    transport::{addrs::AddrPair, listen::Bind, ClientAddr, Local, Remote, ServerAddr},
     Error,
 };
 use std::{collections::HashSet, pin::Pin};
@@ -47,6 +47,7 @@ impl Config {
     where
         B: Bind<ServerConfig>,
         B::Addrs: Param<Remote<ClientAddr>>,
+        B::Addrs: Param<AddrPair>,
     {
         let (registry, server) = tap::new();
         match self {

diff --git a/linkerd/proxy/transport/src/addrs.rs b/linkerd/proxy/transport/src/addrs.rs
@@ -28,6 +28,10 @@ pub struct Local<T>(pub T);
 #[derive(Copy, Clone, Debug, Eq, PartialEq, Hash)]
 pub struct Remote<T>(pub T);
 
+/// Describes a connection from a client to a server.
+#[derive(Copy, Clone, Debug, Eq, PartialEq, Hash)]
+pub struct AddrPair(pub ClientAddr, pub ServerAddr);
+
 // === impl ClientAddr ===
 
 impl std::ops::Deref for ClientAddr {

diff --git a/linkerd/proxy/transport/src/lib.rs b/linkerd/proxy/transport/src/lib.rs
@@ -15,7 +15,7 @@ pub mod listen;
 pub mod orig_dst;
 
 pub use self::{
-    addrs::{ClientAddr, ListenAddr, Local, OrigDstAddr, Remote, ServerAddr},
+    addrs::{AddrPair, ClientAddr, ListenAddr, Local, OrigDstAddr, Remote, ServerAddr},
     connect::ConnectTcp,
     listen::{Bind, BindTcp},
     orig_dst::BindWithOrigDst,

diff --git a/linkerd/proxy/transport/src/listen.rs b/linkerd/proxy/transport/src/listen.rs
@@ -104,3 +104,12 @@ impl Param<Local<ServerAddr>> for Addrs {
         self.server
     }
 }
+
+impl Param<AddrPair> for Addrs {
+    #[inline]
+    fn param(&self) -> AddrPair {
+        let Remote(client) = self.client;
+        let Local(server) = self.server;
+        AddrPair(client, server)
+    }
+}
diff --git a/linkerd/proxy/transport/src/orig_dst.rs b/linkerd/proxy/transport/src/orig_dst.rs
@@ -23,6 +23,7 @@ pub struct Addrs<A = listen::Addrs> {
 // === impl Addrs ===
 
 impl<A> Param<OrigDstAddr> for Addrs<A> {
+    #[inline]
     fn param(&self) -> OrigDstAddr {
         self.orig_dst
     }
@@ -32,11 +33,23 @@ impl<A> Param<Remote<ClientAddr>> for Addrs<A>
 where
     A: Param<Remote<ClientAddr>>,
 {
+    #[inline]
     fn param(&self) -> Remote<ClientAddr> {
         self.inner.param()
     }
 }
 
+impl<A> Param<AddrPair> for Addrs<A>
+where
+    A: Param<Remote<ClientAddr>>,
+{
+    #[inline]
+    fn param(&self) -> AddrPair {
+        let Remote(client) = self.inner.param();
+        AddrPair(client, ServerAddr(self.orig_dst.into()))
+    }
+}
+
 impl<A> Param<Local<ServerAddr>> for Addrs<A>
 where
     A: Param<Local<ServerAddr>>,