feat(outbound): instrument per-parent connection protocol counts (#3733)

The outbound proxy makes protocol decisions based on the discovery response,
keyed on a "parent" reference.

This change adds a `protocol::metrics` middleware that records connection counts
by parent reference.

Author: olix0r

linkerd/app/outbound/src/metrics.rs

@@ -36,6 +36,7 @@ pub struct OutboundMetrics {
 
 #[derive(Clone, Debug, Default)]
 pub(crate) struct PromMetrics {
+    pub(crate) protocol: crate::protocol::MetricsFamilies,
     pub(crate) http_detect: crate::http::DetectMetricsFamilies<ParentRef>,
     pub(crate) http: crate::http::HttpMetrics,
     pub(crate) opaq: crate::opaq::OpaqMetrics,
@@ -89,6 +90,9 @@ where
 
 impl PromMetrics {
     pub fn register(registry: &mut prom::Registry) -> Self {
+        let protocol = crate::protocol::MetricsFamilies::register(
+            registry.sub_registry_with_prefix("tcp_protocol"),
+        );
         let http_detect = crate::http::DetectMetricsFamilies::register(
             // Scoped consistently with the inbound metrics.
             registry.sub_registry_with_prefix("tcp_detect_http"),
@@ -103,6 +107,7 @@ impl PromMetrics {
         let tls = crate::tls::TlsMetrics::register(registry.sub_registry_with_prefix("tls"));
 
         Self {
+            protocol,
             http_detect,
             http,
             opaq,

linkerd/app/outbound/src/protocol.rs

@@ -2,14 +2,20 @@ use crate::{http, Outbound, ParentRef};
 use linkerd_app_core::{io, svc, Error, Infallible};
 use std::{fmt::Debug, hash::Hash};
 
+mod metrics;
+#[cfg(test)]
+mod tests;
+
+pub use self::metrics::MetricsFamilies;
+
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub struct Http<T> {
     version: http::Variant,
     parent: T,
 }
 
 /// Parameter type indicating how the proxy should handle a connection.
-#[derive(Copy, Clone, Debug, Eq, PartialEq)]
+#[derive(Copy, Clone, Debug, Eq, PartialEq, Hash)]
 pub enum Protocol {
     Http1,
     Http2,
@@ -97,7 +103,7 @@ impl<N> Outbound<N> {
             .arc_new_tcp()
         });
 
-        http.map_stack(|_, _, http| {
+        http.map_stack(|_, rt, http| {
             // First separate traffic that needs protocol detection. Then switch
             // between traffic that is known to be HTTP or opaque.
             let known = http.push_switch(
@@ -131,6 +137,7 @@ impl<N> Outbound<N> {
                     },
                     detect.into_inner(),
                 )
+                .push(metrics::NewRecord::layer(rt.metrics.prom.protocol.clone()))
                 .arc_new_tcp()
         })
     }

linkerd/app/outbound/src/protocol/metrics.rs

@@ -0,0 +1,125 @@
+use super::Protocol;
+use crate::ParentRef;
+use linkerd_app_core::{
+    metrics::prom::{self, EncodeLabelSetMut},
+    svc,
+};
+
+#[derive(Clone, Debug)]
+pub struct NewRecord<N> {
+    inner: N,
+    metrics: MetricsFamilies,
+}
+
+#[derive(Clone, Debug)]
+pub struct Record<S> {
+    inner: S,
+    counter: prom::Counter,
+}
+
+#[derive(Clone, Debug, Default)]
+pub struct MetricsFamilies {
+    connections: prom::Family<Labels, prom::Counter>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+struct Labels {
+    protocol: Protocol,
+    parent_ref: ParentRef,
+}
+
+// === impl MetricsFamilies ===
+
+impl MetricsFamilies {
+    pub fn register(reg: &mut prom::Registry) -> Self {
+        let connections = prom::Family::default();
+        reg.register(
+            "connections",
+            "Outbound TCP connections by protocol configuration",
+            connections.clone(),
+        );
+
+        Self { connections }
+    }
+}
+
+// === impl NewRecord ===
+
+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<T, N> svc::NewService<T> for NewRecord<N>
+where
+    T: svc::Param<Protocol>,
+    T: svc::Param<ParentRef>,
+    N: svc::NewService<T>,
+{
+    type Service = Record<N::Service>;
+
+    fn new_service(&self, target: T) -> Self::Service {
+        let counter = (*self.metrics.connections.get_or_create(&Labels {
+            protocol: target.param(),
+            parent_ref: target.param(),
+        }))
+        .clone();
+
+        let inner = self.inner.new_service(target);
+        Record { inner, counter }
+    }
+}
+
+// === impl Record ===
+
+impl<S, I> svc::Service<I> for Record<S>
+where
+    S: svc::Service<I>,
+{
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = S::Future;
+
+    fn poll_ready(
+        &mut self,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, io: I) -> Self::Future {
+        self.counter.inc();
+        self.inner.call(io)
+    }
+}
+
+// === impl Labels ===
+
+impl prom::EncodeLabelSetMut for Labels {
+    fn encode_label_set(&self, enc: &mut prom::encoding::LabelSetEncoder<'_>) -> std::fmt::Result {
+        use prom::encoding::EncodeLabel;
+
+        let protocol = match self.protocol {
+            Protocol::Http1 => "http/1",
+            Protocol::Http2 => "http/2",
+            Protocol::Detect => "detect",
+            Protocol::Opaque => "opaq",
+            Protocol::Tls => "tls",
+        };
+
+        ("protocol", protocol).encode(enc.encode_label())?;
+        self.parent_ref.encode_label_set(enc)?;
+
+        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)
+    }
+}