refactor(http/prom): request::NewRecordBodyData is metrics agnostic

this commit makes a mechanical refactor in the interest of paving ground
towards using the request body metrics middleware in the inbound proxy.

because of minor differences in how we represent protocols like gRPC in
the inbound and outbound proxies, we must account for:

(a): distinct `NewService<T>`'s accepting `Grpc<T>` and `Http<T>`
targets, respectively, each holding a single metrics `Family<L>`.

(b): a single `NewService<T>` that matches on the `Permitted<T>` enum to
select one of two metrics `Family<L>`s.

see #4119 for more background on this
`Permitted<T>` enum.

so, we should make the `NewRecordBodyData` agnostic to its metrics,
rather than holding a concrete `RequestBodyFamilies<L>`.

this commit hoists the metrics out of the middleware, and into the `X`
and `ReqX` extractors used in the `NewService<T>` and `Service<T>`
layers, respectively.

bear particular attention to this part of the diff:

```diff
-    ReqX: ExtractParam<L, Request<ReqB>>,
-    L: linkerd_metrics::prom::encoding::EncodeLabelSet
-        + std::fmt::Debug
-        + std::hash::Hash
-        + Eq
-        + Clone
-        + Send
-        + Sync
-        + 'static,
+    ReqX: ExtractParam<BodyDataMetrics, Request<ReqB>>,
```

in other words, rather than using `X` and subsequently `ReqX` to extract
our labels `L`, we expect these extractors to yield a `BodyDataMetrics`:
one time series that should be incremented accordingly by the
instrumented request.

in even simpler words: `ReqX` now calls `familes.metrics(&labels)` too.

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

Author: cratelyn

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

@@ -5,7 +5,7 @@ use linkerd_app_core::{
     svc,
 };
 use linkerd_http_prom::{
-    body_data::request::{NewRecordBodyData, RequestBodyFamilies},
+    body_data::request::{BodyDataMetrics, NewRecordBodyData, RequestBodyFamilies},
     record_response::{self, StreamLabel},
 };
 
@@ -61,24 +61,36 @@ pub type NewRecordDuration<T, M, N> =
 #[derive(Clone, Debug)]
 pub struct ExtractRecordDurationParams<M>(pub M);
 
+/// Extracts a [`ExtractBodyDataParams`] from a target.
+#[derive(Clone, Debug)]
+pub struct ExtractBodyDataParams(RequestBodyFamilies<labels::Route>);
+
+/// Extracts a single time series from the request body metrics.
+#[derive(Clone, Debug)]
+pub struct ExtractBodyDataMetrics<X> {
+    /// The labeled family of metrics.
+    metrics: RequestBodyFamilies<labels::Route>,
+    /// Extracts [`labels::Route`] to select a time series.
+    extract: X,
+}
+
 pub fn layer<T, N>(
     metrics: &RequestMetrics<T::StreamLabel>,
     body_data: &RequestBodyFamilies<labels::Route>,
 ) -> impl svc::Layer<
     N,
     Service = NewRecordBodyData<
         NewRecordDuration<T, RequestMetrics<T::StreamLabel>, N>,
