Merge TiKV release-8.1

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tikv"
-version = "8.1.1"
+version = "8.1.2"
 authors = ["The TiKV Authors"]
 description = "A distributed transactional key-value database powered by Rust and Raft"
 license = "Apache-2.0"

components/encryption/src/master_key/kms.rs

@@ -141,7 +141,11 @@ impl KmsBackend {
                             self.kms_provider.decrypt_data_key(&ciphertext_key),
                         )
                     }))
-                    .map_err(cloud_convert_error("decrypt encrypted key failed".into()))?;
+                    .map_err(|e| {
+                        Error::WrongMasterKey(box_err!(cloud_convert_error(
+                            "decrypt encrypted key failed".into(),
+                        )(e)))
+                    })?;
                 let data_key = DataKeyPair {
                     encrypted: ciphertext_key,
                     plaintext: PlainKey::new(plaintext, CryptographyType::AesGcm256)
@@ -154,6 +158,12 @@ impl KmsBackend {
             }
         }
     }
+
+    #[cfg(test)]
+    fn clear_state(&mut self) {
+        let mut opt_state = self.state.lock().unwrap();
+        *opt_state = None;
+    }
 }
 
 impl Backend for KmsBackend {
@@ -173,7 +183,10 @@ impl Backend for KmsBackend {
 #[cfg(test)]
 mod fake {
     use async_trait::async_trait;
-    use cloud::{error::Result, kms::KmsProvider};
+    use cloud::{
+        error::{Error as CloudError, KmsError, Result},
+        kms::KmsProvider,
+    };
 
     use super::*;
 
@@ -183,12 +196,14 @@ mod fake {
     #[derive(Debug)]
     pub struct FakeKms {
         plaintext_key: PlainKey,
+        should_decrypt_data_key_fail: bool,
     }
 
     impl FakeKms {
-        pub fn new(plaintext_key: Vec<u8>) -> Self {
+        pub fn new(plaintext_key: Vec<u8>, should_decrypt_data_key_fail: bool) -> Self {
             Self {
                 plaintext_key: PlainKey::new(plaintext_key, CryptographyType::AesGcm256).unwrap(),
+                should_decrypt_data_key_fail,
             }
         }
     }
@@ -204,7 +219,13 @@ mod fake {
         }
 
         async fn decrypt_data_key(&self, _ciphertext: &EncryptedKey) -> Result<Vec<u8>> {
-            Ok(vec![1u8, 32])
+            if self.should_decrypt_data_key_fail {
+                Err(CloudError::KmsError(KmsError::WrongMasterKey(box_err!(
+                    "wrong master key"
+                ))))
+            } else {
+                Ok(vec![1u8, 32])
+            }
         }
 
         fn name(&self) -> &str {
@@ -241,21 +262,36 @@ mod tests {
         assert_eq!(state2.cached(&encrypted2), true);
     }
 
+    const PLAIN_TEXT_HEX: &str = "25431587e9ecffc7c37f8d6d52a9bc3310651d46fb0e3bad2726c8f2db653749";
+    const CIPHER_TEXT_HEX: &str =
+        "84e5f23f95648fa247cb28eef53abec947dbf05ac953734618111583840bd980";
+    const PLAINKEY_HEX: &str = "c3d99825f2181f4808acd2068eac7441a65bd428f14d2aab43fefc0129091139";
+    const IV_HEX: &str = "cafabd9672ca6c79a2fbdc22";
+
+    #[cfg(test)]
+    fn prepare_data_for_encrypt() -> (Iv, Vec<u8>, Vec<u8>, Vec<u8>) {
+        let iv = Vec::from_hex(IV_HEX).unwrap();
+        let iv = Iv::from_slice(iv.as_slice()).unwrap();
+        let pt = Vec::from_hex(PLAIN_TEXT_HEX).unwrap();
+        let plainkey = Vec::from_hex(PLAINKEY_HEX).unwrap();
+        let ct = Vec::from_hex(CIPHER_TEXT_HEX).unwrap();
+        (iv, pt, plainkey, ct)
+    }
+
+    #[cfg(test)]
+    fn prepare_kms_backend(plainkey: Vec<u8>, should_decrypt_data_key_fail: bool) -> KmsBackend {
+        KmsBackend::new(Box::new(FakeKms::new(
+            plainkey,
+            should_decrypt_data_key_fail,
+        )))
+        .unwrap()
+    }
+
     #[test]
     fn test_kms_backend() {
-        // See more http://csrc.nist.gov/groups/STM/cavp/documents/mac/gcmtestvectors.zip
-        let pt = Vec::from_hex("25431587e9ecffc7c37f8d6d52a9bc3310651d46fb0e3bad2726c8f2db653749")
-            .unwrap();
-        let ct = Vec::from_hex("84e5f23f95648fa247cb28eef53abec947dbf05ac953734618111583840bd980")
-            .unwrap();
-        let plainkey =
-            Vec::from_hex("c3d99825f2181f4808acd2068eac7441a65bd428f14d2aab43fefc0129091139")
-                .unwrap();
-
-        let iv = Vec::from_hex("cafabd9672ca6c79a2fbdc22").unwrap();
-
-        let backend = KmsBackend::new(Box::new(FakeKms::new(plainkey))).unwrap();
-        let iv = Iv::from_slice(iv.as_slice()).unwrap();
+        let (iv, pt, plainkey, ct) = prepare_data_for_encrypt();
+        let backend = prepare_kms_backend(plainkey, false);
+
         let encrypted_content = backend.encrypt_content(&pt, iv).unwrap();
         assert_eq!(encrypted_content.get_content(), ct.as_slice());
         let plaintext = backend.decrypt_content(&encrypted_content).unwrap();
@@ -293,4 +329,19 @@ mod tests {
             Error::Other(_)
         );
     }
+
+    #[test]
+    fn test_kms_backend_wrong_key() {
+        let (iv, pt, plainkey, ..) = prepare_data_for_encrypt();
+        let mut backend = prepare_kms_backend(plainkey, true);
+
+        let encrypted_content = backend.encrypt_content(&pt, iv).unwrap();
+        // Clear the cached state to ensure that the subsequent
+        // backend.decrypt_content() invocation bypasses the cache and triggers the
+        // mocked FakeKMS::decrypt_data_key() function.
+        backend.clear_state();
+
+        let err = backend.decrypt_content(&encrypted_content).unwrap_err();
+        assert_matches!(err, Error::WrongMasterKey(_));
+    }
 }

components/engine_rocks/src/properties.rs

@@ -567,6 +567,7 @@ pub fn get_range_stats(
         num_entries,
         num_versions: props.num_versions,
         num_rows: props.num_rows,
+        num_deletes: props.num_deletes,
     })
 }
 

components/engine_traits/src/misc.rs

@@ -57,12 +57,25 @@ pub trait StatisticsReporter<T: ?Sized> {
 
 #[derive(Default)]
 pub struct RangeStats {
-    // The number of entries
+    // The number of entries in write cf.
     pub num_entries: u64,
     // The number of MVCC versions of all rows (num_entries - tombstones).
     pub num_versions: u64,
     // The number of rows.
     pub num_rows: u64,
+    // The number of MVCC deletes of all rows.
+    pub num_deletes: u64,
+}
+
+impl RangeStats {
+    /// The number of redundant keys in the range.
+    /// It's calculated by `num_entries - num_versions + num_deleted`.
+    pub fn redundant_keys(&self) -> u64 {
+        // Consider the number of `mvcc_deletes` as the number of redundant keys.
+        self.num_entries
+            .saturating_sub(self.num_rows)
+            .saturating_add(self.num_deletes)
+    }
 }
 
 pub trait MiscExt: CfNamesExt + FlowControlFactorsExt + WriteBatchExt {

components/raftstore-v2/src/worker/pd/store.rs

@@ -23,6 +23,7 @@ use slog::{error, info, warn};
 use tikv_util::{
     metrics::RecordPairVec,
     store::QueryStats,
+    sys::disk::get_disk_space_stats,
     time::{Duration, Instant as TiInstant, UnixSecs},
     topn::TopN,
 };
@@ -442,7 +443,8 @@ where
 
     /// Returns (capacity, used, available).
     fn collect_engine_size(&self) -> Option<(u64, u64, u64)> {
-        let disk_stats = match fs2::statvfs(self.tablet_registry.tablet_root()) {
+        let (disk_cap, disk_avail) = match get_disk_space_stats(self.tablet_registry.tablet_root())
+        {
             Err(e) => {
                 error!(
                     self.logger,
@@ -452,9 +454,8 @@ where
                 );
                 return None;
             }
-            Ok(stats) => stats,
+            Ok((total_size, available_size)) => (total_size, available_size),
         };
-        let disk_cap = disk_stats.total_space();
         let capacity = if self.cfg.value().capacity.0 == 0 {
             disk_cap
         } else {
@@ -481,7 +482,7 @@ where
         let mut available = capacity.checked_sub(used_size).unwrap_or_default();
         // We only care about rocksdb SST file size, so we should check disk available
         // here.
-        available = cmp::min(available, disk_stats.available_space());
+        available = cmp::min(available, disk_avail);
         Some((capacity, used_size, available))
     }
 }

components/raftstore/src/store/worker/compact.rs

@@ -445,7 +445,7 @@ pub fn need_compact(range_stats: &RangeStats, compact_threshold: &CompactThresho
     // We trigger region compaction when their are to many tombstones as well as
     // redundant keys, both of which can severly impact scan operation:
     let estimate_num_del = range_stats.num_entries - range_stats.num_versions;
-    let redundant_keys = range_stats.num_entries - range_stats.num_rows;
+    let redundant_keys = range_stats.redundant_keys();
     (redundant_keys >= compact_threshold.redundant_rows_threshold
         && redundant_keys * 100
             >= compact_threshold.redundant_rows_percent_threshold * range_stats.num_entries)

components/raftstore/src/store/worker/pd.rs

@@ -44,7 +44,7 @@ use tikv_util::{
     box_err, debug, error, info,
     metrics::ThreadInfoStatistics,
     store::QueryStats,
-    sys::{thread::StdThreadBuildWrapper, SysQuota},
+    sys::{disk::get_disk_space_stats, thread::StdThreadBuildWrapper, SysQuota},
     thd_name,
     time::{Instant as TiInstant, UnixSecs},
     timer::GLOBAL_TIMER_HANDLE,
@@ -2433,7 +2433,7 @@ fn collect_engine_size<EK: KvEngine, ER: RaftEngine>(
         return Some((engine_size.capacity, engine_size.used, engine_size.avail));
     }
     let store_info = store_info.unwrap();
-    let disk_stats = match fs2::statvfs(store_info.kv_engine.path()) {
+    let (disk_cap, disk_avail) = match get_disk_space_stats(store_info.kv_engine.path()) {
         Err(e) => {
             error!(
                 "get disk stat for rocksdb failed";
@@ -2442,9 +2442,8 @@ fn collect_engine_size<EK: KvEngine, ER: RaftEngine>(
             );
             return None;
         }
-        Ok(stats) => stats,
+        Ok((total_size, available_size)) => (total_size, available_size),
     };
-    let disk_cap = disk_stats.total_space();
     let capacity = if store_info.capacity == 0 || disk_cap < store_info.capacity {
         disk_cap
     } else {
@@ -2468,7 +2467,7 @@ fn collect_engine_size<EK: KvEngine, ER: RaftEngine>(
     let mut available = capacity.checked_sub(used_size).unwrap_or_default();
     // We only care about rocksdb SST file size, so we should check disk available
     // here.
-    available = cmp::min(available, disk_stats.available_space());
+    available = cmp::min(available, disk_avail);
     Some((capacity, used_size, available))
 }
 

components/server/Cargo.toml

@@ -15,18 +15,10 @@ sse = ["tikv/sse"]
 memory-engine = []
 mem-profiling = ["tikv/mem-profiling"]
 failpoints = ["tikv/failpoints"]
-test-engine-kv-rocksdb = [
-  "tikv/test-engine-kv-rocksdb"
-]
-test-engine-raft-raft-engine = [
-  "tikv/test-engine-raft-raft-engine"
-]
-test-engines-rocksdb = [
-  "tikv/test-engines-rocksdb",
-]
-test-engines-panic = [
-  "tikv/test-engines-panic",
-]
+test-engine-kv-rocksdb = ["tikv/test-engine-kv-rocksdb"]
+test-engine-raft-raft-engine = ["tikv/test-engine-raft-raft-engine"]
+test-engines-rocksdb = ["tikv/test-engines-rocksdb"]
+test-engines-panic = ["tikv/test-engines-panic"]
 nortcheck = ["engine_rocks/nortcheck"]
 backup-stream-debug = ["backup-stream/backup-stream-debug"]
 
@@ -49,7 +41,6 @@ engine_traits = { workspace = true }
 error_code = { workspace = true }
 fail = "0.5"
 file_system = { workspace = true }
-fs2 = "0.4"
 futures = "0.3"
 grpcio = { workspace = true }
 grpcio-health = { workspace = true }
@@ -59,7 +50,10 @@ hybrid_engine = { workspace = true }
 keys = { workspace = true }
 kvproto = { workspace = true }
 libc = "0.2"
-log = { version = "0.4", features = ["max_level_trace", "release_max_level_debug"] }
+log = { version = "0.4", features = [
+  "max_level_trace",
+  "release_max_level_debug",
+] }
 log_wrappers = { workspace = true }
 pd_client = { workspace = true }
 prometheus = { version = "0.13", features = ["nightly"] }