feat(inbound): record metrics about transport header usage (#3723)

Inbound proxies may receive meshed traffic directly on the proxy's inbound port
with a transport header, informing inbound routing behavior.

This change updates the inbound proxy to record metrics about the usage of
transport headers, including the total number of requests with a transport
header by session protocol and target port.

Author: olix0r

linkerd/app/inbound/src/direct.rs

@@ -15,6 +15,10 @@ use std::fmt::Debug;
 use thiserror::Error;
 use tracing::{debug_span, info_span};
 
+mod metrics;
+
+pub use self::metrics::MetricsFamilies;
+
 /// Creates I/O errors when a connection cannot be forwarded because no transport
 /// header was present.
 #[derive(Debug, Default)]
@@ -108,6 +112,7 @@ impl<N> Inbound<N> {
     {
         self.map_stack(|config, rt, inner| {
             let detect_timeout = config.proxy.detect_protocol_timeout;
+            let metrics = rt.metrics.direct.clone();
 
             let identity = rt
                 .identity
@@ -211,6 +216,7 @@ impl<N> Inbound<N> {
                 )
                 .check_new_service::<(TransportHeader, ClientInfo), _>()
                 // Use ALPN to determine whether a transport header should be read.
+                .push(metrics::NewRecord::layer(metrics))
                 .push(svc::ArcNewService::layer())
                 .push(NewTransportHeaderServer::layer(detect_timeout))
                 .check_new_service::<ClientInfo, _>()

linkerd/app/inbound/src/direct/metrics.rs

@@ -0,0 +1,91 @@
+use super::ClientInfo;
+use linkerd_app_core::{
+    metrics::prom::{self, EncodeLabelSetMut},
+    svc, tls,
+    transport_header::{SessionProtocol, TransportHeader},
+};
+
+#[cfg(test)]
+mod tests;
+
+#[derive(Clone, Debug)]
+pub struct NewRecord<N> {
+    inner: N,
+    metrics: MetricsFamilies,
+}
+
+#[derive(Clone, Debug, Default)]
+pub struct MetricsFamilies {
+    connections: prom::Family<Labels, prom::Counter>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+struct Labels {
+    header: TransportHeader,
+    client_id: tls::ClientId,
+}
+
+impl MetricsFamilies {
+    pub fn register(reg: &mut prom::Registry) -> Self {
+        let connections = prom::Family::default();
+        reg.register(
+            "connections",
+            "TCP connections with transport headers",
+            connections.clone(),
+        );
+
+        Self { connections }
+    }
+}
+
+impl<N> NewRecord<N> {
+    pub fn layer(metrics: MetricsFamilies) -> impl svc::layer::Layer<N, Service = Self> + Clone {
+        svc::layer::mk(move |inner| Self {
+            inner,
+            metrics: metrics.clone(),
+        })
+    }
+}
+
+impl<N> svc::NewService<(TransportHeader, ClientInfo)> for NewRecord<N>
+where
+    N: svc::NewService<(TransportHeader, ClientInfo)>,
+{
+    type Service = N::Service;
+
+    fn new_service(&self, (header, client): (TransportHeader, ClientInfo)) -> Self::Service {
+        self.metrics
+            .connections
+            .get_or_create(&Labels {
+                header: header.clone(),
+                client_id: client.client_id.clone(),
+            })
+            .inc();
+
+        self.inner.new_service((header, client))
+    }
+}
+
+impl prom::EncodeLabelSetMut for Labels {
+    fn encode_label_set(&self, enc: &mut prom::encoding::LabelSetEncoder<'_>) -> std::fmt::Result {
+        use prom::encoding::EncodeLabel;
+        (
+            "session_protocol",
+            self.header.protocol.as_ref().map(|p| match p {
+                SessionProtocol::Http1 => "http/1",
+                SessionProtocol::Http2 => "http/2",
+            }),
+        )
+            .encode(enc.encode_label())?;
+        ("target_port", self.header.port).encode(enc.encode_label())?;
+        ("target_name", self.header.name.as_deref()).encode(enc.encode_label())?;
+        ("client_id", self.client_id.to_str()).encode(enc.encode_label())?;
+        Ok(())
+    }
+}
+
+impl prom::encoding::EncodeLabelSet for Labels {
+    fn encode(&self, mut enc: prom::encoding::LabelSetEncoder<'_>) -> Result<(), std::fmt::Error> {
+        self.encode_label_set(&mut enc)
+    }
+}

linkerd/app/inbound/src/direct/metrics/tests.rs

@@ -0,0 +1,115 @@
+use super::*;
+use crate::direct::ClientInfo;
+use futures::future;
+use linkerd_app_core::{
+    io,
+    metrics::prom,
+    svc, tls,
+    transport::addrs::{ClientAddr, OrigDstAddr, Remote},
+    transport_header::{SessionProtocol, TransportHeader},
+    Error,
+};
+use std::str::FromStr;
+
+fn new_ok<T>() -> svc::ArcNewTcp<T, io::BoxedIo> {
+    svc::ArcNewService::new(|_| svc::BoxService::new(svc::mk(|_| future::ok::<(), Error>(()))))
+}
+
+macro_rules! assert_counted {
+    ($registry:expr, $proto:expr, $port:expr, $name:expr, $value:expr) => {{
+        let mut buf = String::new();
+        prom::encoding::text::encode_registry(&mut buf, $registry).expect("encode registry failed");
+        let metric = format!("connections_total{{session_protocol=\"{}\",target_port=\"{}\",target_name=\"{}\",client_id=\"test.client\"}}", $proto, $port, $name);
+        assert_eq!(
+            buf.split_terminator('\n')
+                .find(|l| l.starts_with(&*metric)),
+            Some(&*format!("{metric} {}", $value)),
+            "metric '{metric}' not found in:\n{buf}"
+        );
+    }};
+}
+
+// Added helper to setup and run the test
+fn run_metric_test(header: TransportHeader) -> prom::Registry {
+    let mut registry = prom::Registry::default();
+    let families = MetricsFamilies::register(&mut registry);
+    let new_record = svc::layer::Layer::layer(&NewRecord::layer(families.clone()), new_ok());
+    // common client info
+    let client_id = tls::ClientId::from_str("test.client").unwrap();
+    let client_addr = Remote(ClientAddr(([127, 0, 0, 1], 40000).into()));
+    let local_addr = OrigDstAddr(([127, 0, 0, 1], 4143).into());
+    let client_info = ClientInfo {
+        client_id: client_id.clone(),
+        alpn: Some(tls::NegotiatedProtocol("transport.l5d.io/v1".into())),
+        client_addr,
+        local_addr,
+    };
+    let _svc = svc::NewService::new_service(&new_record, (header.clone(), client_info.clone()));
+    registry
+}
+
+#[test]
+fn records_metrics_http1_local() {
+    let header = TransportHeader {
+        port: 8080,
+        name: None,
+        protocol: Some(SessionProtocol::Http1),
+    };
+    let registry = run_metric_test(header);
+    assert_counted!(&registry, "http/1", 8080, "", 1);
+}
+
+#[test]
+fn records_metrics_http2_local() {
+    let header = TransportHeader {
+        port: 8081,
+        name: None,
+        protocol: Some(SessionProtocol::Http2),
+    };
+    let registry = run_metric_test(header);
+    assert_counted!(&registry, "http/2", 8081, "", 1);
+}
+
+#[test]
+fn records_metrics_opaq_local() {
+    let header = TransportHeader {
+        port: 8082,
+        name: None,
+        protocol: None,
+    };
+    let registry = run_metric_test(header);
+    assert_counted!(&registry, "", 8082, "", 1);
+}
+
+#[test]
+fn records_metrics_http1_gateway() {
+    let header = TransportHeader {
+        port: 8080,
+        name: Some("mysvc.myns.svc.cluster.local".parse().unwrap()),
+        protocol: Some(SessionProtocol::Http1),
+    };
+    let registry = run_metric_test(header);
+    assert_counted!(&registry, "http/1", 8080, "mysvc.myns.svc.cluster.local", 1);
+}
+
+#[test]
+fn records_metrics_http2_gateway() {
+    let header = TransportHeader {
+        port: 8081,
+        name: Some("mysvc.myns.svc.cluster.local".parse().unwrap()),
+        protocol: Some(SessionProtocol::Http2),
+    };
+    let registry = run_metric_test(header);
+    assert_counted!(&registry, "http/2", 8081, "mysvc.myns.svc.cluster.local", 1);
+}
+
+#[test]
+fn records_metrics_opaq_gateway() {
+    let header = TransportHeader {
+        port: 8082,
+        name: Some("mysvc.myns.svc.cluster.local".parse().unwrap()),
+        protocol: None,
+    };
+    let registry = run_metric_test(header);
+    assert_counted!(&registry, "", 8082, "mysvc.myns.svc.cluster.local", 1);
+}

linkerd/app/inbound/src/metrics.rs

@@ -27,12 +27,16 @@ pub struct InboundMetrics {
     pub proxy: Proxy,
 
     pub detect: crate::detect::MetricsFamilies,
+    pub direct: crate::direct::MetricsFamilies,
 }
 
 impl InboundMetrics {
     pub(crate) fn new(proxy: Proxy, reg: &mut prom::Registry) -> Self {
         let detect =
             crate::detect::MetricsFamilies::register(reg.sub_registry_with_prefix("tcp_detect"));
+        let direct = crate::direct::MetricsFamilies::register(
+            reg.sub_registry_with_prefix("tcp_transport_header"),
+        );
 
         Self {
             http_authz: authz::HttpAuthzMetrics::default(),
@@ -41,6 +45,7 @@ impl InboundMetrics {
             tcp_errors: error::TcpErrorMetrics::default(),
             proxy,
             detect,
+            direct,
         }
     }
 }