Fix error parse of ProxyConfig (#144)

components/proxy_server/Cargo.toml

@@ -80,6 +80,7 @@ serde = "1.0"
 serde_derive = "1.0"
 serde_ignored = "0.1"
 serde_json = "1.0"
+serde_with = "1.4"
 server = { path = "../server", default-features = false }
 slog = { version = "2.3", features = ["max_level_trace", "release_max_level_debug"] }
 slog-global = { version = "0.1", git = "https://github.com/breeswish/slog-global.git", rev = "d592f88e4dbba5eb439998463054f1a44fbf17b9" }

components/proxy_server/src/config.rs

@@ -9,21 +9,60 @@ use std::{
 use itertools::Itertools;
 use online_config::OnlineConfig;
 use serde_derive::{Deserialize, Serialize};
+use serde_with::with_prefix;
 use tikv::config::TiKvConfig;
 use tikv_util::crit;
 
 use crate::fatal;
 
+with_prefix!(prefix_apply "apply-");
+with_prefix!(prefix_store "store-");
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, OnlineConfig)]
 #[serde(default)]
 #[serde(rename_all = "kebab-case")]
-pub struct ProxyConfig {
+pub struct RaftstoreConfig {
     pub snap_handle_pool_size: usize,
+}
+
+impl Default for RaftstoreConfig {
+    fn default() -> Self {
+        RaftstoreConfig {
+            snap_handle_pool_size: 2,
+        }
+    }
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, OnlineConfig)]
+#[serde(default)]
+#[serde(rename_all = "kebab-case")]
+pub struct ServerConfig {
     pub engine_addr: String,
     pub engine_store_version: String,
     pub engine_store_git_hash: String,
 }
 
+impl Default for ServerConfig {
+    fn default() -> Self {
+        ServerConfig {
+            engine_addr: DEFAULT_ENGINE_ADDR.to_string(),
+            engine_store_version: String::default(),
+            engine_store_git_hash: String::default(),
+        }
+    }
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, OnlineConfig)]
+#[serde(default)]
+#[serde(rename_all = "kebab-case")]
+pub struct ProxyConfig {
+    #[online_config(submodule)]
+    pub server: ServerConfig,
+
+    #[online_config(submodule)]
+    #[serde(rename = "raftstore")]
+    pub raft_store: RaftstoreConfig,
+}
+
 pub const DEFAULT_ENGINE_ADDR: &str = if cfg!(feature = "failpoints") {
     "127.0.0.1:20206"
 } else {
@@ -33,10 +72,8 @@ pub const DEFAULT_ENGINE_ADDR: &str = if cfg!(feature = "failpoints") {
 impl Default for ProxyConfig {
     fn default() -> Self {
         ProxyConfig {
-            snap_handle_pool_size: 2,
-            engine_addr: DEFAULT_ENGINE_ADDR.to_string(),
-            engine_store_version: String::default(),
-            engine_store_git_hash: String::default(),
+            raft_store: RaftstoreConfig::default(),
+            server: ServerConfig::default(),
         }
     }
 }

components/proxy_server/src/proxy.rs

@@ -292,6 +292,7 @@ pub unsafe fn run_proxy(
                 })
             });
     check_engine_label(&matches);
+    // Replace config from `match` from TiFlash's side.
     overwrite_config_with_cmd_args(&mut config, &mut proxy_config, &matches);
     config.logger_compatible_adjust();
 

components/proxy_server/src/run.rs

