feat(app/inbound): request body frame size metrics

this commit introduces an additional layer of telemetry to the inbound
proxy's http router.

either http and grpc metrics are used, depending upon the policy that
authorized a given request.

this is based upon #4174, which refactored the
request body telemetry middleware to be metrics agnostic.

see:
* #4165
* #4166
* #4174
* #4127

Signed-off-by: katelyn martin <[email protected]>

Author: cratelyn

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

@@ -18,14 +18,22 @@ use linkerd_http_prom::{
 pub(super) fn layer<N>(
     InboundMetrics {
         request_count,
+        request_body_data,
         response_body_data,
         ..
     }: &InboundMetrics,
 ) -> impl svc::Layer<
     N,
     Service = NewCountRequests<
         ExtractRequestCount,
-        NewRecordBodyData<ExtractRecordBodyDataMetrics, N>,
+        NewRecordBodyData<
+            ExtractRecordBodyDataMetrics,
+            linkerd_http_prom::body_data::request::NewRecordBodyData<
+                N,
+                ExtractRequestBodyDataParams,
+                ExtractRequestBodyDataMetrics,
+            >,
+        >,
     >,
 > {
     use svc::Layer as _;
@@ -40,7 +48,12 @@ pub(super) fn layer<N>(
         NewRecordBodyData::layer_via(extract)
     };
 
-    svc::layer::mk(move |inner| count.layer(body.layer(inner)))
+    let request = {
+        let extract = ExtractRequestBodyDataParams(request_body_data.clone());
+        linkerd_http_prom::body_data::request::NewRecordBodyData::new(extract)
+    };
+
+    svc::layer::mk(move |inner| count.layer(body.layer(request.layer(inner))))
 }
 
 #[derive(Clone, Debug)]
@@ -71,6 +84,23 @@ pub struct ResponseBodyDataLabels {
 #[derive(Clone, Debug)]
 pub struct ExtractRecordBodyDataMetrics(ResponseBodyFamilies);
 
+#[derive(Clone, Debug)]
+pub struct RequestBodyFamilies {
+    grpc: linkerd_http_prom::body_data::request::RequestBodyFamilies<RequestBodyDataLabels>,
+    http: linkerd_http_prom::body_data::request::RequestBodyFamilies<RequestBodyDataLabels>,
+}
+
+#[derive(Clone, Debug, Hash, PartialEq, Eq)]
+pub struct RequestBodyDataLabels {
+    route: RouteLabels,
+}
+
+#[derive(Clone, Debug)]
+pub struct ExtractRequestBodyDataParams(RequestBodyFamilies);
+
+#[derive(Clone, Debug)]
+pub struct ExtractRequestBodyDataMetrics(BodyDataMetrics);
+
 // === impl RequestCountFamilies ===
 
 impl RequestCountFamilies {
@@ -222,3 +252,91 @@ impl<T> svc::ExtractParam<BodyDataMetrics, Permitted<T>> for ExtractRecordBodyDa
         family.metrics(&ResponseBodyDataLabels { route })
     }
 }
+
+// === impl RequestBodyFamilies ===
+
+impl RequestBodyFamilies {
+    /// Registers a new [`RequestBodyDataFamilies`] with the given registry.
+    pub fn register(reg: &mut prom::Registry) -> Self {
+        let grpc = {
+            let reg = reg.sub_registry_with_prefix("grpc");
+            linkerd_http_prom::body_data::request::RequestBodyFamilies::register(reg)
+        };
+
+        let http = {
+            let reg = reg.sub_registry_with_prefix("http");
+            linkerd_http_prom::body_data::request::RequestBodyFamilies::register(reg)
+        };
+
+        Self { grpc, http }
+    }
+
+    /// Fetches the proper body frame metrics family, given a permitted target.
+    fn family<T>(
+        &self,
+        permitted: &Permitted<T>,
+    ) -> &linkerd_http_prom::body_data::request::RequestBodyFamilies<RequestBodyDataLabels> {
+        let Self { grpc, http } = self;
+        match permitted {
+            Permitted::Grpc { .. } => grpc,
+            Permitted::Http { .. } => http,
+        }
+    }
+}
+
+// === impl RequestBodyDataLabels ===
+
+impl EncodeLabelSetMut for RequestBodyDataLabels {
+    fn encode_label_set(&self, enc: &mut LabelSetEncoder<'_>) -> std::fmt::Result {
+        use encoding::EncodeLabel as _;
+
+        let Self {
+            route:
+                RouteLabels {
+                    server: ServerLabel(parent, port),
+                    route,
+                },
+        } = self;
+
+        ("parent_group", parent.group()).encode(enc.encode_label())?;
+        ("parent_kind", parent.kind()).encode(enc.encode_label())?;
+        ("parent_name", parent.name()).encode(enc.encode_label())?;
+        ("parent_port", *port).encode(enc.encode_label())?;
+
+        ("route_group", route.group()).encode(enc.encode_label())?;
+        ("route_kind", route.kind()).encode(enc.encode_label())?;
+        ("route_name", route.name()).encode(enc.encode_label())?;
+
+        Ok(())
+    }
+}
+
+impl EncodeLabelSet for RequestBodyDataLabels {
+    fn encode(&self, mut enc: LabelSetEncoder<'_>) -> std::fmt::Result {
+        self.encode_label_set(&mut enc)
+    }
+}
+
+// === impl ExtractRequestBodyDataParams ===
+
+impl<T> svc::ExtractParam<ExtractRequestBodyDataMetrics, Permitted<T>>
+    for ExtractRequestBodyDataParams
+{
+    fn extract_param(&self, permitted: &Permitted<T>) -> ExtractRequestBodyDataMetrics {
+        let Self(families) = self;
+        let family = families.family(permitted);
+        let route = permitted.route_labels();
+        let metrics = family.metrics(&RequestBodyDataLabels { route });
+        ExtractRequestBodyDataMetrics(metrics)
+    }
+}
+
+// === impl ExtractRequestBodyDataMetrics ===
+
+impl<T> svc::ExtractParam<BodyDataMetrics, T> for ExtractRequestBodyDataMetrics {
+    fn extract_param(&self, _: &T) -> BodyDataMetrics {
+        let Self(metrics) = self;
+
+        metrics.clone()
+    }
+}

linkerd/app/inbound/src/metrics.rs

@@ -11,7 +11,7 @@
 pub(crate) mod authz;
 pub(crate) mod error;
 
-use crate::http::router::{RequestCountFamilies, ResponseBodyFamilies};
+use crate::http::router::{RequestBodyFamilies, RequestCountFamilies, ResponseBodyFamilies};
 pub use linkerd_app_core::metrics::*;
 
 /// Holds LEGACY inbound proxy metrics.
@@ -30,6 +30,7 @@ pub struct InboundMetrics {
     pub detect: crate::detect::MetricsFamilies,
     pub direct: crate::direct::MetricsFamilies,
     pub request_count: RequestCountFamilies,
+    pub request_body_data: RequestBodyFamilies,
     pub response_body_data: ResponseBodyFamilies,
 }
 
@@ -41,6 +42,7 @@ impl InboundMetrics {
             reg.sub_registry_with_prefix("tcp_transport_header"),
         );
         let request_count = RequestCountFamilies::register(reg);
+        let request_body_data = RequestBodyFamilies::register(reg);
         let response_body_data = ResponseBodyFamilies::register(reg);
 
         Self {
@@ -52,6 +54,7 @@ impl InboundMetrics {
             detect,
             direct,
             request_count,
+            request_body_data,
             response_body_data,
         }
     }