xxx(app/inbound): inbound metrics prototype

Cargo.lock

@@ -1508,6 +1508,7 @@ dependencies = [
  "linkerd-http-access-log",
  "linkerd-http-box",
  "linkerd-http-metrics",
+ "linkerd-http-prom",
  "linkerd-idle-cache",
  "linkerd-io",
  "linkerd-meshtls",
@@ -1519,6 +1520,7 @@ dependencies = [
  "linkerd2-proxy-api",
  "once_cell",
  "parking_lot",
+ "prometheus-client",
  "rangemap",
  "thiserror 2.0.16",
  "tokio",

linkerd/app/inbound/Cargo.toml

@@ -23,6 +23,7 @@ futures = { version = "0.3", default-features = false }
 linkerd-app-core = { path = "../core" }
 linkerd-app-test = { path = "../test", optional = true }
 linkerd-http-access-log = { path = "../../http/access-log" }
+linkerd-http-prom = { path = "../../http/prom" }
 linkerd-idle-cache = { path = "../../idle-cache" }
 linkerd-meshtls = { path = "../../meshtls", optional = true, default-features = false }
 linkerd-proxy-client-policy = { path = "../../proxy/client-policy" }
@@ -31,6 +32,7 @@ linkerd-tonic-watch = { path = "../../tonic-watch" }
 linkerd2-proxy-api = { workspace = true, features = ["inbound"] }
 once_cell = "1"
 parking_lot = "0.12"
+prometheus-client = { workspace = true }
 rangemap = "1"
 thiserror = "2"
 tokio = { version = "1", features = ["sync"] }

linkerd/app/inbound/src/http.rs

@@ -1,4 +1,4 @@
-mod router;
+pub(crate) mod router;
 mod server;
 mod set_identity_header;
 #[cfg(test)]

linkerd/app/inbound/src/http/router.rs

@@ -10,6 +10,10 @@ use linkerd_app_core::{
 use std::{fmt, net::SocketAddr};
 use tracing::{debug, debug_span};
 
+pub use self::metrics::RequestCountLabels;
+
+mod metrics;
+
 /// Describes an HTTP client target.
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub struct Http {
@@ -241,6 +245,7 @@ impl<C> Inbound<C> {
                 .push(svc::NewOneshotRoute::layer_via(|t: &policy::Permitted<T>| {
                     LogicalPerRequest::from(t)
                 }))
+                .push(self::metrics::layer(rt.metrics.request_count.clone()))
                 .check_new_service::<policy::Permitted<T>, http::Request<http::BoxBody>>()
                 .push(svc::ArcNewService::layer())
                 .push(policy::NewHttpPolicy::layer(rt.metrics.http_authz.clone()))

linkerd/app/inbound/src/http/router/metrics.rs

@@ -0,0 +1,79 @@
+use linkerd_app_core::{
+    metrics::{prom::EncodeLabelSetMut, ServerLabel},
+    proxy::http,
+    svc::{self, Param},
+    tls,
+    transport::{ClientAddr, Remote, ServerAddr},
+};
+use linkerd_http_prom::{NewCountRequests, RequestCount, RequestCountFamilies};
+use prometheus_client::encoding::{EncodeLabelSet, LabelSetEncoder};
+
+pub(super) fn layer<N>(
+    request_count: RequestCountFamilies<RequestCountLabels>,
+    // TODO(kate): other metrics families will added here.
+) -> impl svc::Layer<N, Service = NewCountRequests<ExtractRequestCount, N>> {
+    svc::layer::mk(move |inner| {
+        use svc::Layer as _;
+        let extract = ExtractRequestCount(request_count.clone());
+        NewCountRequests::layer_via(extract).layer(inner)
+    })
+}
+
+#[derive(Clone, Debug, Hash, PartialEq, Eq)]
+pub struct RequestCountLabels {
+    server: ServerLabel,
+}
+
+#[derive(Clone, Debug)]
+pub struct ExtractRequestCount(pub RequestCountFamilies<RequestCountLabels>);
+
+// === impl ExtractRequestCount ===
+
+impl<T> svc::ExtractParam<RequestCount, T> for ExtractRequestCount
+where
+    T: Param<http::Variant>
+        + Param<Remote<ServerAddr>>
+        + Param<Remote<ClientAddr>>
+        + Param<tls::ConditionalServerTls>
+        + Param<crate::policy::AllowPolicy>,
+{
+    fn extract_param(&self, target: &T) -> RequestCount {
+        let Self(family) = self;
+
+        // XXX(kate): `Param<T>` stubs for labels we can potentially introduce.
+        let policy: crate::policy::AllowPolicy = target.param();
+        let _: http::Variant = target.param();
+        let _: Remote<ServerAddr> = target.param();
+        let _: Remote<ClientAddr> = target.param();
+        let _: tls::ConditionalServerTls = target.param();
+        let _: crate::policy::AllowPolicy = target.param();
+
+        let server = policy.server_label();
+
+        family.metrics(&RequestCountLabels { server })
+    }
+}
+
+// === impl Labels ===
+
+impl EncodeLabelSetMut for RequestCountLabels {
+    fn encode_label_set(&self, enc: &mut LabelSetEncoder<'_>) -> std::fmt::Result {
+        let Self {
+            server: ServerLabel(meta, port),
+        } = self;
+
+        use prometheus_client::encoding::EncodeLabel as _;
+        ("parent_group", meta.group()).encode(enc.encode_label())?;
+        ("parent_kind", meta.kind()).encode(enc.encode_label())?;
+        ("parent_name", meta.name()).encode(enc.encode_label())?;
+        ("parent_port", *port).encode(enc.encode_label())?;
+
+        Ok(())
+    }
+}
+
+impl EncodeLabelSet for RequestCountLabels {
+    fn encode(&self, mut enc: LabelSetEncoder<'_>) -> std::fmt::Result {
+        self.encode_label_set(&mut enc)
+    }
+}

linkerd/app/inbound/src/metrics.rs

@@ -11,7 +11,9 @@
 pub(crate) mod authz;
 pub(crate) mod error;
 
+use crate::http::router::RequestCountLabels;
 pub use linkerd_app_core::metrics::*;
+use linkerd_http_prom::RequestCountFamilies;
 
 /// Holds LEGACY inbound proxy metrics.
 #[derive(Clone, Debug)]
@@ -28,6 +30,7 @@ pub struct InboundMetrics {
 
     pub detect: crate::detect::MetricsFamilies,
     pub direct: crate::direct::MetricsFamilies,
+    pub request_count: RequestCountFamilies<RequestCountLabels>,
 }
 
 impl InboundMetrics {
@@ -37,6 +40,7 @@ impl InboundMetrics {
         let direct = crate::direct::MetricsFamilies::register(
             reg.sub_registry_with_prefix("tcp_transport_header"),
         );
+        let request_count = RequestCountFamilies::register(reg);
 
         Self {
             http_authz: authz::HttpAuthzMetrics::default(),
@@ -46,6 +50,7 @@ impl InboundMetrics {
             proxy,
             detect,
             direct,
+            request_count,
         }
     }
 }