chore: update metriken (#35)

Updates metriken to 0.2.1 and makes necessary code changes to
support the new version.

CHANGELOG.md

@@ -9,5 +9,5 @@
 
 ### Added
 
-- Support Redis pubsub and Momento topics.
+- Support Momento topics.
 - Basic HTTP/1.1 and HTTP/2.0 load generation.

Cargo.toml

@@ -23,7 +23,7 @@ http-body-util = "0.1.0-rc.2"
 hyper = { version = "1.0.0-rc.3", features = ["http1", "http2", "client"]}
 humantime = "2.1.0"
 momento = "0.30.0"
-metriken = "0.1.4"
+metriken = "0.2.1"
 mio = "0.8.5"
 net = { git = "https://github.com/pelikan-io/pelikan" }
 paste = "1.0.12"
@@ -34,7 +34,7 @@ rand_distr = "0.4.3"
 rand_xoshiro = "0.6.0"
 ratelimit = "0.7.0"
 redis = { version = "0.22.3", features = ["tokio-comp"] }
-ringlog = "0.1.1"
+ringlog = "0.2.0"
 serde = "1.0.144"
 serde_json = "1.0.94"
 session = { git = "https://github.com/pelikan-io/pelikan" }

src/admin/mod.rs

@@ -138,8 +138,10 @@ impl<'a> TryFrom<&'a metriken::MetricEntry> for Metric<'a> {
             let percentiles = PERCENTILES
                 .iter()
                 .map(|(label, percentile)| {
-                    let value = heatmap.percentile(*percentile).map(|b| b.high()).ok();
-
+                    let value = match heatmap.percentile(*percentile) {
+                        Some(Ok(bucket)) => Some(bucket.high()),
+                        _ => None,
+                    };
                     (*label, *percentile, value)
                 })
                 .collect();

src/clients/http1.rs

@@ -37,7 +37,7 @@ async fn task(work_receiver: Receiver<WorkItem>, endpoint: String, config: Confi
             if session_requests != 0 {
                 let stop = Instant::now();
                 let lifecycle_ns = (stop - session_start).as_nanos();
-                let _ = SESSION_LIFECYCLE_REQUESTS.increment(stop, lifecycle_ns, 1);
+                let _ = SESSION_LIFECYCLE_REQUESTS.increment(stop, lifecycle_ns);
             }
             CONNECT.increment();
             let stream = match timeout(
@@ -168,8 +168,8 @@ async fn task(work_receiver: Receiver<WorkItem>, endpoint: String, config: Confi
 
                 let latency = stop.duration_since(start).as_nanos();
 
-                let _ = REQUEST_LATENCY.increment(start, latency, 1);
-                let _ = RESPONSE_LATENCY.increment(stop, latency, 1);
+                let _ = REQUEST_LATENCY.increment(start, latency);
+                let _ = RESPONSE_LATENCY.increment(stop, latency);
 
                 if let Some(header) = response
                     .headers()

src/clients/http2.rs

@@ -232,8 +232,8 @@ async fn task(
 
                 let latency = stop.duration_since(start).as_nanos();
 
-                let _ = REQUEST_LATENCY.increment(start, latency, 1);
-                let _ = RESPONSE_LATENCY.increment(stop, latency, 1);
+                let _ = REQUEST_LATENCY.increment(start, latency);
+                let _ = RESPONSE_LATENCY.increment(stop, latency);
 
                 if let Some(header) = response
                     .headers()

src/clients/memcache/mod.rs

@@ -165,8 +165,8 @@ async fn task(work_receiver: Receiver<WorkItem>, endpoint: String, config: Confi
                     // increment success stats and latency
                     RESPONSE_OK.increment();
 
-                    let _ = REQUEST_LATENCY.increment(start, latency_ns, 1);
-                    let _ = RESPONSE_LATENCY.increment(stop, latency_ns, 1);
+                    let _ = REQUEST_LATENCY.increment(start, latency_ns);
+                    let _ = RESPONSE_LATENCY.increment(stop, latency_ns);
 
                     // preserve the connection for the next request
                     stream = Some(s);

src/clients/momento/mod.rs

@@ -172,8 +172,8 @@ async fn task(
 
                 let latency = stop.duration_since(start).as_nanos();
 
-                let _ = REQUEST_LATENCY.increment(start, latency, 1);
-                let _ = RESPONSE_LATENCY.increment(stop, latency, 1);
+                let _ = REQUEST_LATENCY.increment(start, latency);
+                let _ = RESPONSE_LATENCY.increment(stop, latency);
             }
             Err(ResponseError::Exception) => {
                 RESPONSE_EX.increment();

src/clients/ping.rs

@@ -167,8 +167,8 @@ async fn task(work_receiver: Receiver<WorkItem>, endpoint: String, config: Confi
 
                 let latency = stop.duration_since(start).as_nanos();
 
-                let _ = REQUEST_LATENCY.increment(start, latency, 1);
-                let _ = RESPONSE_LATENCY.increment(stop, latency, 1);
+                let _ = REQUEST_LATENCY.increment(start, latency);
+                let _ = RESPONSE_LATENCY.increment(stop, latency);
             }
             Err(ResponseError::Exception) => {
                 // record execption

src/clients/redis/mod.rs

@@ -167,8 +167,8 @@ async fn task(work_receiver: Receiver<WorkItem>, endpoint: String, config: Confi
                 connection = Some(con);
                 RESPONSE_OK.increment();
 
-                let _ = REQUEST_LATENCY.increment(start, latency_ns, 1);
-                let _ = RESPONSE_LATENCY.increment(stop, latency_ns, 1);
+                let _ = REQUEST_LATENCY.increment(start, latency_ns);
+                let _ = RESPONSE_LATENCY.increment(stop, latency_ns);
             }
             Err(ResponseError::Exception) => {
                 CONNECT_CURR.decrement();