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

#3308 introduced tower
middleware to the outbound proxy that records the size and number of
response body's data frames.

this commit introduces an equivalent layer to the inbound proxy's
prometheus metrics.

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

Author: cratelyn

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

@@ -10,7 +10,7 @@ use linkerd_app_core::{
 use std::{fmt, net::SocketAddr};
 use tracing::{debug, debug_span};
 
-pub use self::metrics::RequestCountFamilies;
+pub use self::metrics::*;
 
 mod metrics;
 
@@ -245,7 +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()))
+                .push(self::metrics::layer(rt.metrics.request_count.clone(), rt.metrics.response_body_data.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

@@ -10,16 +10,31 @@ use linkerd_app_core::{
     },
     svc,
 };
-use linkerd_http_prom::count_reqs::{NewCountRequests, RequestCount};
+use linkerd_http_prom::{
+    body_data::response::{BodyDataMetrics, NewRecordBodyData},
+    count_reqs::{NewCountRequests, RequestCount},
+};
 
 pub(super) fn layer<N>(
     request_count: RequestCountFamilies,
-    // TODO(kate): other metrics families will added here.
-) -> impl svc::Layer<N, Service = NewCountRequests<ExtractRequestCount, N>> {
+    response_body_metrics: ResponseBodyFamilies,
+) -> impl svc::Layer<
+    N,
+    Service = NewCountRequests<
+        ExtractRequestCount,
+        NewRecordBodyData<ExtractRecordBodyDataMetrics, N>,
+    >,
+> {
     svc::layer::mk(move |inner| {
         use svc::Layer as _;
         let extract = ExtractRequestCount(request_count.clone());
-        NewCountRequests::layer_via(extract).layer(inner)
+
+        NewCountRequests::layer_via(extract).layer(
+            NewRecordBodyData::layer_via(ExtractRecordBodyDataMetrics(
+                response_body_metrics.clone(),
+            ))
+            .layer(inner),
+        )
     })
 }
 
@@ -37,6 +52,20 @@ pub struct RequestCountLabels {
 #[derive(Clone, Debug)]
 pub struct ExtractRequestCount(pub RequestCountFamilies);
 
+#[derive(Clone, Debug)]
+pub struct ResponseBodyFamilies {
+    grpc: linkerd_http_prom::body_data::response::ResponseBodyFamilies<ResponseBodyDataLabels>,
+    http: linkerd_http_prom::body_data::response::ResponseBodyFamilies<ResponseBodyDataLabels>,
+}
+
+#[derive(Clone, Debug, Hash, PartialEq, Eq)]
+pub struct ResponseBodyDataLabels {
+    route: RouteLabels,
+}
+
+#[derive(Clone, Debug)]
+pub struct ExtractRecordBodyDataMetrics(ResponseBodyFamilies);
+
 // === impl RequestCountFamilies ===
 
 impl RequestCountFamilies {
@@ -112,3 +141,79 @@ impl EncodeLabelSet for RequestCountLabels {
         self.encode_label_set(&mut enc)
     }
 }
+
+// === impl ResponseBodyFamilies ===
+
+impl ResponseBodyFamilies {
+    /// Registers a new [`ResponseBodyDataFamilies`] 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::response::ResponseBodyFamilies::register(reg)
+        };
+
+        let http = {
+            let reg = reg.sub_registry_with_prefix("http");
+            linkerd_http_prom::body_data::response::ResponseBodyFamilies::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::response::ResponseBodyFamilies<ResponseBodyDataLabels> {
+        let Self { grpc, http } = self;
+        match permitted {
+            Permitted::Grpc { .. } => grpc,
+            Permitted::Http { .. } => http,
+        }
+    }
+}
+
+// === impl ResponseBodyDataLabels ===
+
+impl EncodeLabelSetMut for ResponseBodyDataLabels {
+    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 ResponseBodyDataLabels {
+    fn encode(&self, mut enc: LabelSetEncoder<'_>) -> std::fmt::Result {
+        self.encode_label_set(&mut enc)
+    }
+}
+
+// === impl ExtractRecordBodyDataMetrics ===
+
+impl<T> svc::ExtractParam<BodyDataMetrics, Permitted<T>> for ExtractRecordBodyDataMetrics {
+    fn extract_param(&self, permitted: &Permitted<T>) -> BodyDataMetrics {
+        let Self(families) = self;
+        let family = families.family(permitted);
+        let route = permitted.route_labels();
+
+        family.metrics(&ResponseBodyDataLabels { route })
+    }
+}

linkerd/app/inbound/src/metrics.rs

@@ -11,7 +11,7 @@
 pub(crate) mod authz;
 pub(crate) mod error;
 
-use crate::http::router::RequestCountFamilies;
+use crate::http::router::{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 response_body_data: ResponseBodyFamilies,
 }
 
 impl InboundMetrics {
@@ -40,6 +41,7 @@ impl InboundMetrics {
             reg.sub_registry_with_prefix("tcp_transport_header"),
         );
         let request_count = RequestCountFamilies::register(reg);
+        let response_body_data = ResponseBodyFamilies::register(reg);
 
         Self {
             http_authz: authz::HttpAuthzMetrics::default(),
@@ -50,6 +52,7 @@ impl InboundMetrics {
             detect,
             direct,
             request_count,
+            response_body_data,
         }
     }
 }