diff --git a/Cargo.lock b/Cargo.lock index fe0b0581..78ff9201 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2066,6 +2066,7 @@ dependencies = [ "once_cell", "serde", "serde_json", + "uuid", ] [[package]] @@ -2094,6 +2095,7 @@ dependencies = [ "tauri-plugin-vpnservice", "thunk-rs", "tokio", + "uuid", ] [[package]] @@ -9230,21 +9232,23 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" [[package]] name = "uuid" -version = "1.10.0" +version = "1.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "81dfa00651efa65069b0b6b651f4aaa31ba9e3c3ce0137aaad053604ee7e0314" +checksum = "3cf4199d1e5d15ddd86a694e4d0dffa9c323ce759fea589f00fef9d81cc1931d" dependencies = [ - "getrandom 0.2.15", - "rand 0.8.5", + "getrandom 0.3.2", + "js-sys", + "rand 0.9.1", "serde", "uuid-macro-internal", + "wasm-bindgen", ] [[package]] name = "uuid-macro-internal" -version = "1.10.0" +version = "1.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ee1cd046f83ea2c4e920d6ee9f7c3537ef928d75dce5d84a87c2c5d6b3999a3a" +checksum = "26b682e8c381995ea03130e381928e0e005b7c9eb483c6c8682f50e07b33c2b7" dependencies = [ "proc-macro2", "quote", diff --git a/easytier-contrib/easytier-ffi/Cargo.toml b/easytier-contrib/easytier-ffi/Cargo.toml index d86091ce..fff8262a 100644 --- a/easytier-contrib/easytier-ffi/Cargo.toml +++ b/easytier-contrib/easytier-ffi/Cargo.toml @@ -14,3 +14,4 @@ dashmap = "6.0" serde = { version = "1.0", features = ["derive"] } serde_json = "1" +uuid = "1.17.0" diff --git a/easytier-contrib/easytier-ffi/src/lib.rs b/easytier-contrib/easytier-ffi/src/lib.rs index ab8a7b1f..4e2f66ca 100644 --- a/easytier-contrib/easytier-ffi/src/lib.rs +++ b/easytier-contrib/easytier-ffi/src/lib.rs @@ -3,11 +3,14 @@ use std::sync::Mutex; use dashmap::DashMap; use easytier::{ common::config::{ConfigLoader as _, TomlConfigLoader}, - launcher::NetworkInstance, + instance_manager::NetworkInstanceManager, + launcher::ConfigSource, }; -static INSTANCE_MAP: once_cell::sync::Lazy> = +static INSTANCE_NAME_ID_MAP: once_cell::sync::Lazy> = once_cell::sync::Lazy::new(DashMap::new); +static INSTANCE_MANAGER: once_cell::sync::Lazy = + once_cell::sync::Lazy::new(NetworkInstanceManager::new); static ERROR_MSG: once_cell::sync::Lazy>> = once_cell::sync::Lazy::new(|| Mutex::new(Vec::new())); @@ -86,18 +89,20 @@ pub extern "C" fn run_network_instance(cfg_str: *const std::ffi::c_char) -> std: let inst_name = cfg.get_inst_name(); - if INSTANCE_MAP.contains_key(&inst_name) { + if INSTANCE_NAME_ID_MAP.contains_key(&inst_name) { set_error_msg("instance already exists"); return -1; } - let mut instance = NetworkInstance::new(cfg); - if let Err(e) = instance.start().map_err(|e| e.to_string()) { - set_error_msg(&format!("failed to start instance: {}", e)); - return -1; - } + let instance_id = match INSTANCE_MANAGER.run_network_instance(cfg, ConfigSource::FFI) { + Ok(id) => id, + Err(e) => { + set_error_msg(&format!("failed to start instance: {}", e)); + return -1; + } + }; - INSTANCE_MAP.insert(inst_name, instance); + INSTANCE_NAME_ID_MAP.insert(inst_name, instance_id); 0 } @@ -108,7 +113,11 @@ pub extern "C" fn retain_network_instance( length: usize, ) -> std::ffi::c_int { if length == 0 { - INSTANCE_MAP.clear(); + if let Err(e) = INSTANCE_MANAGER.retain_network_instance(Vec::new()) { + set_error_msg(&format!("failed to retain instances: {}", e)); + return -1; + } + INSTANCE_NAME_ID_MAP.clear(); return 0; } @@ -125,7 +134,17 @@ pub extern "C" fn retain_network_instance( .collect::>() }; - let _ = INSTANCE_MAP.retain(|k, _| inst_names.contains(k)); + let inst_ids: Vec = inst_names + .iter() + .filter_map(|name| INSTANCE_NAME_ID_MAP.get(name).map(|id| *id)) + .collect(); + + if let Err(e) = INSTANCE_MANAGER.retain_network_instance(inst_ids) { + set_error_msg(&format!("failed to retain instances: {}", e)); + return -1; + } + + let _ = INSTANCE_NAME_ID_MAP.retain(|k, _| inst_names.contains(k)); 0 } @@ -144,13 +163,20 @@ pub extern "C" fn collect_network_infos( std::slice::from_raw_parts_mut(infos, max_length) }; + let collected_infos = match INSTANCE_MANAGER.collect_network_infos() { + Ok(infos) => infos, + Err(e) => { + set_error_msg(&format!("failed to collect network infos: {}", e)); + return -1; + } + }; + let mut index = 0; - for instance in INSTANCE_MAP.iter() { + for (instance_id, value) in collected_infos.iter() { if index >= max_length { break; } - let key = instance.key(); - let Some(value) = instance.get_running_info() else { + let Some(key) = INSTANCE_MANAGER.get_network_instance_name(instance_id) else { continue; }; // convert value to json string @@ -181,7 +207,6 @@ mod tests { let cfg_str = r#" inst_name = "test" network = "test_network" - fdsafdsa "#; let cstr = std::ffi::CString::new(cfg_str).unwrap(); assert_eq!(parse_config(cstr.as_ptr()), 0); diff --git a/easytier-gui/src-tauri/Cargo.toml b/easytier-gui/src-tauri/Cargo.toml index e23ae8f9..97dfab6f 100644 --- a/easytier-gui/src-tauri/Cargo.toml +++ b/easytier-gui/src-tauri/Cargo.toml @@ -53,6 +53,7 @@ tauri-plugin-positioner = { version = "2.0", features = ["tray-icon"] } tauri-plugin-vpnservice = { path = "../../tauri-plugin-vpnservice" } tauri-plugin-os = "2.0" tauri-plugin-autostart = "2.0" +uuid = "1.17.0" [features] diff --git a/easytier-gui/src-tauri/src/lib.rs b/easytier-gui/src-tauri/src/lib.rs index 517a2716..d42faebe 100644 --- a/easytier-gui/src-tauri/src/lib.rs +++ b/easytier-gui/src-tauri/src/lib.rs @@ -3,10 +3,12 @@ use std::collections::BTreeMap; -use dashmap::DashMap; use easytier::{ - common::config::{ConfigLoader, FileLoggerConfig, TomlConfigLoader}, - launcher::{NetworkConfig, NetworkInstance, NetworkInstanceRunningInfo}, + common::config::{ + ConfigLoader, FileLoggerConfig, LoggingConfigBuilder, + }, + launcher::{ConfigSource, NetworkConfig, NetworkInstanceRunningInfo}, + instance_manager::NetworkInstanceManager, utils::{self, NewFilterSender}, }; @@ -17,8 +19,8 @@ pub const AUTOSTART_ARG: &str = "--autostart"; #[cfg(not(target_os = "android"))] use tauri::tray::{MouseButton, MouseButtonState, TrayIconBuilder, TrayIconEvent}; -static INSTANCE_MAP: once_cell::sync::Lazy> = - once_cell::sync::Lazy::new(DashMap::new); +static INSTANCE_MANAGER: once_cell::sync::Lazy = + once_cell::sync::Lazy::new(NetworkInstanceManager::new); static mut LOGGER_LEVEL_SENDER: once_cell::sync::Lazy> = once_cell::sync::Lazy::new(Default::default); @@ -44,41 +46,39 @@ fn parse_network_config(cfg: NetworkConfig) -> Result { #[tauri::command] fn run_network_instance(cfg: NetworkConfig) -> Result<(), String> { - if INSTANCE_MAP.contains_key(cfg.instance_id()) { - return Err("instance already exists".to_string()); - } let instance_id = cfg.instance_id().to_string(); - let cfg = cfg.gen_config().map_err(|e| e.to_string())?; - let mut instance = NetworkInstance::new(cfg); - instance.start().map_err(|e| e.to_string())?; - + INSTANCE_MANAGER + .run_network_instance(cfg, ConfigSource::GUI) + .map_err(|e| e.to_string())?; println!("instance {} started", instance_id); - INSTANCE_MAP.insert(instance_id, instance); Ok(()) } #[tauri::command] fn retain_network_instance(instance_ids: Vec) -> Result<(), String> { - let _ = INSTANCE_MAP.retain(|k, _| instance_ids.contains(k)); - println!( - "instance {:?} retained", - INSTANCE_MAP - .iter() - .map(|item| item.key().clone()) - .collect::>() - ); + let instance_ids = instance_ids + .into_iter() + .filter_map(|id| uuid::Uuid::parse_str(&id).ok()) + .collect(); + let retained = INSTANCE_MANAGER + .retain_network_instance(instance_ids) + .map_err(|e| e.to_string())?; + println!("instance {:?} retained", retained); Ok(()) } #[tauri::command] fn collect_network_infos() -> Result, String> { + let infos = INSTANCE_MANAGER + .collect_network_infos() + .map_err(|e| e.to_string())?; + let mut ret = BTreeMap::new(); - for instance in INSTANCE_MAP.iter() { - if let Some(info) = instance.get_running_info() { - ret.insert(instance.key().clone(), info); - } + for (uuid, info) in infos { + ret.insert(uuid.to_string(), info); } + Ok(ret) } @@ -97,10 +97,10 @@ fn set_logging_level(level: String) -> Result<(), String> { #[tauri::command] fn set_tun_fd(instance_id: String, fd: i32) -> Result<(), String> { - let mut instance = INSTANCE_MAP - .get_mut(&instance_id) - .ok_or("instance not found")?; - instance.set_tun_fd(fd); + let uuid = uuid::Uuid::parse_str(&instance_id).map_err(|e| e.to_string())?; + INSTANCE_MANAGER + .set_tun_fd(&uuid, fd) + .map_err(|e| e.to_string())?; Ok(()) } @@ -185,13 +185,15 @@ pub fn run() { let Ok(log_dir) = app.path().app_log_dir() else { return Ok(()); }; - let config = TomlConfigLoader::default(); - config.set_file_logger_config(FileLoggerConfig { - dir: Some(log_dir.to_string_lossy().to_string()), - level: None, - file: None, - }); - let Ok(Some(logger_reinit)) = utils::init_logger(config, true) else { + let config = LoggingConfigBuilder::default() + .file_logger(FileLoggerConfig { + dir: Some(log_dir.to_string_lossy().to_string()), + level: None, + file: None, + }) + .build() + .map_err(|e| e.to_string())?; + let Ok(Some(logger_reinit)) = utils::init_logger(&config, true) else { return Ok(()); }; #[allow(static_mut_refs)] diff --git a/easytier-web/src/main.rs b/easytier-web/src/main.rs index 3ad85bb6..852bf775 100644 --- a/easytier-web/src/main.rs +++ b/easytier-web/src/main.rs @@ -8,7 +8,7 @@ use std::sync::Arc; use clap::Parser; use easytier::{ common::{ - config::{ConfigLoader, ConsoleLoggerConfig, FileLoggerConfig, TomlConfigLoader}, + config::{ConsoleLoggerConfig, FileLoggerConfig, LoggingConfigLoader}, constants::EASYTIER_VERSION, error::Error, network::{local_ipv4, local_ipv6}, @@ -101,6 +101,22 @@ struct Cli { api_host: Option, } +impl LoggingConfigLoader for &Cli { + fn get_console_logger_config(&self) -> ConsoleLoggerConfig { + ConsoleLoggerConfig { + level: self.console_log_level.clone(), + } + } + + fn get_file_logger_config(&self) -> FileLoggerConfig { + FileLoggerConfig { + dir: self.file_log_dir.clone(), + level: self.file_log_level.clone(), + file: None, + } + } +} + pub fn get_listener_by_url(l: &url::Url) -> Result, Error> { Ok(match l.scheme() { "tcp" => Box::new(TcpTunnelListener::new(l.clone())), @@ -144,16 +160,7 @@ async fn main() { setup_panic_handler(); let cli = Cli::parse(); - let config = TomlConfigLoader::default(); - config.set_console_logger_config(ConsoleLoggerConfig { - level: cli.console_log_level, - }); - config.set_file_logger_config(FileLoggerConfig { - dir: cli.file_log_dir, - level: cli.file_log_level, - file: None, - }); - init_logger(config, false).unwrap(); + init_logger(&cli, false).unwrap(); // let db = db::Db::new(":memory:").await.unwrap(); let db = db::Db::new(cli.db).await.unwrap(); diff --git a/easytier/src/common/config.rs b/easytier/src/common/config.rs index 9798b323..34004309 100644 --- a/easytier/src/common/config.rs +++ b/easytier/src/common/config.rs @@ -71,11 +71,6 @@ pub trait ConfigLoader: Send + Sync { fn get_listener_uris(&self) -> Vec; - fn get_file_logger_config(&self) -> FileLoggerConfig; - fn set_file_logger_config(&self, config: FileLoggerConfig); - fn get_console_logger_config(&self) -> ConsoleLoggerConfig; - fn set_console_logger_config(&self, config: ConsoleLoggerConfig); - fn get_peers(&self) -> Vec; fn set_peers(&self, peers: Vec); @@ -112,6 +107,12 @@ pub trait ConfigLoader: Send + Sync { fn dump(&self) -> String; } +pub trait LoggingConfigLoader { + fn get_file_logger_config(&self) -> FileLoggerConfig; + + fn get_console_logger_config(&self) -> ConsoleLoggerConfig; +} + pub type NetworkSecretDigest = [u8; 32]; #[derive(Debug, Clone, Deserialize, Serialize, Default, Eq, Hash)] @@ -186,6 +187,24 @@ pub struct ConsoleLoggerConfig { pub level: Option, } +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, derive_builder::Builder)] +pub struct LoggingConfig { + #[builder(setter(into, strip_option), default = None)] + file_logger: Option, + #[builder(setter(into, strip_option), default = None)] + console_logger: Option, +} + +impl LoggingConfigLoader for &LoggingConfig { + fn get_file_logger_config(&self) -> FileLoggerConfig { + self.file_logger.clone().unwrap_or_default() + } + + fn get_console_logger_config(&self) -> ConsoleLoggerConfig { + self.console_logger.clone().unwrap_or_default() + } +} + #[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] pub struct VpnPortalConfig { pub client_cidr: cidr::Ipv4Cidr, @@ -243,9 +262,6 @@ struct Config { peer: Option>, proxy_network: Option>, - file_logger: Option, - console_logger: Option, - rpc_portal: Option, rpc_portal_whitelist: Option>, @@ -486,32 +502,6 @@ impl ConfigLoader for TomlConfigLoader { .unwrap_or_default() } - fn get_file_logger_config(&self) -> FileLoggerConfig { - self.config - .lock() - .unwrap() - .file_logger - .clone() - .unwrap_or_default() - } - - fn set_file_logger_config(&self, config: FileLoggerConfig) { - self.config.lock().unwrap().file_logger = Some(config); - } - - fn get_console_logger_config(&self) -> ConsoleLoggerConfig { - self.config - .lock() - .unwrap() - .console_logger - .clone() - .unwrap_or_default() - } - - fn set_console_logger_config(&self, config: ConsoleLoggerConfig) { - self.config.lock().unwrap().console_logger = Some(config); - } - fn get_peers(&self) -> Vec { self.config.lock().unwrap().peer.clone().unwrap_or_default() } diff --git a/easytier/src/easytier-core.rs b/easytier/src/easytier-core.rs index e209f37a..e733a32b 100644 --- a/easytier/src/easytier-core.rs +++ b/easytier/src/easytier-core.rs @@ -17,20 +17,17 @@ use clap::Parser; use easytier::{ common::{ config::{ - ConfigLoader, ConsoleLoggerConfig, FileLoggerConfig, NetworkIdentity, PeerConfig, - PortForwardConfig, TomlConfigLoader, VpnPortalConfig, + ConfigLoader, ConsoleLoggerConfig, FileLoggerConfig, LoggingConfigLoader, + NetworkIdentity, PeerConfig, PortForwardConfig, TomlConfigLoader, VpnPortalConfig, }, constants::EASYTIER_VERSION, - global_ctx::{EventBusSubscriber, GlobalCtx, GlobalCtxEvent}, - scoped_task::ScopedTask, + global_ctx::GlobalCtx, stun::MockStunInfoCollector, }, connector::create_connector_by_url, - launcher, - proto::{ - self, - common::{CompressionAlgoPb, NatType}, - }, + launcher::ConfigSource, + instance_manager::NetworkInstanceManager, + proto::common::{CompressionAlgoPb, NatType}, tunnel::{IpVersion, PROTO_PORT_OFFSET}, utils::{init_logger, setup_panic_handler}, web_client, @@ -106,10 +103,21 @@ struct Cli { short, long, env = "ET_CONFIG_FILE", - help = t!("core_clap.config_file").to_string() + value_delimiter = ',', + help = t!("core_clap.config_file").to_string(), + num_args = 1.., )] - config_file: Option, + config_file: Option>, + #[command(flatten)] + network_options: NetworkOptions, + + #[command(flatten)] + logging_options: LoggingOptions, +} + +#[derive(Parser, Debug)] +struct NetworkOptions { #[arg( long, env = "ET_NETWORK_NAME", @@ -212,27 +220,6 @@ struct Cli { )] no_listener: bool, - #[arg( - long, - env = "ET_CONSOLE_LOG_LEVEL", - help = t!("core_clap.console_log_level").to_string() - )] - console_log_level: Option, - - #[arg( - long, - env = "ET_FILE_LOG_LEVEL", - help = t!("core_clap.file_log_level").to_string() - )] - file_log_level: Option, - - #[arg( - long, - env = "ET_FILE_LOG_DIR", - help = t!("core_clap.file_log_dir").to_string() - )] - file_log_dir: Option, - #[arg( long, env = "ET_HOSTNAME", @@ -470,6 +457,30 @@ struct Cli { private_mode: Option, } +#[derive(Parser, Debug)] +struct LoggingOptions { + #[arg( + long, + env = "ET_CONSOLE_LOG_LEVEL", + help = t!("core_clap.console_log_level").to_string() + )] + console_log_level: Option, + + #[arg( + long, + env = "ET_FILE_LOG_LEVEL", + help = t!("core_clap.file_log_level").to_string() + )] + file_log_level: Option, + + #[arg( + long, + env = "ET_FILE_LOG_DIR", + help = t!("core_clap.file_log_dir").to_string() + )] + file_log_dir: Option, +} + rust_i18n::i18n!("locales", fallback = "en"); impl Cli { @@ -527,43 +538,47 @@ impl Cli { } } -impl TryFrom<&Cli> for TomlConfigLoader { - type Error = anyhow::Error; - - fn try_from(cli: &Cli) -> Result { - let cfg = if let Some(config_file) = &cli.config_file { - TomlConfigLoader::new(config_file) - .with_context(|| format!("failed to load config file: {:?}", cli.config_file))? - } else { - TomlConfigLoader::default() +impl NetworkOptions { + fn can_merge(&self, cfg: &TomlConfigLoader, config_file_count: usize) -> bool { + if config_file_count == 1{ + return true; + } + let Some(network_name) = &self.network_name else { + return false; }; + if cfg.get_network_identity().network_name == *network_name { + return true; + } + false + } - if cli.hostname.is_some() { - cfg.set_hostname(cli.hostname.clone()); + fn merge_into(&self, cfg: &mut TomlConfigLoader) -> anyhow::Result<()> { + if self.hostname.is_some() { + cfg.set_hostname(self.hostname.clone()); } let old_ns = cfg.get_network_identity(); - let network_name = cli.network_name.clone().unwrap_or(old_ns.network_name); - let network_secret = cli + let network_name = self.network_name.clone().unwrap_or(old_ns.network_name); + let network_secret = self .network_secret .clone() .unwrap_or(old_ns.network_secret.unwrap_or_default()); cfg.set_network_identity(NetworkIdentity::new(network_name, network_secret)); - if let Some(dhcp) = cli.dhcp { + if let Some(dhcp) = self.dhcp { cfg.set_dhcp(dhcp); } - if let Some(ipv4) = &cli.ipv4 { + if let Some(ipv4) = &self.ipv4 { cfg.set_ipv4(Some(ipv4.parse().with_context(|| { format!("failed to parse ipv4 address: {}", ipv4) })?)) } - if !cli.peers.is_empty() { + if !self.peers.is_empty() { let mut peers = cfg.get_peers(); - peers.reserve(peers.len() + cli.peers.len()); - for p in &cli.peers { + peers.reserve(peers.len() + self.peers.len()); + for p in &self.peers { peers.push(PeerConfig { uri: p .parse() @@ -573,9 +588,9 @@ impl TryFrom<&Cli> for TomlConfigLoader { cfg.set_peers(peers); } - if cli.no_listener || !cli.listeners.is_empty() { + if self.no_listener || !self.listeners.is_empty() { cfg.set_listeners( - Cli::parse_listeners(cli.no_listener, cli.listeners.clone())? + Cli::parse_listeners(self.no_listener, self.listeners.clone())? .into_iter() .map(|s| s.parse().unwrap()) .collect(), @@ -589,9 +604,9 @@ impl TryFrom<&Cli> for TomlConfigLoader { ); } - if !cli.mapped_listeners.is_empty() { + if !self.mapped_listeners.is_empty() { cfg.set_mapped_listeners(Some( - cli.mapped_listeners + self.mapped_listeners .iter() .map(|s| { s.parse() @@ -608,14 +623,14 @@ impl TryFrom<&Cli> for TomlConfigLoader { )); } - for n in cli.proxy_networks.iter() { + for n in self.proxy_networks.iter() { cfg.add_proxy_cidr( n.parse() .with_context(|| format!("failed to parse proxy network: {}", n))?, ); } - let rpc_portal = if let Some(r) = &cli.rpc_portal { + let rpc_portal = if let Some(r) = &self.rpc_portal { Cli::parse_rpc_portal(r.clone()) .with_context(|| format!("failed to parse rpc portal: {}", r))? } else if let Some(r) = cfg.get_rpc_portal() { @@ -625,9 +640,9 @@ impl TryFrom<&Cli> for TomlConfigLoader { }; cfg.set_rpc_portal(rpc_portal); - cfg.set_rpc_portal_whitelist(cli.rpc_portal_whitelist.clone()); + cfg.set_rpc_portal_whitelist(self.rpc_portal_whitelist.clone()); - if let Some(external_nodes) = cli.external_node.as_ref() { + if let Some(external_nodes) = self.external_node.as_ref() { let mut old_peers = cfg.get_peers(); old_peers.push(PeerConfig { uri: external_nodes.parse().with_context(|| { @@ -637,37 +652,11 @@ impl TryFrom<&Cli> for TomlConfigLoader { cfg.set_peers(old_peers); } - if cli.console_log_level.is_some() { - cfg.set_console_logger_config(ConsoleLoggerConfig { - level: cli.console_log_level.clone(), - }); - } - - if let Some(inst_name) = &cli.instance_name { + if let Some(inst_name) = &self.instance_name { cfg.set_inst_name(inst_name.clone()); } - if cli.file_log_dir.is_some() || cli.file_log_level.is_some() { - let inst_name = cfg.get_inst_name(); - let old_fl = cfg.get_file_logger_config(); - let file_log_dir = if cli.file_log_dir.is_some() { - &cli.file_log_dir - } else { - &old_fl.dir - }; - let file_log_level = if cli.file_log_level.is_some() { - &cli.file_log_level - } else { - &old_fl.level - }; - cfg.set_file_logger_config(FileLoggerConfig { - level: file_log_level.clone(), - dir: file_log_dir.clone(), - file: Some(format!("easytier-{}", inst_name)), - }); - } - - if let Some(vpn_portal) = cli.vpn_portal.as_ref() { + if let Some(vpn_portal) = self.vpn_portal.as_ref() { let url: url::Url = vpn_portal .parse() .with_context(|| format!("failed to parse vpn portal url: {}", vpn_portal))?; @@ -687,7 +676,7 @@ impl TryFrom<&Cli> for TomlConfigLoader { }); } - if let Some(manual_routes) = cli.manual_routes.as_ref() { + if let Some(manual_routes) = self.manual_routes.as_ref() { let mut routes = Vec::::with_capacity(manual_routes.len()); for r in manual_routes { routes.push( @@ -699,7 +688,7 @@ impl TryFrom<&Cli> for TomlConfigLoader { } #[cfg(feature = "socks5")] - if let Some(socks5_proxy) = cli.socks5 { + if let Some(socks5_proxy) = self.socks5 { cfg.set_socks5_portal(Some( format!("socks5://0.0.0.0:{}", socks5_proxy) .parse() @@ -708,7 +697,7 @@ impl TryFrom<&Cli> for TomlConfigLoader { } #[cfg(feature = "socks5")] - for port_forward in cli.port_forward.iter() { + for port_forward in self.port_forward.iter() { let example_str = ", example: udp://0.0.0.0:12345/10.126.126.1:12345"; let bind_addr = format!( @@ -742,38 +731,38 @@ impl TryFrom<&Cli> for TomlConfigLoader { } let mut f = cfg.get_flags(); - if let Some(default_protocol) = &cli.default_protocol { + if let Some(default_protocol) = &self.default_protocol { f.default_protocol = default_protocol.clone() }; - if let Some(v) = cli.disable_encryption { + if let Some(v) = self.disable_encryption { f.enable_encryption = !v; } - if let Some(v) = cli.disable_ipv6 { + if let Some(v) = self.disable_ipv6 { f.enable_ipv6 = !v; } - f.latency_first = cli.latency_first.unwrap_or(f.latency_first); - if let Some(dev_name) = &cli.dev_name { + f.latency_first = self.latency_first.unwrap_or(f.latency_first); + if let Some(dev_name) = &self.dev_name { f.dev_name = dev_name.clone() } - if let Some(mtu) = cli.mtu { + if let Some(mtu) = self.mtu { f.mtu = mtu as u32; } - f.enable_exit_node = cli.enable_exit_node.unwrap_or(f.enable_exit_node); - f.proxy_forward_by_system = cli + f.enable_exit_node = self.enable_exit_node.unwrap_or(f.enable_exit_node); + f.proxy_forward_by_system = self .proxy_forward_by_system .unwrap_or(f.proxy_forward_by_system); - f.no_tun = cli.no_tun.unwrap_or(f.no_tun) || cfg!(not(feature = "tun")); - f.use_smoltcp = cli.use_smoltcp.unwrap_or(f.use_smoltcp); - if let Some(wl) = cli.relay_network_whitelist.as_ref() { + f.no_tun = self.no_tun.unwrap_or(f.no_tun) || cfg!(not(feature = "tun")); + f.use_smoltcp = self.use_smoltcp.unwrap_or(f.use_smoltcp); + if let Some(wl) = self.relay_network_whitelist.as_ref() { f.relay_network_whitelist = wl.join(" "); } - f.disable_p2p = cli.disable_p2p.unwrap_or(f.disable_p2p); - f.disable_udp_hole_punching = cli + f.disable_p2p = self.disable_p2p.unwrap_or(f.disable_p2p); + f.disable_udp_hole_punching = self .disable_udp_hole_punching .unwrap_or(f.disable_udp_hole_punching); - f.relay_all_peer_rpc = cli.relay_all_peer_rpc.unwrap_or(f.relay_all_peer_rpc); - f.multi_thread = cli.multi_thread.unwrap_or(f.multi_thread); - if let Some(compression) = &cli.compression { + f.relay_all_peer_rpc = self.relay_all_peer_rpc.unwrap_or(f.relay_all_peer_rpc); + f.multi_thread = self.multi_thread.unwrap_or(f.multi_thread); + if let Some(compression) = &self.compression { f.data_compress_algo = match compression.as_str() { "none" => CompressionAlgoPb::None, "zstd" => CompressionAlgoPb::Zstd, @@ -784,154 +773,35 @@ impl TryFrom<&Cli> for TomlConfigLoader { } .into(); } - f.bind_device = cli.bind_device.unwrap_or(f.bind_device); - f.enable_kcp_proxy = cli.enable_kcp_proxy.unwrap_or(f.enable_kcp_proxy); - f.disable_kcp_input = cli.disable_kcp_input.unwrap_or(f.disable_kcp_input); - f.accept_dns = cli.accept_dns.unwrap_or(f.accept_dns); - f.private_mode = cli.private_mode.unwrap_or(f.private_mode); + f.bind_device = self.bind_device.unwrap_or(f.bind_device); + f.enable_kcp_proxy = self.enable_kcp_proxy.unwrap_or(f.enable_kcp_proxy); + f.disable_kcp_input = self.disable_kcp_input.unwrap_or(f.disable_kcp_input); + f.accept_dns = self.accept_dns.unwrap_or(f.accept_dns); + f.private_mode = self.private_mode.unwrap_or(f.private_mode); cfg.set_flags(f); - if !cli.exit_nodes.is_empty() { - cfg.set_exit_nodes(cli.exit_nodes.clone()); + if !self.exit_nodes.is_empty() { + cfg.set_exit_nodes(self.exit_nodes.clone()); } - Ok(cfg) + Ok(()) } } -fn print_event(msg: String) { - println!( - "{}: {}", - chrono::Local::now().format("%Y-%m-%d %H:%M:%S"), - msg - ); -} - -fn peer_conn_info_to_string(p: proto::cli::PeerConnInfo) -> String { - format!( - "my_peer_id: {}, dst_peer_id: {}, tunnel_info: {:?}", - p.my_peer_id, p.peer_id, p.tunnel - ) -} - -#[tracing::instrument] -pub fn handle_event(mut events: EventBusSubscriber) -> tokio::task::JoinHandle<()> { - tokio::spawn(async move { - loop { - if let Ok(e) = events.recv().await { - match e { - GlobalCtxEvent::PeerAdded(p) => { - print_event(format!("new peer added. peer_id: {}", p)); - } - - GlobalCtxEvent::PeerRemoved(p) => { - print_event(format!("peer removed. peer_id: {}", p)); - } - - GlobalCtxEvent::PeerConnAdded(p) => { - print_event(format!( - "new peer connection added. conn_info: {}", - peer_conn_info_to_string(p) - )); - } - - GlobalCtxEvent::PeerConnRemoved(p) => { - print_event(format!( - "peer connection removed. conn_info: {}", - peer_conn_info_to_string(p) - )); - } - - GlobalCtxEvent::ListenerAddFailed(p, msg) => { - print_event(format!( - "listener add failed. listener: {}, msg: {}", - p, msg - )); - } - - GlobalCtxEvent::ListenerAcceptFailed(p, msg) => { - print_event(format!( - "listener accept failed. listener: {}, msg: {}", - p, msg - )); - } - - GlobalCtxEvent::ListenerAdded(p) => { - if p.scheme() == "ring" { - continue; - } - print_event(format!("new listener added. listener: {}", p)); - } - - GlobalCtxEvent::ConnectionAccepted(local, remote) => { - print_event(format!( - "new connection accepted. local: {}, remote: {}", - local, remote - )); - } - - GlobalCtxEvent::ConnectionError(local, remote, err) => { - print_event(format!( - "connection error. local: {}, remote: {}, err: {}", - local, remote, err - )); - } - - GlobalCtxEvent::TunDeviceReady(dev) => { - print_event(format!("tun device ready. dev: {}", dev)); - } - - GlobalCtxEvent::TunDeviceError(err) => { - print_event(format!("tun device error. err: {}", err)); - } - - GlobalCtxEvent::Connecting(dst) => { - print_event(format!("connecting to peer. dst: {}", dst)); - } - - GlobalCtxEvent::ConnectError(dst, ip_version, err) => { - print_event(format!( - "connect to peer error. dst: {}, ip_version: {}, err: {}", - dst, ip_version, err - )); - } - - GlobalCtxEvent::VpnPortalClientConnected(portal, client_addr) => { - print_event(format!( - "vpn portal client connected. portal: {}, client_addr: {}", - portal, client_addr - )); - } - - GlobalCtxEvent::VpnPortalClientDisconnected(portal, client_addr) => { - print_event(format!( - "vpn portal client disconnected. portal: {}, client_addr: {}", - portal, client_addr - )); - } - - GlobalCtxEvent::DhcpIpv4Changed(old, new) => { - print_event(format!("dhcp ip changed. old: {:?}, new: {:?}", old, new)); - } - - GlobalCtxEvent::DhcpIpv4Conflicted(ip) => { - print_event(format!("dhcp ip conflict. ip: {:?}", ip)); - } - - GlobalCtxEvent::PortForwardAdded(cfg) => { - print_event(format!( - "port forward added. local: {}, remote: {}, proto: {}", - cfg.bind_addr.unwrap().to_string(), - cfg.dst_addr.unwrap().to_string(), - cfg.socket_type().as_str_name() - )); - } - } - } else { - events = events.resubscribe(); - } +impl LoggingConfigLoader for &LoggingOptions { + fn get_console_logger_config(&self) -> ConsoleLoggerConfig { + ConsoleLoggerConfig { + level: self.console_log_level.clone(), } - }) + } + + fn get_file_logger_config(&self) -> FileLoggerConfig { + FileLoggerConfig { + level: self.file_log_level.clone(), + dir: self.file_log_dir.clone(), + file: None, + } + } } #[cfg(target_os = "windows")] @@ -1046,8 +916,7 @@ fn win_service_main(arg: Vec) { } async fn run_main(cli: Cli) -> anyhow::Result<()> { - let cfg = TomlConfigLoader::try_from(&cli)?; - init_logger(&cfg, false)?; + init_logger(&cli.logging_options, false)?; if cli.config_server.is_some() { let config_server_url_s = cli.config_server.clone().unwrap(); @@ -1088,7 +957,7 @@ async fn run_main(cli: Cli) -> anyhow::Result<()> { let mut flags = global_ctx.get_flags(); flags.bind_device = false; global_ctx.set_flags(flags); - let hostname = match cli.hostname { + let hostname = match cli.network_options.hostname { None => gethostname::gethostname().to_string_lossy().to_string(), Some(hostname) => hostname.to_string(), }; @@ -1100,19 +969,47 @@ async fn run_main(cli: Cli) -> anyhow::Result<()> { tokio::signal::ctrl_c().await.unwrap(); return Ok(()); } + let manager = NetworkInstanceManager::new(); + let mut crate_cli_network = + cli.config_file.is_none() || cli.network_options.network_name.is_some(); + if let Some(config_files) = cli.config_file { + let config_file_count = config_files.len(); + for config_file in config_files { + let mut cfg = TomlConfigLoader::new(&config_file) + .with_context(|| format!("failed to load config file: {:?}", config_file))?; - println!("Starting easytier with config:"); - println!("############### TOML ###############\n"); - println!("{}", cfg.dump()); - println!("-----------------------------------"); - - let mut l = launcher::NetworkInstance::new(cfg).set_fetch_node_info(false); - let _t = ScopedTask::from(handle_event(l.start().unwrap())); - tokio::select! { - e = l.wait() => { - if let Some(e) = e { - eprintln!("launcher error: {}", e); + if cli.network_options.can_merge(&cfg, config_file_count) { + cli.network_options.merge_into(&mut cfg).with_context(|| { + format!("failed to merge config from cli: {:?}", config_file) + })?; + crate_cli_network = false; } + + println!( + "Starting easytier from config file {:?} with config:", + config_file + ); + println!("############### TOML ###############\n"); + println!("{}", cfg.dump()); + println!("-----------------------------------"); + manager.run_network_instance(cfg, ConfigSource::File)?; + } + } + + if crate_cli_network { + let mut cfg = TomlConfigLoader::default(); + cli.network_options + .merge_into(&mut cfg) + .with_context(|| format!("failed to create config from cli"))?; + println!("Starting easytier from cli with config:"); + println!("############### TOML ###############\n"); + println!("{}", cfg.dump()); + println!("-----------------------------------"); + manager.run_network_instance(cfg, ConfigSource::Cli)?; + } + + tokio::select! { + _ = manager.wait() => { } _ = tokio::signal::ctrl_c() => { println!("ctrl-c received, exiting..."); diff --git a/easytier/src/instance_manager.rs b/easytier/src/instance_manager.rs new file mode 100644 index 00000000..29eef009 --- /dev/null +++ b/easytier/src/instance_manager.rs @@ -0,0 +1,491 @@ +use std::{collections::BTreeMap, sync::Arc}; + +use dashmap::DashMap; + +use crate::{ + common::{ + config::{ConfigLoader, TomlConfigLoader}, + global_ctx::{EventBusSubscriber, GlobalCtxEvent}, + scoped_task::ScopedTask, + }, + launcher::{ConfigSource, NetworkInstance, NetworkInstanceRunningInfo}, + proto, +}; + +pub struct NetworkInstanceManager { + instance_map: Arc>, + instance_stop_tasks: Arc>>, + stop_check_notifier: Arc, +} + +impl NetworkInstanceManager { + pub fn new() -> Self { + NetworkInstanceManager { + instance_map: Arc::new(DashMap::new()), + instance_stop_tasks: Arc::new(DashMap::new()), + stop_check_notifier: Arc::new(tokio::sync::Notify::new()), + } + } + + fn start_instance_task(&self, instance_id: uuid::Uuid) -> Result<(), anyhow::Error> { + let instance = self + .instance_map + .get(&instance_id) + .ok_or_else(|| anyhow::anyhow!("instance {} not found", instance_id))?; + + if instance.get_config_source() == ConfigSource::FFI { + // FFI have no tokio runtime, so we don't need to spawn a task, and instance should be managed by the caller. + return Ok(()); + } + + let instance_stop_notifier = instance.get_stop_notifier(); + let instance_config_source = instance.get_config_source(); + let instance_event_receiver = match instance.get_config_source() { + ConfigSource::Cli | ConfigSource::File => Some(instance.subscribe_event()), + _ => None, + }; + + let instance_map = self.instance_map.clone(); + let instance_stop_tasks = self.instance_stop_tasks.clone(); + + let stop_check_notifier = self.stop_check_notifier.clone(); + self.instance_stop_tasks.insert( + instance_id, + ScopedTask::from(tokio::spawn(async move { + let Some(instance_stop_notifier) = instance_stop_notifier else { + return; + }; + let _t = if let Some(event) = instance_event_receiver.flatten() { + Some(ScopedTask::from(handle_event(instance_id, event))) + } else { + None + }; + instance_stop_notifier.notified().await; + if let Some(instance) = instance_map.get(&instance_id) { + if let Some(e) = instance.get_latest_error_msg() { + tracing::error!(?e, ?instance_id, "instance stopped with error"); + eprintln!("instance {} stopped with error: {}", instance_id, e); + } + } + match instance_config_source { + ConfigSource::Cli | ConfigSource::File => { + instance_map.remove(&instance_id); + } + ConfigSource::Web | ConfigSource::GUI | ConfigSource::FFI => {} + } + instance_stop_tasks.remove(&instance_id); + stop_check_notifier.notify_waiters(); + })), + ); + Ok(()) + } + + pub fn run_network_instance( + &self, + cfg: TomlConfigLoader, + source: ConfigSource, + ) -> Result { + let instance_id = cfg.get_id(); + if self.instance_map.contains_key(&instance_id) { + anyhow::bail!("instance {} already exists", instance_id); + } + + let mut instance = NetworkInstance::new(cfg, source); + instance.start()?; + + self.instance_map.insert(instance_id, instance); + self.start_instance_task(instance_id)?; + Ok(instance_id) + } + + pub fn retain_network_instance( + &self, + instance_ids: Vec, + ) -> Result, anyhow::Error> { + self.instance_map.retain(|k, _| instance_ids.contains(k)); + Ok(self.list_network_instance_ids()) + } + + pub fn delete_network_instance( + &self, + instance_ids: Vec, + ) -> Result, anyhow::Error> { + self.instance_map.retain(|k, _| !instance_ids.contains(k)); + Ok(self.list_network_instance_ids()) + } + + pub fn collect_network_infos( + &self, + ) -> Result, anyhow::Error> { + let mut ret = BTreeMap::new(); + for instance in self.instance_map.iter() { + if let Some(info) = instance.get_running_info() { + ret.insert(instance.key().clone(), info); + } + } + Ok(ret) + } + + pub fn list_network_instance_ids(&self) -> Vec { + self.instance_map + .iter() + .map(|item| item.key().clone()) + .collect() + } + + pub fn get_network_instance_name(&self, instance_id: &uuid::Uuid) -> Option { + self.instance_map + .get(instance_id) + .map(|instance| instance.value().get_inst_name()) + } + + pub fn set_tun_fd(&self, instance_id: &uuid::Uuid, fd: i32) -> Result<(), anyhow::Error> { + let mut instance = self + .instance_map + .get_mut(instance_id) + .ok_or_else(|| anyhow::anyhow!("instance not found"))?; + instance.set_tun_fd(fd); + Ok(()) + } + + pub async fn wait(&self) { + while self.instance_map.len() > 0 { + self.stop_check_notifier.notified().await; + } + } +} + +#[tracing::instrument] +fn handle_event( + instance_id: uuid::Uuid, + mut events: EventBusSubscriber, +) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + loop { + if let Ok(e) = events.recv().await { + match e { + GlobalCtxEvent::PeerAdded(p) => { + print_event(instance_id, format!("new peer added. peer_id: {}", p)); + } + + GlobalCtxEvent::PeerRemoved(p) => { + print_event(instance_id, format!("peer removed. peer_id: {}", p)); + } + + GlobalCtxEvent::PeerConnAdded(p) => { + print_event( + instance_id, + format!( + "new peer connection added. conn_info: {}", + peer_conn_info_to_string(p) + ), + ); + } + + GlobalCtxEvent::PeerConnRemoved(p) => { + print_event( + instance_id, + format!( + "peer connection removed. conn_info: {}", + peer_conn_info_to_string(p) + ), + ); + } + + GlobalCtxEvent::ListenerAddFailed(p, msg) => { + print_event( + instance_id, + format!("listener add failed. listener: {}, msg: {}", p, msg), + ); + } + + GlobalCtxEvent::ListenerAcceptFailed(p, msg) => { + print_event( + instance_id, + format!("listener accept failed. listener: {}, msg: {}", p, msg), + ); + } + + GlobalCtxEvent::ListenerAdded(p) => { + if p.scheme() == "ring" { + continue; + } + print_event(instance_id, format!("new listener added. listener: {}", p)); + } + + GlobalCtxEvent::ConnectionAccepted(local, remote) => { + print_event( + instance_id, + format!( + "new connection accepted. local: {}, remote: {}", + local, remote + ), + ); + } + + GlobalCtxEvent::ConnectionError(local, remote, err) => { + print_event( + instance_id, + format!( + "connection error. local: {}, remote: {}, err: {}", + local, remote, err + ), + ); + } + + GlobalCtxEvent::TunDeviceReady(dev) => { + print_event(instance_id, format!("tun device ready. dev: {}", dev)); + } + + GlobalCtxEvent::TunDeviceError(err) => { + print_event(instance_id, format!("tun device error. err: {}", err)); + } + + GlobalCtxEvent::Connecting(dst) => { + print_event(instance_id, format!("connecting to peer. dst: {}", dst)); + } + + GlobalCtxEvent::ConnectError(dst, ip_version, err) => { + print_event( + instance_id, + format!( + "connect to peer error. dst: {}, ip_version: {}, err: {}", + dst, ip_version, err + ), + ); + } + + GlobalCtxEvent::VpnPortalClientConnected(portal, client_addr) => { + print_event( + instance_id, + format!( + "vpn portal client connected. portal: {}, client_addr: {}", + portal, client_addr + ), + ); + } + + GlobalCtxEvent::VpnPortalClientDisconnected(portal, client_addr) => { + print_event( + instance_id, + format!( + "vpn portal client disconnected. portal: {}, client_addr: {}", + portal, client_addr + ), + ); + } + + GlobalCtxEvent::DhcpIpv4Changed(old, new) => { + print_event( + instance_id, + format!("dhcp ip changed. old: {:?}, new: {:?}", old, new), + ); + } + + GlobalCtxEvent::DhcpIpv4Conflicted(ip) => { + print_event(instance_id, format!("dhcp ip conflict. ip: {:?}", ip)); + } + + GlobalCtxEvent::PortForwardAdded(cfg) => { + print_event( + instance_id, + format!( + "port forward added. local: {}, remote: {}, proto: {}", + cfg.bind_addr.unwrap().to_string(), + cfg.dst_addr.unwrap().to_string(), + cfg.socket_type().as_str_name() + ), + ); + } + } + } else { + events = events.resubscribe(); + } + } + }) +} + +fn print_event(instance_id: uuid::Uuid, msg: String) { + println!( + "{}: [{}] {}", + chrono::Local::now().format("%Y-%m-%d %H:%M:%S"), + instance_id, + msg + ); +} + +fn peer_conn_info_to_string(p: proto::cli::PeerConnInfo) -> String { + format!( + "my_peer_id: {}, dst_peer_id: {}, tunnel_info: {:?}", + p.my_peer_id, p.peer_id, p.tunnel + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::common::config::*; + + #[tokio::test] + async fn it_works() { + let manager = NetworkInstanceManager::new(); + let cfg_str = r#" + listeners = [] + "#; + + let port = crate::utils::find_free_tcp_port(10012..65534).expect("no free tcp port found"); + + let instance_id1 = manager + .run_network_instance( + TomlConfigLoader::new_from_str(cfg_str) + .map(|c| { + c.set_listeners(vec![format!("tcp://0.0.0.0:{}", port).parse().unwrap()]); + c + }) + .unwrap(), + ConfigSource::Cli, + ) + .unwrap(); + let instance_id2 = manager + .run_network_instance( + TomlConfigLoader::new_from_str(cfg_str).unwrap(), + ConfigSource::File, + ) + .unwrap(); + let instance_id3 = manager + .run_network_instance( + TomlConfigLoader::new_from_str(cfg_str).unwrap(), + ConfigSource::GUI, + ) + .unwrap(); + let instance_id4 = manager + .run_network_instance( + TomlConfigLoader::new_from_str(cfg_str).unwrap(), + ConfigSource::Web, + ) + .unwrap(); + let instance_id5 = manager + .run_network_instance( + TomlConfigLoader::new_from_str(cfg_str).unwrap(), + ConfigSource::FFI, + ) + .unwrap(); + + tokio::time::sleep(std::time::Duration::from_secs(1)).await; // to make instance actually started + + assert!(!crate::utils::check_tcp_available(port)); + + assert!(manager.instance_map.contains_key(&instance_id1)); + assert!(manager.instance_map.contains_key(&instance_id2)); + assert!(manager.instance_map.contains_key(&instance_id3)); + assert!(manager.instance_map.contains_key(&instance_id4)); + assert!(manager.instance_map.contains_key(&instance_id5)); + assert_eq!(manager.list_network_instance_ids().len(), 5); + assert_eq!(manager.instance_stop_tasks.len(), 4); // FFI instance does not have a stop task + + manager + .delete_network_instance(vec![instance_id3, instance_id4, instance_id5]) + .unwrap(); + assert!(!manager.instance_map.contains_key(&instance_id3)); + assert!(!manager.instance_map.contains_key(&instance_id4)); + assert!(!manager.instance_map.contains_key(&instance_id5)); + assert_eq!(manager.list_network_instance_ids().len(), 2); + } + + #[tokio::test] + async fn test_single_instance_failed() { + let free_tcp_port = + crate::utils::find_free_tcp_port(10012..65534).expect("no free tcp port found"); + + for config_source in [ConfigSource::Cli, ConfigSource::File] { + let _port_holder = + std::net::TcpListener::bind(format!("0.0.0.0:{}", free_tcp_port)).unwrap(); + + let cfg_str = format!( + r#" + listeners = ["tcp://0.0.0.0:{}"] + "#, + free_tcp_port + ); + + let manager = NetworkInstanceManager::new(); + manager + .run_network_instance( + TomlConfigLoader::new_from_str(cfg_str.as_str()).unwrap(), + config_source.clone(), + ) + .unwrap(); + + tokio::select! { + _ = manager.wait() => { + assert_eq!(manager.list_network_instance_ids().len(), 0); + } + _ = tokio::time::sleep(std::time::Duration::from_secs(5)) => { + panic!("instance manager with single failed instance({:?}) should not running", config_source); + } + } + } + for config_source in [ConfigSource::Web, ConfigSource::GUI, ConfigSource::FFI] { + let _port_holder = + std::net::TcpListener::bind(format!("0.0.0.0:{}", free_tcp_port)).unwrap(); + + let cfg_str = format!( + r#" + listeners = ["tcp://0.0.0.0:{}"] + "#, + free_tcp_port + ); + + let manager = NetworkInstanceManager::new(); + manager + .run_network_instance( + TomlConfigLoader::new_from_str(cfg_str.as_str()).unwrap(), + config_source.clone(), + ) + .unwrap(); + + assert_eq!(manager.list_network_instance_ids().len(), 1); + } + } + + #[tokio::test] + async fn test_multiple_instances_one_failed() { + let free_tcp_port = + crate::utils::find_free_tcp_port(10012..65534).expect("no free tcp port found"); + + let manager = NetworkInstanceManager::new(); + let cfg_str = format!( + r#" + listeners = ["tcp://0.0.0.0:{}"] + [flags] + enable_ipv6 = false + "#, + free_tcp_port + ); + + manager + .run_network_instance( + TomlConfigLoader::new_from_str(cfg_str.as_str()).unwrap(), + ConfigSource::Cli, + ) + .unwrap(); + + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + + manager + .run_network_instance( + TomlConfigLoader::new_from_str(cfg_str.as_str()).unwrap(), + ConfigSource::Cli, + ) + .unwrap(); + + tokio::select! { + _ = manager.wait() => { + panic!("instance manager with multiple instances one failed should still running"); + } + _ = tokio::time::sleep(std::time::Duration::from_secs(2)) => { + assert_eq!(manager.list_network_instance_ids().len(), 1); + } + } + } +} diff --git a/easytier/src/launcher.rs b/easytier/src/launcher.rs index bd5ba726..90e9ac0a 100644 --- a/easytier/src/launcher.rs +++ b/easytier/src/launcher.rs @@ -1,6 +1,5 @@ use std::{ collections::VecDeque, - net::SocketAddr, sync::{atomic::AtomicBool, Arc, RwLock}, }; @@ -214,24 +213,18 @@ impl EasyTierLauncher { Ok(()) } - fn check_tcp_available(port: u16) -> bool { - let s = format!("0.0.0.0:{}", port).parse::().unwrap(); - std::net::TcpListener::bind(s).is_ok() - } - fn select_proper_rpc_port(cfg: &TomlConfigLoader) { let Some(mut f) = cfg.get_rpc_portal() else { return; }; if f.port() == 0 { - for i in 15888..15900 { - if Self::check_tcp_available(i) { - f.set_port(i); - cfg.set_rpc_portal(f); - break; - } - } + let Some(port) = crate::utils::find_free_tcp_port(15888..15900) else { + tracing::warn!("No free port found for RPC portal, skipping setting RPC portal"); + return; + }; + f.set_port(port); + cfg.set_rpc_portal(f); } } @@ -343,25 +336,40 @@ impl Drop for EasyTierLauncher { pub type NetworkInstanceRunningInfo = crate::proto::web::NetworkInstanceRunningInfo; +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ConfigSource { + Cli, + File, + Web, + GUI, + FFI, +} + pub struct NetworkInstance { config: TomlConfigLoader, launcher: Option, - fetch_node_info: bool, + config_source: ConfigSource, } impl NetworkInstance { - pub fn new(config: TomlConfigLoader) -> Self { + pub fn new(config: TomlConfigLoader, source: ConfigSource) -> Self { Self { config, launcher: None, - fetch_node_info: true, + config_source: source, } } - pub fn set_fetch_node_info(mut self, fetch_node_info: bool) -> Self { - self.fetch_node_info = fetch_node_info; - self + fn get_fetch_node_info(&self) -> bool { + match self.config_source { + ConfigSource::Cli | ConfigSource::File => false, + ConfigSource::Web | ConfigSource::GUI | ConfigSource::FFI => true, + } + } + + pub fn get_config_source(&self) -> ConfigSource { + self.config_source.clone() } pub fn is_easytier_running(&self) -> bool { @@ -395,6 +403,10 @@ impl NetworkInstance { }) } + pub fn get_inst_name(&self) -> String { + self.config.get_inst_name() + } + pub fn set_tun_fd(&mut self, tun_fd: i32) { if let Some(launcher) = self.launcher.as_ref() { launcher.data.tun_fd.write().unwrap().replace(tun_fd); @@ -406,7 +418,7 @@ impl NetworkInstance { return Ok(self.subscribe_event().unwrap()); } - let launcher = EasyTierLauncher::new(self.fetch_node_info); + let launcher = EasyTierLauncher::new(self.get_fetch_node_info()); self.launcher = Some(launcher); let ev = self.subscribe_event().unwrap(); @@ -418,7 +430,7 @@ impl NetworkInstance { Ok(ev) } - fn subscribe_event(&self) -> Option> { + pub fn subscribe_event(&self) -> Option> { if let Some(launcher) = self.launcher.as_ref() { Some(launcher.data.event_subscriber.read().unwrap().subscribe()) } else { @@ -426,9 +438,16 @@ impl NetworkInstance { } } - pub async fn wait(&self) -> Option { + pub fn get_stop_notifier(&self) -> Option> { + if let Some(launcher) = self.launcher.as_ref() { + Some(launcher.data.instance_stop_notifier.clone()) + } else { + None + } + } + + pub fn get_latest_error_msg(&self) -> Option { if let Some(launcher) = self.launcher.as_ref() { - launcher.data.instance_stop_notifier.notified().await; launcher.error_msg.read().unwrap().clone() } else { None diff --git a/easytier/src/lib.rs b/easytier/src/lib.rs index 23fa00bf..793cf8fd 100644 --- a/easytier/src/lib.rs +++ b/easytier/src/lib.rs @@ -9,6 +9,7 @@ mod vpn_portal; pub mod common; pub mod connector; pub mod launcher; +pub mod instance_manager; pub mod peers; pub mod proto; pub mod tunnel; diff --git a/easytier/src/utils.rs b/easytier/src/utils.rs index e36365c8..e705ca01 100644 --- a/easytier/src/utils.rs +++ b/easytier/src/utils.rs @@ -4,7 +4,7 @@ use anyhow::Context; use tracing::level_filters::LevelFilter; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter, Layer}; -use crate::common::{config::ConfigLoader, get_logger_timer_rfc3339}; +use crate::common::{config::LoggingConfigLoader, get_logger_timer_rfc3339}; pub type PeerRoutePair = crate::proto::cli::PeerRoutePair; @@ -23,7 +23,7 @@ pub fn float_to_str(f: f64, precision: usize) -> String { pub type NewFilterSender = std::sync::mpsc::Sender; pub fn init_logger( - config: impl ConfigLoader, + config: impl LoggingConfigLoader, need_reload: bool, ) -> Result, anyhow::Error> { let file_config = config.get_file_logger_config(); @@ -211,6 +211,21 @@ pub fn setup_panic_handler() { })); } +pub fn check_tcp_available(port: u16) -> bool { + use std::net::TcpListener; + let s = std::net::SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED), port); + TcpListener::bind(s).is_ok() +} + +pub fn find_free_tcp_port(range: std::ops::Range) -> Option { + for port in range { + if check_tcp_available(port) { + return Some(port); + } + } + None +} + #[cfg(test)] mod tests { use crate::common::config::{self}; @@ -219,7 +234,7 @@ mod tests { async fn test_logger_reload() { println!("current working dir: {:?}", std::env::current_dir()); - let config = config::TomlConfigLoader::default(); + let config = config::LoggingConfigBuilder::default().build().unwrap(); let s = init_logger(&config, true).unwrap(); tracing::debug!("test not display debug"); s.unwrap().send(LevelFilter::DEBUG.to_string()).unwrap(); diff --git a/easytier/src/web_client/controller.rs b/easytier/src/web_client/controller.rs index 655dbf99..45249c79 100644 --- a/easytier/src/web_client/controller.rs +++ b/easytier/src/web_client/controller.rs @@ -1,11 +1,5 @@ -use std::collections::BTreeMap; - -use dashmap::DashMap; - use crate::{ - common::config::{ConfigLoader, TomlConfigLoader}, - launcher::NetworkInstance, - proto::{ + common::config::ConfigLoader, launcher::ConfigSource, instance_manager::NetworkInstanceManager, proto::{ rpc_types::{self, controller::BaseController}, web::{ CollectNetworkInfoRequest, CollectNetworkInfoResponse, DeleteNetworkInstanceRequest, @@ -14,13 +8,13 @@ use crate::{ RetainNetworkInstanceResponse, RunNetworkInstanceRequest, RunNetworkInstanceResponse, ValidateConfigRequest, ValidateConfigResponse, WebClientService, }, - }, + } }; pub struct Controller { token: String, hostname: String, - instance_map: DashMap, + manager: NetworkInstanceManager, } impl Controller { @@ -28,55 +22,12 @@ impl Controller { Controller { token, hostname, - instance_map: DashMap::new(), + manager: NetworkInstanceManager::new(), } } - pub fn run_network_instance(&self, cfg: TomlConfigLoader) -> Result<(), anyhow::Error> { - let instance_id = cfg.get_id(); - if self.instance_map.contains_key(&instance_id) { - anyhow::bail!("instance {} already exists", instance_id); - } - - let mut instance = NetworkInstance::new(cfg); - instance.start()?; - - println!("instance {} started", instance_id); - self.instance_map.insert(instance_id, instance); - Ok(()) - } - - pub fn retain_network_instance( - &self, - instance_ids: Vec, - ) -> Result { - self.instance_map.retain(|k, _| instance_ids.contains(k)); - let remain = self - .instance_map - .iter() - .map(|item| item.key().clone().into()) - .collect::>(); - println!("instance {:?} retained", remain); - Ok(RetainNetworkInstanceResponse { - remain_inst_ids: remain, - }) - } - - pub fn collect_network_infos(&self) -> Result { - let mut map = BTreeMap::new(); - for instance in self.instance_map.iter() { - if let Some(info) = instance.get_running_info() { - map.insert(instance.key().to_string(), info); - } - } - Ok(NetworkInstanceRunningInfoMap { map }) - } - pub fn list_network_instance_ids(&self) -> Vec { - self.instance_map - .iter() - .map(|item| item.key().clone()) - .collect() + self.manager.list_network_instance_ids() } pub fn token(&self) -> String { @@ -114,7 +65,8 @@ impl WebClientService for Controller { if let Some(inst_id) = req.inst_id { cfg.set_id(inst_id.into()); } - self.run_network_instance(cfg)?; + self.manager.run_network_instance(cfg, ConfigSource::Web)?; + println!("instance {} started", id); Ok(RunNetworkInstanceResponse { inst_id: Some(id.into()), }) @@ -125,7 +77,13 @@ impl WebClientService for Controller { _: BaseController, req: RetainNetworkInstanceRequest, ) -> Result { - Ok(self.retain_network_instance(req.inst_ids.into_iter().map(Into::into).collect())?) + let remain = self + .manager + .retain_network_instance(req.inst_ids.into_iter().map(Into::into).collect())?; + println!("instance {:?} retained", remain); + Ok(RetainNetworkInstanceResponse { + remain_inst_ids: remain.iter().map(|item| (*item).into()).collect(), + }) } async fn collect_network_info( @@ -133,7 +91,14 @@ impl WebClientService for Controller { _: BaseController, req: CollectNetworkInfoRequest, ) -> Result { - let mut ret = self.collect_network_infos()?; + let mut ret = NetworkInstanceRunningInfoMap { + map: self + .manager + .collect_network_infos()? + .into_iter() + .map(|(k, v)| (k.to_string(), v)) + .collect(), + }; let include_inst_ids = req .inst_ids .iter() @@ -163,6 +128,7 @@ impl WebClientService for Controller { ) -> Result { Ok(ListNetworkInstanceResponse { inst_ids: self + .manager .list_network_instance_ids() .into_iter() .map(Into::into) @@ -176,11 +142,12 @@ impl WebClientService for Controller { _: BaseController, req: DeleteNetworkInstanceRequest, ) -> Result { - let mut inst_ids = self.list_network_instance_ids(); - inst_ids.retain(|id| !req.inst_ids.contains(&(id.clone().into()))); - self.retain_network_instance(inst_ids.clone())?; + let remain_inst_ids = self + .manager + .delete_network_instance(req.inst_ids.into_iter().map(Into::into).collect())?; + println!("instance {:?} retained", remain_inst_ids); Ok(DeleteNetworkInstanceResponse { - remain_inst_ids: inst_ids.into_iter().map(Into::into).collect(), + remain_inst_ids: remain_inst_ids.into_iter().map(Into::into).collect(), }) } }