Make metrics collection optional/faster

scylla/src/client/session.rs

@@ -997,11 +997,14 @@ where
             identity: config.identity,
         };
 
+        let metrics = Arc::new(Metrics::new());
+
         let pool_config = PoolConfig {
             connection_config,
             pool_size: config.connection_pool_size,
             can_use_shard_aware_port: !config.disallow_shard_aware_port,
             keepalive_interval: config.keepalive_interval,
+            metrics: Some(metrics.clone()),
         };
 
         let cluster = Cluster::new(
@@ -1021,7 +1024,7 @@ where
             cluster,
             default_execution_profile_handle,
             schema_agreement_interval: config.schema_agreement_interval,
-            metrics: Arc::new(Metrics::new()),
+            metrics,
             schema_agreement_timeout: config.schema_agreement_timeout,
             schema_agreement_automatic_waiting: config.schema_agreement_automatic_waiting,
             refresh_metadata_on_auto_schema_agreement: config
@@ -1978,6 +1981,7 @@ where
             Some(timeout) => tokio::time::timeout(timeout, runner)
                 .await
                 .unwrap_or_else(|e| {
+                    self.metrics.inc_request_timeouts();
                     Err(QueryError::RequestTimeout(format!(
                         "Request took longer than {}ms: {}",
                         timeout.as_millis(),

scylla/src/cluster/metadata.rs

@@ -762,6 +762,10 @@ impl MetadataReader {
             // The shard-aware port won't be used with PerHost pool size anyway,
             // so explicitly disable it here
             can_use_shard_aware_port: false,
+
+            // TODO: This should probably be changed or removed depending on
+            //       how we want to pass the metrics to the PoolRefiller
+            metrics: None,
         };
 
         NodeConnectionPool::new(endpoint, pool_config, None, refresh_requester)

scylla/src/network/connection.rs

@@ -21,6 +21,7 @@ use crate::frame::{
     server_event_type::EventType,
     FrameParams, SerializedRequest,
 };
+use crate::observability::metrics::Metrics;
 use crate::policies::address_translator::AddressTranslator;
 use crate::query::Query;
 use crate::response::query_result::QueryResult;
@@ -1903,13 +1904,22 @@ pub(super) async fn open_connection_to_shard_aware_port(
     shard: Shard,
     sharder: Sharder,
     connection_config: &ConnectionConfig,
+    metrics: Option<Arc<Metrics>>,
 ) -> Result<(Connection, ErrorReceiver), ConnectionError> {
     // Create iterator over all possible source ports for this shard
     let source_port_iter = sharder.iter_source_ports_for_shard(shard);
 
     for port in source_port_iter {
         let connect_result = open_connection(endpoint.clone(), Some(port), connection_config).await;
 
+        if let Some(metrics) = &metrics {
+            if connect_result.is_ok() {
+                metrics.inc_total_connections();
+            } else if let Err(ConnectionError::ConnectTimeout) = &connect_result {
+                metrics.inc_connection_timeouts();
+            }
+        }
+
         match connect_result {
             Err(err) if err.is_address_unavailable_for_use() => continue, // If we can't use this port, try the next one
             result => return result,

scylla/src/network/connection_pool.rs

@@ -12,6 +12,8 @@ use crate::routing::{Shard, ShardCount, Sharder};
 
 use crate::cluster::metadata::{PeerEndpoint, UntranslatedEndpoint};
 
+use crate::observability::metrics::Metrics;
+
 #[cfg(feature = "cloud")]
 use crate::cluster::node::resolve_hostname;
 
@@ -61,6 +63,11 @@ pub(crate) struct PoolConfig {
     pub(crate) pool_size: PoolSize,
     pub(crate) can_use_shard_aware_port: bool,
     pub(crate) keepalive_interval: Option<Duration>,
+    // TODO: The metrics should definitely not be stored here,
+    //       but it was the easiest way to pass it to the refiller.
+    //       It could be refactored to be passed as a parameter to the refiller,
+    //       but it would require a lot of changes in the code.
+    pub(crate) metrics: Option<Arc<Metrics>>,
 }
 
 impl Default for PoolConfig {
@@ -70,6 +77,7 @@ impl Default for PoolConfig {
             pool_size: Default::default(),
             can_use_shard_aware_port: true,
             keepalive_interval: None,
+            metrics: None,
         }
     }
 }
@@ -922,6 +930,8 @@ impl PoolRefiller {
         // As this may may involve resolving a hostname, the whole operation is async.
         let endpoint_fut = self.maybe_translate_for_serverless(endpoint);
 
+        let metrics = self.pool_config.metrics.clone();
+
         let fut = match (self.sharder.clone(), self.shard_aware_port, shard) {
             (Some(sharder), Some(port), Some(shard)) => async move {
                 let shard_aware_endpoint = {
@@ -934,6 +944,7 @@ impl PoolRefiller {
                     shard,
                     sharder.clone(),
                     &cfg,
+                    metrics,
                 )
                 .await;
                 OpenedConnectionEvent {
@@ -946,6 +957,15 @@ impl PoolRefiller {
             _ => async move {
                 let non_shard_aware_endpoint = endpoint_fut.await;
                 let result = open_connection(non_shard_aware_endpoint, None, &cfg).await;
+
+                if let Some(metrics) = metrics {
+                    if result.is_ok() {
+                        metrics.inc_total_connections();
+                    } else if let Err(ConnectionError::ConnectTimeout) = &result {
+                        metrics.inc_connection_timeouts();
+                    }
+                }
+
                 OpenedConnectionEvent {
                     result,
                     requested_shard: None,
@@ -1022,6 +1042,11 @@ impl PoolRefiller {
             match maybe_idx {
                 Some(idx) => {
                     v.swap_remove(idx);
+                    self.pool_config
+                        .metrics
+                        .as_ref()
+                        .unwrap()
+                        .dec_total_connections();
                     true
                 }
                 None => false,
@@ -1242,6 +1267,7 @@ mod tests {
                 0,
                 sharder.clone(),
                 &connection_config,
+                None,
             ));
         }
 

scylla/src/observability/metrics.rs

@@ -180,6 +180,9 @@ pub struct Metrics {
     retries_num: AtomicU64,
     histogram: Arc<LockFreeHistogram>,
     meter: Arc<Meter>,
+    total_connections: AtomicU64,
+    connection_timeouts: AtomicU64,
+    request_timeouts: AtomicU64,
 }
 
 impl Metrics {
@@ -192,6 +195,9 @@ impl Metrics {
             retries_num: AtomicU64::new(0),
             histogram: Arc::new(LockFreeHistogram::default()),
             meter: Arc::new(Meter::new()),
+            total_connections: AtomicU64::new(0),
+            connection_timeouts: AtomicU64::new(0),
+            request_timeouts: AtomicU64::new(0),
         }
     }
 
@@ -223,6 +229,26 @@ impl Metrics {
         self.retries_num.fetch_add(1, ORDER_TYPE);
     }
 
+    /// Increments counter for total number of connections
+    pub(crate) fn inc_total_connections(&self) {
+        self.total_connections.fetch_add(1, ORDER_TYPE);
+    }
+
+    /// Decrements counter for total number of connections
+    pub(crate) fn dec_total_connections(&self) {
+        self.total_connections.fetch_sub(1, ORDER_TYPE);
+    }
+
+    /// Increments counter for connection timeouts
+    pub(crate) fn inc_connection_timeouts(&self) {
+        self.connection_timeouts.fetch_add(1, ORDER_TYPE);
+    }
+
+    /// Increments counter for request timeouts
+    pub(crate) fn inc_request_timeouts(&self) {
+        self.request_timeouts.fetch_add(1, ORDER_TYPE);
+    }
+
     /// Saves to histogram latency of completing single query.
     /// For paged queries it should log latency for every page.
     ///
@@ -324,6 +350,21 @@ impl Metrics {
         self.meter.fifteen_minute_rate()
     }
 
+    /// Returns total number of active connections
+    pub fn get_total_connections(&self) -> u64 {
+        self.total_connections.load(ORDER_TYPE)
+    }
+
+    /// Returns counter for connection timeouts
+    pub fn get_connection_timeouts(&self) -> u64 {
+        self.connection_timeouts.load(ORDER_TYPE)
+    }
+
+    /// Returns counter for request timeouts
+    pub fn get_request_timeouts(&self) -> u64 {
+        self.request_timeouts.load(ORDER_TYPE)
+    }
+
     // Metric implementations
 
     fn mean(h: Histogram) -> Result<u64, MetricsError> {