feat: unify config reading behavior

Author: ziqonghuang

Cargo.lock

@@ -118,7 +118,7 @@ slog-stdlog = "4.1.0"
 num-bigint = "0.4.6"
 bigdecimal = "0.4.8"
 bitvec = "1.0.1"
-config = "0.13.4"
+config = "0.15.9"
 tempfile = "3.21.0"
 md-5 = "0.10.6"
 bitflags = "2.10.0"

Cargo.toml

@@ -23,12 +23,12 @@ use curvine_client::unified::UnifiedFileSystem;
 use curvine_common::conf::ClusterConf;
 use curvine_common::version;
 use orpc::common::{Logger, Utils};
-use orpc::io::net::InetAddr;
 use orpc::runtime::RpcRuntime;
-use orpc::{err_box, CommonResult};
+use orpc::CommonResult;
+use serde::Serialize;
 use std::sync::Arc;
 
-#[derive(Parser, Debug)]
+#[derive(Parser, Debug, Serialize)]
 #[command(author, version, about, long_about = None)]
 pub struct CurvineArgs {
     /// Configuration file path (optional)
@@ -48,58 +48,20 @@ pub struct CurvineArgs {
     )]
     pub master_addrs: Option<String>,
 
+    #[serde(skip_serializing)]
     #[command(subcommand)]
     command: Commands,
 }
 
 impl CurvineArgs {
-    /// Get cluster configuration with priority: CLI args > config file > env vars > defaults
     pub fn get_conf(&self) -> CommonResult<ClusterConf> {
-        // Priority 1: Try to load from config file (CLI arg or env var)
+        // Try to load from config file (CLI arg or env var)
         let conf_path = self
             .conf
             .clone()
             .or_else(|| std::env::var(ClusterConf::ENV_CONF_FILE).ok());
 
-        let mut conf = if let Some(path) = conf_path {
-            match ClusterConf::from(&path) {
-                Ok(c) => c,
-                Err(e) => {
-                    eprintln!("Warning: Failed to load config file '{}': {}", path, e);
-                    eprintln!("Using default configuration");
-                    Self::create_default_conf()
-                }
-            }
-        } else {
-            println!("No config file specified, using default configuration");
-            Self::create_default_conf()
-        };
-
-        // Priority 2: Override with CLI master_addrs if provided
-        if let Some(master_addrs) = &self.master_addrs {
-            let mut vec = vec![];
-            for node in master_addrs.split(',') {
-                let tmp: Vec<&str> = node.split(':').collect();
-                if tmp.len() != 2 {
-                    return err_box!("Invalid master_addrs format: '{}'. Expected format: 'host1:port1,host2:port2'", master_addrs);
-                }
-                let hostname = tmp[0].to_string();
-                let port: u16 = tmp[1]
-                    .parse()
-                    .map_err(|_| format!("Invalid port number in master_addrs: '{}'", tmp[1]))?;
-                vec.push(InetAddr::new(hostname, port));
-            }
-            conf.client.master_addrs = vec;
-        }
-
-        // Initialize configuration (parse string values to actual types)
-        conf.client.init()?;
-
-        Ok(conf)
-    }
-
-    fn create_default_conf() -> ClusterConf {
-        ClusterConf::default()
+        ClusterConf::from(conf_path.unwrap(), Some(&self), Some("cli"))
     }
 }
 

curvine-cli/src/main.rs

@@ -40,6 +40,7 @@ slog = { workspace = true }
 slog-stdlog  = { workspace = true }
 chrono = { workspace = true }
 trait-variant = { workspace = true }
+config = { workspace = true }
 
 anyhow = { workspace = true }
 tracing = { workspace = true }

curvine-common/Cargo.toml

@@ -16,17 +16,17 @@ use crate::conf::CliConf;
 use crate::conf::{ClientConf, FuseConf, JobConf, JournalConf, MasterConf, WorkerConf};
 use crate::rocksdb::DBConf;
 use crate::version;
+use config::{Config, Environment, File, FileFormat};
 use log::info;
 use orpc::client::{ClientConf as RpcConf, ClientFactory, SyncClient};
 use orpc::common::{LogConf, Utils};
 use orpc::io::net::{InetAddr, NodeAddr};
 use orpc::io::retry::TimeBondedRetryBuilder;
 use orpc::server::ServerConf;
-use orpc::{err_box, try_err, CommonResult};
+use orpc::{err_box, CommonResult};
 use serde::{Deserialize, Serialize};
-use std::env;
+use serde_json::{json, Value};
 use std::fmt::{Display, Formatter};
-use std::fs::read_to_string;
 use std::time::Duration;
 
 // Cluster configuration files.