-        (),
-        T,
-        labels::Route,
+        ExtractBodyDataParams,
+        ExtractBodyDataMetrics<T>,
     >,
 >
 where
     T: Clone + MkStreamLabel,
     T: svc::ExtractParam<labels::Route, http::Request<http::BoxBody>>,
 {
     let record = NewRecordDuration::layer_via(ExtractRecordDurationParams(metrics.clone()));
-    let body_data = NewRecordBodyData::new((), body_data.clone());
+    let body_data = NewRecordBodyData::new(ExtractBodyDataParams(body_data.clone()));
 
     svc::layer::mk(move |inner| {
         use svc::Layer;
@@ -172,6 +184,31 @@ where
     }
 }
 
+// === impl ExtractRecordBodyDataParams ===
+
+impl<T: Clone> svc::ExtractParam<ExtractBodyDataMetrics<T>, T> for ExtractBodyDataParams {
+    fn extract_param(&self, t: &T) -> ExtractBodyDataMetrics<T> {
+        let Self(metrics) = self;
+        ExtractBodyDataMetrics {
+            metrics: metrics.clone(),
+            extract: t.clone(),
+        }
+    }
+}
+
+// === impl ExtractRecordBodyDataMetrics ===
+
+impl<B, X> svc::ExtractParam<BodyDataMetrics, http::Request<B>> for ExtractBodyDataMetrics<X>
+where
+    X: svc::ExtractParam<labels::Route, http::Request<B>>,
+{
+    fn extract_param(&self, req: &http::Request<B>) -> BodyDataMetrics {
+        let Self { metrics, extract } = self;
+        let labels = extract.extract_param(req);
+        metrics.metrics(&labels)
+    }
+}
+
 // === impl LabelHttpRsp ===
 
 impl<P> From<P> for LabelHttpRsp<P> {

linkerd/http/prom/src/body_data/request.rs

@@ -10,90 +10,71 @@ use std::marker::PhantomData;
 
 /// A [`NewService<T>`] that creates [`RecordBodyData`] services.
 #[derive(Clone, Debug)]
-pub struct NewRecordBodyData<N, X, ReqX, L> {
+pub struct NewRecordBodyData<N, X, ReqX> {
     /// The inner [`NewService<T>`].
     inner: N,
     extract: X,
-    metrics: RequestBodyFamilies<L>,
     marker: PhantomData<ReqX>,
 }
 
 /// Tracks body frames for an inner `S`-typed [`Service`].
 #[derive(Clone, Debug)]
-pub struct RecordBodyData<S, ReqX, L> {
+pub struct RecordBodyData<S, ReqX> {
     /// The inner [`Service<T>`].
     inner: S,
     extract: ReqX,
-    metrics: RequestBodyFamilies<L>,
 }
 
 // === impl NewRecordBodyData ===
 
-impl<N, X, ReqX, L> NewRecordBodyData<N, X, ReqX, L>
+impl<N, X, ReqX> NewRecordBodyData<N, X, ReqX>
 where
     X: Clone,
-    L: Clone,
 {
     /// Returns a [`Layer<S>`] that tracks body chunks.
     ///
     /// This uses an `X`-typed [`ExtractParam<P, T>`] implementation to extract service parameters
     /// from a `T`-typed target.
-    pub fn new(extract: X, metrics: RequestBodyFamilies<L>) -> impl Layer<N, Service = Self> {
+    pub fn new(extract: X) -> impl Layer<N, Service = Self> {
         svc::layer::mk(move |inner| Self {
             inner,
             extract: extract.clone(),
-            metrics: metrics.clone(),
             marker: PhantomData,
         })
     }
 }
 
-impl<T, N, X, ReqX, L> NewService<T> for NewRecordBodyData<N, X, ReqX, L>
+impl<T, N, X, ReqX> NewService<T> for NewRecordBodyData<N, X, ReqX>
 where
     N: NewService<T>,
     X: ExtractParam<ReqX, T>,
-    L: Clone,
 {
-    type Service = RecordBodyData<N::Service, ReqX, L>;
+    type Service = RecordBodyData<N::Service, ReqX>;
 
     fn new_service(&self, target: T) -> Self::Service {
         let Self {
             inner,
             extract,
-            metrics,
             marker: _,
         } = self;
 
         let extract = extract.extract_param(&target);
         let inner = inner.new_service(target);
-        let metrics = metrics.clone();
 
-        RecordBodyData {
-            inner,
-            extract,
-            metrics,
-        }
+        RecordBodyData { inner, extract }
     }
 }
 
 // === impl RecordBodyData ===
 
-impl<ReqB, RespB, S, ReqX, L> Service<Request<ReqB>> for RecordBodyData<S, ReqX, L>
+impl<ReqB, RespB, S, ReqX> Service<Request<ReqB>> for RecordBodyData<S, ReqX>
 where
     S: Service<Request<BoxBody>, Response = Response<RespB>>,
     S::Future: Send + 'static,
     ReqB: http_body::Body + Send + 'static,
     ReqB::Data: Send + 'static,
     ReqB::Error: Into<Error>,
-    ReqX: ExtractParam<L, Request<ReqB>>,
-    L: linkerd_metrics::prom::encoding::EncodeLabelSet
-        + std::fmt::Debug
-        + std::hash::Hash
-        + Eq
-        + Clone
-        + Send
-        + Sync
-        + 'static,
+    ReqX: ExtractParam<BodyDataMetrics, Request<ReqB>>,
 {
     type Response = S::Response;
     type Error = S::Error;
@@ -108,15 +89,10 @@ where
     }
 
     fn call(&mut self, req: Request<ReqB>) -> Self::Future {
-        let Self {
-            inner,
-            extract,
-            metrics,
-        } = self;
+        let Self { inner, extract } = self;
 
         let req = {
-            let labels = extract.extract_param(&req);
-            let metrics = metrics.metrics(&labels);
+            let metrics = extract.extract_param(&req);
             let instrument = |b| super::body::Body::new(b, metrics);
             req.map(instrument).map(BoxBody::new)
         };