metrics: factor out NewService for target-scoped metrics (#1117)

hawkw · web-flow · commit 0d0c5e735a7a · 2021-07-14T11:50:21.000-07:00
While working on adding a target port label to the TLS detection failure metrics added in #1114, I realized that there was some boilerplate code for scoping metrics based on a label from a target that was basically repeated between `linkerd-http-metrics` and `linkerd-transport`'s `metrics` module, and that this boilerplate code would have to be rewritten a third time for adding target port labels to the TLS detection metrics. Instead of rewriting this code a third time, I factored it out into a `NewMetrics` type in the `linkerd-metrics` crate, which given a registry of scopes, a `NewService`, and a metrics service type that can be constructed from the `NewService`'s service and the registry's metrics, implements `NewService` by wrapping the inner service with a metrics service scoped for the target's labels. The `linkerd-http-metrics` crate's `NewHttpMetrics` and the `linkerd-transport` crate's `metrics::MakeAccept` can both be replaced with the new `NewMetrics` type. Signed-off-by: Eliza Weisman <eliza@buoyant.io>
diff --git a/Cargo.lock b/Cargo.lock
@@ -1035,6 +1035,7 @@ dependencies = [
  "hdrhistogram",
  "http",
  "hyper",
+ "linkerd-stack",
  "parking_lot",
  "quickcheck",
  "tokio",
diff --git a/linkerd/app/admin/src/stack.rs b/linkerd/app/admin/src/stack.rs
@@ -62,7 +62,7 @@ impl Config {
         let (ready, latch) = crate::server::Readiness::new();
         let admin = crate::server::Admin::new(report, ready, shutdown, trace);
         let admin = svc::stack(admin)
-            .push(metrics.http_endpoint.to_layer::<classify::Response, _>())
+            .push(metrics.http_endpoint.to_layer::<classify::Response, _, Target>())
             .push_on_response(
                 svc::layers()
                     .push(metrics.http_errors.clone())
diff --git a/linkerd/app/core/src/control.rs b/linkerd/app/core/src/control.rs
@@ -93,7 +93,7 @@ impl Config {
             .push(self::resolve::layer(dns, resolve_backoff))
             .push_on_response(self::control::balance::layer())
             .into_new_service()
-            .push(metrics.to_layer::<classify::Response, _>())
+            .push(metrics.to_layer::<classify::Response, _, _>())
             .push(self::add_origin::layer())
             .push_on_response(svc::layers().push_spawn_buffer(self.buffer_capacity))
             .push_map_target(move |()| addr.clone())
diff --git a/linkerd/app/inbound/src/http/mod.rs b/linkerd/app/inbound/src/http/mod.rs
@@ -149,7 +149,11 @@ where
             // Registers the stack to be tapped.
             .push(tap::NewTapHttp::layer(rt.tap.clone()))
             // Records metrics for each `Target`.
-            .push(rt.metrics.http_endpoint.to_layer::<classify::Response, _>())
+            .push(
+                rt.metrics
+                    .http_endpoint
+                    .to_layer::<classify::Response, _, _>(),
+            )
             .push_on_response(http_tracing::client(rt.span_sink.clone(), trace_labels()))
             .push_on_response(http::BoxResponse::layer())
             .check_new_service::<Target, http::Request<_>>();
@@ -174,7 +178,11 @@ where
                     // by tap.
                     .push_http_insert_target::<dst::Route>()
                     // Records per-route metrics.
-                    .push(rt.metrics.http_route.to_layer::<classify::Response, _>())
+                    .push(
+                        rt.metrics
+                            .http_route
+                            .to_layer::<classify::Response, _, dst::Route>(),
+                    )
                     // Sets the per-route response classifier as a request
                     // extension.
                     .push(classify::NewClassify::layer())
diff --git a/linkerd/app/outbound/src/http/endpoint.rs b/linkerd/app/outbound/src/http/endpoint.rs
@@ -45,7 +45,11 @@ impl<C> Outbound<C> {
             .into_new_service()
             .push_new_reconnect(backoff)
             .push(tap::NewTapHttp::layer(rt.tap.clone()))
-            .push(rt.metrics.http_endpoint.to_layer::<classify::Response, _>())
+            .push(
+                rt.metrics
+                    .http_endpoint
+                    .to_layer::<classify::Response, _, _>(),
+            )
             .push_on_response(http_tracing::client(
                 rt.span_sink.clone(),
                 crate::trace_labels(),
diff --git a/linkerd/app/outbound/src/http/logical.rs b/linkerd/app/outbound/src/http/logical.rs
@@ -1,7 +1,7 @@
 use super::{CanonicalDstHeader, Concrete, Endpoint, Logical};
 use crate::{endpoint, resolve, stack_labels, Outbound};
 use linkerd_app_core::{
-    classify, config, profiles,
+    classify, config, dst, profiles,
     proxy::{
         api_resolve::{ConcreteAddr, Metadata},
         core::Resolve,
@@ -118,7 +118,7 @@ impl<E> Outbound<E> {
                     .push(
                         rt.metrics
                             .http_route_actual
-                            .to_layer::<classify::Response, _>(),
+                            .to_layer::<classify::Response, _, dst::Route>(),
                     )
                     // Depending on whether or not the request can be retried,
                     // it may have one of two `Body` types. This layer unifies
@@ -131,7 +131,7 @@ impl<E> Outbound<E> {
                     // Sets an optional request timeout.
                     .push(http::MakeTimeoutLayer::default())
                     // Records per-route metrics.
-                    .push(rt.metrics.http_route.to_layer::<classify::Response, _>())
+                    .push(rt.metrics.http_route.to_layer::<classify::Response, _, _>())
                     // Sets the per-route response classifier as a request
                     // extension.
                     .push(classify::NewClassify::layer())
diff --git a/linkerd/http-metrics/Cargo.toml b/linkerd/http-metrics/Cargo.toml
@@ -14,7 +14,7 @@ http-body = "0.4"
 hyper = { version = "0.14.10", features = ["http1", "http2"] }
 linkerd-error = { path = "../error" }
 linkerd-http-classify = { path = "../http-classify" }
-linkerd-metrics = { path = "../metrics" }
+linkerd-metrics = { path = "../metrics", features = ["linkerd-stack"] }
 linkerd-stack = { path = "../stack" }
 parking_lot = "0.11"
 pin-project = "1"
diff --git a/linkerd/http-metrics/src/lib.rs b/linkerd/http-metrics/src/lib.rs
@@ -3,14 +3,14 @@
 #![allow(clippy::inconsistent_struct_constructor)]
 
 pub use self::{requests::Requests, retries::Retries};
-use linkerd_metrics::{LastUpdate, Store};
+use linkerd_metrics::SharedStore;
 use parking_lot::Mutex;
-use std::{fmt, hash::Hash, sync::Arc, time::Duration};
+use std::{fmt, hash::Hash, time::Duration};
 
 pub mod requests;
 pub mod retries;
 
-type Registry<T, M> = Store<T, Mutex<M>>;
+type Registry<T, M> = SharedStore<T, Mutex<M>>;
 
 /// Reports metrics for prometheus.
 #[derive(Debug)]
@@ -19,7 +19,7 @@ where
     T: Hash + Eq,
 {
     prefix: &'static str,
-    registry: Arc<Mutex<Registry<T, M>>>,
+    registry: Registry<T, M>,
     /// The amount of time metrics with no updates should be retained for reports
     retain_idle: Duration,
     /// Whether latencies should be reported.
@@ -46,7 +46,7 @@ impl<T, M> Report<T, M>
 where
     T: Hash + Eq,
 {
-    fn new(retain_idle: Duration, registry: Arc<Mutex<Registry<T, M>>>) -> Self {
+    fn new(retain_idle: Duration, registry: Registry<T, M>) -> Self {
         Self {
             prefix: "",
             registry,
diff --git a/linkerd/http-metrics/src/requests/mod.rs b/linkerd/http-metrics/src/requests/mod.rs
@@ -1,25 +1,22 @@
 mod report;
 mod service;
 
-use super::{LastUpdate, Registry, Report};
+pub use self::service::{NewHttpMetrics, ResponseBody};
+use super::Report;
 use linkerd_http_classify::ClassifyResponse;
-use linkerd_metrics::{latency, Counter, FmtMetrics, Histogram};
-use linkerd_stack::layer;
-use parking_lot::Mutex;
+use linkerd_metrics::{latency, Counter, FmtMetrics, Histogram, LastUpdate, NewMetrics};
+use linkerd_stack::{self as svc, layer};
 use std::{
     collections::HashMap,
     fmt::Debug,
     hash::Hash,
-    sync::Arc,
     time::{Duration, Instant},
 };
 
-pub use self::service::{NewHttpMetrics, ResponseBody};
-
-type SharedRegistry<T, C> = Arc<Mutex<Registry<T, Metrics<C>>>>;
+type Registry<T, C> = super::Registry<T, Metrics<C>>;
 
 #[derive(Debug)]
-pub struct Requests<T, C>(SharedRegistry<T, C>)
+pub struct Requests<T, C>(Registry<T, C>)
 where
     T: Hash + Eq,
     C: Hash + Eq;
@@ -52,7 +49,7 @@ pub struct ClassMetrics {
 
 impl<T: Hash + Eq, C: Hash + Eq> Default for Requests<T, C> {
     fn default() -> Self {
-        Requests(Arc::new(Mutex::new(Registry::default())))
+        Requests(Registry::default())
     }
 }
 
@@ -64,12 +61,15 @@ impl<T: Hash + Eq, C: Hash + Eq> Requests<T, C> {
         Report::new(retain_idle, self.0)
     }
 
-    pub fn to_layer<L, N>(&self) -> impl layer::Layer<N, Service = NewHttpMetrics<N, T, L>> + Clone
+    pub fn to_layer<L, N, Tgt>(
+        &self,
+    ) -> impl layer::Layer<N, Service = NewHttpMetrics<N, T, L, C, N::Service>> + Clone
     where
         L: ClassifyResponse<Class = C> + Send + Sync + 'static,
+        N: svc::NewService<Tgt>,
     {
         let reg = self.0.clone();
-        layer::mk(move |inner| NewHttpMetrics::new(reg.clone(), inner))
+        NewMetrics::layer(reg)
     }
 }
 
diff --git a/linkerd/http-metrics/src/requests/report.rs b/linkerd/http-metrics/src/requests/report.rs
@@ -1,6 +1,9 @@
 use super::{ClassMetrics, Metrics, StatusMetrics};
-use crate::{Prefixed, Registry, Report};
-use linkerd_metrics::{latency, Counter, FmtLabels, FmtMetric, FmtMetrics, Histogram, Metric};
+use crate::{Prefixed, Report};
+use linkerd_metrics::{
+    latency, Counter, FmtLabels, FmtMetric, FmtMetrics, Histogram, Metric, Store,
+};
+use parking_lot::Mutex;
 use std::{fmt, hash::Hash, time::Instant};
 use tracing::trace;
 
@@ -43,7 +46,7 @@ where
     C: FmtLabels + Hash + Eq,
 {
     fn fmt_by_target<N, M>(
-        registry: &Registry<T, Metrics<C>>,
+        registry: &Store<T, Mutex<Metrics<C>>>,
         f: &mut fmt::Formatter<'_>,
         metric: Metric<'_, N, M>,
         get_metric: impl Fn(&Metrics<C>) -> &M,
@@ -56,7 +59,7 @@ where
     }
 
     fn fmt_by_status<N, M>(
-        registry: &Registry<T, Metrics<C>>,
+        registry: &Store<T, Mutex<Metrics<C>>>,
         f: &mut fmt::Formatter<'_>,
         metric: Metric<'_, N, M>,
         get_metric: impl Fn(&StatusMetrics<C>) -> &M,
@@ -78,7 +81,7 @@ where
     }
 
     fn fmt_by_class<N, M>(
-        registry: &Registry<T, Metrics<C>>,
+        registry: &Store<T, Mutex<Metrics<C>>>,
         f: &mut fmt::Formatter<'_>,
         metric: Metric<'_, N, M>,
         get_metric: impl Fn(&ClassMetrics) -> &M,
diff --git a/linkerd/http-metrics/src/requests/service.rs b/linkerd/http-metrics/src/requests/service.rs
@@ -1,9 +1,10 @@
-use super::{ClassMetrics, Metrics, SharedRegistry, StatusMetrics};
+use super::{ClassMetrics, Metrics, StatusMetrics};
 use futures::{ready, TryFuture};
 use http_body::Body;
 use linkerd_error::Error;
 use linkerd_http_classify::{ClassifyEos, ClassifyResponse};
-use linkerd_stack::{NewService, Param, Proxy};
+use linkerd_metrics::NewMetrics;
+use linkerd_stack::Proxy;
 use parking_lot::Mutex;
 use pin_project::{pin_project, pinned_drop};
 use std::{
@@ -17,17 +18,8 @@ use std::{
 };
 
 /// Wraps services to record metrics.
-#[derive(Debug)]
-pub struct NewHttpMetrics<N, K, C>
-where
-    K: Hash + Eq,
-    C: ClassifyResponse,
-    C::Class: Hash + Eq,
-{
-    registry: SharedRegistry<K, C::Class>,
-    inner: N,
-    _p: PhantomData<fn() -> C>,
-}
+pub type NewHttpMetrics<N, K, C, Class, S> =
+    NewMetrics<N, K, Mutex<Metrics<Class>>, HttpMetrics<S, C>>;
 
 /// A middleware that records HTTP metrics.
 #[pin_project]
@@ -85,71 +77,22 @@ where
     inner: B,
 }
 
-// === impl NewHttpMetrics ===
+// === impl HttpMetrics ===
 
-impl<N, K, C> NewHttpMetrics<N, K, C>
+impl<S, C> From<(S, Arc<Mutex<Metrics<C::Class>>>)> for HttpMetrics<S, C>
 where
-    K: Hash + Eq,
     C: ClassifyResponse,
     C::Class: Hash + Eq,
 {
-    pub(crate) fn new(registry: SharedRegistry<K, C::Class>, inner: N) -> Self {
-        Self {
-            inner,
-            registry,
-            _p: PhantomData,
-        }
-    }
-}
-
-impl<N, K, C> Clone for NewHttpMetrics<N, K, C>
-where
-    N: Clone,
-    K: Clone + Hash + Eq,
-    C: ClassifyResponse + Send + Sync + 'static,
-    C::Class: Hash + Eq,
-{
-    fn clone(&self) -> Self {
+    fn from((inner, metrics): (S, Arc<Mutex<Metrics<C::Class>>>)) -> Self {
         Self {
-            inner: self.inner.clone(),
-            registry: self.registry.clone(),
-            _p: PhantomData,
-        }
-    }
-}
-
-impl<T, N, K, C> NewService<T> for NewHttpMetrics<N, K, C>
-where
-    T: Param<K>,
-    K: Hash + Eq,
-    N: NewService<T>,
-    C: ClassifyResponse + Default + Send + Sync + 'static,
-    C::Class: Hash + Eq,
-{
-    type Service = HttpMetrics<N::Service, C>;
-
-    fn new_service(&mut self, target: T) -> Self::Service {
-        let metrics = {
-            let mut r = self.registry.lock();
-            Some(
-                r.entry(target.param())
-                    .or_insert_with(|| Arc::new(Mutex::new(Metrics::default())))
-                    .clone(),
-            )
-        };
-
-        let inner = self.inner.new_service(target);
-
-        HttpMetrics {
+            metrics: Some(metrics),
             inner,
-            metrics,
             _p: PhantomData,
         }
     }
 }
 
-// === impl HttpMetrics ===
-
 impl<S, C> Clone for HttpMetrics<S, C>
 where
     S: Clone,
diff --git a/linkerd/http-metrics/src/retries.rs b/linkerd/http-metrics/src/retries.rs
@@ -1,5 +1,5 @@
-use super::{LastUpdate, Prefixed, Registry, Report};
-use linkerd_metrics::{Counter, FmtLabels, FmtMetric, FmtMetrics, Metric};
+use super::{Prefixed, Registry, Report};
+use linkerd_metrics::{Counter, FmtLabels, FmtMetric, FmtMetrics, LastUpdate, Metric};
 use parking_lot::Mutex;
 use std::{
     fmt,
@@ -10,7 +10,7 @@ use std::{
 use tracing::trace;
 
 #[derive(Debug)]
-pub struct Retries<T>(Arc<Mutex<Registry<T, Metrics>>>)
+pub struct Retries<T>(Registry<T, Metrics>)
 where
     T: Hash + Eq;
 
@@ -30,7 +30,7 @@ struct NoBudgetLabel;
 
 impl<T: Hash + Eq> Default for Retries<T> {
     fn default() -> Self {
-        Retries(Arc::new(Mutex::new(Registry::default())))
+        Retries(Registry::default())
     }
 }
 
diff --git a/linkerd/metrics/Cargo.toml b/linkerd/metrics/Cargo.toml
@@ -16,6 +16,7 @@ deflate = { version = "0.9.1", features = ["gzip"] }
 hdrhistogram = { version = "7.3", default-features = false, optional = true }
 http = "0.2"
 hyper = { version = "0.14.10", features = ["http1", "http2"] }
+linkerd-stack = { path = "../stack", optional = true }
 parking_lot = "0.11"
 tokio = { version = "1", features = ["time"], optional = true }
 tracing = "0.1.26"
diff --git a/linkerd/metrics/src/lib.rs b/linkerd/metrics/src/lib.rs
diff --git a/linkerd/metrics/src/new_metrics.rs b/linkerd/metrics/src/new_metrics.rs
diff --git a/linkerd/metrics/src/store.rs b/linkerd/metrics/src/store.rs
diff --git a/linkerd/proxy/transport/Cargo.toml b/linkerd/proxy/transport/Cargo.toml
diff --git a/linkerd/proxy/transport/src/metrics.rs b/linkerd/proxy/transport/src/metrics.rs
