refactor(outbound): simplify http route metric labeling (#3604)

The outbound HTTP route labels use multiple ExtractParam implementations to
convert a matched route target to set of labels (using an HTTP request).

This change removes the RouteLabelExtract type, in favor of implementing label
extraction on the route target type directly.

Author: olix0r

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

@@ -88,6 +88,7 @@ where
     Self: svc::Param<classify::Request>,
     Self: svc::Param<extensions::Params>,
     Self: metrics::MkStreamLabel,
+    Self: svc::ExtractParam<metrics::labels::Route, http::Request<http::BoxBody>>,
     MatchedBackend<T, M, F>: filters::Apply,
     MatchedBackend<T, M, F>: metrics::MkStreamLabel,
 {
@@ -123,21 +124,13 @@ where
                 // leaking tasks onto the runtime.
                 .push_on_service(svc::LoadShed::layer())
                 .push(filters::NewApplyFilters::<Self, _, _>::layer())
-                .push({
-                    let mk = Self::label_extractor;
-                    let metrics = metrics.retry.clone();
-                    retry::NewHttpRetry::layer_via_mk(mk, metrics)
-                })
+                .push(retry::NewHttpRetry::<Self, _>::layer(metrics.retry.clone()))
                 .check_new::<Self>()
                 .check_new_service::<Self, http::Request<http::BoxBody>>()
                 // Set request extensions based on the route configuration
                 // AND/OR headers
                 .push(extensions::NewSetExtensions::layer())
-                .push(metrics::layer(
-                    &metrics.requests,
-                    Self::label_extractor,
-                    &metrics.body_data,
-                ))
+                .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.
@@ -176,6 +169,16 @@ impl<T> filters::Apply for Http<T> {
     }
 }
 
+impl<B, T> svc::ExtractParam<metrics::labels::Route, http::Request<B>> for Http<T> {
+    fn extract_param(&self, req: &http::Request<B>) -> metrics::labels::Route {
+        metrics::labels::Route::new(
+            self.params.parent_ref.clone(),
+            self.params.route_ref.clone(),
+            req.uri(),
+        )
+    }
+}
+
 impl<T> metrics::MkStreamLabel for Http<T> {
     type StatusLabels = metrics::labels::HttpRouteRsp;
     type DurationLabels = metrics::labels::Route;
@@ -232,6 +235,16 @@ impl<T> filters::Apply for Grpc<T> {
     }
 }
 
+impl<B, T> svc::ExtractParam<metrics::labels::Route, http::Request<B>> for Grpc<T> {
+    fn extract_param(&self, req: &http::Request<B>) -> metrics::labels::Route {
+        metrics::labels::Route::new(
+            self.params.parent_ref.clone(),
+            self.params.route_ref.clone(),
+            req.uri(),
+        )
+    }
+}
+
 impl<T> metrics::MkStreamLabel for Grpc<T> {
     type StatusLabels = metrics::labels::GrpcRouteRsp;
     type DurationLabels = metrics::labels::Route;

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

@@ -1,6 +1,7 @@
 use super::{backend::metrics as backend, retry};
 use linkerd_app_core::{
     metrics::prom::{self, EncodeLabelSetMut},
+    proxy::http,
     svc,
 };
 use linkerd_http_prom::{
@@ -60,25 +61,24 @@ pub type NewRecordDuration<T, M, N> =
 #[derive(Clone, Debug)]
 pub struct ExtractRecordDurationParams<M>(pub M);
 
-pub fn layer<T, N, X>(
+pub fn layer<T, N>(
     metrics: &RequestMetrics<T::StreamLabel>,
-    mk: X,
     body_data: &RequestBodyFamilies<labels::Route>,
 ) -> impl svc::Layer<
     N,
     Service = NewRecordBodyData<
         NewRecordDuration<T, RequestMetrics<T::StreamLabel>, N>,
-        X,
-        labels::RouteLabelExtract,
+        (),
+        T,
         labels::Route,
     >,
 >
 where
     T: Clone + MkStreamLabel,
-    X: Clone,
+    T: svc::ExtractParam<labels::Route, http::Request<http::BoxBody>>,
 {
     let record = NewRecordDuration::layer_via(ExtractRecordDurationParams(metrics.clone()));
-    let body_data = NewRecordBodyData::new(mk, body_data.clone());
+    let body_data = NewRecordBodyData::new((), body_data.clone());
 
     svc::layer::mk(move |inner| {
         use svc::Layer;

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

@@ -1,7 +1,6 @@
 //! Prometheus label types.
 use linkerd_app_core::{
-    dns, errors, metrics::prom::EncodeLabelSetMut, proxy::http, svc::ExtractParam,
-    Error as BoxError,
+    dns, errors, metrics::prom::EncodeLabelSetMut, proxy::http, Error as BoxError,
 };
 use prometheus_client::encoding::*;
 
@@ -40,9 +39,6 @@ pub struct GrpcRsp {
     pub error: Option<Error>,
 }
 
-#[derive(Clone, Debug)]
-pub struct RouteLabelExtract(pub ParentRef, pub RouteRef);
-
 #[derive(Copy, Clone, Debug, Hash, PartialEq, Eq)]
 pub enum Error {
     FailFast,
@@ -215,37 +211,6 @@ impl EncodeLabelSet for GrpcRsp {
     }
 }
 
-// === impl MatchedRoute ===
-
-impl<T, M, F, P> super::super::MatchedRoute<T, M, F, P> {
-    /// Returns a [`RouteLabelExtract`].
-    ///
-    /// The extractor returned by this function provides a [`ExtractParam<P, T>`] implementation
-    /// that can be used to acquire the route-level labels corresponding to a given outbound
-    /// request.
-    pub(crate) fn label_extractor(&self) -> RouteLabelExtract {
-        use super::super::Route;
-        let Route {
-            parent_ref,
-            route_ref,
-            ..
-        } = &self.params;
-
-        RouteLabelExtract(parent_ref.clone(), route_ref.clone())
-    }
-}
-
-// === impl RouteLabelExtract ===
-
-impl<B> ExtractParam<Route, http::Request<B>> for RouteLabelExtract {
-    fn extract_param(&self, t: &http::Request<B>) -> Route {
-        let Self(parent, route) = self;
-        let uri = t.uri();
-
-        Route::new(parent.clone(), route.clone(), uri)
-    }
-}
-
 // === impl Error ===
 
 impl Error {

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

@@ -1,5 +1,3 @@
-use crate::http::policy::route::MatchedRoute;
-
 use super::{
     super::{Grpc, Http, Route},
     labels,
@@ -593,10 +591,8 @@ pub fn mock_http_route_metrics(
         &req,
     )
     .expect("find default route");
-
-    let extract = MatchedRoute::label_extractor;
     let (tx, handle) = tower_test::mock::pair::<http::Request<BoxBody>, http::Response<BoxBody>>();
-    let svc = super::layer(metrics, extract, body_data)
+    let svc = super::layer(metrics, body_data)
         .layer(move |_t: Http<()>| tx.clone())
         .new_service(Http {
             r#match,
@@ -642,9 +638,8 @@ pub fn mock_grpc_route_metrics(
     )
     .expect("find default route");
 
-    let extract = MatchedRoute::label_extractor;
     let (tx, handle) = tower_test::mock::pair::<http::Request<BoxBody>, http::Response<BoxBody>>();
-    let svc = super::layer(metrics, extract, body_data)
+    let svc = super::layer(metrics, body_data)
         .layer(move |_t: Grpc<()>| tx.clone())
         .new_service(Grpc {
             r#match,

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

@@ -1,7 +1,4 @@
-use super::{
-    extensions,
-    metrics::labels::{Route as RouteLabels, RouteLabelExtract},
-};
+use super::{extensions, metrics::labels::Route as RouteLabels};
 use futures::future::{Either, Ready};
 use linkerd_app_core::{
     cause_ref, classify,
@@ -15,8 +12,7 @@ use linkerd_http_retry::{self as retry, peek_trailers::PeekTrailersBody};
 use linkerd_proxy_client_policy as policy;
 use tokio::time;
 
-pub type NewHttpRetry<F, N> =
-    retry::NewHttpRetry<RetryPolicy, RouteLabels, F, RouteLabelExtract, N>;
+pub type NewHttpRetry<X, N> = retry::NewHttpRetry<RetryPolicy, RouteLabels, (), X, N>;
 
 #[derive(Clone, Debug)]
 pub struct RetryPolicy {

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

@@ -65,7 +65,8 @@ where
     route::MatchedRoute<T, M::Summary, F, P>: route::filters::Apply
         + svc::Param<classify::Request>
         + svc::Param<route::extensions::Params>
-        + route::metrics::MkStreamLabel,
+        + route::metrics::MkStreamLabel
+        + svc::ExtractParam<route::metrics::labels::Route, http::Request<http::BoxBody>>,
     route::MatchedBackend<T, M::Summary, F>: route::filters::Apply + route::metrics::MkStreamLabel,
 {
     /// Builds a stack that applies routes to distribute requests over a cached

linkerd/http/retry/src/lib.rs

@@ -85,6 +85,14 @@ struct Metrics {
 
 // === impl NewHttpRetry ===
 
+impl<P, L: Clone, ReqX, N> NewHttpRetry<P, L, (), ReqX, N> {
+    pub fn layer(
+        metrics: MetricFamilies<L>,
+    ) -> impl tower::layer::Layer<N, Service = Self> + Clone {
+        Self::layer_via_mk((), metrics)
+    }
+}
+
 impl<P, L: Clone, X: Clone, ReqX, N> NewHttpRetry<P, L, X, ReqX, N> {
     pub fn layer_via_mk(
         extract: X,