Add metrics to TxPool

CHANGELOG.md

@@ -6,6 +6,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## [Unreleased]
 
+### Added
+- [2321](https://github.com/FuelLabs/fuel-core/pull/2321): New metrics for the txpool: "The size of transactions in the txpool" (`txpool_tx_size`), "The time spent by a transaction in the txpool in seconds" (`txpool_tx_time_in_txpool_seconds`), The number of transactions in the txpool (`txpool_number_of_transactions`), "The number of transactions pending verification before entering the txpool" (`txpool_number_of_transactions_pending_verification`), "The number of executable transactions in the txpool" (`txpool_number_of_executable_transactions`), "The time it took to select transactions for inclusion in a block in nanoseconds" (`txpool_select_transaction_time_nanoseconds`), The time it took to insert a transaction in the txpool in milliseconds (`txpool_insert_transaction_time_milliseconds`).
+
 ## [Version 0.40.0]
 
 ### Added

bin/fuel-core/src/cli/run.rs

@@ -298,7 +298,7 @@ impl Command {
             p2p_args,
             #[cfg(feature = "p2p")]
             sync_args,
-            disabled_metrics: metrics,
+            disabled_metrics,
             max_da_lag,
             max_wait_time,
             tx_pool,
@@ -309,7 +309,7 @@ impl Command {
             profiling: _,
         } = self;
 
-        let enabled_metrics = metrics.list_of_enabled();
+        let enabled_metrics = disabled_metrics.list_of_enabled();
 
         if !enabled_metrics.is_empty() {
             info!("`{:?}` metrics are enabled", enabled_metrics);
@@ -334,7 +334,7 @@ impl Command {
         #[cfg(feature = "p2p")]
         let p2p_cfg = p2p_args.into_config(
             chain_config.chain_name.clone(),
-            metrics.is_enabled(Module::P2P),
+            disabled_metrics.is_enabled(Module::P2P),
         )?;
 
         let trigger: Trigger = poa_trigger.into();
@@ -444,7 +444,7 @@ impl Command {
         };
 
         let block_importer = fuel_core::service::config::fuel_core_importer::Config::new(
-            metrics.is_enabled(Module::Importer),
+            disabled_metrics.is_enabled(Module::Importer),
         );
 
         let da_compression = match da_compression {
@@ -562,10 +562,11 @@ impl Command {
                 pool_limits,
                 heavy_work: pool_heavy_work_config,
                 service_channel_limits,
+                metrics: disabled_metrics.is_enabled(Module::TxPool),
             },
             block_producer: ProducerConfig {
                 coinbase_recipient,
-                metrics: metrics.is_enabled(Module::Producer),
+                metrics: disabled_metrics.is_enabled(Module::Producer),
             },
             starting_gas_price,
             gas_price_change_percent,

crates/metrics/src/buckets.rs

@@ -9,6 +9,8 @@ use strum_macros::EnumIter;
 #[cfg_attr(test, derive(EnumIter))]
 pub(crate) enum Buckets {
     Timing,
+    TimingCoarseGrained,
+    TransactionSize,
 }
 static BUCKETS: OnceLock<HashMap<Buckets, Vec<f64>>> = OnceLock::new();
 pub(crate) fn buckets(b: Buckets) -> impl Iterator<Item = f64> {
@@ -34,6 +36,46 @@ fn initialize_buckets() -> HashMap<Buckets, Vec<f64>> {
                10.000,
             ],
         ),
+        (
+            Buckets::TimingCoarseGrained,
+            vec![
+                5.0,
+                10.0,
+                25.0,
+                50.0,
+                100.0,
+                250.0,
+                500.0,
+                1000.0,
+                2500.0, 
+                5000.0, 
+                10000.0,
+            ],
+        ),
+        (
+            // We consider blocks up to 256kb in size and single transaction can take any of this space.
+            Buckets::TransactionSize,
+            vec![
+                  1.0 * 1024.0,
+                  2.0 * 1024.0,
+                  3.0 * 1024.0,
+                  4.0 * 1024.0,
+                  5.0 * 1024.0,
+                  7.0 * 1024.0,
+                 10.0 * 1024.0,
+                 13.0 * 1024.0,
+                 18.0 * 1024.0,
+                 24.0 * 1024.0,
+                 33.0 * 1024.0,
+                 44.0 * 1024.0,
+                 59.0 * 1024.0,
+                 79.0 * 1024.0,
+                106.0 * 1024.0,
+                142.0 * 1024.0,
+                191.0 * 1024.0,
+                256.0 * 1024.0,
+            ]
+        ),
     ]
     .into_iter()
     .collect()

crates/metrics/src/txpool_metrics.rs

@@ -1,24 +1,97 @@
-use crate::global_registry;
-use prometheus_client::metrics::histogram::Histogram;
+use crate::{
+    buckets::{
+        buckets,
+        Buckets,
+    },
+    global_registry,
+};
+use prometheus_client::metrics::{
+    gauge::Gauge,
+    histogram::Histogram,
+};
 use std::sync::OnceLock;
 
 pub struct TxPoolMetrics {
-    pub tx_size_histogram: Histogram,
+    /// Size of transactions in the txpool in bytes
+    pub tx_size: Histogram,
+    /// Number of transactions in the txpool
+    pub number_of_transactions: Gauge,
+    /// Number of transactions pending verification, before being inserted in the txpool
+    pub number_of_transactions_pending_verification: Gauge,
+    /// Number of transactions that can be included in the next block
+    pub number_of_executable_transactions: Gauge,
+    /// Time of transactions in the txpool in seconds
+    pub transaction_time_in_txpool_secs: Histogram,
+    /// Time actively spent by transaction insertion in the thread pool
+    pub transaction_insertion_time_in_thread_pool_milliseconds: Histogram,
+    /// How long it took for the selection algorithm to select transactions
+    pub select_transaction_time_nanoseconds: Histogram,
 }
 
 impl Default for TxPoolMetrics {
     fn default() -> Self {
-        let tx_sizes = Vec::new();
+        let tx_size = Histogram::new(buckets(Buckets::TransactionSize));
+        let transaction_time_in_txpool_secs = Histogram::new(buckets(Buckets::Timing));
+        let select_transaction_time_nanoseconds =
+            Histogram::new(buckets(Buckets::TimingCoarseGrained));
+        let transaction_insertion_time_in_thread_pool_milliseconds =
+            Histogram::new(buckets(Buckets::TimingCoarseGrained));
 
-        let tx_size_histogram = Histogram::new(tx_sizes.into_iter());
+        let number_of_transactions = Gauge::default();
+        let number_of_transactions_pending_verification = Gauge::default();
+        let number_of_executable_transactions = Gauge::default();
 
-        let metrics = TxPoolMetrics { tx_size_histogram };
+        let metrics = TxPoolMetrics {
+            tx_size,
+            number_of_transactions,
+            number_of_transactions_pending_verification,
+            number_of_executable_transactions,
+            transaction_time_in_txpool_secs,
+            transaction_insertion_time_in_thread_pool_milliseconds,
+            select_transaction_time_nanoseconds,
+        };
 
         let mut registry = global_registry().registry.lock();
         registry.register(
-            "Tx_Size_Histogram",
-            "A Histogram keeping track of the size of txs",
-            metrics.tx_size_histogram.clone(),
+            "txpool_tx_size",
+            "The size of transactions in the txpool",
+            metrics.tx_size.clone(),
+        );
+
+        registry.register(
+            "txpool_tx_time_in_txpool_seconds",
+            "The time spent by a transaction in the txpool in seconds",
+            metrics.transaction_time_in_txpool_secs.clone(),
+        );
+
+        registry.register(
+            "txpool_number_of_transactions",
+            "The number of transactions in the txpool",
+            metrics.number_of_transactions.clone(),
+        );
+
+        registry.register(
+            "txpool_number_of_executable_transactions",
+            "The number of executable transactions in the txpool",
+            metrics.number_of_executable_transactions.clone(),
+        );
+
+        registry.register(
+            "txpool_number_of_transactions_pending_verification",
+            "The number of transactions pending verification before entering the txpool",
+            metrics.number_of_transactions_pending_verification.clone(),
+        );
+
+        registry.register(
+            "txpool_select_transaction_time_nanoseconds",
+            "The time it took to select transactions for inclusion in a block in nanoseconds",
+            metrics.select_transaction_time_nanoseconds.clone(),
+        );
+
+        registry.register(
+            "txpool_insert_transaction_time_milliseconds",
+            "The time it took to insert a transaction in the txpool in milliseconds",
+            metrics.select_transaction_time_nanoseconds.clone(),
         );
 
         metrics

crates/services/txpool_v2/Cargo.toml

@@ -14,6 +14,7 @@ description = "Transaction pool that manages transactions and their dependencies
 anyhow = { workspace = true }
 async-trait = { workspace = true }
 derive_more = { workspace = true }
+fuel-core-metrics = { workspace = true }
 fuel-core-services = { workspace = true, features = ["sync-processor"] }
 fuel-core-storage = { workspace = true, features = ["std"] }
 fuel-core-types = { workspace = true, features = ["test-helpers"] }

crates/services/txpool_v2/src/config.rs

@@ -140,6 +140,8 @@ pub struct Config {
     pub heavy_work: HeavyWorkConfig,
     /// Blacklist. Transactions with blacklisted inputs will not be accepted.
     pub black_list: BlackList,
+    /// Enable metrics when set to true
+    pub metrics: bool,
 }
 
 #[derive(Clone, Debug)]
@@ -199,6 +201,7 @@ impl Default for Config {
                 max_pending_write_pool_requests: 1000,
                 max_pending_read_pool_requests: 1000,
             },
+            metrics: false,
         }
     }
 }

crates/services/txpool_v2/src/pool.rs

@@ -3,7 +3,10 @@ mod collisions;
 use std::{
     collections::HashMap,
     iter,
-    time::SystemTime,
+    time::{
+        Instant,
+        SystemTime,
+    },
 };
 
 use collisions::CollisionsExt;
@@ -228,6 +231,22 @@ where
         Ok(can_store_transaction)
     }
 
+    fn record_transaction_time_in_txpool(tx: &StorageData) {
+        if let Ok(elapsed) = tx.creation_instant.elapsed() {
+            fuel_core_metrics::txpool_metrics::txpool_metrics()
+                .transaction_time_in_txpool_secs
+                .observe(elapsed.as_secs_f64());
+        } else {
+            tracing::warn!("Failed to calculate transaction time in txpool");
+        }
+    }
+
+    fn record_select_transaction_time_in_nanoseconds(start: Instant) {
+        let elapsed = start.elapsed().as_nanos() as f64;
+        fuel_core_metrics::txpool_metrics::txpool_metrics()
+            .select_transaction_time_nanoseconds
+            .observe(elapsed);
+    }
     // TODO: Use block space also (https://github.com/FuelLabs/fuel-core/issues/2133)
     /// Extract transactions for a block.
     /// Returns a list of transactions that were selected for the block
@@ -236,9 +255,23 @@ where
         &mut self,
         constraints: Constraints,
     ) -> Vec<ArcPoolTx> {
-        self.selection_algorithm
-            .gather_best_txs(constraints, &mut self.storage)
+        let start = std::time::Instant::now();
+        let metrics = self.config.metrics;
+        let best_txs = self
+            .selection_algorithm
+            .gather_best_txs(constraints, &mut self.storage);
+
+        if metrics {
+            Self::record_select_transaction_time_in_nanoseconds(start)
+        };
+
+        best_txs
             .into_iter()
+            .inspect(|storage_data| {
+                if metrics {
+                    Self::record_transaction_time_in_txpool(storage_data)
+                }
+            })
             .map(|storage_entry| {
                 self.update_components_and_caches_on_removal(iter::once(&storage_entry));
 

crates/services/txpool_v2/src/selection_algorithms/mod.rs

@@ -37,6 +37,9 @@ pub trait SelectionAlgorithm {
         store_entry: &StorageData,
     );
 
+    /// Returns the number of executable transactions
+    fn number_of_executable_transactions(&self) -> usize;
+
     /// Get less worth transactions iterator
     fn get_less_worth_txs(&self) -> impl Iterator<Item = &Self::StorageIndex>;
 

crates/services/txpool_v2/src/selection_algorithms/ratio_tip_gas.rs

@@ -252,4 +252,8 @@ where
         let key = Self::key(storage_entry);
         self.on_removed_transaction_inner(key)
     }
+
+    fn number_of_executable_transactions(&self) -> usize {
+        self.executable_transactions_sorted_tip_gas_ratio.len()
+    }
 }