Merge pull request #1263 from wprzytula/330-make-metrics-collection-optional/faster

Make metrics collection optional/faster

Author: wprzytula

.github/workflows/rust.yml

@@ -50,6 +50,8 @@ jobs:
     # Various (de)serialization features.
     - name: Cargo check with all serialization features
       run: cargo check --all-targets --manifest-path "scylla/Cargo.toml" --features "full-serialization"
+    - name: Cargo check with metrics feature
+      run: cargo check --all-targets --manifest-path "scylla/Cargo.toml" --features "metrics"
     - name: Cargo check with secrecy-08 feature
       run: cargo check --all-targets --manifest-path "scylla/Cargo.toml" --features "secrecy-08"
     - name: Cargo check with chrono-04 feature

Cargo.lock.msrv

@@ -1,5 +1,7 @@
 # Driver metrics
 
+This feature is available only under the crate feature `metrics`.
+
 During operation the driver collects various metrics.
 
 They can be accessed at any moment using `Session::get_metrics()`
@@ -11,6 +13,9 @@ They can be accessed at any moment using `Session::get_metrics()`
 * Total number of paged queries
 * Number of errors during paged queries
 * Number of retries
+* Latency histogram statistics (min, max, mean, standard deviation, percentiles)
+* Rates of queries per second in various time frames
+* Number of active connections, and connection and request timeouts
 
 ### Example
 ```rust
@@ -24,11 +29,32 @@ println!("Queries requested: {}", metrics.get_queries_num());
 println!("Iter queries requested: {}", metrics.get_queries_iter_num());
 println!("Errors occurred: {}", metrics.get_errors_num());
 println!("Iter errors occurred: {}", metrics.get_errors_iter_num());
-println!("Average latency: {}", metrics.get_latency_avg_ms().unwrap());
+println!("Average latency: {}", metrics.get_latency_avg_ms()?);
 println!(
     "99.9 latency percentile: {}",
-    metrics.get_latency_percentile_ms(99.9).unwrap()
+    metrics.get_latency_percentile_ms(99.9)?
 );
+
+let snapshot = metrics.get_snapshot()?;
+println!("Min: {}", snapshot.min);
+println!("Max: {}", snapshot.max);
+println!("Mean: {}", snapshot.mean);
+println!("Standard deviation: {}", snapshot.stddev);
+println!("Median: {}", snapshot.median);
+println!("75th percentile: {}", snapshot.percentile_75);
+println!("95th percentile: {}", snapshot.percentile_95);
+println!("98th percentile: {}", snapshot.percentile_98);
+println!("99th percentile: {}", snapshot.percentile_99);
+println!("99.9th percentile: {}", snapshot.percentile_99_9);
+
+println!("Mean rate: {}", metrics.get_mean_rate());
+println!("One minute rate: {}", metrics.get_one_minute_rate());
+println!("Five minute rate: {}", metrics.get_five_minute_rate());
+println!("Fifteen minute rate: {}", metrics.get_fifteen_minute_rate());
+
+println!("Total connections: {}", metrics.get_total_connections());
+println!("Connection timeouts: {}", metrics.get_connection_timeouts());
+println!("Requests timeouts: {}", metrics.get_request_timeouts());
 # Ok(())
 # }
 ```

docs/source/metrics/metrics.md

@@ -4,6 +4,8 @@ This policy has access to `Metrics` shared with session, and triggers
 speculative execution when the request to the current host is above a
 given percentile.
 
+This policy requires enabling crate feature `"metrics"` to be available.
+
 
 ### Example
 To use this policy in `Session`:

docs/source/speculative-execution/percentile.md

@@ -19,6 +19,7 @@ scylla = { path = "../scylla", features = [
     "num-bigint-03",
     "num-bigint-04",
     "bigdecimal-04",
+    "metrics",
 ] }
 tokio = { version = "1.34", features = ["full"] }
 tracing = { version = "0.1.25", features = ["log"] }

examples/Cargo.toml

@@ -94,12 +94,33 @@ async fn main() -> Result<()> {
     println!("Iter queries requested: {}", metrics.get_queries_iter_num());
     println!("Errors occurred: {}", metrics.get_errors_num());
     println!("Iter errors occurred: {}", metrics.get_errors_iter_num());
-    println!("Average latency: {}", metrics.get_latency_avg_ms().unwrap());
+    println!("Average latency: {}", metrics.get_latency_avg_ms()?);
     println!(
         "99.9 latency percentile: {}",
-        metrics.get_latency_percentile_ms(99.9).unwrap()
+        metrics.get_latency_percentile_ms(99.9)?
     );
 
+    let snapshot = metrics.get_snapshot()?;
+    println!("Min: {}", snapshot.min);
+    println!("Max: {}", snapshot.max);
+    println!("Mean: {}", snapshot.mean);
+    println!("Standard deviation: {}", snapshot.stddev);
+    println!("Median: {}", snapshot.median);
+    println!("75th percentile: {}", snapshot.percentile_75);
+    println!("95th percentile: {}", snapshot.percentile_95);
+    println!("98th percentile: {}", snapshot.percentile_98);
+    println!("99th percentile: {}", snapshot.percentile_99);
+    println!("99.9th percentile: {}", snapshot.percentile_99_9);
+
+    println!("Mean rate: {}", metrics.get_mean_rate());
+    println!("One minute rate: {}", metrics.get_one_minute_rate());
+    println!("Five minute rate: {}", metrics.get_five_minute_rate());
+    println!("Fifteen minute rate: {}", metrics.get_fifteen_minute_rate());
+
+    println!("Total connections: {}", metrics.get_total_connections());
+    println!("Connection timeouts: {}", metrics.get_connection_timeouts());
+    println!("Requests timeouts: {}", metrics.get_request_timeouts());
+
     println!("Ok.");
 
     Ok(())

examples/basic.rs

@@ -39,6 +39,7 @@ full-serialization = [
     "num-bigint-04",
     "bigdecimal-04",
 ]
+metrics = ["dep:histogram"]
 unstable-testing = []
 
 [dependencies]
@@ -47,7 +48,7 @@ byteorder = "1.3.4"
 bytes = "1.0.1"
 futures = "0.3.6"
 hashbrown = "0.14"
-histogram = "0.6.9"
+histogram = { version = "0.11.1", optional = true }
 tokio = { version = "1.34", features = [
     "net",
     "time",

scylla/Cargo.toml

@@ -32,6 +32,7 @@ use crate::frame::response::result;
 use crate::network::Connection;
 use crate::observability::driver_tracing::RequestSpan;
 use crate::observability::history::{self, HistoryListener};
+#[cfg(feature = "metrics")]
 use crate::observability::metrics::Metrics;
 use crate::policies::load_balancing::{self, RoutingInfo};
 use crate::policies::retry::{RequestInfo, RetryDecision, RetrySession};
@@ -66,6 +67,7 @@ pub(crate) struct PreparedIteratorConfig {
     pub(crate) values: SerializedValues,
     pub(crate) execution_profile: Arc<ExecutionProfileInner>,
     pub(crate) cluster_state: Arc<ClusterState>,
+    #[cfg(feature = "metrics")]
     pub(crate) metrics: Arc<Metrics>,
 }
 
@@ -139,6 +141,7 @@ struct PagerWorker<'a, QueryFunc, SpanCreatorFunc> {
     query_consistency: Consistency,
     retry_session: Box<dyn RetrySession>,
     execution_profile: Arc<ExecutionProfileInner>,
+    #[cfg(feature = "metrics")]
     metrics: Arc<Metrics>,
 
     paging_state: PagingState,
@@ -237,11 +240,13 @@ where
 
                 match retry_decision {
                     RetryDecision::RetrySameTarget(cl) => {
+                        #[cfg(feature = "metrics")]
                         self.metrics.inc_retries_num();
                         current_consistency = cl.unwrap_or(current_consistency);
                         continue 'same_node_retries;
                     }
                     RetryDecision::RetryNextTarget(cl) => {
+                        #[cfg(feature = "metrics")]
                         self.metrics.inc_retries_num();
                         current_consistency = cl.unwrap_or(current_consistency);
                         continue 'nodes_in_plan;
@@ -301,6 +306,7 @@ where
         node: NodeRef<'_>,
         request_span: &RequestSpan,
     ) -> Result<ControlFlow<PageSendAttemptedProof, ()>, RequestAttemptError> {
+        #[cfg(feature = "metrics")]
         self.metrics.inc_total_paged_queries();
         let query_start = std::time::Instant::now();
 
@@ -326,6 +332,7 @@ where
                 tracing_id,
                 ..
             }) => {
+                #[cfg(feature = "metrics")]
                 let _ = self.metrics.log_query_latency(elapsed.as_millis() as u64);
                 self.log_attempt_success();
                 self.log_request_success();
@@ -361,6 +368,7 @@ where
                 Ok(ControlFlow::Continue(()))
             }
             Err(err) => {
+                #[cfg(feature = "metrics")]
                 self.metrics.inc_failed_paged_queries();
                 self.execution_profile
                     .load_balancing_policy
@@ -380,6 +388,7 @@ where
                 Ok(ControlFlow::Break(proof))
             }
             Ok(response) => {
+                #[cfg(feature = "metrics")]
                 self.metrics.inc_failed_paged_queries();
                 let err =
                     RequestAttemptError::UnexpectedResponse(response.response.to_response_kind());
@@ -664,7 +673,7 @@ impl QueryPager {
         statement: Statement,
         execution_profile: Arc<ExecutionProfileInner>,
         cluster_state: Arc<ClusterState>,
-        metrics: Arc<Metrics>,
+        #[cfg(feature = "metrics")] metrics: Arc<Metrics>,
     ) -> Result<Self, NextPageError> {
         let (sender, receiver) = mpsc::channel::<Result<ReceivedPage, NextPageError>>(1);
 
@@ -727,6 +736,7 @@ impl QueryPager {
                 query_consistency: consistency,
                 retry_session,
                 execution_profile,
+                #[cfg(feature = "metrics")]
                 metrics,
                 paging_state: PagingState::start(),
                 history_listener: statement.config.history_listener.clone(),
@@ -847,6 +857,7 @@ impl QueryPager {
                 query_consistency: consistency,
                 retry_session,
                 execution_profile: config.execution_profile,
+                #[cfg(feature = "metrics")]
                 metrics: config.metrics,
                 paging_state: PagingState::start(),
                 history_listener: config.prepared.config.history_listener.clone(),