feat: add Key-Value GET and SET latency metrics for cache workloads (#391)

Add Key-Value GET and SET latency metrics for cache workloads

Author: krispraws

src/clients/cache/memcache/mod.rs

@@ -103,6 +103,16 @@ async fn task(
 
         let request = request.unwrap();
 
+        let mut latency_histograms = vec![&RESPONSE_LATENCY];
+        match &request.request {
+            Request::Get(_) => {
+                latency_histograms.push(&KVGET_RESPONSE_LATENCY);
+            }
+            Request::Set(_) => {
+                latency_histograms.push(&KVSET_RESPONSE_LATENCY);
+            }
+            _ => {}
+        }
         // compose request
         REQUEST_OK.increment();
         let _ = protocol.compose_request(&request.request, &mut write_buffer);
@@ -166,7 +176,6 @@ async fn task(
         match response {
             Ok(response) => {
                 let latency_ns = stop.duration_since(start).as_nanos() as u64;
-
                 // check if the response is valid
                 if (request.validator)(response).is_err() {
                     // increment error stats, connection will be dropped
@@ -175,9 +184,9 @@ async fn task(
                 } else {
                     // increment success stats and latency
                     RESPONSE_OK.increment();
-
-                    let _ = RESPONSE_LATENCY.increment(latency_ns);
-
+                    for hist in latency_histograms {
+                        let _ = hist.increment(latency_ns);
+                    }
                     // preserve the connection for the next request
                     stream = Some(s);
                 }

src/clients/cache/momento/mod.rs

@@ -78,13 +78,20 @@ async fn task(
 
         REQUEST.increment();
         let start = Instant::now();
+        let mut latency_histograms = vec![&RESPONSE_LATENCY];
         let result = match work_item {
             ClientWorkItemKind::Request { request, .. } => match request {
                 /*
                  * KEY-VALUE
                  */
-                ClientRequest::Get(r) => get(&mut client, &config, cache_name, r).await,
-                ClientRequest::Set(r) => set(&mut client, &config, cache_name, r).await,
+                ClientRequest::Get(r) => {
+                    latency_histograms.push(&KVGET_RESPONSE_LATENCY);
+                    get(&mut client, &config, cache_name, r).await
+                }
+                ClientRequest::Set(r) => {
+                    latency_histograms.push(&KVSET_RESPONSE_LATENCY);
+                    set(&mut client, &config, cache_name, r).await
+                }
                 ClientRequest::Delete(r) => delete(&mut client, &config, cache_name, r).await,
 
                 /*
@@ -182,8 +189,9 @@ async fn task(
                 RESPONSE_OK.increment();
 
                 let latency = stop.duration_since(start).as_nanos() as u64;
-
-                let _ = RESPONSE_LATENCY.increment(latency);
+                for hist in latency_histograms {
+                    let _ = hist.increment(latency);
+                }
             }
             Err(ResponseError::Exception) => {
                 RESPONSE_EX.increment();

src/clients/cache/redis/mod.rs

@@ -120,6 +120,7 @@ async fn task(
 
         REQUEST.increment();
         let start = Instant::now();
+        let mut latency_histograms = vec![&RESPONSE_LATENCY];
         let result = match work_item {
             ClientWorkItemKind::Request { request, .. } => match request {
                 /*
@@ -132,9 +133,15 @@ async fn task(
                  */
                 ClientRequest::Add(r) => add(&mut con, &config, r).await,
                 ClientRequest::Delete(r) => delete(&mut con, &config, r).await,
-                ClientRequest::Get(r) => get(&mut con, &config, r).await,
+                ClientRequest::Get(r) => {
+                    latency_histograms.push(&KVGET_RESPONSE_LATENCY);
+                    get(&mut con, &config, r).await
+                }
                 ClientRequest::Replace(r) => replace(&mut con, &config, r).await,
-                ClientRequest::Set(r) => set(&mut con, &config, r).await,
+                ClientRequest::Set(r) => {
+                    latency_histograms.push(&KVSET_RESPONSE_LATENCY);
+                    set(&mut con, &config, r).await
+                }
 
                 /*
                  * HASHES (DICTIONARIES)
@@ -202,8 +209,9 @@ async fn task(
             Ok(_) => {
                 connection = Some(con);
                 RESPONSE_OK.increment();
-
-                let _ = RESPONSE_LATENCY.increment(latency_ns);
+                for hist in latency_histograms {
+                    let _ = hist.increment(latency_ns);
+                }
             }
             Err(ResponseError::Exception) => {
                 RESPONSE_EX.increment();

src/metrics/mod.rs

@@ -351,6 +351,22 @@ macro_rules! request {
 pub static RESPONSE_LATENCY: AtomicHistogram = AtomicHistogram::new(7, 64);
 pub static RESPONSE_LATENCY_HISTOGRAM: &str = "response_latency";
 
+#[metric(
+    name = KVGET_RESPONSE_LATENCY_HISTOGRAM,
+    description = "Distribution of response latencies for key-value GET requests",
+    metadata = { unit = "nanoseconds" }
+)]
+pub static KVGET_RESPONSE_LATENCY: AtomicHistogram = AtomicHistogram::new(7, 64);
+pub static KVGET_RESPONSE_LATENCY_HISTOGRAM: &str = "kvget_response_latency";
+
+#[metric(
+    name = KVSET_RESPONSE_LATENCY_HISTOGRAM,
+    description = "Distribution of response latencies for key-value SET requests",
+    metadata = { unit = "nanoseconds" }
+)]
+pub static KVSET_RESPONSE_LATENCY: AtomicHistogram = AtomicHistogram::new(7, 64);
+pub static KVSET_RESPONSE_LATENCY_HISTOGRAM: &str = "kvset_response_latency";
+
 #[metric(
     name = RESPONSE_TTFB_HISTOGRAM,
     description = "Distribution of time-to-first-byte for responses",

src/output/mod.rs

@@ -110,6 +110,8 @@ fn client_stats(snapshot: &mut MetricsSnapshot) {
     let connect_sr = 100.0 * connect_ok / connect_total;
 
     let response_latency = snapshot.percentiles(RESPONSE_LATENCY_HISTOGRAM);
+    let kvset_response_latency = snapshot.percentiles(KVSET_RESPONSE_LATENCY_HISTOGRAM);
+    let kvget_response_latency = snapshot.percentiles(KVGET_RESPONSE_LATENCY_HISTOGRAM);
     let response_ttfb = snapshot.percentiles(RESPONSE_TTFB_HISTOGRAM);
 
     output!(
@@ -168,6 +170,24 @@ fn client_stats(snapshot: &mut MetricsSnapshot) {
 
     output!("{latencies}");
 
+    let mut kvset_latencies = "Client Key-Value SET Response Latency (us):".to_owned();
+
+    for (label, _percentile, nanoseconds) in kvset_response_latency {
+        let microseconds = nanoseconds / 1000;
+        kvset_latencies.push_str(&format!(" {label}: {microseconds}"))
+    }
+
+    output!("{kvset_latencies}");
+
+    let mut kvget_latencies = "Client Key-Value GET Response Latency (us):".to_owned();
+
+    for (label, _percentile, nanoseconds) in kvget_response_latency {
+        let microseconds = nanoseconds / 1000;
+        kvget_latencies.push_str(&format!(" {label}: {microseconds}"))
+    }
+
+    output!("{kvget_latencies}");
+
     let mut latencies = "Client Time-to-First-Byte (us):".to_owned();
 
     for (label, _percentile, nanoseconds) in response_ttfb {