metrics: add target_addr label for accepted transport metrics (#861)

This branch adds a `target_addr` label to the Prometheus metrics
generated for accepted (`peer="src"`) connections, containing the
original destination address of the accepted connection.

When we add the proxy's admin listener to the inbound metrics, 
this will allow distinguishing between admin and inbound proxy
traffic. Additionally, it allows partitioning transport metrics by
the client's target address.

Author: hawkw

linkerd/app/core/src/transport/labels.rs

@@ -2,7 +2,7 @@ pub use crate::metrics::{Direction, OutboundEndpointLabels};
 use linkerd_conditional::Conditional;
 use linkerd_metrics::FmtLabels;
 use linkerd_tls as tls;
-use std::fmt;
+use std::{fmt, net::SocketAddr};
 
 /// Describes a class of transport.
 ///
@@ -11,7 +11,11 @@ use std::fmt;
 /// Implements `FmtLabels`.
 #[derive(Clone, Debug, Eq, PartialEq, Hash)]
 pub enum Key {
-    Accept(Direction, tls::server::ConditionalTls),
+    Accept {
+        direction: Direction,
+        id: tls::server::ConditionalTls,
+        target_addr: SocketAddr,
+    },
     OutboundConnect(OutboundEndpointLabels),
     InboundConnect,
 }
@@ -25,18 +29,30 @@ pub struct TlsConnect<'t>(&'t tls::ConditionalServerId);
 // === impl Key ===
 
 impl Key {
-    pub fn accept(direction: Direction, id: tls::server::ConditionalTls) -> Self {
-        Self::Accept(direction, id)
+    pub fn accept(
+        direction: Direction,
+        id: tls::server::ConditionalTls,
+        target_addr: SocketAddr,
+    ) -> Self {
+        Self::Accept {
+            direction,
+            id,
+            target_addr,
+        }
     }
 }
 
 impl FmtLabels for Key {
     fn fmt_labels(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
-            Self::Accept(direction, identity) => {
+            Self::Accept {
+                direction,
+                id,
+                target_addr,
+            } => {
                 direction.fmt_labels(f)?;
-                write!(f, ",peer=\"src\",")?;
-                TlsAccept::from(identity).fmt_labels(f)
+                write!(f, ",peer=\"src\",target_addr=\"{}\",", target_addr)?;
+                TlsAccept::from(id).fmt_labels(f)
             }
             Self::OutboundConnect(endpoint) => {
                 Direction::Out.fmt_labels(f)?;

linkerd/app/inbound/src/target.rs

@@ -82,7 +82,11 @@ impl Into<SocketAddr> for &'_ TcpAccept {
 
 impl Into<transport::labels::Key> for &'_ TcpAccept {
     fn into(self) -> transport::labels::Key {
-        transport::labels::Key::accept(transport::labels::Direction::In, self.client_id.clone())
+        transport::labels::Key::accept(
+            transport::labels::Direction::In,
+            self.client_id.clone(),
+            self.target_addr,
+        )
     }
 }
 

linkerd/app/integration/src/tcp.rs

@@ -153,6 +153,10 @@ impl TcpServer {
 }
 
 impl TcpConn {
+    pub fn target_addr(&self) -> SocketAddr {
+        self.addr
+    }
+
     pub async fn read(&self) -> Vec<u8> {
         self.try_read()
             .await

linkerd/app/integration/src/tests/telemetry.rs

@@ -922,12 +922,15 @@ mod transport {
             .await;
     }
 
-    async fn test_tcp_accept(fixture: impl Future<Output = TcpFixture>, labels: metrics::Labels) {
+    async fn test_tcp_accept(
+        fixture: impl Future<Output = TcpFixture>,
+        labels: impl Fn(&proxy::Listening) -> metrics::Labels,
+    ) {
         let _trace = trace_init();
         let TcpFixture {
             client,
             metrics,
-            proxy: _proxy,
+            proxy,
             dst: _dst,
             profile: _profile,
         } = fixture.await;
@@ -937,7 +940,7 @@ mod transport {
         tcp_client.write(TcpFixture::HELLO_MSG).await;
         assert_eq!(tcp_client.read().await, TcpFixture::BYE_MSG.as_bytes());
 
-        let labels = labels.label("peer", "src");
+        let labels = labels(&proxy).label("peer", "src");
         let mut opens = labels.metric("tcp_open_total").value(1u64);
         let mut closes = labels
             .metric("tcp_close_total")
@@ -1088,10 +1091,12 @@ mod transport {
 
     #[tokio::test]
     async fn inbound_http_accept() {
-        test_http_accept(Fixture::inbound(), |_| {
+        test_http_accept(Fixture::inbound(), |proxy| {
+            let orig_dst = proxy.inbound_server.as_ref().unwrap().addr;
             metrics::labels()
                 .label("direction", "inbound")
                 .label("tls", "disabled")
+                .label("target_addr", orig_dst)
         })
         .await;
     }
@@ -1109,11 +1114,13 @@ mod transport {
 
     #[tokio::test]
     async fn outbound_http_accept() {
-        test_http_accept(Fixture::outbound(), |_| {
+        test_http_accept(Fixture::outbound(), |proxy| {
+            let orig_dst = proxy.outbound_server.as_ref().unwrap().addr;
             metrics::labels()
                 .label("direction", "outbound")
                 .label("tls", "no_identity")
                 .label("no_tls_reason", "loopback")
+                .label("target_addr", orig_dst)
         })
         .await;
     }
@@ -1147,12 +1154,13 @@ mod transport {
 
     #[tokio::test]
     async fn inbound_tcp_accept() {
-        test_tcp_accept(
-            TcpFixture::inbound(),
+        test_tcp_accept(TcpFixture::inbound(), |proxy| {
+            let orig_dst = proxy.inbound_server.as_ref().unwrap().addr;
             metrics::labels()
                 .label("direction", "inbound")
-                .label("tls", "disabled"),
-        )
+                .label("tls", "disabled")
+                .label("target_addr", orig_dst)
+        })
         .await
     }
 
@@ -1170,13 +1178,15 @@ mod transport {
 
     #[tokio::test]
     async fn outbound_tcp_accept() {
-        test_tcp_accept(
-            TcpFixture::outbound(),
+        test_tcp_accept(TcpFixture::outbound(), |proxy| {
+            let orig_dst = proxy.outbound_server.as_ref().unwrap().addr;
+
             metrics::labels()
                 .label("direction", "outbound")
                 .label("tls", "no_identity")
-                .label("no_tls_reason", "loopback"),
-        )
+                .label("no_tls_reason", "loopback")
+                .label("target_addr", orig_dst)
+        })
         .await;
     }
 

linkerd/app/outbound/src/target.rs

@@ -69,15 +69,20 @@ impl<P> Into<Addr> for &'_ Accept<P> {
 impl<P> Into<transport::labels::Key> for &'_ Accept<P> {
     fn into(self) -> transport::labels::Key {
         const NO_TLS: tls::server::ConditionalTls = Conditional::None(tls::server::NoTls::Loopback);
-        transport::labels::Key::accept(transport::labels::Direction::Out, NO_TLS)
+        transport::labels::Key::accept(transport::labels::Direction::Out, NO_TLS, self.orig_dst)
     }
 }
 
 // === impl Logical ===
 
 impl<P> From<(Option<profiles::Receiver>, Accept<P>)> for Logical<P> {
     fn from(
-        (profile, Accept { orig_dst, protocol }): (Option<profiles::Receiver>, Accept<P>),
+        (
+            profile,
+            Accept {
+                orig_dst, protocol, ..
+            },
+        ): (Option<profiles::Receiver>, Accept<P>),
     ) -> Self {
         Self {
             profile,