@@ -75,37 +75,110 @@ impl ClusterConf {
     pub const ENV_CLIENT_HOSTNAME: &'static str = "CURVINE_CLIENT_HOSTNAME";
     pub const ENV_CONF_FILE: &'static str = "CURVINE_CONF_FILE";
 
-    pub fn from<T: AsRef<str>>(path: T) -> CommonResult<Self> {
-        let str = try_err!(read_to_string(path.as_ref()));
-        let mut conf = try_err!(toml::from_str::<Self>(&str));
-
-        if let Ok(v) = env::var(Self::ENV_MASTER_HOSTNAME) {
-            conf.master.hostname = v.to_owned();
-            conf.journal.hostname = v;
-        }
+    pub fn argstostr<U: Serialize>(args_name: &str, args: &U) -> String {
+        let args_value: Value = serde_json::to_value(args).unwrap();
+
+        let result = if let Some(master_addrs_str) =
+            args_value.get("master_addrs").and_then(|v| v.as_str())
+        {
+            let addrs: Vec<Value> = master_addrs_str
+                .split(',')
+                .filter(|s| !s.trim().is_empty())
+                .map(|addr| {
+                    let addr = addr.trim();
+                    let parts: Vec<&str> = addr.split(':').collect();
+
+                    if parts.len() == 2 {
+                        let hostname = parts[0];
+                        let port = parts[1]
+                            .parse::<u16>()
+                            .map_err(|_| format!("Invalid port: {}", parts[1]))?;
+
+                        Ok(json!({
+                            "hostname": hostname,
+                            "port": port
+                        }))
+                    } else {
+                        Err(format!("Invalid address format: {}", addr))
+                    }
+                })
+                .collect::<Result<Vec<_>, _>>()
+                .unwrap();
+
+            json!({
+                args_name: args_value,
+                "client": {
+                    "master_addrs": addrs
+                }
+            })
+        } else {
+            json!({
+                args_name: args_value
+            })
+        };
 
-        // Apply worker hostname from environment variable (used by worker process)
-        if let Ok(v) = env::var(Self::ENV_WORKER_HOSTNAME) {
-            conf.worker.hostname = v;
-        }
+        serde_json::to_string_pretty(&result).unwrap()
+    }
 
-        // Apply client hostname from environment variable
-        if let Ok(v) = env::var(Self::ENV_CLIENT_HOSTNAME) {
-            conf.client.hostname = v;
+    pub fn from<T: AsRef<str>, U: Serialize>(
+        path: T,
+        args: Option<&U>,
+        args_name: Option<&str>,
+    ) -> CommonResult<Self> {
+        let cli_source = if let (Some(name), Some(args_value)) = (args_name, args) {
+            ClusterConf::argstostr(name, args_value)
+        } else {
+            "{}".to_string()
+        };
+
+        // Get cluster configuration with priority: CLI args > env vars > config file > defaults
+        let config = Config::builder()
+            .add_source(File::new(path.as_ref(), FileFormat::Toml))
+            .add_source(Environment::with_prefix("CURVINE").separator("_"))
+            .add_source(File::from_str(&cli_source, FileFormat::Json));
+
+        let mut conf;
+        match config.build() {
+            Ok(config) => {
+                conf = Self {
+                    format_master: true,
+                    format_worker: true,
+                    testing: false,
+                    cluster_id: "curvine".to_string(),
+                    master: config.get::<MasterConf>("master").unwrap_or_default(),
+                    journal: config.get::<JournalConf>("journal").unwrap_or_default(),
+                    worker: config.get::<WorkerConf>("worker").unwrap_or_default(),
+                    log: config.get::<LogConf>("log").unwrap_or_default(),
+                    client: config.get::<ClientConf>("client").unwrap_or_default(),
+                    fuse: config.get::<FuseConf>("fuse").unwrap_or_default(),
+                    s3_gateway: config
+                        .get::<S3GatewayConf>("s3_gateway")
+                        .unwrap_or_default(),
+                    job: config.get::<JobConf>("job").unwrap_or_default(),
+                    cli: config.get::<CliConf>("cli").unwrap_or_default(),
+                };
+            }
+            Err(e) => {
+                eprintln!(
+                    "Warning: Failed to load config file '{}': {}",
+                    path.as_ref(),
+                    e
+                );
+                eprintln!("Using default configuration");
+                conf = ClusterConf::default();
+            }
         }
 
         conf.master.init()?;
         conf.client.init()?;
         conf.fuse.init()?;
         conf.job.init()?;
-
         if conf.client.master_addrs.is_empty() {
             for peer in &mut conf.journal.journal_addrs {
                 let node = InetAddr::new(&peer.hostname, conf.master.rpc_port);
                 conf.client.master_addrs.push(node);
             }
         }
-
         Ok(conf)
     }
 
@@ -268,7 +341,7 @@ impl ClusterConf {
 
 impl Default for ClusterConf {
     fn default() -> Self {
-        Self {
+        let mut conf = Self {
             format_master: true,
             format_worker: true,
             testing: false,
@@ -282,7 +355,14 @@ impl Default for ClusterConf {
             s3_gateway: Default::default(),
             job: Default::default(),
             cli: Default::default(),
-        }
+        };
+
+        conf.master.init().unwrap();
+        conf.client.init().unwrap();
+        conf.fuse.init().unwrap();
+        conf.job.init().unwrap();
+
+        conf
     }
 }
 

curvine-common/src/conf/cluster_conf.rs

@@ -55,7 +55,7 @@ mod tests {
     #[test]
     fn cluster() {
         let path = "../etc/curvine-cluster.toml";
-        let conf = ClusterConf::from(path).unwrap();
+        let conf = ClusterConf::from(path, None::<&()>, None).unwrap();
         println!("conf = {:#?}", conf)
     }
 

curvine-common/src/conf/mod.rs

@@ -27,6 +27,8 @@ mini-moka = { workspace = true }
 fxhash = { workspace = true }
 dashmap = { workspace = true }
 bitflags = { workspace = true }
+serde = { workspace = true }
+serde_with = { workspace = true }
 axum = { workspace = true }
 tracing = { workspace = true }
 tracing-appender = { workspace = true }