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/raftstore/src/store/metrics.rs b/components/raftstore/src/store/metrics.rs
index 9428c5025db..38716c5b277 100644
--- a/components/raftstore/src/store/metrics.rs
+++ b/components/raftstore/src/store/metrics.rs
@@ -271,6 +271,12 @@ make_static_metric! {
         finished,
     }
 
+    pub label_enum SnapshotGenerateBytesType {
+        kv,
+        sst,
+        plain,
+    }
+
     pub struct SnapshotBrWaitApplyEvent : IntCounter {
         "event" => SnapshotBrWaitApplyEventType
     }
@@ -339,6 +345,10 @@ make_static_metric! {
             applystore_busy,
         },
     }
+
+    pub struct SnapshotGenerateBytesTypeVec: IntCounter {
+        "type" => SnapshotGenerateBytesType,
+    }
 }
 
 lazy_static! {
@@ -947,9 +957,11 @@ lazy_static! {
         &["type"]
     ).unwrap();
 
-    pub static ref SNAPSHOT_LIMIT_GENERATE_BYTES: IntCounter = register_int_counter!(
+    pub static ref SNAPSHOT_LIMIT_GENERATE_BYTES_VEC: SnapshotGenerateBytesTypeVec = register_static_int_counter_vec!(
+        SnapshotGenerateBytesTypeVec,
         "tikv_snapshot_limit_generate_bytes",
         "Total snapshot generate limit used",
+        &["type"],
     )
     .unwrap();
 
diff --git a/components/raftstore/src/store/snap.rs b/components/raftstore/src/store/snap.rs
index 20b9fba22a1..72b3ce1bf40 100644
--- a/components/raftstore/src/store/snap.rs
+++ b/components/raftstore/src/store/snap.rs
@@ -902,7 +902,15 @@ impl Snapshot {
                     for_balance,
                 )?
             };
-            SNAPSHOT_LIMIT_GENERATE_BYTES.inc_by(cf_stat.total_size as u64);
+            SNAPSHOT_LIMIT_GENERATE_BYTES_VEC
+                .kv
+                .inc_by(cf_stat.total_kv_size as u64);
+            SNAPSHOT_LIMIT_GENERATE_BYTES_VEC
+                .sst
+                .inc_by(cf_stat.total_sst_size as u64);
+            SNAPSHOT_LIMIT_GENERATE_BYTES_VEC
+                .plain
+                .inc_by(cf_stat.total_plain_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
@@ -925,14 +933,16 @@ impl Snapshot {
                 .observe(cf_stat.key_count as f64);
             SNAPSHOT_CF_SIZE
                 .get(*cf_enum)
-                .observe(cf_stat.total_size as f64);
+                .observe(cf_stat.total_kv_size as f64);
             info!(
                 "scan snapshot of one cf";
                 "region_id" => region.get_id(),
                 "snapshot" => self.path(),
                 "cf" => cf,
                 "key_count" => cf_stat.key_count,
-                "size" => cf_stat.total_size,
+                "size" => cf_stat.total_kv_size,
+                "sst_size" => cf_stat.total_sst_size,
+                "plain_size" => cf_stat.total_plain_size,
             );
         }
 
diff --git a/components/raftstore/src/store/snap/io.rs b/components/raftstore/src/store/snap/io.rs
index 1d1597e95e2..03bc6caee1e 100644
--- a/components/raftstore/src/store/snap/io.rs
+++ b/components/raftstore/src/store/snap/io.rs
@@ -9,8 +9,8 @@ use std::{
 
 use encryption::{DataKeyManager, DecrypterReader, EncrypterWriter, Iv};
 use engine_traits::{
-    CfName, Error as EngineError, IterOptions, Iterable, Iterator, KvEngine, Mutable, RefIterable,
-    SstCompressionType, SstReader, SstWriter, SstWriterBuilder, WriteBatch,
+    CfName, Error as EngineError, ExternalSstFileInfo, IterOptions, Iterable, Iterator, KvEngine,
+    Mutable, RefIterable, SstCompressionType, SstReader, SstWriter, SstWriterBuilder, WriteBatch,
 };
 use fail::fail_point;
 #[cfg(not(test))]
@@ -36,10 +36,24 @@ pub trait StaleDetector {
     fn is_stale(&self) -> bool;
 }
 
+/// Statistics for tracking the process of building SST files.
 #[derive(Clone, Copy, Default)]
 pub struct BuildStatistics {
+    /// The total number of keys processed during the build.
     pub key_count: usize,
-    pub total_size: usize,
+
+    /// The total size (in bytes) of key-value pairs processed.
+    /// This represents the combined size of keys and values before any
+    /// compression.
+    pub total_kv_size: usize,
+
+    /// The total size (in bytes) of the generated SST files after compression.
+    /// This reflects the on-disk size of the output files.
+    pub total_sst_size: usize,
+
+    /// 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,
 }
 
 /// Build a snapshot file for the given column family in plain format.
@@ -88,7 +102,7 @@ where
     let mut stats = BuildStatistics::default();
     box_try!(snap.scan(cf, start_key, end_key, false, |key, value| {
         stats.key_count += 1;
-        stats.total_size += key.len() + value.len();
+        stats.total_kv_size += key.len() + value.len();
         box_try!(BytesEncoder::encode_compact_bytes(&mut writer, key));
         box_try!(BytesEncoder::encode_compact_bytes(&mut writer, value));
         Ok(true)
@@ -103,6 +117,8 @@ where
             encrypted_file.unwrap().finalize().unwrap()
         };
         box_try!(file.sync_all());
+        let metadata = box_try!(file.metadata());
+        stats.total_plain_size += metadata.len() as usize;
     } else {
         drop(file);
         box_try!(fs::remove_file(path));
@@ -163,8 +179,8 @@ where
     let finish_sst_writer = |sst_writer: E::SstWriter,
                              path: String,
                              key_mgr: Option<Arc<DataKeyManager>>|
-     -> Result<(), Error> {
-        sst_writer.finish()?;
+     -> Result<u64, Error> {
+        let info = sst_writer.finish()?;
         (|| {
             fail_point!("inject_sst_file_corruption", |_| {
                 static CALLED: std::sync::atomic::AtomicBool =
@@ -199,7 +215,7 @@ where
             return Err(io::Error::new(io::ErrorKind::InvalidData, e).into());
         }
         File::open(&path).and_then(|f| f.sync_all())?;
-        Ok(())
+        Ok(info.file_size())
     };
 
     let instant = Instant::now();
@@ -209,7 +225,7 @@ where
         IoType::Replication
     });
     let mut prev_io_bytes = get_thread_io_bytes_stats().unwrap();
-    let mut next_io_check_size = stats.total_size + SCAN_BYTES_PER_IO_LIMIT_CHECK;
+    let mut next_io_check_size = stats.total_kv_size + SCAN_BYTES_PER_IO_LIMIT_CHECK;
     let handle_read_io_usage = |prev_io_bytes: &mut IoBytes, remained_quota: &mut usize| {
         let cur_io_bytes = get_thread_io_bytes_stats().unwrap();
         let read_delta = (cur_io_bytes.read - prev_io_bytes.read) as usize;
@@ -239,7 +255,9 @@ where
             match result {
                 Ok(new_sst_writer) => {
                     let old_writer = sst_writer.replace(new_sst_writer);
-                    box_try!(finish_sst_writer(old_writer, prev_path, key_mgr.clone()));
+                    stats.total_sst_size +=
+                        box_try!(finish_sst_writer(old_writer, prev_path, key_mgr.clone()))
+                            as usize;
                 }
                 Err(e) => {
                     let io_error = io::Error::new(io::ErrorKind::Other, e);
@@ -249,13 +267,13 @@ where
         }
 
         stats.key_count += 1;
-        stats.total_size += entry_len;
+        stats.total_kv_size += entry_len;
 
-        if stats.total_size >= next_io_check_size {
+        if stats.total_kv_size >= next_io_check_size {
             // TODO(@hhwyt): Consider incorporating snapshot file write I/O into the
             // limiting mechanism.
             handle_read_io_usage(&mut prev_io_bytes, &mut remained_quota);
-            next_io_check_size = stats.total_size + SCAN_BYTES_PER_IO_LIMIT_CHECK;
+            next_io_check_size = stats.total_kv_size + SCAN_BYTES_PER_IO_LIMIT_CHECK;
         }
 
         if let Err(e) = sst_writer.borrow_mut().put(key, value) {
@@ -270,14 +288,16 @@ where
     handle_read_io_usage(&mut prev_io_bytes, &mut remained_quota);
 
     if stats.key_count > 0 {
-        box_try!(finish_sst_writer(sst_writer.into_inner(), path, key_mgr));
+        stats.total_sst_size +=
+            box_try!(finish_sst_writer(sst_writer.into_inner(), path, key_mgr)) as usize;
         cf_file.add_file(file_id);
         info!(
-            "build_sst_cf_file_list builds {} files in cf {}. Total keys {}, total size {}. raw_size_per_file {}, total takes {:?}",
+            "build_sst_cf_file_list builds {} files in cf {}. Total keys {}, total kv size {}, total sst size {}. raw_size_per_file {}, total takes {:?}",
             file_id + 1,
             cf,
             stats.key_count,
-            stats.total_size,
+            stats.total_kv_size,
+            stats.total_sst_size,
             raw_size_per_file,
             instant.saturating_elapsed(),
         );
@@ -695,7 +715,7 @@ mod tests {
             true,
         )
         .unwrap();
-        assert_eq!(stats.total_size, 11890);
+        assert_eq!(stats.total_kv_size, 11890);
         assert!(start.saturating_elapsed_secs() > 1_f64);
     }
 }
diff --git a/metrics/grafana/tikv_details.dashboard.py b/metrics/grafana/tikv_details.dashboard.py
index 497e97fe9ba..b8475b92228 100644
--- a/metrics/grafana/tikv_details.dashboard.py
+++ b/metrics/grafana/tikv_details.dashboard.py
@@ -4090,6 +4090,15 @@ def Snapshot() -> RowPanel:
                         legend_format="clean-region-by-{{type}}",
                         additional_groupby=True,
                     ),
+                    target(
+                        expr=expr_sum_delta(
+                            "tikv_server_snapshot_task_total",
+                            range_selector="1m",
+                            by_labels=["type"],
+                        ),
+                        legend_format="{{type}}",
+                        additional_groupby=True,
+                    ),
                 ],
             ),
             graph_panel(
@@ -4101,13 +4110,14 @@ def Snapshot() -> RowPanel:
                         expr=expr_sum_rate(
                             "tikv_snapshot_limit_transport_bytes",
                             by_labels=["instance", "type"],
-                        ),
+                        )
                     ),
                     target(
                         expr=expr_sum_rate(
                             "tikv_snapshot_limit_generate_bytes",
+                            by_labels=["instance", "type"],
                         ),
-                        legend_format="{{instance}}-generate",
+                        legend_format="{{instance}}-generate-{{type}}",
                     ),
                 ],
             ),
diff --git a/metrics/grafana/tikv_details.json b/metrics/grafana/tikv_details.json
index 7dd84579fca..3c5da8cb2d0 100644
--- a/metrics/grafana/tikv_details.json
+++ b/metrics/grafana/tikv_details.json
@@ -58908,6 +58908,21 @@
               "refId": "",
               "step": 10,
               "target": ""
+            },
+            {
+              "datasource": "${DS_TEST-CLUSTER}",
+              "expr": "sum(delta(\n    tikv_server_snapshot_task_total\n    {k8s_cluster=\"$k8s_cluster\",tidb_cluster=\"$tidb_cluster\",instance=~\"$instance\"}\n    [1m]\n)) by (type, $additional_groupby) ",
+              "format": "time_series",
+              "hide": false,
+              "instant": false,
+              "interval": "",
+              "intervalFactor": 1,
+              "legendFormat": "{{type}} {{$additional_groupby}}",
+              "metric": "",
+              "query": "sum(delta(\n    tikv_server_snapshot_task_total\n    {k8s_cluster=\"$k8s_cluster\",tidb_cluster=\"$tidb_cluster\",instance=~\"$instance\"}\n    [1m]\n)) by (type, $additional_groupby) ",
+              "refId": "",
+              "step": 10,
+              "target": ""
             }
           ],
           "thresholds": [],
@@ -59044,15 +59059,15 @@
             },
             {
               "datasource": "${DS_TEST-CLUSTER}",
-              "expr": "sum(rate(\n    tikv_snapshot_limit_generate_bytes\n    {k8s_cluster=\"$k8s_cluster\",tidb_cluster=\"$tidb_cluster\",instance=~\"$instance\"}\n    [$__rate_interval]\n)) by (instance) ",
+              "expr": "sum(rate(\n    tikv_snapshot_limit_generate_bytes\n    {k8s_cluster=\"$k8s_cluster\",tidb_cluster=\"$tidb_cluster\",instance=~\"$instance\"}\n    [$__rate_interval]\n)) by (instance, type) ",
               "format": "time_series",
               "hide": false,
               "instant": false,
               "interval": "",
               "intervalFactor": 1,
-              "legendFormat": "{{instance}}-generate",
+              "legendFormat": "{{instance}}-generate-{{type}}",
               "metric": "",
-              "query": "sum(rate(\n    tikv_snapshot_limit_generate_bytes\n    {k8s_cluster=\"$k8s_cluster\",tidb_cluster=\"$tidb_cluster\",instance=~\"$instance\"}\n    [$__rate_interval]\n)) by (instance) ",
+              "query": "sum(rate(\n    tikv_snapshot_limit_generate_bytes\n    {k8s_cluster=\"$k8s_cluster\",tidb_cluster=\"$tidb_cluster\",instance=~\"$instance\"}\n    [$__rate_interval]\n)) by (instance, type) ",
               "refId": "",
               "step": 10,
               "target": ""
diff --git a/metrics/grafana/tikv_details.json.sha256 b/metrics/grafana/tikv_details.json.sha256
index 352119f2dfd..317e72944b6 100644
--- a/metrics/grafana/tikv_details.json.sha256
+++ b/metrics/grafana/tikv_details.json.sha256
@@ -1,1 +1,1 @@
-428ee69bad88bb97968f235beded66f7b14ae90758e57c67384f831ee28879b4  ./metrics/grafana/tikv_details.json
+4725fc14b18b65f93f28dce4428f83419708f14fe51a8788d5618fc52797ebfd  ./metrics/grafana/tikv_details.json


Test patch: