metrics: add target_addr label to TCP accept error metrics (#1118)

The TCP accept error metrics do not include any context about the target
address/port, which makes them difficult to act on.

This change adds a `target_addr` label to the
`inbound_tcp_accept_errors_total` and
`outbound_tcp_accept_errors_total` metrics.

Co-authored-by: Oliver Gould <[email protected]>

Author: hawkw

Cargo.lock

@@ -701,6 +701,7 @@ dependencies = [
  "linkerd-trace-context",
  "linkerd-tracing",
  "linkerd-transport-header",
+ "parking_lot",
  "pin-project",
  "quickcheck",
  "regex",

linkerd/app/core/Cargo.toml

@@ -36,7 +36,7 @@ linkerd-http-metrics = { path = "../../http-metrics" }
 linkerd-http-retry = { path = "../../http-retry" }
 linkerd-identity = { path = "../../identity" }
 linkerd-io = { path = "../../io" }
-linkerd-metrics = { path = "../../metrics" }
+linkerd-metrics = { path = "../../metrics", features = ["linkerd-stack"] }
 linkerd-transport-header = { path = "../../transport-header" }
 linkerd-opencensus = { path = "../../opencensus" }
 linkerd-proxy-core = { path = "../../proxy/core" }
@@ -66,6 +66,7 @@ tokio = { version = "1", features = ["macros", "sync", "parking_lot"]}
 tokio-stream = { version = "0.1.7", features = ["time"] }
 tonic = { version = "0.5", default-features = false, features = ["prost"] }
 tracing = "0.1.26"
+parking_lot = "0.11"
 pin-project = "1"
 
 [dependencies.tower]

linkerd/app/core/src/metrics/mod.rs

@@ -1,23 +1,23 @@
 mod tcp_accept_errors;
 
-pub use crate::{
+use crate::{
     classify::{Class, SuccessOrFailure},
-    control, dst, errors, http_metrics, http_metrics as metrics, opencensus, proxy,
-    proxy::identity,
-    stack_metrics,
+    control, dst, errors, http_metrics, http_metrics as metrics, opencensus, stack_metrics,
     svc::Param,
     telemetry, tls,
     transport::{
         self,
-        labels::{TlsAccept, TlsConnect},
+        labels::{TargetAddr, TlsAccept, TlsConnect},
     },
 };
 use linkerd_addr::Addr;
 use linkerd_metrics::FmtLabels;
 pub use linkerd_metrics::*;
-use std::fmt::{self, Write};
-use std::net::SocketAddr;
-use std::time::{Duration, SystemTime};
+use std::{
+    fmt::{self, Write},
+    net::SocketAddr,
+    time::{Duration, SystemTime},
+};
 
 pub type ControlHttp = http_metrics::Requests<ControlLabels, Class>;
 
@@ -38,7 +38,7 @@ pub struct Proxy {
     pub http_errors: errors::MetricsLayer,
     pub stack: Stack,
     pub transport: transport::Metrics,
-    pub tcp_accept_errors: tcp_accept_errors::Layer,
+    pub tcp_accept_errors: tcp_accept_errors::Registry,
 }
 
 #[derive(Clone, Debug)]
@@ -173,7 +173,7 @@ impl Metrics {
                 http_errors: http_errors.inbound(),
                 stack: stack.clone(),
                 transport: transport.clone(),
-                tcp_accept_errors: inbound_tcp_accept_errors.layer(),
+                tcp_accept_errors: inbound_tcp_accept_errors.clone(),
             },
             outbound: Proxy {
                 http_endpoint,
@@ -183,7 +183,7 @@ impl Metrics {
                 http_errors: http_errors.outbound(),
                 stack: stack.clone(),
                 transport,
-                tcp_accept_errors: outbound_tcp_accept_errors.layer(),
+                tcp_accept_errors: outbound_tcp_accept_errors.clone(),
             },
             control,
             opencensus,
@@ -282,9 +282,7 @@ impl FmtLabels for InboundEndpointLabels {
             write!(f, ",")?;
         }
 
-        write!(f, "target_addr=\"{}\",", self.target_addr)?;
-
-        TlsAccept::from(&self.tls).fmt_labels(f)?;
+        (TargetAddr(self.target_addr), TlsAccept::from(&self.tls)).fmt_labels(f)?;
 
         Ok(())
     }
@@ -297,9 +295,9 @@ impl FmtLabels for OutboundEndpointLabels {
             write!(f, ",")?;
         }
 
-        write!(f, "target_addr=\"{}\",", self.target_addr)?;
-
-        TlsConnect::from(&self.server_id).fmt_labels(f)?;
+        let ta = TargetAddr(self.target_addr);
+        let tls = TlsConnect::from(&self.server_id);
+        (ta, tls).fmt_labels(f)?;
 
         if let Some(labels) = self.labels.as_ref() {
             write!(f, ",{}", labels)?;

linkerd/app/core/src/metrics/tcp_accept_errors.rs

@@ -1,10 +1,15 @@
+use crate::{
+    metrics::{self, Counter, FmtMetrics},
+    svc,
+    transport::{labels, OrigDstAddr},
+};
 use linkerd_error::Error;
-use linkerd_error_metrics::{FmtLabels, LabelError, RecordErrorLayer};
-use linkerd_metrics::{metrics, Counter, FmtMetrics};
+use linkerd_error_metrics::{FmtLabels, LabelError, RecordError};
 use linkerd_tls::server::DetectTimeout as TlsDetectTimeout;
-use std::fmt;
+use parking_lot::Mutex;
+use std::{collections::HashMap, fmt};
 
-metrics! {
+metrics::metrics! {
     inbound_tcp_accept_errors_total: Counter {
         "The total number of inbound TCP connections that could not be processed due to a proxy error."
     },
@@ -15,9 +20,17 @@ metrics! {
 }
 
 #[derive(Clone, Debug)]
-pub struct Registry(linkerd_error_metrics::Registry<AcceptErrors>);
+pub struct Registry {
+    scopes: metrics::SharedStore<OrigDstAddr, Scope>,
+    metric: linkerd_error_metrics::Metric,
+}
+
+type Scope = Mutex<HashMap<AcceptErrors, metrics::Counter>>;
+
+type NewErrorMetrics<N, S> =
+    metrics::NewMetrics<N, OrigDstAddr, Scope, RecordError<LabelAcceptErrors, AcceptErrors, S>>;
 
-#[derive(Clone, Copy, Debug)]
+#[derive(Clone, Copy, Debug, Default)]
 pub struct LabelAcceptErrors(());
 
 #[derive(Clone, Debug, Eq, PartialEq, Hash)]
@@ -27,32 +40,44 @@ pub enum AcceptErrors {
     Other,
 }
 
-pub type Layer = RecordErrorLayer<LabelAcceptErrors, AcceptErrors>;
-
 // === impl Registry ===
 
 impl Registry {
     pub fn inbound() -> Self {
-        Self(linkerd_error_metrics::Registry::new(
-            inbound_tcp_accept_errors_total,
-        ))
+        Self {
+            metric: inbound_tcp_accept_errors_total,
+            scopes: Default::default(),
+        }
     }
 
     pub fn outbound() -> Self {
-        Self(linkerd_error_metrics::Registry::new(
-            outbound_tcp_accept_errors_total,
-        ))
+        Self {
+            metric: outbound_tcp_accept_errors_total,
+            scopes: Default::default(),
+        }
     }
 
-    pub fn layer(&self) -> Layer {
-        self.0.layer(LabelAcceptErrors(()))
+    pub fn layer<T, N: svc::NewService<T>>(
+        &self,
+    ) -> impl svc::Layer<N, Service = NewErrorMetrics<N, N::Service>> + Clone {
+        metrics::NewMetrics::layer(self.scopes.clone())
     }
 }
 
 impl FmtMetrics for Registry {
-    #[inline]
     fn fmt_metrics(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        self.0.fmt_metrics(f)
+        use metrics::FmtMetric;
+        let errors = self.scopes.lock();
+
+        self.metric.fmt_help(f)?;
+        for (OrigDstAddr(a), ms) in errors.iter() {
+            let ta = labels::TargetAddr(*a);
+            for (e, m) in ms.lock().iter() {
+                m.fmt_metric_labeled(f, self.metric.name, (ta, e))?;
+            }
+        }
+
+        Ok(())
     }
 }
 

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

@@ -21,10 +21,13 @@ pub enum Key {
 }
 
 #[derive(Clone, Debug, Eq, PartialEq, Hash)]
-pub struct TlsAccept<'t>(&'t tls::ConditionalServerTls);
+pub(crate) struct TlsAccept<'t>(&'t tls::ConditionalServerTls);
 
 #[derive(Clone, Debug, Eq, PartialEq, Hash)]
-pub struct TlsConnect<'t>(&'t tls::ConditionalClientTls);
+pub(crate) struct TlsConnect<'t>(&'t tls::ConditionalClientTls);
+
+#[derive(Copy, Clone, Debug, Eq, PartialEq, Hash)]
+pub(crate) struct TargetAddr(pub(crate) SocketAddr);
 
 // === impl Key ===
 
@@ -51,8 +54,8 @@ impl FmtLabels for Key {
                 target_addr,
             } => {
                 direction.fmt_labels(f)?;
-                write!(f, ",peer=\"src\",target_addr=\"{}\",", target_addr)?;
-                TlsAccept::from(tls).fmt_labels(f)
+                f.write_str(",peer=\"src\",")?;
+                (TargetAddr(*target_addr), TlsAccept::from(tls)).fmt_labels(f)
             }
             Self::OutboundConnect(endpoint) => {
                 Direction::Out.fmt_labels(f)?;
@@ -122,3 +125,11 @@ impl<'t> FmtLabels for TlsConnect<'t> {
         }
     }
 }
+
+// === impl TargetAddr ===
+
+impl FmtLabels for TargetAddr {
+    fn fmt_labels(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "target_addr=\"{}\"", self.0)
+    }
+}

linkerd/app/inbound/src/direct.rs

@@ -69,7 +69,8 @@ impl<N> Inbound<N> {
         gateway: G,
     ) -> Inbound<svc::BoxNewService<T, svc::BoxService<I, (), Error>>>
     where
-        T: Param<Remote<ClientAddr>> + Param<OrigDstAddr> + Clone + Send + 'static,
+        T: Param<Remote<ClientAddr>> + Param<OrigDstAddr>,
+        T: Clone + Send + 'static,
         I: io::AsyncRead + io::AsyncWrite + io::Peek + io::PeerAddr,
         I: Debug + Send + Sync + Unpin + 'static,
         N: svc::NewService<TcpEndpoint, Service = NSvc> + Clone + Send + Sync + Unpin + 'static,

linkerd/app/inbound/src/lib.rs

@@ -204,7 +204,8 @@ where
         gateway: G,
     ) -> svc::BoxNewService<T, svc::BoxService<I, (), Error>>
     where
