feat(app/inbound): introduce request duration metrics

this commit introduces a new middleware layer to the inbound proxy.
this instruments inbound traffic with Prometheus telemetry that records
request body latency, and emits a histogram of request body durations.

as in #4420, the buckets are chosen to mimic the
request and response buckets emitted by the outbound proxy, with their
granularity flipped. in other words, the inbound proxy is more
interested in fine-grained request body metrics than response body
metrics, while the outbound proxy is more interested in fine-grained
response body metrics than request body metrics.

* #4420

Signed-off-by: katelyn martin <kate@buoyant.io>

Author: cratelyn

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

@@ -2,12 +2,14 @@ use crate::InboundMetrics;
 use linkerd_app_core::svc;
 
 pub use self::{
-    count_reqs::*, labels::RouteLabels, req_body::*, rsp_body::*, rsp_duration::*, status::*,
+    count_reqs::*, labels::RouteLabels, req_body::*, req_duration::*, rsp_body::*, rsp_duration::*,
+    status::*,
 };
 
 mod count_reqs;
 mod labels;
 mod req_body;
+mod req_duration;
 mod rsp_body;
 mod rsp_duration;
 mod status;
@@ -16,6 +18,7 @@ pub(super) fn layer<N>(
     InboundMetrics {
         request_count,
         request_body_data,
+        request_duration,
         response_body_data,
         response_duration,
         status_codes,
@@ -34,6 +37,11 @@ pub(super) fn layer<N>(
         NewResponseDuration::layer_via(extract)
     };
 
+    let request_duration = {
+        let extract = ExtractRequestDurationMetrics(request_duration.clone());
+        NewRequestDuration::layer_via(extract)
+    };
+
     let response_body = {
         let extract = ExtractResponseBodyDataMetrics::new(response_body_data.clone());
         NewRecordResponseBodyData::layer_via(extract)
@@ -50,15 +58,17 @@ pub(super) fn layer<N>(
     };
 
     svc::layer::mk(move |inner| {
-        count.layer(
-            response_duration.layer(response_body.layer(request_body.layer(status.layer(inner)))),
-        )
+        count.layer(response_duration.layer(
+            request_duration.layer(response_body.layer(request_body.layer(status.layer(inner)))),
+        ))
     })
 }
 
 /// An `N`-typed service instrumented with metrics middleware.
 type Instrumented<N> = NewCountRequests<
     NewResponseDuration<
-        NewRecordResponseBodyData<NewRecordRequestBodyData<NewRecordStatusCode<N>>>,
+        NewRequestDuration<
+            NewRecordResponseBodyData<NewRecordRequestBodyData<NewRecordStatusCode<N>>>,
+        >,
     >,
 >;

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

@@ -0,0 +1,102 @@
+use super::RouteLabels;
+use crate::policy::PermitVariant;
+use linkerd_app_core::{
+    metrics::prom::{self, EncodeLabelSetMut},
+    svc,
+};
+use linkerd_http_prom::{
+    record_response::{self, Params},
+    stream_label::with::MkWithLabels,
+};
+
+pub type NewRequestDuration<N> =
+    record_response::NewRequestDuration<MkLabelDuration, ExtractRequestDurationMetrics, N>;
+
+pub type RequestDurationParams =
+    Params<MkLabelDuration, record_response::RequestMetrics<RequestDurationLabels>>;
+
+#[derive(Clone, Debug)]
+pub struct ExtractRequestDurationMetrics(pub RequestDurationFamilies);
+
+#[derive(Clone, Debug)]
+pub struct RequestDurationFamilies {
+    grpc: record_response::RequestMetrics<RequestDurationLabels>,
+    http: record_response::RequestMetrics<RequestDurationLabels>,
+}
+
+#[derive(Clone, Debug, Hash, PartialEq, Eq)]
+pub struct RequestDurationLabels {
+    route: RouteLabels,
+}
+
+pub type MkLabelDuration = MkWithLabels<RequestDurationLabels>;
+
+// === impl RequestDurationFamilies ===
+
+impl RequestDurationFamilies {
+    /// Registers a new [`RequestDurationFamilies`] with the given registry.
+    pub fn register(
+        reg: &mut prom::Registry,
+        histo: impl Clone + IntoIterator<Item = f64>,
+    ) -> Self {
+        let grpc = {
+            let reg = reg.sub_registry_with_prefix("grpc");
+            record_response::RequestMetrics::register(reg, histo.clone())
+        };
+
+        let http = {
+            let reg = reg.sub_registry_with_prefix("http");
+            record_response::RequestMetrics::register(reg, histo)
+        };
+
+        Self { grpc, http }
+    }
+}
+
+// === impl ExtractRequestDurationMetrics ===
+
+impl<T> svc::ExtractParam<RequestDurationParams, T> for ExtractRequestDurationMetrics
+where
+    T: svc::Param<PermitVariant> + svc::Param<RouteLabels>,
+{
+    fn extract_param(&self, target: &T) -> RequestDurationParams {
+        let Self(families) = self;
+
+        let labeler = {
+            let route: RouteLabels = target.param();
+            let labels = RequestDurationLabels { route };
+            MkLabelDuration::new(labels)
+        };
+
+        let metric = {
+            let variant: PermitVariant = target.param();
+            let RequestDurationFamilies { grpc, http } = families;
+            match variant {
+                PermitVariant::Grpc => grpc,
+                PermitVariant::Http => http,
+            }
+            .clone()
+        };
+
+        RequestDurationParams { labeler, metric }
+    }
+}
+
+// === impl RequestDurationLabels ===
+
+impl prom::EncodeLabelSetMut for RequestDurationLabels {
+    fn encode_label_set(
+        &self,
+        encoder: &mut prom::encoding::LabelSetEncoder<'_>,
+    ) -> std::fmt::Result {
+        let Self { route } = self;
+        route.encode_label_set(encoder)?;
+        Ok(())
+    }
+}
+
+impl prom::encoding::EncodeLabelSet for RequestDurationLabels {
+    fn encode(&self, mut encoder: prom::encoding::LabelSetEncoder<'_>) -> std::fmt::Result {
+        self.encode_label_set(&mut encoder)
+    }
+}

linkerd/app/inbound/src/metrics.rs

@@ -12,8 +12,8 @@ pub(crate) mod authz;
 pub(crate) mod error;
 
 use crate::http::router::metrics::{
-    RequestBodyFamilies, RequestCountFamilies, ResponseBodyFamilies, ResponseDurationFamilies,
-    StatusCodeFamilies,
+    RequestBodyFamilies, RequestCountFamilies, RequestDurationFamilies, ResponseBodyFamilies,
+    ResponseDurationFamilies, StatusCodeFamilies,
 };
 pub use linkerd_app_core::metrics::*;
 
@@ -34,6 +34,7 @@ pub struct InboundMetrics {
     pub direct: crate::direct::MetricsFamilies,
     pub request_count: RequestCountFamilies,
     pub request_body_data: RequestBodyFamilies,
+    pub request_duration: RequestDurationFamilies,
     pub response_body_data: ResponseBodyFamilies,
     pub response_duration: ResponseDurationFamilies,
     pub status_codes: StatusCodeFamilies,
@@ -48,6 +49,8 @@ impl InboundMetrics {
         );
         let request_count = RequestCountFamilies::register(reg);
         let request_body_data = RequestBodyFamilies::register(reg);
+        let request_duration =
+            RequestDurationFamilies::register(reg, Self::REQUEST_BUCKETS.iter().copied());
         let response_body_data = ResponseBodyFamilies::register(reg);
         let response_duration =
             ResponseDurationFamilies::register(reg, Self::RESPONSE_BUCKETS.iter().copied());
@@ -63,24 +66,32 @@ impl InboundMetrics {
             direct,
             request_count,
             request_body_data,
+            request_duration,
             response_body_data,
             response_duration,
             status_codes,
         }
     }
 
     // There are two histograms for which we need to register metrics:
-    //   (1) request durations, which are measured on routes. TODO(kate): forthcoming.
+    //   (1) request durations, which are measured on routes.
     //   (2) response durations, which are measured on route-backends.
     //
     // Should these change in the future, be sure to consider the outbound proxy's corresponding
     // constants measuring request and response latency for *outgoing* traffic.
 
+    /// Histogram buckets for request latency.
+    ///
+    /// Because response duration is the more meaningful metric operationally for the inbound
+    /// proxy, we opt to preserve higher fidelity for request durations (especially for lower
+    /// values).
+    const REQUEST_BUCKETS: &'static [f64] = &[0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 10.0];
+
     /// Histogram buckets for response latency.
     ///
-    /// These buckets for this histogram are coarse, eliding several buckets for short response
-    /// durations to be conservative about the costs of tracking two histograms' respective time
-    /// series.
+    /// These buckets for this histogram are coarse than those of [`Self::REQUEST_BUCKETS`],
+    /// eliding several buckets for short response durations to be conservative about the costs of
+    /// tracking two histograms' respective time series.
     const RESPONSE_BUCKETS: &'static [f64] = &[0.05, 0.5, 1.0, 10.0];
 }