Repository: tikv/tikv (language: rust)

Issue:
Enhance raft snapshot metrics
## Development Task

Currently, in TiKV->Snapshot->Snapshot transport seed, we only have metrics for logical KV transport throughput. To enhance diagnostic efficiency, we need to add metrics to reflect the physical SST data throughput and actual read I/O throughput from RocksDB scans.


Discussion / hints:
/assign @hhwyt 

Fix patch:
diff --git a/components/file_system/src/lib.rs b/components/file_system/src/lib.rs
index f9240e998f6..01bc0105ddf 100644
--- a/components/file_system/src/lib.rs
+++ b/components/file_system/src/lib.rs
@@ -206,6 +206,8 @@ pub struct IoBytesTracker {
     initialized: bool,
     // Stores the previous successfully fetched I/O bytes. Used to calculate deltas.
     prev_io_bytes: IoBytes,
+    // Stores the initial successfully fetched I/O bytes.
+    initial_io_bytes: IoBytes,
 }
 impl IoBytesTracker {
     /// Creates a new `IoBytesTracker` and attempts to initialize it.
@@ -218,6 +220,7 @@ impl IoBytesTracker {
         let mut tracker = IoBytesTracker {
             initialized: false,
             prev_io_bytes: IoBytes::default(),
+            initial_io_bytes: IoBytes::default(),
         };
         tracker.update(); // Attempt to initialize immediately
         tracker
@@ -242,6 +245,7 @@ impl IoBytesTracker {
                     // Initialize on the first successful fetch
                     self.prev_io_bytes = current_io_bytes;
                     self.initialized = true;
+                    self.initial_io_bytes = current_io_bytes;
                     None // No delta to report yet
                 }
             }
@@ -251,6 +255,14 @@ impl IoBytesTracker {
             }
         }
     }