-        T: svc::Param<Remote<ClientAddr>> + svc::Param<OrigDstAddr> + Clone + Send + 'static,
+        T: svc::Param<Remote<ClientAddr>> + svc::Param<OrigDstAddr>,
+        T: Clone + Send + 'static,
         I: io::AsyncRead + io::AsyncWrite + io::Peek + io::PeerAddr,
         I: Debug + Send + Sync + Unpin + 'static,
         G: svc::NewService<direct::GatewayConnection, Service = GSvc>,
@@ -292,7 +293,7 @@ where
                 let OrigDstAddr(target_addr) = a.param();
                 info_span!("server", port = target_addr.port())
             })
-            .push_on_response(rt.metrics.tcp_accept_errors)
+            .push(rt.metrics.tcp_accept_errors.layer())
             .push_on_response(svc::BoxService::layer())
             .push(svc::BoxNewService::layer())
             .into_inner()

linkerd/app/inbound/src/target.rs

@@ -92,6 +92,12 @@ impl Param<SocketAddr> for TcpAccept {
     }
 }
 
+impl Param<OrigDstAddr> for TcpAccept {
+    fn param(&self) -> OrigDstAddr {
+        OrigDstAddr(self.target_addr)
+    }
+}
+
 impl Param<transport::labels::Key> for TcpAccept {
     fn param(&self) -> transport::labels::Key {
         transport::labels::Key::accept(

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

@@ -60,6 +60,10 @@ async fn run_proxy(
     (proxy, admin_client)
 }
 
+fn metric(proxy: &proxy::Listening) -> metrics::MetricMatch {
+    metrics::metric(METRIC).label("target_addr", proxy.inbound_server.as_ref().unwrap().addr)
+}
+
 /// Tests that the detect metric is labeled and incremented on timeout.
 #[tokio::test]
 async fn inbound_timeout() {
@@ -73,7 +77,7 @@ async fn inbound_timeout() {
     tokio::time::sleep(TIMEOUT + Duration::from_millis(15)) // just in case
         .await;
 
-    metrics::metric(METRIC)
+    metric(&proxy)
         .label("error", "tls_detect_timeout")
         .value(1u64)
         .assert_in(&metrics)
@@ -93,7 +97,7 @@ async fn inbound_io_err() {
     tcp_client.write(TcpFixture::HELLO_MSG).await;
     drop(tcp_client);
 
-    metrics::metric(METRIC)
+    metric(&proxy)
         .label("error", "io")
         .value(1u64)
         .assert_in(&metrics)
@@ -119,7 +123,7 @@ async fn inbound_success() {
     );
     let no_tls_client = client::tcp(proxy.inbound);
 
-    let metric = metrics::metric(METRIC)
+    let metric = metric(&proxy)
         .label("error", "tls_detect_timeout")
         .value(1u64);
 
@@ -150,8 +154,9 @@ async fn inbound_multi() {
     let (proxy, metrics) = default_proxy().await;
     let client = client::tcp(proxy.inbound);
 
-    let timeout_metric = metrics::metric(METRIC).label("error", "tls_detect_timeout");
-    let io_metric = metrics::metric(METRIC).label("error", "io");
+    let metric = metric(&proxy);
+    let timeout_metric = metric.clone().label("error", "tls_detect_timeout");
+    let io_metric = metric.label("error", "io");
 
     let tcp_client = client.connect().await;
 
@@ -200,8 +205,9 @@ async fn inbound_direct_multi() {
     let (proxy, metrics) = run_proxy(proxy, identity).await;
     let client = client::tcp(proxy.inbound);
 
-    let timeout_metric = metrics::metric(METRIC).label("error", "tls_detect_timeout");
-    let no_tls_metric = metrics::metric(METRIC).label("error", "other");
+    let metric = metrics::metric(METRIC).label("target_addr", proxy.inbound);
+    let timeout_metric = metric.clone().label("error", "tls_detect_timeout");
+    let no_tls_metric = metric.clone().label("error", "other");
 
     let tcp_client = client.connect().await;
 
@@ -248,7 +254,11 @@ async fn inbound_direct_success() {
     // connections require mutual authentication.
     let auth = "bar.ns1.svc.cluster.local";
     let ctrl = controller::new();
-    let dst = format!("{}:{}", auth, proxy1.inbound.port());
+    let dst = format!(
+        "{}:{}",
+        auth,
+        proxy1.inbound_server.as_ref().unwrap().addr.port()
+    );
     let _profile_out = ctrl.profile_tx_default(proxy1.inbound, auth);
     let dst = ctrl.destination_tx(dst);
     dst.send(controller::destination_add_tls(
@@ -265,6 +275,7 @@ async fn inbound_direct_success() {
     let no_tls_client = client::tcp(proxy1.inbound);
 
     let metric = metrics::metric(METRIC)
+        .label("target_addr", proxy1.inbound)
         .label("error", "tls_detect_timeout")
         .value(1u64);
 

linkerd/app/outbound/src/discover.rs

@@ -76,7 +76,7 @@ impl<N> Outbound<N> {
                 .push_cache(config.proxy.cache_max_idle_age)
                 .instrument(|a: &tcp::Accept| info_span!("server", orig_dst = %a.orig_dst))
                 .push_request_filter(|t: T| tcp::Accept::try_from(t.param()))
-                .push_on_response(rt.metrics.tcp_accept_errors.clone())
+                .push(rt.metrics.tcp_accept_errors.layer())
                 .push(svc::BoxNewService::layer())
                 .check_new_service::<T, I>()
         })