metrics: Use parking_lot::Mutex in telemetry (#1142)

PR #1117 includes a change to move some uses of `std::sync::Mutex` to
`parking_lot::Mutex`, as the latter implementation is generally more
efficient.

This change updates all telemetry-related uses of Mutex with the
implementation in `parking_lot`.

Co-authored-by: Eliza Weisman <[email protected]>

Author: olix0r

Cargo.lock

@@ -911,6 +911,7 @@ version = "0.1.0"
 dependencies = [
  "futures",
  "linkerd-metrics",
+ "parking_lot",
  "pin-project",
  "tower",
 ]
@@ -974,6 +975,7 @@ dependencies = [
  "linkerd-http-classify",
  "linkerd-metrics",
  "linkerd-stack",
+ "parking_lot",
  "pin-project",
  "tower",
  "tracing",
@@ -1200,6 +1202,7 @@ dependencies = [
  "linkerd-stack",
  "linkerd-tls",
  "linkerd2-proxy-api",
+ "parking_lot",
  "pin-project",
  "prost-types",
  "quickcheck",
@@ -1237,6 +1240,7 @@ dependencies = [
  "linkerd-io",
  "linkerd-metrics",
  "linkerd-stack",
+ "parking_lot",
  "pin-project",
  "socket2 0.4.0",
  "tokio",
@@ -1331,6 +1335,7 @@ name = "linkerd-stack-metrics"
 version = "0.1.0"
 dependencies = [
  "linkerd-metrics",
+ "parking_lot",
  "tokio",
  "tower",
 ]

linkerd/error-metrics/Cargo.toml

@@ -10,5 +10,6 @@ publish = false
 [dependencies]
 futures = { version = "0.3", default-features = false }
 linkerd-metrics = { path = "../metrics" }
+parking_lot = "0.11"
 pin-project = "1"
 tower = { version = "0.4.8", default-features = false }

linkerd/error-metrics/src/layer.rs

@@ -1,10 +1,7 @@
 use crate::RecordError;
 use linkerd_metrics::Counter;
-use std::{
-    collections::HashMap,
-    hash::Hash,
-    sync::{Arc, Mutex},
-};
+use parking_lot::Mutex;
+use std::{collections::HashMap, hash::Hash, sync::Arc};
 
 #[derive(Debug)]
 pub struct RecordErrorLayer<L, K: Hash + Eq> {

linkerd/error-metrics/src/lib.rs

@@ -9,12 +9,8 @@ pub use self::layer::RecordErrorLayer;
 pub use self::service::RecordError;
 pub use linkerd_metrics::FmtLabels;
 use linkerd_metrics::{self as metrics, Counter, FmtMetrics};
-use std::{
-    collections::HashMap,
-    fmt,
-    hash::Hash,
-    sync::{Arc, Mutex},
-};
+use parking_lot::Mutex;
+use std::{collections::HashMap, fmt, hash::Hash, sync::Arc};
 
 pub trait LabelError<E> {
     type Labels: FmtLabels + Hash + Eq;
@@ -58,10 +54,7 @@ impl<K: Hash + Eq> Clone for Registry<K> {
 
 impl<K: FmtLabels + Hash + Eq> FmtMetrics for Registry<K> {
     fn fmt_metrics(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        let errors = match self.errors.lock() {
-            Ok(errors) => errors,
-            Err(_) => return Ok(()),
-        };
+        let errors = self.errors.lock();
         if errors.is_empty() {
             return Ok(());
         }

linkerd/error-metrics/src/service.rs

@@ -1,13 +1,14 @@
 use crate::LabelError;
 use futures::TryFuture;
 use linkerd_metrics::{Counter, FmtLabels};
+use parking_lot::Mutex;
 use pin_project::pin_project;
 use std::{
     collections::HashMap,
     future::Future,
     hash::Hash,
     pin::Pin,
-    sync::{Arc, Mutex},
+    sync::Arc,
     task::{Context, Poll},
 };
 
@@ -38,9 +39,11 @@ impl<L, K: FmtLabels + Hash + Eq, S> RecordError<L, K, S> {
         L: LabelError<E, Labels = K> + Clone,
     {
         let labels = label.label_error(&err);
-        if let Ok(mut errors) = errors.lock() {
-            errors.entry(labels).or_insert_with(Default::default).incr();
-        }
+        errors
+            .lock()
+            .entry(labels)
+            .or_insert_with(Default::default)
+            .incr();
     }
 }
 

linkerd/http-metrics/Cargo.toml

@@ -15,7 +15,8 @@ hyper = { version = "0.14.10", features = ["http1", "http2"] }
 linkerd-error = { path = "../error" }
 linkerd-http-classify = { path = "../http-classify" }
 linkerd-metrics = { path = "../metrics" }
-linkerd-stack = { path = "../stack" } 
+linkerd-stack = { path = "../stack" }
+parking_lot = "0.11"
+pin-project = "1"
 tower = "0.4.8"
 tracing = "0.1.26"
-pin-project = "1"

linkerd/http-metrics/src/lib.rs

@@ -4,10 +4,8 @@
 
 pub use self::{requests::Requests, retries::Retries};
 use linkerd_metrics::{LastUpdate, Store};
-use std::fmt;
-use std::hash::Hash;
-use std::sync::{Arc, Mutex};
-use std::time::Duration;
+use parking_lot::Mutex;
+use std::{fmt, hash::Hash, sync::Arc, time::Duration};
 
 pub mod requests;
 pub mod retries;

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

@@ -5,11 +5,12 @@ use super::{LastUpdate, Registry, Report};
 use linkerd_http_classify::ClassifyResponse;
 use linkerd_metrics::{latency, Counter, FmtMetrics, Histogram};
 use linkerd_stack::layer;
+use parking_lot::Mutex;
 use std::{
     collections::HashMap,
     fmt::Debug,
     hash::Hash,
-    sync::{Arc, Mutex},
+    sync::Arc,
     time::{Duration, Instant},
 };
 
@@ -144,7 +145,7 @@ mod tests {
         let retain_idle_for = Duration::from_secs(1);
         let r = super::Requests::<Target, Class>::default();
         let report = r.clone().into_report(retain_idle_for);
-        let mut registry = r.0.lock().unwrap();
+        let mut registry = r.0.lock();
 
         let before_update = Instant::now();
         let metrics = registry

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

@@ -66,12 +66,11 @@ where
         M: FmtMetric,
     {
         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));
-                    let labels = (tgt, status);
-                    get_metric(&*m).fmt_metric_labeled(f, &metric.name, labels)?;
-                }
+            let tm = tm.lock();
+            for (status, m) in &tm.by_status {
+                let status = status.as_ref().map(|s| Status(*s));
+                let labels = (tgt, status);
+                get_metric(&*m).fmt_metric_labeled(f, &metric.name, labels)?;
             }
         }
 
@@ -89,13 +88,12 @@ where
         M: FmtMetric,
     {
         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 {
-                        let status = status.as_ref().map(|s| Status(*s));
-                        let labels = (tgt, (status, cls));
-                        get_metric(&*m).fmt_metric_labeled(f, &metric.name, labels)?;
-                    }
+            let tm = tm.lock();
+            for (status, sm) in &tm.by_status {
+                for (cls, m) in &sm.by_class {
+                    let status = status.as_ref().map(|s| Status(*s));
+                    let labels = (tgt, (status, cls));
+                    get_metric(&*m).fmt_metric_labeled(f, &metric.name, labels)?;
                 }
             }
         }
@@ -110,10 +108,7 @@ where
     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,
-        };
+        let mut registry = self.registry.lock();
         trace!(
             prefix = self.prefix,
             targets = registry.len(),

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

@@ -4,14 +4,17 @@ use http_body::Body;
 use linkerd_error::Error;
 use linkerd_http_classify::{ClassifyEos, ClassifyResponse};
 use linkerd_stack::{NewService, Param, Proxy};
+use parking_lot::Mutex;
 use pin_project::{pin_project, pinned_drop};
-use std::fmt::Debug;
-use std::hash::Hash;
-use std::marker::PhantomData;
-use std::pin::Pin;
-use std::sync::{Arc, Mutex};
-use std::task::{Context, Poll};
-use std::time::Instant;
+use std::{
+    fmt::Debug,
+    hash::Hash,
+    marker::PhantomData,
+    pin::Pin,
+    sync::Arc,
+    task::{Context, Poll},
+    time::Instant,
+};
 
 /// Wraps services to record metrics.
 #[derive(Debug)]
@@ -126,13 +129,13 @@ where
     type Service = HttpMetrics<N::Service, C>;
 
     fn new_service(&mut self, target: T) -> Self::Service {
-        let metrics = match self.registry.lock() {
-            Ok(mut r) => Some(
+        let metrics = {
+            let mut r = self.registry.lock();
+            Some(
                 r.entry(target.param())
                     .or_insert_with(|| Arc::new(Mutex::new(Metrics::default())))
                     .clone(),
-            ),
-            Err(_) => None,
+            )
         };
 
         let inner = self.inner.new_service(target);
@@ -182,10 +185,9 @@ where
         if req.body().is_end_stream() {
             if let Some(lock) = req_metrics.take() {
                 let now = Instant::now();
-                if let Ok(mut metrics) = lock.lock() {
-                    (*metrics).last_update = now;
-                    (*metrics).total.incr();
-                }
+                let mut metrics = lock.lock();
+                (*metrics).last_update = now;
+                (*metrics).total.incr();
             }
         }
 
@@ -232,10 +234,9 @@ where
         if req.body().is_end_stream() {
             if let Some(lock) = req_metrics.take() {
                 let now = Instant::now();
-                if let Ok(mut metrics) = lock.lock() {
-                    (*metrics).last_update = now;
-                    (*metrics).total.incr();
-                }
+                let mut metrics = lock.lock();
+                (*metrics).last_update = now;
+                (*metrics).total.incr();
             }
         }
 
@@ -324,10 +325,9 @@ where
 
         if let Some(lock) = this.metrics.take() {
             let now = Instant::now();
-            if let Ok(mut metrics) = lock.lock() {
-                (*metrics).last_update = now;
-                (*metrics).total.incr();
-            }
+            let mut metrics = lock.lock();
+            (*metrics).last_update = now;
+            (*metrics).total.incr();
         }
 
         Poll::Ready(frame)
@@ -390,10 +390,7 @@ where
             Some(lock) => lock,
             None => return,
         };
-        let mut metrics = match lock.lock() {
-            Ok(m) => m,
-            Err(_) => return,
-        };
+        let mut metrics = lock.lock();
 
         (*metrics).last_update = now;
 
@@ -434,10 +431,7 @@ fn measure_class<C: Hash + Eq>(
     status: Option<http::StatusCode>,
 ) {
     let now = Instant::now();
-    let mut metrics = match lock.lock() {
-        Ok(m) => m,
-        Err(_) => return,
-    };
+    let mut metrics = lock.lock();
 
     (*metrics).last_update = now;