transport: expire idle TCP metrics (#781)

The proxy now exposes transport metrics with per-endpoint metadata; but,
unlike HTTP metrics, these metrics are never dropped. So an application
that has communicated with a now-defunct pod will continue to expose
metrics about that pod indefinitely.

This branch factors out the existing code for expiring idle HTTP metrics
so that it can be used for transport metrics as well, and adds eviction
for transport metrics. 

I also fixed a couple issues in the `linkerd2_metrics::summary` module
that were making the crate fail to compile with `--all-features`.

Closes linkerd/linkerd2#5372.

Signed-off-by: Eliza Weisman <[email protected]>

Author: hawkw

linkerd/app/core/src/metrics.rs

@@ -140,7 +140,7 @@ impl Metrics {
 
         let stack = stack_metrics::Registry::default();
 
-        let (transport, transport_report) = transport::metrics::new();
+        let (transport, transport_report) = transport::metrics::new(retain_idle);
 
         let (opencensus, opencensus_report) = opencensus::metrics::new();
 

linkerd/http-metrics/src/lib.rs

@@ -1,22 +1,16 @@
 #![deny(warnings, rust_2018_idioms)]
 
 pub use self::{requests::Requests, retries::Retries};
-use indexmap::IndexMap;
+use linkerd2_metrics::{LastUpdate, Store};
 use std::fmt;
 use std::hash::Hash;
 use std::sync::{Arc, Mutex};
-use std::time::{Duration, Instant};
+use std::time::Duration;
 
 pub mod requests;
 pub mod retries;
 
-#[derive(Debug)]
-struct Registry<T, M>
-where
-    T: Hash + Eq,
-{
-    by_target: IndexMap<T, Arc<Mutex<M>>>,
-}
+type Registry<T, M> = Store<T, Mutex<M>>;
 
 /// Reports metrics for prometheus.
 #[derive(Debug)]
@@ -26,7 +20,7 @@ where
 {
     prefix: &'static str,
     registry: Arc<Mutex<Registry<T, M>>>,
-    /// The amount time metrics with no updates should be retained for reports
+    /// The amount of time metrics with no updates should be retained for reports
     retain_idle: Duration,
     /// Whether latencies should be reported.
     include_latencies: bool,
@@ -48,35 +42,6 @@ struct Prefixed<'p, N: fmt::Display> {
     name: N,
 }
 
-trait LastUpdate {
-    fn last_update(&self) -> Instant;
-}
-
-impl<T, M> Default for Registry<T, M>
-where
-    T: Hash + Eq,
-{
-    fn default() -> Self {
-        Self {
-            by_target: IndexMap::default(),
-        }
-    }
-}
-
-impl<T, M> Registry<T, M>
-where
-    T: Hash + Eq,
-    M: LastUpdate,
-{
-    /// Retains metrics for all targets that (1) no longer have an active
-    /// reference to the `RequestMetrics` structure and (2) have not been updated since `epoch`.
-    fn retain_since(&mut self, epoch: Instant) {
-        self.by_target.retain(|_, m| {
-            Arc::strong_count(&m) > 1 || m.lock().map(|m| m.last_update() >= epoch).unwrap_or(false)
-        })
-    }
-}
-
 impl<T, M> Report<T, M>
 where
     T: Hash + Eq,

linkerd/http-metrics/src/requests/layer.rs

@@ -172,8 +172,7 @@ where
     fn new_service(&mut self, target: T) -> Self::Service {
         let metrics = match self.registry.lock() {
             Ok(mut r) => Some(
-                r.by_target
-                    .entry((&target).into())
+                r.entry((&target).into())
                     .or_insert_with(|| Arc::new(Mutex::new(Metrics::default())))
                     .clone(),
             ),
@@ -208,12 +207,7 @@ where
 
     fn call(&mut self, target: T) -> Self::Future {
         let metrics = match self.registry.lock() {
-            Ok(mut r) => Some(
-                r.by_target
-                    .entry((&target).into())
-                    .or_insert_with(|| Arc::new(Mutex::new(Metrics::default())))
-                    .clone(),
-            ),
+            Ok(mut r) => Some(r.entry((&target).into()).or_default().clone()),
             Err(_) => None,
         };
 

linkerd/http-metrics/src/requests/mod.rs

@@ -143,31 +143,30 @@ mod tests {
 
         let before_update = Instant::now();
         let metrics = registry
-            .by_target
             .entry(Target(123))
             .or_insert_with(Default::default)
             .clone();
-        assert_eq!(registry.by_target.len(), 1, "target should be registered");
+        assert_eq!(registry.len(), 1, "target should be registered");
         let after_update = Instant::now();
 
         registry.retain_since(after_update);
         assert_eq!(
-            registry.by_target.len(),
+            registry.len(),
             1,
             "target should not be evicted by time alone"
         );
 
         drop(metrics);
         registry.retain_since(before_update);
         assert_eq!(
-            registry.by_target.len(),
+            registry.len(),
             1,
             "target should not be evicted by availability alone"
         );
 
         registry.retain_since(after_update);
         assert_eq!(
-            registry.by_target.len(),
+            registry.len(),
             0,
             "target should be evicted by time once the handle is dropped"
         );

linkerd/http-metrics/src/requests/report.rs

@@ -1,11 +1,10 @@
 use super::{ClassMetrics, Metrics, StatusMetrics};
 use crate::{Prefixed, Registry, Report};
 use linkerd2_metrics::{latency, Counter, FmtLabels, FmtMetric, FmtMetrics, Histogram, Metric};
-use std::fmt;
-use std::hash::Hash;
-use std::time::Instant;
+use std::{fmt, hash::Hash, time::Instant};
 use tracing::trace;
 
+#[derive(Copy, Clone)]
 struct Status(http::StatusCode);
 
 impl<T, C> Report<T, Metrics<C>>
@@ -38,84 +37,35 @@ where
     }
 }
 
-impl<T, C> FmtMetrics for Report<T, Metrics<C>>
-where
-    T: FmtLabels + Hash + Eq,
-    C: FmtLabels + Hash + Eq,
-{
-    fn fmt_metrics(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        let mut registry = match self.registry.lock() {
-            Err(_) => return Ok(()),
-            Ok(r) => r,
-        };
-        trace!(
-            prefix = self.prefix,
-            targets = registry.by_target.len(),
-            include_latencies = self.include_latencies,
-            "Formatting HTTP request metrics",
-        );
-
-        if registry.by_target.is_empty() {
-            return Ok(());
-        }
-
-        let metric = self.request_total();
-        metric.fmt_help(f)?;
-        registry.fmt_by_target(f, metric, |s| &s.total)?;
-
-        if self.include_latencies {
-            let metric = self.response_latency_ms();
-            metric.fmt_help(f)?;
-            registry.fmt_by_status(f, metric, |s| &s.latency)?;
-        }
-
-        let metric = self.response_total();
-        metric.fmt_help(f)?;
-        registry.fmt_by_class(f, metric, |s| &s.total)?;
-
-        registry.retain_since(Instant::now() - self.retain_idle);
-
-        Ok(())
-    }
-}
-
-impl<T, C> Registry<T, Metrics<C>>
+impl<T, C> Report<T, Metrics<C>>
 where
     T: FmtLabels + Hash + Eq,
     C: FmtLabels + Hash + Eq,
 {
-    fn fmt_by_target<N, V, F>(
-        &self,
+    fn fmt_by_target<N, M>(
+        registry: &Registry<T, Metrics<C>>,
         f: &mut fmt::Formatter<'_>,
-        metric: Metric<'_, N, V>,
-        get_metric: F,
+        metric: Metric<'_, N, M>,
+        get_metric: impl Fn(&Metrics<C>) -> &M,
     ) -> fmt::Result
     where
         N: fmt::Display,
-        V: FmtMetric,
-        F: Fn(&Metrics<C>) -> &V,
+        M: FmtMetric,
     {
-        for (tgt, tm) in &self.by_target {
-            if let Ok(m) = tm.lock() {
-                get_metric(&*m).fmt_metric_labeled(f, &metric.name, tgt)?;
-            }
-        }
-
-        Ok(())
+        registry.fmt_by_locked(f, metric, get_metric)
     }
 
-    fn fmt_by_status<N, M, F>(
-        &self,
+    fn fmt_by_status<N, M>(
+        registry: &Registry<T, Metrics<C>>,
         f: &mut fmt::Formatter<'_>,
         metric: Metric<'_, N, M>,
-        get_metric: F,
+        get_metric: impl Fn(&StatusMetrics<C>) -> &M,
     ) -> fmt::Result
     where
         N: fmt::Display,
         M: FmtMetric,
-        F: Fn(&StatusMetrics<C>) -> &M,
     {
-        for (tgt, tm) in &self.by_target {
+        for (tgt, tm) in registry.iter() {
             if let Ok(tm) = tm.lock() {
                 for (status, m) in &tm.by_status {
                     let status = status.as_ref().map(|s| Status(*s));
@@ -128,18 +78,17 @@ where
         Ok(())
     }
 
-    fn fmt_by_class<N, M, F>(
-        &self,
+    fn fmt_by_class<N, M>(
+        registry: &Registry<T, Metrics<C>>,
         f: &mut fmt::Formatter<'_>,
         metric: Metric<'_, N, M>,
-        get_metric: F,
+        get_metric: impl Fn(&ClassMetrics) -> &M,
     ) -> fmt::Result
     where
         N: fmt::Display,
         M: FmtMetric,
-        F: Fn(&ClassMetrics) -> &M,
     {
-        for (tgt, tm) in &self.by_target {
+        for (tgt, tm) in registry.iter() {
             if let Ok(tm) = tm.lock() {
                 for (status, sm) in &tm.by_status {
                     for (cls, m) in &sm.by_class {
@@ -155,6 +104,47 @@ where
     }
 }
 
+impl<T, C> FmtMetrics for Report<T, Metrics<C>>
+where
+    T: FmtLabels + Hash + Eq,
+    C: FmtLabels + Hash + Eq,
+{
+    fn fmt_metrics(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let mut registry = match self.registry.lock() {
+            Err(_) => return Ok(()),
+            Ok(r) => r,
+        };
+        trace!(
+            prefix = self.prefix,
+            targets = registry.len(),
+            include_latencies = self.include_latencies,
+            "Formatting HTTP request metrics",
+        );
+
+        if registry.is_empty() {
+            return Ok(());
+        }
+
+        let metric = self.request_total();
+        metric.fmt_help(f)?;
+        Self::fmt_by_target(&registry, f, metric, |s| &s.total)?;
+
+        if self.include_latencies {
+            let metric = self.response_latency_ms();
+            metric.fmt_help(f)?;
+            Self::fmt_by_status(&registry, f, metric, |s| &s.latency)?;
+        }
+
+        let metric = self.response_total();
+        metric.fmt_help(f)?;
+        Self::fmt_by_class(&registry, f, metric, |s| &s.total)?;
+
+        registry.retain_since(Instant::now() - self.retain_idle);
+
+        Ok(())
+    }
+}
+
 impl FmtLabels for Status {
     fn fmt_labels(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(f, "status_code=\"{}\"", self.0.as_u16())

linkerd/http-metrics/src/retries.rs

@@ -38,12 +38,7 @@ impl<T: Hash + Eq> Retries<T> {
 
     pub fn get_handle(&self, target: impl Into<T>) -> Handle {
         let mut reg = self.0.lock().expect("retry metrics registry poisoned");
-        Handle(
-            reg.by_target
-                .entry(target.into())
-                .or_insert_with(|| Arc::new(Mutex::new(Metrics::default())))
-                .clone(),
-        )
+        Handle(reg.entry(target.into()).or_default().clone())
     }
 }
 
@@ -110,17 +105,17 @@ where
         };
         trace!(
             prfefix = %self.prefix,
-            targets = %registry.by_target.len(),
+            targets = %registry.len(),
             "Formatting HTTP retry metrics",
         );
 
-        if registry.by_target.is_empty() {
+        if registry.is_empty() {
             return Ok(());
         }
 
         let metric = self.retryable_total();
         metric.fmt_help(f)?;
-        for (tgt, tm) in &registry.by_target {
+        for (tgt, tm) in registry.iter() {
             if let Ok(m) = tm.lock() {
                 m.retryable.fmt_metric_labeled(f, &metric.name, tgt)?;
                 m.no_budget

linkerd/metrics/Cargo.toml

@@ -22,4 +22,4 @@ tracing = "0.1.2"
 
 [dev-dependencies]
 quickcheck = { version = "0.9", default-features = false }
-tokio = { version = "0.3", features = ["macros", "rt-multi-thread", "test-util", "time"] }
+tokio = { version = "0.3", features = ["rt", "macros", "test-util", "time"] }

linkerd/metrics/src/lib.rs

@@ -9,6 +9,7 @@ pub mod latency;
 mod prom;
 mod scopes;
 mod serve;
+mod store;
 #[cfg(feature = "summary")]
 mod summary;
 
@@ -18,6 +19,7 @@ pub use self::histogram::Histogram;
 pub use self::prom::{FmtLabels, FmtMetric, FmtMetrics, Metric};
 pub use self::scopes::Scopes;
 pub use self::serve::Serve;
+pub use self::store::{LastUpdate, Store};
 #[cfg(feature = "summary")]
 pub use self::summary::Summary;