implement stream restore for tikv side. (#12)

* sst-importer: impl apply for br stream restore Signed-off-by: 3pointer <[email protected]> * move br-stream::codec outside for sst-importer Signed-off-by: 3pointer <[email protected]> * perform rewrite Signed-off-by: 3pointer <[email protected]> * make clippy Signed-off-by: 3pointer <[email protected]> * add test Signed-off-by: 3pointer <[email protected]> * address some somments Signed-off-by: 3pointer <[email protected]> * address comment Signed-off-by: 3pointer <[email protected]> * address comment Signed-off-by: 3pointer <[email protected]> * address comment Signed-off-by: 3pointer <[email protected]> * fix valid
3pointer · Feb 10, 2022 · 2c0c168 · 2c0c168
1 parent 6eaa365
commit 2c0c168
Show file tree

Hide file tree

Showing 10 changed files with 432 additions and 118 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/components/br-stream/src/codec.rs b/components/br-stream/src/codec.rs
diff --git a/components/br-stream/src/lib.rs b/components/br-stream/src/lib.rs
@@ -1,6 +1,5 @@
 // Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0.
 #![feature(inherent_ascii_escape)]
-mod codec;
 pub mod config;
 mod endpoint;
 pub mod errors;

diff --git a/components/br-stream/src/router.rs b/components/br-stream/src/router.rs
@@ -13,7 +13,6 @@ use std::{
 
 use crate::{
     annotate,
-    codec::Encoder,
     endpoint::Task,
     errors::Error,
     metadata::StreamTask,
@@ -41,7 +40,9 @@ use slog_global::debug;
 use tidb_query_datatype::codec::table::decode_table_id;
 
 use tikv_util::{
-    box_err, defer, error, info,
+    box_err,
+    codec::stream_event::EventEncoder,
+    defer, info, error,
     time::{Instant, Limiter},
     warn,
     worker::Scheduler,
@@ -825,7 +826,7 @@ impl DataFile {
     async fn on_event(&mut self, mut kv: ApplyEvent) -> Result<usize> {
         let now = Instant::now_coarse();
         let _entry_size = kv.size();
-        let encoded = Encoder::encode_event(&kv.key, &kv.value);
+        let encoded = EventEncoder::encode_event(&kv.key, &kv.value);
         let mut size = 0;
         for slice in encoded {
             let slice = slice.as_ref();

diff --git a/components/sst_importer/src/import_file.rs b/components/sst_importer/src/import_file.rs
@@ -219,8 +219,7 @@ impl ImportDir {
         })
     }
 
-    pub fn join(&self, meta: &SstMeta) -> Result<ImportPath> {
-        let file_name = sst_meta_to_path(meta)?;
+    pub fn get_import_path(&self, file_name: &str) -> Result<ImportPath> {
         let save_path = self.root_dir.join(&file_name);
         let temp_path = self.temp_dir.join(&file_name);
         let clone_path = self.clone_dir.join(&file_name);
@@ -231,6 +230,11 @@ impl ImportDir {
         })
     }
 
+    pub fn join(&self, meta: &SstMeta) -> Result<ImportPath> {
+        let file_name = sst_meta_to_path(meta)?;
+        self.get_import_path(file_name.to_str().unwrap())
+    }
+
     pub fn create(
         &self,
         meta: &SstMeta,

diff --git a/components/sst_importer/src/metrics.rs b/components/sst_importer/src/metrics.rs
@@ -83,4 +83,12 @@ lazy_static! {
         &["type", "error"]
     )
     .unwrap();
+    pub static ref IMPORTER_APPLY_DURATION: HistogramVec = register_histogram_vec!(
+        "tikv_import_apply_duration",
+        "Bucketed histogram of importer apply duration",
+        &["type"],
+        // Start from 10ms.
+        exponential_buckets(0.01, 2.0, 20).unwrap()
+    )
+    .unwrap();
 }