refine br-stream to backup-stream

Cargo.toml

@@ -256,7 +256,7 @@ members = [
   "components/error_code",
   "components/concurrency_manager",
   "components/server",
-  "components/br-stream",
+  "components/backup-stream",
   "components/file_system",
   "components/collections",
   "components/coprocessor_plugin_api",

components/br-stream/Cargo.toml → components/backup-stream/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "br-stream"
+name = "backup-stream"
 version = "0.1.0"
 edition = "2018"
 

components/br-stream/bin/playground.rs → components/backup-stream/bin/playground.rs

@@ -1,5 +1,5 @@
 // Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0.
-use br_stream::{
+use backup_stream::{
     errors::Result,
     metadata::{store::EtcdStore, MetadataClient},
 };

components/br-stream/src/endpoint.rs → components/backup-stream/src/endpoint.rs

@@ -81,7 +81,7 @@ where
         router: RT,
         pd_client: Arc<PDC>,
     ) -> Endpoint<EtcdStore, R, E, RT, PDC> {
-        let pool = create_tokio_runtime(config.num_threads, "br-stream")
+        let pool = create_tokio_runtime(config.num_threads, "backup-stream")
             .expect("failed to create tokio runtime for backup stream worker.");
 
         // TODO consider TLS?
@@ -97,7 +97,7 @@ where
         };
 
         let range_router = Router::new(
-            PathBuf::from(config.streaming_path.clone()),
+            PathBuf::from(config.temp_path.clone()),
             scheduler.clone(),
             config.temp_file_size_limit_per_task.0,
         );
@@ -116,7 +116,7 @@ where
             pool.spawn(Self::starts_flush_ticks(range_router.clone()));
         }
 
-        info!("the endpoint of stream backup started"; "path" => %config.streaming_path);
+        info!("the endpoint of stream backup started"; "path" => %config.temp_path);
         Endpoint {
             config,
             meta_client,
@@ -395,7 +395,7 @@ where
                 }
                 if let Err(err) = pd_cli
                     .update_service_safe_point(
-                        format!("br-stream-{}-{}", task, store_id),
+                        format!("backup-stream-{}-{}", task, store_id),
                         TimeStamp::new(rts),
                         Duration::from_secs(600),
                     )
@@ -479,7 +479,7 @@ where
     /// Modify observe over some region.
     /// This would register the region to the RaftStore.
     pub fn on_modify_observe(&self, op: ObserveOp) {
-        info!("br-stream: on_modify_observe"; "op" => ?op);
+        info!("backup stream: on_modify_observe"; "op" => ?op);
         match op {
             ObserveOp::Start {
                 region,

components/br-stream/src/router.rs → components/backup-stream/src/router.rs

@@ -28,7 +28,7 @@ use engine_traits::{CfName, CF_DEFAULT, CF_LOCK, CF_WRITE};
 use external_storage::{BackendConfig, UnpinReader};
 use external_storage_export::{create_storage, ExternalStorage};
 
-use futures::io::{AllowStdIo, Cursor};
+use futures::io::Cursor;
 use kvproto::{
     brpb::{DataFileInfo, FileType, Metadata},
     raft_cmdpb::CmdType,

components/server/Cargo.toml

@@ -28,7 +28,7 @@ nortcheck = ["engine_rocks/nortcheck"]
 
 [dependencies]
 backup = { path = "../backup", default-features = false }
-br-stream = { path = "../br-stream", default-features = false }
+backup-stream = { path = "../backup-stream", default-features = false }
 cdc = { path = "../cdc", default-features = false }
 chrono = "0.4"
 tempfile = "3.0"

components/server/src/server.rs

@@ -25,8 +25,8 @@ use std::{
     u64,
 };
 
-use br_stream::config::BackupStreamConfigManager;
-use br_stream::observer::BackupStreamObserver;
+use backup_stream::config::BackupStreamConfigManager;
+use backup_stream::observer::BackupStreamObserver;
 use cdc::{CdcConfigManager, MemoryQuota};
 use concurrency_manager::ConcurrencyManager;
 use encryption_export::{data_key_manager_from_config, DataKeyManager};
@@ -797,9 +797,9 @@ impl<ER: RaftEngine> TiKVServer<ER> {
         );
 
         // Start backup stream
-        if self.config.backup_stream.enable_streaming {
+        if self.config.backup_stream.enable {
             // Create backup stream.
-            let mut backup_stream_worker = Box::new(LazyWorker::new("br-stream"));
+            let mut backup_stream_worker = Box::new(LazyWorker::new("backup-stream"));
             let backup_stream_scheduler = backup_stream_worker.scheduler();
 
             // Register backup-stream observer.
@@ -811,7 +811,7 @@ impl<ER: RaftEngine> TiKVServer<ER> {
                 Box::new(BackupStreamConfigManager(backup_stream_worker.scheduler())),
             );
 
-            let backup_stream_endpoint = br_stream::Endpoint::new::<String>(
+            let backup_stream_endpoint = backup_stream::Endpoint::new::<String>(
                 node.id(),
                 &self.config.pd.endpoints,
                 self.config.backup_stream.clone(),

src/config.rs

@@ -2280,8 +2280,9 @@ impl Default for BackupConfig {
 #[serde(rename_all = "kebab-case")]
 pub struct BackupStreamConfig {
     pub num_threads: usize,
-    pub enable_streaming: bool,
-    pub streaming_path: String,
+    #[online_config(skip)]
+    pub enable: bool,
+    pub temp_path: String,
     pub temp_file_size_limit_per_task: ReadableSize,
 }
 
@@ -2300,9 +2301,9 @@ impl Default for BackupStreamConfig {
         Self {
             // use at most 50% of vCPU by default
             num_threads: (cpu_num * 0.5).clamp(1.0, 8.0) as usize,
-            enable_streaming: false,
+            enable: false,
             // TODO: may be use raft store directory
-            streaming_path: env::temp_dir().into_os_string().into_string().unwrap(),
+            temp_path: env::temp_dir().into_os_string().into_string().unwrap(),
             temp_file_size_limit_per_task: ReadableSize::mb(128),
         }
     }

src/import/sst_service.rs

@@ -522,7 +522,7 @@ where
                             let mut import_err = kvproto::import_sstpb::Error::default();
                             let err = r.response.get_header().get_error();
                             import_err
-                                .set_message(format!("failed to complete raft command"));
+                                .set_message("failed to complete raft command".to_string());
                             // FIXME: if there are many errors, we may lose some of them here.
                             import_err 
                                 .set_store_error(err.clone());