@@ -522,6 +522,7 @@ impl<ER: RaftEngine> TiKvServer<ER> {
 
         // Initialize and check config
         let cfg_controller = Self::init_config(config);
+        info!("using proxy config"; "config" => ?proxy_config);
         let config = cfg_controller.get_current();
 
         let store_path = Path::new(&config.storage.data_dir).to_owned();
@@ -1081,11 +1082,11 @@ impl<ER: RaftEngine> TiKvServer<ER> {
         let health_service = HealthService::default();
         let mut default_store = kvproto::metapb::Store::default();
 
-        if !self.proxy_config.engine_store_version.is_empty() {
-            default_store.set_version(self.proxy_config.engine_store_version.clone());
+        if !self.proxy_config.server.engine_store_version.is_empty() {
+            default_store.set_version(self.proxy_config.server.engine_store_version.clone());
         }
-        if !self.proxy_config.engine_store_git_hash.is_empty() {
-            default_store.set_git_hash(self.proxy_config.engine_store_git_hash.clone());
+        if !self.proxy_config.server.engine_store_git_hash.is_empty() {
+            default_store.set_git_hash(self.proxy_config.server.engine_store_git_hash.clone());
         }
         // addr -> store.peer_address
         if self.config.server.advertise_addr.is_empty() {
@@ -1094,8 +1095,8 @@ impl<ER: RaftEngine> TiKvServer<ER> {
             default_store.set_peer_address(self.config.server.advertise_addr.clone())
         }
         // engine_addr -> store.addr
-        if !self.proxy_config.engine_addr.is_empty() {
-            default_store.set_address(self.proxy_config.engine_addr.clone());
+        if !self.proxy_config.server.engine_addr.is_empty() {
+            default_store.set_address(self.proxy_config.server.engine_addr.clone());
         } else {
             panic!("engine address is empty");
         }
@@ -1213,7 +1214,7 @@ impl<ER: RaftEngine> TiKvServer<ER> {
             node.id(),
             self.engines.as_ref().unwrap().engines.kv.clone(),
             importer.clone(),
-            self.proxy_config.snap_handle_pool_size,
+            self.proxy_config.raft_store.snap_handle_pool_size,
         );
         tiflash_ob.register_to(self.coprocessor_host.as_mut().unwrap());
 

components/proxy_server/src/setup.rs

@@ -52,21 +52,21 @@ pub fn overwrite_config_with_cmd_args(
     }
 
     if let Some(engine_store_version) = matches.value_of("engine-version") {
-        proxy_config.engine_store_version = engine_store_version.to_owned();
+        proxy_config.server.engine_store_version = engine_store_version.to_owned();
     }
 
     if let Some(engine_store_git_hash) = matches.value_of("engine-git-hash") {
-        proxy_config.engine_store_git_hash = engine_store_git_hash.to_owned();
+        proxy_config.server.engine_store_git_hash = engine_store_git_hash.to_owned();
     }
 
-    if proxy_config.engine_addr.is_empty() {
+    if proxy_config.server.engine_addr.is_empty() {
         if let Some(engine_addr) = matches.value_of("engine-addr") {
-            proxy_config.engine_addr = engine_addr.to_owned();
+            proxy_config.server.engine_addr = engine_addr.to_owned();
         }
     }
 
     if let Some(engine_addr) = matches.value_of("advertise-engine-addr") {
-        proxy_config.engine_addr = engine_addr.to_owned();
+        proxy_config.server.engine_addr = engine_addr.to_owned();
     }
 
     if let Some(data_dir) = matches.value_of("data-dir") {

new-mock-engine-store/src/mock_cluster.rs

@@ -283,7 +283,7 @@ impl<T: Simulator<TiFlashEngine>> Cluster<T> {
 
         engines.kv.init(
             helper_ptr,
-            self.cfg.proxy_cfg.snap_handle_pool_size,
+            self.cfg.proxy_cfg.raft_store.snap_handle_pool_size,
             Some(ffi_hub),
         );
 

new-mock-engine-store/src/node.rs

@@ -300,7 +300,7 @@ impl Simulator<TiFlashEngine> for NodeCluster {
             node_id,
             engines.kv.clone(),
             importer.clone(),
-            cfg.proxy_cfg.snap_handle_pool_size,
+            cfg.proxy_cfg.raft_store.snap_handle_pool_size,
         );
         tiflash_ob.register_to(&mut coprocessor_host);
 

tests/proxy/normal.rs

@@ -61,7 +61,7 @@ use crate::proxy::*;
 #[test]
 fn test_config() {
     let mut file = tempfile::NamedTempFile::new().unwrap();
-    let text = "memory-usage-high-water=0.65\nsnap-handle-pool-size=4\n[nosense]\nfoo=2\n[rocksdb]\nmax-open-files = 111\nz=1";
+    let text = "memory-usage-high-water=0.65\n[server]\nengine-addr=\"1.2.3.4:5\"\n[raftstore]\nsnap-handle-pool-size=4\n[nosense]\nfoo=2\n[rocksdb]\nmax-open-files = 111\nz=1";
     write!(file, "{}", text).unwrap();
     let path = file.path();
 
@@ -76,7 +76,11 @@ fn test_config() {
 
     let mut proxy_unrecognized_keys = Vec::new();
     let proxy_config = ProxyConfig::from_file(path, Some(&mut proxy_unrecognized_keys)).unwrap();
-    assert_eq!(proxy_config.snap_handle_pool_size, 4);
+    assert_eq!(proxy_config.raft_store.snap_handle_pool_size, 4);
+    assert_eq!(proxy_config.server.engine_addr, "1.2.3.4:5");
+    assert!(proxy_unrecognized_keys.contains(&"rocksdb".to_string()));
+    assert!(proxy_unrecognized_keys.contains(&"memory-usage-high-water".to_string()));
+    assert!(proxy_unrecognized_keys.contains(&"nosense".to_string()));
     let v1 = vec!["a.b", "b"]
         .iter()
         .map(|e| String::from(*e))
@@ -97,16 +101,16 @@ fn test_config() {
 
     // Will not override ProxyConfig
     let proxy_config_new = ProxyConfig::from_file(path, None).unwrap();
-    assert_eq!(proxy_config_new.snap_handle_pool_size, 4);
+    assert_eq!(proxy_config_new.raft_store.snap_handle_pool_size, 4);
 }
 
 #[test]
-fn test_config_addr() {
+fn test_config_default_addr() {
     let mut file = tempfile::NamedTempFile::new().unwrap();
     let text = "memory-usage-high-water=0.65\nsnap-handle-pool-size=4\n[nosense]\nfoo=2\n[rocksdb]\nmax-open-files = 111\nz=1";
     write!(file, "{}", text).unwrap();
     let path = file.path();
-    let mut args: Vec<&str> = vec![];
+    let args: Vec<&str> = vec![];
     let matches = App::new("RaftStore Proxy")
         .arg(
             Arg::with_name("config")