feat(app): Outbount proxy uses route request frame metrics (wip)

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

linkerd/app/outbound/src/http/logical/policy/route.rs

@@ -133,7 +133,7 @@ where
                 // Set request extensions based on the route configuration
                 // AND/OR headers
                 .push(extensions::NewSetExtensions::layer())
-                .push(metrics::layer(&metrics.requests))
+                .push(metrics::layer(&metrics.requests, &metrics.body_data))
                 .check_new::<Self>()
                 .check_new_service::<Self, http::Request<http::BoxBody>>()
                 // Configure a classifier to use in the endpoint stack.

linkerd/app/outbound/src/http/logical/policy/route/metrics.rs

@@ -1,9 +1,13 @@
 use super::{backend::metrics as backend, retry};
+use crate::{ParentRef, RouteRef};
 use linkerd_app_core::{
     metrics::prom::{self, EncodeLabelSetMut},
     svc,
 };
-use linkerd_http_prom::record_response::{self, StreamLabel};
+use linkerd_http_prom::{
+    body_data::request::{BodyDataMetrics, NewRecordBodyData, RequestBodyFamilies},
+    record_response::{self, StreamLabel},
+};
 
 pub use linkerd_http_prom::record_response::MkStreamLabel;
 
@@ -23,6 +27,7 @@ pub struct RouteMetrics<R: StreamLabel, B: StreamLabel> {
     pub(super) retry: retry::RouteRetryMetrics,
     pub(super) requests: RequestMetrics<R>,
     pub(super) backend: backend::RouteBackendMetrics<B>,
+    pub(super) body_data: RequestBodyFamilies<labels::Route>,
 }
 
 pub type HttpRouteMetrics = RouteMetrics<LabelHttpRouteRsp, LabelHttpRouteBackendRsp>;
@@ -56,13 +61,26 @@ pub type NewRecordDuration<T, M, N> =
 #[derive(Clone, Debug)]
 pub struct ExtractRecordDurationParams<M>(pub M);
 
+#[derive(Clone, Debug)]
+pub struct ExtractRecordBodyDataParams(RequestBodyFamilies<labels::Route>);
+
 pub fn layer<T, N>(
     metrics: &RequestMetrics<T::StreamLabel>,
-) -> impl svc::Layer<N, Service = NewRecordDuration<T, RequestMetrics<T::StreamLabel>, N>> + Clone
+    body_data: &RequestBodyFamilies<labels::Route>,
+) -> impl svc::Layer<
+    N,
+    Service = NewRecordBodyData<
+        ExtractRecordBodyDataParams,
+        NewRecordDuration<T, RequestMetrics<T::StreamLabel>, N>,
+    >,
+>
 where
     T: Clone + MkStreamLabel,
 {
-    NewRecordDuration::layer_via(ExtractRecordDurationParams(metrics.clone()))
+    let record = NewRecordDuration::layer_via(ExtractRecordDurationParams(metrics.clone()));
+    let body_data = NewRecordBodyData::layer_via(ExtractRecordBodyDataParams(body_data.clone()));
+
+    svc::layers().push(record).push(body_data)
 }
 
 // === impl RouteMetrics ===
@@ -89,6 +107,7 @@ impl<R: StreamLabel, B: StreamLabel> Default for RouteMetrics<R, B> {
             requests: Default::default(),
             backend: Default::default(),
             retry: Default::default(),
+            body_data: Default::default(),
         }
     }
 }
@@ -99,6 +118,7 @@ impl<R: StreamLabel, B: StreamLabel> Clone for RouteMetrics<R, B> {
             requests: self.requests.clone(),
             backend: self.backend.clone(),
             retry: self.retry.clone(),
+            body_data: self.body_data.clone(),
         }
     }
 }
@@ -113,11 +133,13 @@ impl<R: StreamLabel, B: StreamLabel> RouteMetrics<R, B> {
         );
 
         let retry = retry::RouteRetryMetrics::register(reg.sub_registry_with_prefix("retry"));
+        let body_data = RequestBodyFamilies::register(reg);
 
         Self {
             requests,
             backend,
             retry,
+            body_data,
         }
     }
 
@@ -147,6 +169,27 @@ where
     }
 }
 
+// === impl ExtractRecordBodyDataParams ===
+
+impl<T> svc::ExtractParam<BodyDataMetrics, T> for ExtractRecordBodyDataParams
+where
+    T: svc::Param<ParentRef> + svc::Param<RouteRef>,
+{
+    fn extract_param(&self, target: &T) -> BodyDataMetrics {
+        let Self(families) = self;
+
+        #[allow(unreachable_code)] // XXX(kate); remove this once `todo!()` is addressed.
+        let labels = {
+            let parent = target.param();
+            let route = target.param();
+            let uri = todo!();
+            labels::Route::new(parent, route, uri)
+        };
+
+        families.get(&labels)
+    }
+}
+
 // === impl LabelHttpRsp ===
 
 impl<P> From<P> for LabelHttpRsp<P> {