+
+    /// Returns the total accumulated I/O bytes.
+    pub fn get_total_io_bytes(&self) -> IoBytes {
+        IoBytes {
+            read: self.prev_io_bytes.read - self.initial_io_bytes.read,
+            write: self.prev_io_bytes.write - self.initial_io_bytes.write,
+        }
+    }
 }
 
 impl Default for IoBytesTracker {
@@ -773,11 +785,9 @@ mod tests {
         assert!(!file.exists());
     }
 
+    #[cfg(feature = "failpoints")]
     #[test]
     fn test_io_bytes_tracker_normal() {
-        #[cfg(not(feature = "failpoints"))]
-        return;
-
         fail::cfg("delta_read_io_bytes", "return(100)").unwrap();
         fail::cfg("delta_write_io_bytes", "return(50)").unwrap();
         let mut io_tracker = IoBytesTracker::new();
@@ -789,13 +799,19 @@ mod tests {
         assert_eq!(io_bytes.unwrap().write, 50);
         assert_eq!(io_tracker.prev_io_bytes.read, 200);
         assert_eq!(io_tracker.prev_io_bytes.write, 100);
+
+        let total_io_bytes = io_tracker.get_total_io_bytes();
+        assert_eq!(total_io_bytes.read, 100);
+        assert_eq!(total_io_bytes.write, 50);
+        let _ = io_tracker.update();
+        let total_io_bytes = io_tracker.get_total_io_bytes();
+        assert_eq!(total_io_bytes.read, 200);
+        assert_eq!(total_io_bytes.write, 100);
     }
 
+    #[cfg(feature = "failpoints")]
     #[test]
     fn test_io_bytes_tracker_initialization_failure() {
-        #[cfg(not(feature = "failpoints"))]
-        return;
-
         fail::cfg("failed_to_get_thread_io_bytes_stats", "1*return").unwrap();
         fail::cfg("delta_read_io_bytes", "return(100)").unwrap();
         fail::cfg("delta_write_io_bytes", "return(50)").unwrap();
diff --git a/components/raftstore/Cargo.toml b/components/raftstore/Cargo.toml
index ce017f9330f..aad9e798dcf 100644
--- a/components/raftstore/Cargo.toml
+++ b/components/raftstore/Cargo.toml
@@ -36,7 +36,7 @@ engine_rocks = { workspace = true, optional = true }
 engine_traits = { workspace = true }
 error_code = { workspace = true }
 fail = "0.5"
-file_system = { workspace = true, features = ["testexport"] }
+file_system = { workspace = true }
 futures = "0.3"
 futures-util = { version = "0.3.1", default-features = false, features = [
   "io",
@@ -88,5 +88,6 @@ encryption_export = { workspace = true }
 engine_panic = { workspace = true }
 engine_rocks = { workspace = true }
 engine_test = { workspace = true }
+file_system = { workspace = true, features = ["testexport"] }
 panic_hook = { workspace = true }
 test_sst_importer = { workspace = true }
diff --git a/components/raftstore/src/store/metrics.rs b/components/raftstore/src/store/metrics.rs
index 38716c5b277..bccc557cdae 100644
--- a/components/raftstore/src/store/metrics.rs
+++ b/components/raftstore/src/store/metrics.rs
@@ -275,6 +275,7 @@ make_static_metric! {
         kv,
         sst,
         plain,
+        io,
     }
 
     pub struct SnapshotBrWaitApplyEvent : IntCounter {
diff --git a/components/raftstore/src/store/snap.rs b/components/raftstore/src/store/snap.rs
index 72b3ce1bf40..32bd389a7e9 100644
--- a/components/raftstore/src/store/snap.rs
+++ b/components/raftstore/src/store/snap.rs
@@ -911,6 +911,9 @@ impl Snapshot {
             SNAPSHOT_LIMIT_GENERATE_BYTES_VEC
                 .plain
                 .inc_by(cf_stat.total_plain_size as u64);
+            SNAPSHOT_LIMIT_GENERATE_BYTES_VEC
+                .io
+                .inc_by(cf_stat.total_io_size as u64);
             cf_file.kv_count = cf_stat.key_count as u64;
             if cf_file.kv_count > 0 {
                 // Use `kv_count` instead of file size to check empty files because encrypted
diff --git a/components/raftstore/src/store/snap/io.rs b/components/raftstore/src/store/snap/io.rs
index 74a5739f2c6..bf1218fa09b 100644
--- a/components/raftstore/src/store/snap/io.rs
+++ b/components/raftstore/src/store/snap/io.rs
@@ -52,6 +52,10 @@ pub struct BuildStatistics {
     /// The total size (in bytes) of the raw data in plain text format.
     /// This represents the uncompressed size of the CF_LOCK data.
     pub total_plain_size: usize,
+
+    /// The total size (in bytes) of the IO used for read data.
+    /// This does not include usage from CF_LOCK.
+    pub total_io_size: usize,
 }
 
 /// Build a snapshot file for the given column family in plain format.
@@ -264,6 +268,7 @@ where
     // Handle the IO generated by the remaining key-value pairs less than
     // SCAN_BYTES_PER_IO_LIMIT_CHECK.
     handle_read_io_usage(&mut io_tracker, &mut remained_quota);
+    stats.total_io_size = io_tracker.get_total_io_bytes().read as usize;
 
     if stats.key_count > 0 {
         stats.total_sst_size +=
@@ -665,11 +670,9 @@ mod tests {
     // function returns 4096 bytes of I/O usage, resulting in total of 8192 bytes.
     // With the 8000 bytes/sec limitation, we assert that the elapsed time must
     // exceed 1 second.
+    #[cfg(feature = "failpoints")]
     #[test]
     fn test_build_sst_with_io_limiter() {
-        #[cfg(not(feature = "failpoints"))]
-        return;
-
         let dir = Builder::new().prefix("test-io-limiter").tempdir().unwrap();
         let db = open_test_db_with_nkeys(dir.path(), None, None, 1000).unwrap();
         // The max throughput is 8000 bytes/sec.
@@ -698,6 +701,8 @@ mod tests {
             true,
         )
         .unwrap();
+        // 8192 represents the mocked total I/O bytes.
+        assert_eq!(stats.total_io_size, 8192);
         assert_eq!(stats.total_kv_size, 11890);
         // Must exceed 1 second!
         assert!(start.saturating_elapsed_secs() > 1_f64);
diff --git a/components/resource_control/Cargo.toml b/components/resource_control/Cargo.toml
index 6ce694387a0..c800c05c86f 100644
--- a/components/resource_control/Cargo.toml
+++ b/components/resource_control/Cargo.toml
@@ -13,7 +13,7 @@ collections = { workspace = true }
 crossbeam = { workspace = true }
 dashmap = "5.1"
 fail = "0.5"
-file_system = { workspace = true, features = ["testexport"] }
+file_system = { workspace = true }
 futures = { version = "0.3", features = ["compat"] }
 kvproto = { workspace = true }
 lazy_static = "1.0"
@@ -33,5 +33,6 @@ tokio-timer = { workspace = true }
 yatp = { git = "https://github.com/tikv/yatp.git", branch = "master" }
 
 [dev-dependencies]
+file_system = { workspace = true, features = ["testexport"] }
 rand = "0.8"
 test_pd = { workspace = true }
diff --git a/components/resource_control/src/future.rs b/components/resource_control/src/future.rs
index c115fcffef4..c48a138b59a 100644
--- a/components/resource_control/src/future.rs
+++ b/components/resource_control/src/future.rs
@@ -242,11 +242,9 @@ mod tests {
     #[allow(clippy::unused_async)]
     async fn empty() {}
 
+    #[cfg(feature = "failpoints")]
     #[test]
     fn test_limited_future() {
-        #[cfg(not(feature = "failpoints"))]
-        return;
-
         let pool = YatpPoolBuilder::new(DefaultTicker::default())
             .thread_count(1, 1, 1)
             .name_prefix("test")


Test patch: