From da28c8badc30d186fef780617db625d3d514f381 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9F=A9=E5=98=89=E4=B9=90?= Date: Sun, 7 Jun 2026 16:18:54 +0800 Subject: [PATCH] =?UTF-8?q?[OHOS]=20fix:=20=E4=BF=AE=E5=A4=8D=E5=86=85?= =?UTF-8?q?=E5=AD=98=E6=B3=84=E9=9C=B2=E9=97=AE=E9=A2=98=EF=BC=8C=E5=B9=B6?= =?UTF-8?q?=E9=87=8D=E6=9E=84=E6=97=A5=E5=BF=97=E7=AE=A1=E7=90=86=EF=BC=8C?= =?UTF-8?q?=E9=A2=84=E9=98=B2=E6=80=A7=E4=BF=AE=E5=A4=8D=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E5=BA=93=E5=88=9D=E5=A7=8B=E5=8C=96=E5=BC=82=E5=B8=B8=E9=97=AE?= =?UTF-8?q?=E9=A2=98=20(#2328)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: leak memory feat: new log manager * fix: fail to init db * fix: fail to init db * fix: cargo format --- easytier-contrib/easytier-ohrs/Cargo.lock | 176 ++++++---- easytier-contrib/easytier-ohrs/Cargo.toml | 1 - .../src/config/storage/config_meta.rs | 328 +++++++++++++----- .../src/config/types/stored_config.rs | 8 - .../easytier-ohrs/src/config_repo.rs | 154 ++++++-- .../src/config_repo/field_store.rs | 5 +- .../src/config_repo/legacy_migration.rs | 5 +- .../easytier-ohrs/src/exports/config_api.rs | 4 + .../easytier-ohrs/src/exports/runtime_api.rs | 124 +++++-- .../easytier-ohrs/src/kernel_bridge.rs | 4 +- .../src/kernel_bridge/protocol.rs | 34 ++ .../src/kernel_bridge/routing.rs | 33 +- .../src/kernel_bridge/socket_server.rs | 198 +++++++++-- easytier-contrib/easytier-ohrs/src/lib.rs | 97 ++++-- .../src/platform/logging/log_manager.rs | 266 ++++++++++++++ .../easytier-ohrs/src/platform/logging/mod.rs | 1 + .../src/platform/logging/native_log.rs | 72 ++-- .../src/runtime/state/runtime_state.rs | 171 +++++++++ 18 files changed, 1325 insertions(+), 356 deletions(-) create mode 100644 easytier-contrib/easytier-ohrs/src/platform/logging/log_manager.rs diff --git a/easytier-contrib/easytier-ohrs/Cargo.lock b/easytier-contrib/easytier-ohrs/Cargo.lock index 2731b4e6..bd13d008 100644 --- a/easytier-contrib/easytier-ohrs/Cargo.lock +++ b/easytier-contrib/easytier-ohrs/Cargo.lock @@ -1237,18 +1237,18 @@ dependencies = [ "ordered_hash_map", "parking_lot", "paste", + "pbjson", + "pbjson-build", "percent-encoding", - "petgraph 0.8.2", + "petgraph", "pin-project-lite", "pnet", "prefix-trie", "proc-macro2", - "prost", + "prost 0.14.3", "prost-build", - "prost-reflect", + "prost-reflect 0.16.4", "prost-reflect-build", - "prost-wkt", - "prost-wkt-build", "prost-wkt-types", "quinn", "quinn-plaintext", @@ -1318,9 +1318,8 @@ dependencies = [ "napi-build-ohos", "napi-derive-ohos", "napi-ohos", - "ohos-hilog-binding", "once_cell", - "prost-reflect", + "prost-reflect 0.14.7", "rusqlite", "serde", "serde_json", @@ -3165,22 +3164,6 @@ dependencies = [ "libc", ] -[[package]] -name = "ohos-hilog-binding" -version = "0.1.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "860d1e3c2c5e3217d819a16c815d2d4dcbc7610285d2612d08745a29c353a503" -dependencies = [ - "libc", - "ohos-hilogs-sys", -] - -[[package]] -name = "ohos-hilogs-sys" -version = "0.0.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ed07615005d0f8d7bcf901f89c8ff4870666a9bdb00382f588af383f40c160b7" - [[package]] name = "once_cell" version = "1.21.3" @@ -3308,6 +3291,28 @@ version = "1.0.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a" +[[package]] +name = "pbjson" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8edd1efdd8ab23ba9cb9ace3d9987a72663d5d7c9f74fa00b51d6213645cf6c" +dependencies = [ + "base64 0.22.1", + "serde", +] + +[[package]] +name = "pbjson-build" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2ed4d5c6ae95e08ac768883c8401cf0e8deb4e6e1d6a4e1fd3d2ec4f0ec63200" +dependencies = [ + "heck 0.5.0", + "itertools 0.14.0", + "prost 0.14.3", + "prost-types 0.14.3", +] + [[package]] name = "pbkdf2" version = "0.12.2" @@ -3334,16 +3339,6 @@ version = "2.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" -[[package]] -name = "petgraph" -version = "0.7.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3672b37090dbd86368a4145bc067582552b29c27377cad4e0a306c97f9bd7772" -dependencies = [ - "fixedbitset", - "indexmap", -] - [[package]] name = "petgraph" version = "0.8.2" @@ -3631,24 +3626,33 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2796faa41db3ec313a31f7624d9286acf277b52de526150b7e69f3debf891ee5" dependencies = [ "bytes", - "prost-derive", + "prost-derive 0.13.5", +] + +[[package]] +name = "prost" +version = "0.14.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2ea70524a2f82d518bce41317d0fae74151505651af45faf1ffbd6fd33f0568" +dependencies = [ + "bytes", + "prost-derive 0.14.3", ] [[package]] name = "prost-build" -version = "0.13.5" +version = "0.14.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf" +checksum = "343d3bd7056eda839b03204e68deff7d1b13aba7af2b2fd16890697274262ee7" dependencies = [ "heck 0.5.0", "itertools 0.14.0", "log", "multimap", - "once_cell", - "petgraph 0.7.1", + "petgraph", "prettyplease", - "prost", - "prost-types", + "prost 0.14.3", + "prost-types 0.14.3", "regex", "syn 2.0.106", "tempfile", @@ -3667,6 +3671,19 @@ dependencies = [ "syn 2.0.106", ] +[[package]] +name = "prost-derive" +version = "0.14.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b" +dependencies = [ + "anyhow", + "itertools 0.14.0", + "proc-macro2", + "quote", + "syn 2.0.106", +] + [[package]] name = "prost-reflect" version = "0.14.7" @@ -3674,19 +3691,30 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7b5edd582b62f5cde844716e66d92565d7faf7ab1445c8cebce6e00fba83ddb2" dependencies = [ "once_cell", - "prost", - "prost-reflect-derive", - "prost-types", + "prost 0.13.5", + "prost-reflect-derive 0.14.0", + "prost-types 0.13.5", +] + +[[package]] +name = "prost-reflect" +version = "0.16.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "590aa145fee8f7a26b5a6055365e7c5e89a5c1caae9869de76ec0ee73181a2f9" +dependencies = [ + "prost 0.14.3", + "prost-reflect-derive 0.16.0", + "prost-types 0.14.3", ] [[package]] name = "prost-reflect-build" -version = "0.14.0" +version = "0.16.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "50e2537231d94dd2778920c2ada37dd9eb1ac0325bb3ee3ee651bd44c1134123" +checksum = "8214ae2c30bbac390db0134d08300e770ef89b6d4e5abf855e8d300eded87e28" dependencies = [ "prost-build", - "prost-reflect", + "prost-reflect 0.16.4", ] [[package]] @@ -3700,24 +3728,44 @@ dependencies = [ "syn 2.0.106", ] +[[package]] +name = "prost-reflect-derive" +version = "0.16.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b6d90e29fa6c0d13c2c19ba5e4b3fb0efbf5975d27bcf4e260b7b15455bcabe" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.106", +] + [[package]] name = "prost-types" version = "0.13.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52c2c1bf36ddb1a1c396b3601a3cec27c2462e45f07c386894ec3ccf5332bd16" dependencies = [ - "prost", + "prost 0.13.5", +] + +[[package]] +name = "prost-types" +version = "0.14.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8991c4cbdb8bc5b11f0b074ffe286c30e523de90fee5ba8132f1399f23cb3dd7" +dependencies = [ + "prost 0.14.3", ] [[package]] name = "prost-wkt" -version = "0.6.1" +version = "0.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "497e1e938f0c09ef9cabe1d49437b4016e03e8f82fbbe5d1c62a9b61b9decae1" +checksum = "cd3de5e9c9e84fcb5efa204b8e283d23e615a8bc8c777bf1d6622bb01dc61445" dependencies = [ "chrono", "inventory", - "prost", + "prost 0.14.3", "serde", "serde_derive", "serde_json", @@ -3726,27 +3774,27 @@ dependencies = [ [[package]] name = "prost-wkt-build" -version = "0.6.1" +version = "0.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "07b8bf115b70a7aa5af1fd5d6e9418492e9ccb6e4785e858c938e28d132a884b" +checksum = "fe500dc80e757a75e1e8fb7290e448d62dfba3105ece1d058579cb00b58151cd" dependencies = [ "heck 0.5.0", - "prost", + "prost 0.14.3", "prost-build", - "prost-types", + "prost-types 0.14.3", "quote", ] [[package]] name = "prost-wkt-types" -version = "0.6.1" +version = "0.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c8cdde6df0a98311c839392ca2f2f0bcecd545f86a62b4e3c6a49c336e970fe5" +checksum = "13807eaa7e15833d06e899008371926201cdcd11d74b6d490f49130cdb3f415e" dependencies = [ "chrono", - "prost", + "prost 0.14.3", "prost-build", - "prost-types", + "prost-types 0.14.3", "prost-wkt", "prost-wkt-build", "regex", @@ -3835,9 +3883,9 @@ dependencies = [ [[package]] name = "quote" -version = "1.0.40" +version = "1.0.45" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1885c039570dc00dcb4ff087a89e185fd56bae234ddc7f056a945bf36467248d" +checksum = "41f2619966050689382d2b44f664f4bc593e129785a36d6ee376ddf37259b924" dependencies = [ "proc-macro2", ] @@ -3974,9 +4022,9 @@ dependencies = [ [[package]] name = "regex" -version = "1.11.2" +version = "1.12.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "23d7fd106d8c02486a8d64e778353d1cffe08ce79ac2e82f540c86d0facf6912" +checksum = "e10754a14b9137dd7b1e3e5b0493cc9171fdd105e0ab477f51b72e7f3ac0e276" dependencies = [ "aho-corasick", "memchr", @@ -3986,9 +4034,9 @@ dependencies = [ [[package]] name = "regex-automata" -version = "0.4.10" +version = "0.4.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6b9458fa0bfeeac22b5ca447c63aaf45f28439a709ccd244698632f9aa6394d6" +checksum = "6e1dd4122fc1595e8162618945476892eefca7b88c52820e74af6262213cae8f" dependencies = [ "aho-corasick", "memchr", diff --git a/easytier-contrib/easytier-ohrs/Cargo.toml b/easytier-contrib/easytier-ohrs/Cargo.toml index b1d65d9e..6400d2cc 100644 --- a/easytier-contrib/easytier-ohrs/Cargo.toml +++ b/easytier-contrib/easytier-ohrs/Cargo.toml @@ -11,7 +11,6 @@ async-trait = "0.1" base64 = "0.22" flate2 = "1.1" gethostname = "1.1" -ohos-hilog-binding = {version = "*", features = ["redirect"]} easytier = { path = "../../easytier" } napi-derive-ohos = "1.1" napi-ohos = { version = "1.1", default-features = false, features = [ diff --git a/easytier-contrib/easytier-ohrs/src/config/storage/config_meta.rs b/easytier-contrib/easytier-ohrs/src/config/storage/config_meta.rs index bed9fe3f..74d39a9b 100644 --- a/easytier-contrib/easytier-ohrs/src/config/storage/config_meta.rs +++ b/easytier-contrib/easytier-ohrs/src/config/storage/config_meta.rs @@ -1,15 +1,49 @@ use crate::config::types::stored_config::{ SnapshotImportResult, StoredConfigList, StoredConfigMeta, }; -use ohos_hilog_binding::{hilog_debug, hilog_error}; +use once_cell::sync::Lazy; use rusqlite::{Connection, OptionalExtension, params}; +use std::collections::HashSet; +use std::ops::{Deref, DerefMut}; use std::path::{Path, PathBuf}; -use std::sync::Mutex; +use std::sync::{Mutex, MutexGuard}; use std::time::{SystemTime, UNIX_EPOCH}; static CONFIG_DB_PATH: Mutex> = Mutex::new(None); +static CONFIG_DB_CONNECTION: Lazy>> = Lazy::new(|| Mutex::new(None)); const CONFIG_DB_FILE_NAME: &str = "easytier-config-store.db"; +struct CachedConfigDb { + path: PathBuf, + conn: Connection, +} + +pub(crate) struct ConfigDbGuard<'a> { + guard: MutexGuard<'a, Option>, +} + +impl Deref for ConfigDbGuard<'_> { + type Target = Connection; + + fn deref(&self) -> &Self::Target { + &self + .guard + .as_ref() + .expect("config db connection guard must contain a connection") + .conn + } +} + +impl DerefMut for ConfigDbGuard<'_> { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self + .guard + .as_mut() + .expect("config db connection guard must contain a connection") + .conn + } +} + #[derive(Debug, Clone)] struct StoredConfigMetaRecord { config_id: String, @@ -79,31 +113,176 @@ fn init_schema(conn: &Connection) -> rusqlite::Result<()> { ); CREATE INDEX IF NOT EXISTS idx_stored_config_fields_config_id ON stored_config_fields(config_id);", - ) + )?; + + ensure_column( + conn, + "stored_configs", + "favorite", + "ALTER TABLE stored_configs ADD COLUMN favorite INTEGER NOT NULL DEFAULT 0;", + )?; + ensure_column( + conn, + "stored_configs", + "temporary", + "ALTER TABLE stored_configs ADD COLUMN temporary INTEGER NOT NULL DEFAULT 0;", + )?; + ensure_column( + conn, + "stored_config_fields", + "updated_at", + "ALTER TABLE stored_config_fields ADD COLUMN updated_at TEXT NOT NULL DEFAULT '0';", + )?; + + if !validate_store_schema(conn)? { + return Err(rusqlite::Error::InvalidQuery); + } + + conn.execute_batch("PRAGMA user_version = 1;") } -pub(crate) fn open_db() -> Option { - let path = db_file_path()?; - let conn = match Connection::open(&path) { +fn table_columns(conn: &Connection, table_name: &str) -> rusqlite::Result> { + let mut stmt = conn.prepare(&format!("PRAGMA table_info({})", table_name))?; + let rows = stmt.query_map([], |row| row.get::<_, String>(1))?; + let mut columns = HashSet::new(); + for row in rows { + columns.insert(row?); + } + Ok(columns) +} + +fn ensure_column( + conn: &Connection, + table_name: &str, + column_name: &str, + alter_sql: &str, +) -> rusqlite::Result<()> { + let columns = table_columns(conn, table_name)?; + if !columns.contains(column_name) { + conn.execute_batch(alter_sql)?; + } + Ok(()) +} + +fn validate_store_schema(conn: &Connection) -> rusqlite::Result { + let meta_columns = table_columns(conn, "stored_configs")?; + let field_columns = table_columns(conn, "stored_config_fields")?; + let required_meta = [ + "config_id", + "display_name", + "created_at", + "updated_at", + "favorite", + "temporary", + ]; + let required_fields = ["config_id", "field_name", "field_json", "updated_at"]; + + Ok(required_meta + .iter() + .all(|column| meta_columns.contains(*column)) + && required_fields + .iter() + .all(|column| field_columns.contains(*column))) +} + +fn move_db_file_if_exists(path: &Path) -> bool { + if !path.exists() { + return true; + } + let target = PathBuf::from(format!( + "{}.corrupt.{}", + path.to_string_lossy(), + now_ts_string() + )); + match std::fs::rename(path, &target) { + Ok(_) => true, + Err(e) => { + ohrs_log_error!( + "[Rust] failed to move corrupt config db {} to {}: {}", + path.display(), + target.display(), + e + ); + false + } + } +} + +fn recover_config_db_files(path: &Path) -> bool { + let main_ok = move_db_file_if_exists(path); + let wal_ok = move_db_file_if_exists(Path::new(&format!("{}-wal", path.to_string_lossy()))); + let shm_ok = move_db_file_if_exists(Path::new(&format!("{}-shm", path.to_string_lossy()))); + main_ok && wal_ok && shm_ok +} + +fn open_connection(path: &Path) -> Option { + let conn = match Connection::open(path) { Ok(conn) => conn, Err(e) => { - hilog_error!("[Rust] failed to open config db {}: {}", path.display(), e); + ohrs_log_error!("[Rust] failed to open config db {}: {}", path.display(), e); return None; } }; if let Err(e) = init_schema(&conn) { - hilog_error!( + ohrs_log_error!( "[Rust] failed to initialize config db {}: {}", path.display(), e ); - return None; + drop(conn); + if !recover_config_db_files(path) { + return None; + } + + let recovered = match Connection::open(path) { + Ok(conn) => conn, + Err(e) => { + ohrs_log_error!( + "[Rust] failed to open recovered config db {}: {}", + path.display(), + e + ); + return None; + } + }; + if let Err(e) = init_schema(&recovered) { + ohrs_log_error!( + "[Rust] failed to initialize recovered config db {}: {}", + path.display(), + e + ); + return None; + } + return Some(recovered); } Some(conn) } +pub(crate) fn open_db() -> Option> { + let path = db_file_path()?; + let mut guard = match CONFIG_DB_CONNECTION.lock() { + Ok(guard) => guard, + Err(e) => { + ohrs_log_error!("[Rust] failed to lock config db connection: {}", e); + return None; + } + }; + + let should_open = guard + .as_ref() + .map(|cached| cached.path != path || !cached.path.exists()) + .unwrap_or(true); + if should_open { + let conn = open_connection(&path)?; + + *guard = Some(CachedConfigDb { path, conn }); + } + + Some(ConfigDbGuard { guard }) +} + fn row_to_meta(row: &rusqlite::Row<'_>) -> rusqlite::Result { Ok(StoredConfigMetaRecord { config_id: row.get(0)?, @@ -250,7 +429,7 @@ fn ensure_parent_dir(path: &Path) -> bool { Some(parent) => match std::fs::create_dir_all(parent) { Ok(_) => true, Err(e) => { - hilog_error!( + ohrs_log_error!( "[Rust] failed to create snapshot parent {}: {}", parent.display(), e @@ -276,7 +455,7 @@ fn to_meta(record: StoredConfigMetaRecord) -> StoredConfigMeta { pub fn init_config_meta_store(root_dir: String) -> bool { let root = PathBuf::from(root_dir); if let Err(e) = std::fs::create_dir_all(&root) { - hilog_error!( + ohrs_log_error!( "[Rust] failed to create config db dir {}: {}", root.display(), e @@ -290,7 +469,7 @@ pub fn init_config_meta_store(root_dir: String) -> bool { *guard = Some(db_path.clone()); } Err(e) => { - hilog_error!("[Rust] failed to lock config db path: {}", e); + ohrs_log_error!("[Rust] failed to lock config db path: {}", e); return false; } } @@ -299,7 +478,7 @@ pub fn init_config_meta_store(root_dir: String) -> bool { return false; } - hilog_debug!("[Rust] initialized config db at {}", db_path.display()); + ohrs_log_debug!("[Rust] initialized config db at {}", db_path.display()); true } @@ -314,7 +493,7 @@ pub fn export_config_store_snapshot(target_path: String) -> bool { let mut dst = match Connection::open(&target) { Ok(conn) => conn, Err(e) => { - hilog_error!( + ohrs_log_error!( "[Rust] failed to open snapshot target {}: {}", target.display(), e @@ -323,7 +502,7 @@ pub fn export_config_store_snapshot(target_path: String) -> bool { } }; if let Err(e) = init_schema(&dst) { - hilog_error!( + ohrs_log_error!( "[Rust] failed to init snapshot schema {}: {}", target.display(), e @@ -333,7 +512,7 @@ pub fn export_config_store_snapshot(target_path: String) -> bool { match copy_snapshot_tables(&src, &mut dst) { Ok(_) => true, Err(e) => { - hilog_error!( + ohrs_log_error!( "[Rust] failed to export snapshot {}: {}", target.display(), e @@ -348,7 +527,7 @@ pub fn import_config_store_snapshot_with_result(source_path: String) -> Snapshot let src = match Connection::open(&source) { Ok(conn) => conn, Err(e) => { - hilog_error!( + ohrs_log_error!( "[Rust] failed to open snapshot source {}: {}", source.display(), e @@ -357,7 +536,7 @@ pub fn import_config_store_snapshot_with_result(source_path: String) -> Snapshot } }; if !validate_snapshot_schema(&src) { - hilog_error!("[Rust] invalid snapshot schema {}", source.display()); + ohrs_log_error!("[Rust] invalid snapshot schema {}", source.display()); return snapshot_import_err( "invalid_snapshot_schema", format!("invalid snapshot schema: {}", source.display()), @@ -367,7 +546,7 @@ pub fn import_config_store_snapshot_with_result(source_path: String) -> Snapshot let (meta_rows, field_rows) = match read_snapshot_tables(&src) { Ok(rows) => rows, Err(e) => { - hilog_error!( + ohrs_log_error!( "[Rust] failed to read snapshot source {}: {}", source.display(), e @@ -385,7 +564,7 @@ pub fn import_config_store_snapshot_with_result(source_path: String) -> Snapshot match write_snapshot_tables(&mut dst, meta_rows, field_rows) { Ok(_) => snapshot_import_ok(), Err(e) => { - hilog_error!( + ohrs_log_error!( "[Rust] failed to import snapshot {}: {}", source.display(), e @@ -399,6 +578,41 @@ pub fn import_config_store_snapshot(source_path: String) -> bool { import_config_store_snapshot_with_result(source_path).ok } +pub fn reset_config_meta_store() -> bool { + let Some(conn) = open_db() else { + return false; + }; + let tx = match conn.unchecked_transaction() { + Ok(tx) => tx, + Err(e) => { + ohrs_log_error!( + "[Rust] failed to start config store reset transaction: {}", + e + ); + return false; + } + }; + + if let Err(e) = tx.execute("DELETE FROM stored_config_fields", []) { + ohrs_log_error!("[Rust] failed to reset config fields: {}", e); + let _ = tx.rollback(); + return false; + } + if let Err(e) = tx.execute("DELETE FROM stored_configs", []) { + ohrs_log_error!("[Rust] failed to reset config meta: {}", e); + let _ = tx.rollback(); + return false; + } + + match tx.commit() { + Ok(_) => true, + Err(e) => { + ohrs_log_error!("[Rust] failed to commit config store reset: {}", e); + false + } + } +} + pub fn list_config_meta_entries() -> StoredConfigList { let Some(conn) = open_db() else { return StoredConfigList { configs: vec![] }; @@ -411,7 +625,7 @@ pub fn list_config_meta_entries() -> StoredConfigList { ) { Ok(stmt) => stmt, Err(e) => { - hilog_error!("[Rust] failed to prepare list meta query: {}", e); + ohrs_log_error!("[Rust] failed to prepare list meta query: {}", e); return StoredConfigList { configs: vec![] }; } }; @@ -419,7 +633,7 @@ pub fn list_config_meta_entries() -> StoredConfigList { let rows = match stmt.query_map([], row_to_meta) { Ok(rows) => rows, Err(e) => { - hilog_error!("[Rust] failed to list config meta rows: {}", e); + ohrs_log_error!("[Rust] failed to list config meta rows: {}", e); return StoredConfigList { configs: vec![] }; } }; @@ -438,59 +652,6 @@ pub fn get_config_meta(config_id: &str) -> Option { load_meta_record(&conn, config_id).map(to_meta) } -pub fn upsert_config_meta( - config_id: String, - display_name: String, - favorite: bool, - temporary: bool, -) -> StoredConfigMeta { - let now = now_ts_string(); - let Some(conn) = open_db() else { - return StoredConfigMeta { - config_id, - display_name, - created_at: now.clone(), - updated_at: now, - favorite, - temporary, - }; - }; - - let created_at = load_meta_record(&conn, &config_id) - .map(|record| record.created_at) - .unwrap_or_else(|| now.clone()); - - if let Err(e) = conn.execute( - "INSERT INTO stored_configs ( - config_id, display_name, created_at, updated_at, favorite, temporary - ) VALUES (?1, ?2, ?3, ?4, ?5, ?6) - ON CONFLICT(config_id) DO UPDATE SET - display_name = excluded.display_name, - updated_at = excluded.updated_at, - favorite = excluded.favorite, - temporary = excluded.temporary", - params![ - config_id, - display_name, - created_at, - now, - if favorite { 1 } else { 0 }, - if temporary { 1 } else { 0 } - ], - ) { - hilog_error!("[Rust] failed to upsert config meta: {}", e); - } - - get_config_meta(&config_id).unwrap_or(StoredConfigMeta { - config_id, - display_name, - created_at, - updated_at: now, - favorite, - temporary, - }) -} - pub(crate) fn upsert_config_meta_in_tx( tx: &rusqlite::Transaction<'_>, config_id: String, @@ -614,20 +775,3 @@ pub fn set_config_favorite(config_id: String, favorite: bool) -> Option bool { - let Some(conn) = open_db() else { - return false; - }; - - match conn.execute( - "DELETE FROM stored_configs WHERE config_id = ?1", - params![config_id], - ) { - Ok(rows) => rows > 0, - Err(e) => { - hilog_error!("[Rust] failed to delete config meta {}: {}", config_id, e); - false - } - } -} diff --git a/easytier-contrib/easytier-ohrs/src/config/types/stored_config.rs b/easytier-contrib/easytier-ohrs/src/config/types/stored_config.rs index 4b91168e..0ba42541 100644 --- a/easytier-contrib/easytier-ohrs/src/config/types/stored_config.rs +++ b/easytier-contrib/easytier-ohrs/src/config/types/stored_config.rs @@ -35,14 +35,6 @@ pub struct ExportTomlResult { pub toml_text: String, } -#[derive(Debug, Clone, Serialize)] -#[serde(rename_all = "camelCase")] -#[napi(object)] -pub struct StoredConfigSummary { - pub config_id: String, - pub display_name: String, -} - #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] #[napi(object)] diff --git a/easytier-contrib/easytier-ohrs/src/config_repo.rs b/easytier-contrib/easytier-ohrs/src/config_repo.rs index d415c502..aabdfd97 100644 --- a/easytier-contrib/easytier-ohrs/src/config_repo.rs +++ b/easytier-contrib/easytier-ohrs/src/config_repo.rs @@ -1,21 +1,74 @@ use super::{field_store, import_export, legacy_migration, validation}; use crate::config::storage::config_meta::{ - delete_config_meta, get_config_meta, init_config_meta_store, list_config_meta_entries, open_db, - upsert_config_meta_in_tx, + get_config_meta, init_config_meta_store, list_config_meta_entries, open_db, + reset_config_meta_store, upsert_config_meta_in_tx, }; use crate::config::types::stored_config::{ExportTomlResult, StoredConfigRecord}; -use easytier::common::config::ConfigLoader; use easytier::proto::api::manage::NetworkConfig; -use ohos_hilog_binding::{hilog_debug, hilog_error}; +use once_cell::sync::Lazy; use rusqlite::params; use serde_json::Value; +use std::collections::HashMap; use std::path::PathBuf; use std::sync::Mutex; +use std::time::Instant; static CONFIG_ROOT_DIR: Mutex> = Mutex::new(None); +static RUNTIME_CONFIG_SNAPSHOTS: Lazy>> = + Lazy::new(|| Mutex::new(HashMap::new())); pub(crate) const CONFIG_DIR_NAME: &str = "easytier-configs"; pub(crate) const KERNEL_SOCKET_FILE_NAME: &str = "easytier-kernel.sock"; +#[derive(Clone)] +pub(crate) struct RuntimeConfigSnapshot { + pub display_name: String, + pub config: NetworkConfig, +} + +pub(crate) fn cache_runtime_config_snapshot( + config_id: String, + display_name: String, + config: NetworkConfig, +) { + if let Ok(mut guard) = RUNTIME_CONFIG_SNAPSHOTS.lock() { + guard.insert( + config_id, + RuntimeConfigSnapshot { + display_name, + config, + }, + ); + } +} + +pub(crate) fn clear_runtime_config_snapshot(config_id: &str) { + if let Ok(mut guard) = RUNTIME_CONFIG_SNAPSHOTS.lock() { + guard.remove(config_id); + } +} + +pub(crate) fn get_runtime_config_snapshot(config_id: &str) -> Option { + RUNTIME_CONFIG_SNAPSHOTS + .lock() + .ok() + .and_then(|guard| guard.get(config_id).cloned()) +} + +pub(crate) fn get_runtime_config_route_overrides(config_id: &str) -> (Vec, Vec) { + RUNTIME_CONFIG_SNAPSHOTS + .lock() + .ok() + .and_then(|guard| { + guard.get(config_id).map(|snapshot| { + ( + snapshot.config.routes.clone(), + snapshot.config.proxy_cidrs.clone(), + ) + }) + }) + .unwrap_or_default() +} + pub(crate) fn config_root_dir() -> Option { CONFIG_ROOT_DIR .lock() @@ -35,7 +88,7 @@ pub fn init_config_store(root_dir: String) -> bool { let root = PathBuf::from(root_dir); let configs_dir = root.join(CONFIG_DIR_NAME); if let Err(e) = std::fs::create_dir_all(&configs_dir) { - hilog_error!( + ohrs_log_error!( "[Rust] failed to create config dir {}: {}", configs_dir.display(), e @@ -48,7 +101,7 @@ pub fn init_config_store(root_dir: String) -> bool { *guard = Some(root.clone()); } Err(e) => { - hilog_error!("[Rust] failed to lock config root dir: {}", e); + ohrs_log_error!("[Rust] failed to lock config root dir: {}", e); return false; } } @@ -57,13 +110,23 @@ pub fn init_config_store(root_dir: String) -> bool { return false; } - hilog_debug!( + ohrs_log_debug!( "[Rust] initialized config repo at {}", configs_dir.display() ); true } +pub fn reset_config_store() -> bool { + if !reset_config_meta_store() { + return false; + } + if let Ok(mut guard) = RUNTIME_CONFIG_SNAPSHOTS.lock() { + guard.clear(); + } + true +} + fn migrate_legacy_file_if_needed(config_id: &str) -> Option<()> { if validation::validate_config_id(config_id).is_err() { return None; @@ -84,7 +147,7 @@ pub fn save_config_record( let config = match validation::validate_config_json(&config_json, config_id.clone()) { Ok(config) => config, Err(e) => { - hilog_error!("[Rust] save_config_record failed {}", e); + ohrs_log_error!("[Rust] save_config_record failed {}", e); return None; } }; @@ -92,7 +155,7 @@ pub fn save_config_record( let normalized_json = match serde_json::to_string(&config) { Ok(raw) => raw, Err(e) => { - hilog_error!( + ohrs_log_error!( "[Rust] failed to serialize normalized config {}: {}", config_id, e @@ -108,15 +171,15 @@ pub fn save_config_record( let conn = open_db()?; let tx = conn.unchecked_transaction().ok()?; - let existing_meta = get_config_meta(&config_id); - let favorite = existing_meta - .as_ref() - .map(|meta| meta.favorite) - .unwrap_or(false); - let temporary = existing_meta - .as_ref() - .map(|meta| meta.temporary) - .unwrap_or(false); + let existing_meta = tx + .query_row( + "SELECT favorite, temporary FROM stored_configs WHERE config_id = ?1", + params![config_id.clone()], + |row| Ok((row.get::<_, i64>(0)? != 0, row.get::<_, i64>(1)? != 0)), + ) + .ok(); + let favorite = existing_meta.map(|meta| meta.0).unwrap_or(false); + let temporary = existing_meta.map(|meta| meta.1).unwrap_or(false); let meta = upsert_config_meta_in_tx(&tx, config_id.clone(), display_name, favorite, temporary)?; field_store::replace_config_fields(&tx, &config_id, fields)?; @@ -150,16 +213,32 @@ pub fn get_config_record(config_id: &str) -> Option { } pub fn get_config_field_value(config_id: &str, field: &str) -> Option { + let total_start = Instant::now(); validation::validate_config_id(config_id).ok()?; migrate_legacy_file_if_needed(config_id)?; + let open_start = Instant::now(); let conn = open_db()?; - conn.query_row( - "SELECT field_json FROM stored_config_fields + let open_elapsed = open_start.elapsed(); + let query_start = Instant::now(); + let result = conn + .query_row( + "SELECT field_json FROM stored_config_fields WHERE config_id = ?1 AND field_name = ?2", - params![config_id, field], - |row| row.get::<_, String>(0), - ) - .ok() + params![config_id, field], + |row| row.get::<_, String>(0), + ) + .ok(); + ohrs_log_debug!( + "[Rust] get_config_field_value config={} field={} found={} open_ms={} query_ms={} total_ms={} len={}", + config_id, + field, + result.is_some(), + open_elapsed.as_millis(), + query_start.elapsed().as_millis(), + total_start.elapsed().as_millis(), + result.as_ref().map(|value| value.len()).unwrap_or(0) + ); + result } pub fn set_config_field_value(config_id: &str, field: &str, json_value: &str) -> bool { @@ -200,11 +279,6 @@ pub fn set_config_field_value(config_id: &str, field: &str, json_value: &str) -> save_config_record(config_id.to_string(), display_name, normalized).is_some() } -pub fn get_display_name(config_id: &str) -> Option { - validation::validate_config_id(config_id).ok()?; - get_config_meta(config_id).map(|meta| meta.display_name) -} - pub fn get_default_config_json() -> Option { crate::build_default_network_config_json().ok() } @@ -226,7 +300,14 @@ pub fn start_kernel_with_config_id(config_id: &str) -> bool { Some(raw) => raw, None => return false, }; - crate::run_network_instance_from_json(&raw) + let display_name = get_config_meta(config_id) + .map(|meta| meta.display_name) + .unwrap_or_else(|| config_id.to_string()); + let started = crate::run_network_instance_from_json(&raw); + if started && let Ok(config) = serde_json::from_str::(&raw) { + cache_runtime_config_snapshot(config_id.to_string(), display_name, config); + } + started } pub fn list_config_meta_json() -> String { @@ -251,11 +332,20 @@ pub fn delete_config_record(config_id: &str) -> bool { "DELETE FROM stored_config_fields WHERE config_id = ?1", params![config_id], ) { - hilog_error!("[Rust] failed to delete config fields {}: {}", config_id, e); + ohrs_log_error!("[Rust] failed to delete config fields {}: {}", config_id, e); return false; } - delete_config_meta(config_id) + match conn.execute( + "DELETE FROM stored_configs WHERE config_id = ?1", + params![config_id], + ) { + Ok(rows) => rows > 0, + Err(e) => { + ohrs_log_error!("[Rust] failed to delete config meta {}: {}", config_id, e); + false + } + } } pub fn export_config_toml(config_id: &str) -> Option { diff --git a/easytier-contrib/easytier-ohrs/src/config_repo/field_store.rs b/easytier-contrib/easytier-ohrs/src/config_repo/field_store.rs index c954ac02..2cf69ce9 100644 --- a/easytier-contrib/easytier-ohrs/src/config_repo/field_store.rs +++ b/easytier-contrib/easytier-ohrs/src/config_repo/field_store.rs @@ -1,5 +1,4 @@ use crate::config::storage::config_meta::{now_ts_string, open_db}; -use ohos_hilog_binding::hilog_error; use rusqlite::{Connection, params}; use serde_json::{Map, Value}; @@ -43,7 +42,7 @@ pub(super) fn replace_config_fields( "DELETE FROM stored_config_fields WHERE config_id = ?1", params![config_id], ) { - hilog_error!( + ohrs_log_error!( "[Rust] failed to clear existing config fields {}: {}", config_id, e @@ -58,7 +57,7 @@ pub(super) fn replace_config_fields( VALUES (?1, ?2, ?3, ?4)", params![config_id, field_name, field_json, now_ts_string()], ) { - hilog_error!("[Rust] failed to persist config field {}: {}", config_id, e); + ohrs_log_error!("[Rust] failed to persist config field {}: {}", config_id, e); return None; } } diff --git a/easytier-contrib/easytier-ohrs/src/config_repo/legacy_migration.rs b/easytier-contrib/easytier-ohrs/src/config_repo/legacy_migration.rs index 4dbee7c8..c3482f29 100644 --- a/easytier-contrib/easytier-ohrs/src/config_repo/legacy_migration.rs +++ b/easytier-contrib/easytier-ohrs/src/config_repo/legacy_migration.rs @@ -1,5 +1,4 @@ use crate::config::storage::config_meta::get_config_meta; -use ohos_hilog_binding::hilog_error; use std::path::PathBuf; use super::validation; @@ -10,7 +9,7 @@ pub(super) fn legacy_config_file_path( config_id: &str, ) -> Option { if !validation::is_valid_config_id(config_id) { - hilog_error!("[Rust] invalid legacy config_id {}", config_id); + ohrs_log_error!("[Rust] invalid legacy config_id {}", config_id); return None; } root_dir.as_ref().map(|root| { @@ -41,7 +40,7 @@ pub(super) fn migrate_legacy_file_if_needed( save_config_record(config_id.to_string(), display_name, raw)?; if let Err(e) = std::fs::remove_file(&legacy_path) { - hilog_error!( + ohrs_log_error!( "[Rust] failed to remove legacy config file {}: {}", legacy_path.display(), e diff --git a/easytier-contrib/easytier-ohrs/src/exports/config_api.rs b/easytier-contrib/easytier-ohrs/src/exports/config_api.rs index 31d55842..02986cb8 100644 --- a/easytier-contrib/easytier-ohrs/src/exports/config_api.rs +++ b/easytier-contrib/easytier-ohrs/src/exports/config_api.rs @@ -5,6 +5,10 @@ pub(crate) fn init_config_store(root_dir: String) -> bool { config::repository::init_config_store(root_dir) } +pub(crate) fn reset_config_store() -> bool { + config::repository::reset_config_store() +} + pub(crate) fn list_configs() -> String { config::repository::list_config_meta_json() } diff --git a/easytier-contrib/easytier-ohrs/src/exports/runtime_api.rs b/easytier-contrib/easytier-ohrs/src/exports/runtime_api.rs index 6ca437c4..e28368e0 100644 --- a/easytier-contrib/easytier-ohrs/src/exports/runtime_api.rs +++ b/easytier-contrib/easytier-ohrs/src/exports/runtime_api.rs @@ -1,18 +1,15 @@ -use crate::config::repository::load_config_json; -use crate::config::storage::config_meta::get_config_display_name; +use crate::config::repository::{clear_runtime_config_snapshot, get_runtime_config_snapshot}; use crate::config::types::stored_config::KeyValuePair; use crate::kernel_bridge::{ aggregate_requested_tun_routes, start_local_socket_server as start_local_socket_server_inner, stop_local_socket_server as stop_local_socket_server_inner, }; use crate::runtime::state::runtime_state::{ - RuntimeAggregateState, TunAggregateState, clear_tun_attached, mark_tun_attached, + RuntimeAggregateState, RuntimeInstanceState, TunAggregateState, clear_tun_attached, + is_tun_attached, mark_tun_attached, runtime_instance_from_config_snapshot, runtime_instance_from_running_info, }; -use crate::{ASYNC_RUNTIME, EASYTIER_VERSION, INSTANCE_MANAGER, WEB_CLIENTS}; -use easytier::proto::api::manage::NetworkConfig; -use ohos_hilog_binding::{hilog_error, hilog_info}; -use std::sync::Arc; +use crate::{ASYNC_RUNTIME, INSTANCE_MANAGER, WEB_CLIENTS}; pub(crate) fn start_kernel( config_id: String, @@ -29,9 +26,12 @@ pub(crate) fn stop_kernel( ) -> bool { clear_tun_attached(&config_id); if stop_web_client(&config_id) { + clear_runtime_config_snapshot(&config_id); return true; } + let _ = stop_local_socket_server_inner(); + let Some(instance_id) = parse_instance_uuid(&config_id) else { return false; }; @@ -40,9 +40,20 @@ pub(crate) fn stop_kernel( .delete_network_instance(vec![instance_id]) .map(|_| true) .unwrap_or_else(|err| { - hilog_error!("[Rust] stop_kernel failed {}: {}", config_id, err); + ohrs_log_error!("[Rust] stop_kernel failed {}: {}", config_id, err); false }); + if ret { + clear_runtime_config_snapshot(&config_id); + } + let has_active_instances = !INSTANCE_MANAGER.list_network_instance_ids().is_empty(); + let has_web_clients = WEB_CLIENTS + .lock() + .map(|guard| !guard.is_empty()) + .unwrap_or(false); + if has_active_instances || has_web_clients { + let _ = start_local_socket_server_inner(); + } maybe_stop_local_socket_server(); ret } @@ -59,10 +70,10 @@ pub(crate) fn stop_network_instance( } pub(crate) fn collect_network_infos() -> Vec { - let infos = match INSTANCE_MANAGER.collect_network_infos_sync() { + let infos = match ASYNC_RUNTIME.block_on(INSTANCE_MANAGER.collect_network_infos()) { Ok(infos) => infos, Err(err) => { - hilog_error!("[Rust] collect network infos failed {}", err); + ohrs_log_error!("[Rust] collect network infos failed {}", err); return vec![]; } }; @@ -86,7 +97,7 @@ pub(crate) fn set_tun_fd( parse_instance_uuid: impl Fn(&str) -> Option, ) -> bool { let Some(instance_id) = parse_instance_uuid(&config_id) else { - hilog_error!("[Rust] set_tun_fd invalid instance id: {}", config_id); + ohrs_log_error!("[Rust] set_tun_fd invalid instance id: {}", config_id); return false; }; @@ -94,7 +105,7 @@ pub(crate) fn set_tun_fd( .set_tun_fd(&instance_id, fd) .map(|_| { mark_tun_attached(&config_id); - hilog_info!( + ohrs_log_info!( "[Rust] set_tun_fd success instance={} fd={} marked_attached=true", config_id, fd @@ -102,7 +113,7 @@ pub(crate) fn set_tun_fd( true }) .unwrap_or_else(|err| { - hilog_error!("[Rust] set_tun_fd failed {}: {}", config_id, err); + ohrs_log_error!("[Rust] set_tun_fd failed {}: {}", config_id, err); false }) } @@ -112,10 +123,10 @@ pub(crate) fn get_runtime_snapshot() -> RuntimeAggregateState { } pub(crate) fn get_runtime_snapshot_inner() -> RuntimeAggregateState { - let infos = match INSTANCE_MANAGER.collect_network_infos_sync() { + let infos = match ASYNC_RUNTIME.block_on(INSTANCE_MANAGER.collect_network_infos()) { Ok(infos) => infos, Err(err) => { - hilog_error!("[Rust] collect network infos failed {}", err); + ohrs_log_error!("[Rust] collect network infos failed {}", err); return RuntimeAggregateState { instances: vec![], tun: TunAggregateState { @@ -129,30 +140,67 @@ pub(crate) fn get_runtime_snapshot_inner() -> RuntimeAggregateState { }; } }; + let mut live_infos = infos + .into_iter() + .map(|(instance_id, info)| (instance_id.to_string(), info)) + .collect::>(); + let mut active_config_ids = live_infos.keys().cloned().collect::>(); + if let Ok(guard) = WEB_CLIENTS.lock() { + for config_id in guard.keys() { + if !active_config_ids.iter().any(|value| value == config_id) { + active_config_ids.push(config_id.clone()); + } + } + } - let mut instances = Vec::with_capacity(infos.len()); - for (instance_uuid, info) in infos { - let config_id = instance_uuid.to_string(); - let display_name = get_config_display_name(&config_id).unwrap_or_else(|| config_id.clone()); - let config_json = load_config_json(&config_id); - let stored_config = config_json - .as_deref() - .and_then(|raw| serde_json::from_str::(raw).ok()); - let magic_dns_enabled = stored_config - .as_ref() - .and_then(|cfg| cfg.enable_magic_dns) - .unwrap_or(false); - let need_exit_node = stored_config - .as_ref() - .map(|cfg| !cfg.exit_nodes.is_empty()) - .unwrap_or(false); - instances.push(runtime_instance_from_running_info( - config_id, - display_name, - magic_dns_enabled, - need_exit_node, - info, - )); + let mut instances = Vec::with_capacity(active_config_ids.len()); + for config_id in active_config_ids { + if let Some(info) = live_infos.remove(&config_id) { + let snapshot = get_runtime_config_snapshot(&config_id); + let display_name = snapshot + .as_ref() + .map(|snapshot| snapshot.display_name.clone()) + .unwrap_or_else(|| config_id.clone()); + let magic_dns_enabled = snapshot + .as_ref() + .and_then(|snapshot| snapshot.config.enable_magic_dns) + .unwrap_or(false); + let need_exit_node = snapshot + .as_ref() + .map(|snapshot| !snapshot.config.exit_nodes.is_empty()) + .unwrap_or(false); + instances.push(runtime_instance_from_running_info( + config_id, + display_name, + magic_dns_enabled, + need_exit_node, + info, + )); + } else if let Some(snapshot) = get_runtime_config_snapshot(&config_id) { + instances.push(runtime_instance_from_config_snapshot( + config_id, + snapshot.display_name, + snapshot.config, + true, + )); + } else { + let tun_attached = is_tun_attached(&config_id); + instances.push(RuntimeInstanceState { + config_id: config_id.clone(), + instance_id: config_id.clone(), + display_name: config_id.clone(), + running: true, + tun_required: tun_attached, + tun_attached, + magic_dns_enabled: false, + need_exit_node: false, + error_message: None, + my_node_info: None, + events: Vec::new(), + routes: Vec::new(), + peers: Vec::new(), + }); + } } instances.sort_by(|a, b| { diff --git a/easytier-contrib/easytier-ohrs/src/kernel_bridge.rs b/easytier-contrib/easytier-ohrs/src/kernel_bridge.rs index f01868b0..c415e930 100644 --- a/easytier-contrib/easytier-ohrs/src/kernel_bridge.rs +++ b/easytier-contrib/easytier-ohrs/src/kernel_bridge.rs @@ -3,4 +3,6 @@ mod routing; mod socket_server; pub(crate) use routing::aggregate_requested_tun_routes; -pub use socket_server::{start_local_socket_server, stop_local_socket_server}; +pub use socket_server::{ + set_snapshot_broadcast_enabled, start_local_socket_server, stop_local_socket_server, +}; diff --git a/easytier-contrib/easytier-ohrs/src/kernel_bridge/protocol.rs b/easytier-contrib/easytier-ohrs/src/kernel_bridge/protocol.rs index 13cc6aed..5d503232 100644 --- a/easytier-contrib/easytier-ohrs/src/kernel_bridge/protocol.rs +++ b/easytier-contrib/easytier-ohrs/src/kernel_bridge/protocol.rs @@ -48,3 +48,37 @@ pub(crate) fn broadcast_local_socket_message( *clients = active_clients; delivered } + +pub(crate) fn send_local_socket_json_payload_message( + stream: &mut UnixStream, + message_type: &str, + payload_json: &str, +) -> std::io::Result<()> { + let message_type_json = serde_json::to_string(message_type) + .map_err(|err| Error::new(ErrorKind::InvalidData, err.to_string()))?; + let mut raw = Vec::with_capacity(message_type_json.len() + payload_json.len() + 38); + raw.extend_from_slice(b"{\"messageType\":"); + raw.extend_from_slice(message_type_json.as_bytes()); + raw.extend_from_slice(b",\"payloadJson\":"); + raw.extend_from_slice(payload_json.as_bytes()); + raw.extend_from_slice(b"}\n"); + stream.write_all(&raw)?; + Ok(()) +} + +pub(crate) fn broadcast_local_socket_json_payload_message( + clients: &mut Vec, + message_type: &str, + payload_json: &str, +) -> bool { + let mut active_clients = Vec::with_capacity(clients.len()); + let mut delivered = false; + for mut client in clients.drain(..) { + if send_local_socket_json_payload_message(&mut client, message_type, payload_json).is_ok() { + delivered = true; + active_clients.push(client); + } + } + *clients = active_clients; + delivered +} diff --git a/easytier-contrib/easytier-ohrs/src/kernel_bridge/routing.rs b/easytier-contrib/easytier-ohrs/src/kernel_bridge/routing.rs index 52a1649f..e910d6fa 100644 --- a/easytier-contrib/easytier-ohrs/src/kernel_bridge/routing.rs +++ b/easytier-contrib/easytier-ohrs/src/kernel_bridge/routing.rs @@ -1,20 +1,12 @@ -use crate::config::repository::load_config_json; +use crate::config::repository::get_runtime_config_route_overrides; use crate::runtime::state::runtime_state::RuntimeInstanceState; -use easytier::proto::api::manage::NetworkConfig; use ipnet::IpNet; -use ohos_hilog_binding::hilog_debug; use std::collections::HashSet; use std::net::IpAddr; -pub(crate) fn load_manual_routes(config_id: &str) -> Vec { - load_config_json(config_id) - .and_then(|raw| serde_json::from_str::(&raw).ok()) - .map(|config| config.routes) - .unwrap_or_default() -} - fn normalize_route_cidr(route: &str) -> Option { - route + let normalized = route.split("->").next().unwrap_or(route).trim(); + normalized .parse::() .ok() .map(|network| match network { @@ -22,7 +14,7 @@ fn normalize_route_cidr(route: &str) -> Option { IpNet::V6(net) => net.trunc().to_string(), }) .or_else(|| { - route.parse::().ok().map(|addr| match addr { + normalized.parse::().ok().map(|addr| match addr { IpAddr::V4(ip) => format!("{}/32", ip), IpAddr::V6(ip) => format!("{}/128", ip), }) @@ -67,8 +59,9 @@ pub(crate) fn aggregate_tun_routes(instance: &RuntimeInstanceState) -> Vec Vec Vec { diff --git a/easytier-contrib/easytier-ohrs/src/kernel_bridge/socket_server.rs b/easytier-contrib/easytier-ohrs/src/kernel_bridge/socket_server.rs index f91372cc..a992901c 100644 --- a/easytier-contrib/easytier-ohrs/src/kernel_bridge/socket_server.rs +++ b/easytier-contrib/easytier-ohrs/src/kernel_bridge/socket_server.rs @@ -1,8 +1,11 @@ -use super::protocol::{TunRequestPayload, broadcast_local_socket_message}; +use super::protocol::{ + TunRequestPayload, broadcast_local_socket_json_payload_message, broadcast_local_socket_message, +}; +use crate::INSTANCE_MANAGER; use crate::config::repository::kernel_socket_path; use crate::get_runtime_snapshot_inner; use crate::kernel_bridge::routing::aggregate_tun_routes; -use ohos_hilog_binding::{hilog_error, hilog_info}; +use easytier::common::global_ctx::{EventBusSubscriber, GlobalCtxEvent}; use once_cell::sync::Lazy; use std::collections::{HashMap, HashSet}; use std::io::ErrorKind; @@ -11,7 +14,7 @@ use std::path::PathBuf; use std::sync::Mutex; use std::sync::atomic::{AtomicBool, Ordering}; use std::thread::{self, JoinHandle}; -use std::time::Duration; +use std::time::{Duration, Instant}; struct LocalSocketState { stop_flag: std::sync::Arc, @@ -20,12 +23,77 @@ struct LocalSocketState { } static LOCAL_SOCKET_STATE: Lazy>> = Lazy::new(|| Mutex::new(None)); +static SNAPSHOT_BROADCAST_ENABLED: AtomicBool = AtomicBool::new(true); +const SOCKET_TICK_INTERVAL: Duration = Duration::from_millis(250); +const TUN_FAST_CHECK_WINDOW: Duration = Duration::from_secs(8); +const EVENT_RECEIVER_SYNC_INTERVAL: Duration = Duration::from_secs(1); + +pub fn set_snapshot_broadcast_enabled(enabled: bool) { + SNAPSHOT_BROADCAST_ENABLED.store(enabled, Ordering::Relaxed); +} + +fn sync_tun_event_receivers(receivers: &mut HashMap) { + let mut active_instance_ids = HashSet::new(); + for instance in INSTANCE_MANAGER.iter() { + let instance_id = instance.key().to_string(); + active_instance_ids.insert(instance_id.clone()); + if !receivers.contains_key(&instance_id) + && let Some(receiver) = instance.value().subscribe_event() + { + receivers.insert(instance_id, receiver); + } + } + receivers.retain(|instance_id, _| active_instance_ids.contains(instance_id)); +} + +fn event_needs_tun_refresh(event: &GlobalCtxEvent) -> bool { + matches!( + event, + GlobalCtxEvent::DhcpIpv4Changed(_, _) + | GlobalCtxEvent::DhcpIpv4Conflicted(_) + | GlobalCtxEvent::PublicIpv6Changed(_, _) + | GlobalCtxEvent::PublicIpv6RoutesUpdated(_, _) + | GlobalCtxEvent::ProxyCidrsUpdated(_, _) + | GlobalCtxEvent::ConfigPatched(_) + | GlobalCtxEvent::PeerAdded(_) + | GlobalCtxEvent::PeerRemoved(_) + | GlobalCtxEvent::PeerConnAdded(_) + | GlobalCtxEvent::PeerConnRemoved(_) + ) +} + +fn drain_tun_refresh_events(receivers: &mut HashMap) -> bool { + let mut refresh_needed = false; + let mut closed_receivers = Vec::new(); + for (instance_id, receiver) in receivers.iter_mut() { + loop { + match receiver.try_recv() { + Ok(event) => { + refresh_needed = event_needs_tun_refresh(&event) || refresh_needed; + } + Err(tokio::sync::broadcast::error::TryRecvError::Empty) => break, + Err(tokio::sync::broadcast::error::TryRecvError::Lagged(_)) => { + refresh_needed = true; + continue; + } + Err(tokio::sync::broadcast::error::TryRecvError::Closed) => { + closed_receivers.push(instance_id.clone()); + break; + } + } + } + } + for instance_id in closed_receivers { + receivers.remove(&instance_id); + } + refresh_needed +} pub fn start_local_socket_server() -> bool { let socket_path = match kernel_socket_path() { Some(path) => path, None => { - hilog_error!("[Rust] kernel socket path unavailable"); + ohrs_log_error!("[Rust] kernel socket path unavailable"); return false; } }; @@ -34,7 +102,7 @@ pub fn start_local_socket_server() -> bool { Ok(guard) if guard.is_some() => return true, Ok(_) => {} Err(err) => { - hilog_error!("[Rust] lock localsocket state failed: {}", err); + ohrs_log_error!("[Rust] lock localsocket state failed: {}", err); return false; } } @@ -46,7 +114,7 @@ pub fn start_local_socket_server() -> bool { let listener = match UnixListener::bind(&socket_path) { Ok(listener) => listener, Err(err) => { - hilog_error!( + ohrs_log_error!( "[Rust] bind localsocket failed {}: {}", socket_path.display(), err @@ -55,7 +123,7 @@ pub fn start_local_socket_server() -> bool { } }; if let Err(err) = listener.set_nonblocking(true) { - hilog_error!("[Rust] set localsocket nonblocking failed: {}", err); + ohrs_log_error!("[Rust] set localsocket nonblocking failed: {}", err); let _ = std::fs::remove_file(&socket_path); return false; } @@ -66,6 +134,10 @@ pub fn start_local_socket_server() -> bool { let mut last_snapshot_json = String::new(); let mut delivered_tun_requests = HashSet::new(); let mut last_tun_route_signatures = HashMap::::new(); + let mut tun_fast_until = Instant::now() + TUN_FAST_CHECK_WINDOW; + let mut tun_bootstrap_done = false; + let mut last_event_receiver_sync_at: Option = None; + let mut tun_event_receivers = HashMap::::new(); let mut clients = Vec::::new(); while !worker_stop_flag.load(Ordering::Relaxed) { @@ -75,36 +147,86 @@ pub fn start_local_socket_server() -> bool { Ok((stream, _addr)) => { accepted_client = true; clients.push(stream); + tun_fast_until = Instant::now() + TUN_FAST_CHECK_WINDOW; + tun_bootstrap_done = false; } Err(err) if err.kind() == ErrorKind::WouldBlock => break, Err(err) => { - hilog_error!("[Rust] accept localsocket failed: {}", err); + ohrs_log_error!("[Rust] accept localsocket failed: {}", err); break; } } } - let snapshot = get_runtime_snapshot_inner(); - let snapshot_json = match serde_json::to_string(&snapshot) { - Ok(json) => json, - Err(err) => { - hilog_error!("[Rust] serialize runtime snapshot failed: {}", err); - thread::sleep(Duration::from_millis(250)); - continue; + let snapshot_enabled = SNAPSHOT_BROADCAST_ENABLED.load(Ordering::Relaxed); + if clients.is_empty() { + if !last_snapshot_json.is_empty() { + last_snapshot_json.clear(); } - }; - - if accepted_client || snapshot_json != last_snapshot_json { - let _ = broadcast_local_socket_message( - &mut clients, - "runtime_snapshot", - &snapshot_json, - ); - last_snapshot_json = snapshot_json; + delivered_tun_requests.clear(); + last_tun_route_signatures.clear(); + tun_event_receivers.clear(); + last_event_receiver_sync_at = None; + tun_bootstrap_done = false; + thread::sleep(SOCKET_TICK_INTERVAL); + continue; } + let now = Instant::now(); + let should_sync_event_receivers = accepted_client + || last_event_receiver_sync_at + .map(|last| now.duration_since(last) >= EVENT_RECEIVER_SYNC_INTERVAL) + .unwrap_or(true); + if should_sync_event_receivers { + sync_tun_event_receivers(&mut tun_event_receivers); + last_event_receiver_sync_at = Some(now); + } + if drain_tun_refresh_events(&mut tun_event_receivers) { + tun_bootstrap_done = false; + tun_fast_until = now + TUN_FAST_CHECK_WINDOW; + } + let should_collect_snapshot = snapshot_enabled + || accepted_client + || (!tun_bootstrap_done && now < tun_fast_until); + if !should_collect_snapshot { + if !last_snapshot_json.is_empty() { + last_snapshot_json.clear(); + } + thread::sleep(SOCKET_TICK_INTERVAL); + continue; + } + + let snapshot = get_runtime_snapshot_inner(); + if snapshot_enabled { + let snapshot_json = match serde_json::to_string(&snapshot) { + Ok(json) => json, + Err(err) => { + ohrs_log_error!("[Rust] serialize runtime snapshot failed: {}", err); + thread::sleep(SOCKET_TICK_INTERVAL); + continue; + } + }; + + if accepted_client || snapshot_json != last_snapshot_json { + let _ = broadcast_local_socket_json_payload_message( + &mut clients, + "runtime_snapshot", + &snapshot_json, + ); + last_snapshot_json = snapshot_json; + } + } else if !last_snapshot_json.is_empty() { + last_snapshot_json.clear(); + } + + let mut saw_running_instance = false; + let mut saw_tun_candidate = false; for instance in snapshot.instances.iter() { + if instance.running { + saw_running_instance = true; + } if instance.running && instance.tun_required { + saw_tun_candidate = true; let virtual_ipv4 = instance .my_node_info .as_ref() @@ -120,9 +242,16 @@ pub fn start_local_socket_server() -> bool { continue; } let aggregated_routes = aggregate_tun_routes(instance); - let route_signature = serde_json::to_string(&aggregated_routes) - .unwrap_or_else(|_| "[]".to_string()); - let should_send = !delivered_tun_requests.contains(&instance.instance_id) + let route_signature = serde_json::to_string(&( + &virtual_ipv4, + &virtual_ipv4_cidr, + &aggregated_routes, + instance.magic_dns_enabled, + instance.need_exit_node, + )) + .unwrap_or_else(|_| "[]".to_string()); + let should_send = accepted_client + || !delivered_tun_requests.contains(&instance.instance_id) || last_tun_route_signatures .get(&instance.instance_id) .map(|value| value != &route_signature) @@ -143,7 +272,7 @@ pub fn start_local_socket_server() -> bool { let payload_json = match serde_json::to_string(&payload) { Ok(json) => json, Err(err) => { - hilog_error!("[Rust] serialize tun request failed: {}", err); + ohrs_log_error!("[Rust] serialize tun request failed: {}", err); continue; } }; @@ -157,8 +286,15 @@ pub fn start_local_socket_server() -> bool { last_tun_route_signatures.remove(&instance.instance_id); } } + if !snapshot_enabled + && (!delivered_tun_requests.is_empty() + || (saw_running_instance && !saw_tun_candidate) + || now >= tun_fast_until) + { + tun_bootstrap_done = true; + } - thread::sleep(Duration::from_millis(250)); + thread::sleep(SOCKET_TICK_INTERVAL); } }); @@ -172,7 +308,7 @@ pub fn start_local_socket_server() -> bool { true } Err(err) => { - hilog_error!("[Rust] lock localsocket state failed: {}", err); + ohrs_log_error!("[Rust] lock localsocket state failed: {}", err); false } } @@ -182,7 +318,7 @@ pub fn stop_local_socket_server() -> bool { let state = match LOCAL_SOCKET_STATE.lock() { Ok(mut guard) => guard.take(), Err(err) => { - hilog_error!("[Rust] lock localsocket state failed: {}", err); + ohrs_log_error!("[Rust] lock localsocket state failed: {}", err); return false; } }; diff --git a/easytier-contrib/easytier-ohrs/src/lib.rs b/easytier-contrib/easytier-ohrs/src/lib.rs index a9c3e1ff..ad222865 100644 --- a/easytier-contrib/easytier-ohrs/src/lib.rs +++ b/easytier-contrib/easytier-ohrs/src/lib.rs @@ -1,14 +1,46 @@ +macro_rules! ohrs_log_error { + ($($arg:tt)*) => {{ + if $crate::platform::logging::log_manager::app_log_enabled(5) { + $crate::platform::logging::log_manager::record_app_log( + 5, + "RustOhrs", + &std::format!($($arg)*), + ); + } + }}; +} + +macro_rules! ohrs_log_info { + ($($arg:tt)*) => {{ + if $crate::platform::logging::log_manager::app_log_enabled(4) { + $crate::platform::logging::log_manager::record_app_log( + 4, + "RustOhrs", + &std::format!($($arg)*), + ); + } + }}; +} + +macro_rules! ohrs_log_debug { + ($($arg:tt)*) => {{ + if $crate::platform::logging::log_manager::app_log_enabled(3) { + $crate::platform::logging::log_manager::record_app_log( + 3, + "RustOhrs", + &std::format!($($arg)*), + ); + } + }}; +} + mod config; mod exports; mod kernel_bridge; mod platform; mod runtime; -use config::repository::{ - create_config_record, delete_config_record, export_config_toml, get_config_field_value, - get_default_config_json, import_toml_config, init_config_store as init_repo_store, - list_config_meta_json, save_config_record, set_config_field_value, start_kernel_with_config_id, -}; +use config::repository::{cache_runtime_config_snapshot, start_kernel_with_config_id}; use config::services::schema_service::{ ConfigFieldMapping, NetworkConfigSchema, get_network_config_field_mappings as build_network_config_field_mappings, @@ -31,15 +63,11 @@ use easytier::proto::api::manage::NetworkConfig; use easytier::proto::api::manage::NetworkingMethod; use easytier::web_client::{WebClient, WebClientHooks, run_web_client}; use kernel_bridge::{ - aggregate_requested_tun_routes, start_local_socket_server as start_local_socket_server_inner, + set_snapshot_broadcast_enabled, start_local_socket_server as start_local_socket_server_inner, stop_local_socket_server as stop_local_socket_server_inner, }; use napi_derive_ohos::napi; -use ohos_hilog_binding::{hilog_error, hilog_info}; -use runtime::state::runtime_state::{ - RuntimeAggregateState, TunAggregateState, clear_tun_attached, mark_tun_attached, - runtime_instance_from_running_info, -}; +use runtime::state::runtime_state::RuntimeAggregateState; use std::collections::{HashMap, HashSet}; use std::format; use std::sync::{Arc, Mutex}; @@ -101,7 +129,7 @@ fn stop_web_client(config_id: &str) -> bool { let managed = match WEB_CLIENTS.lock() { Ok(mut guard) => guard.remove(config_id), Err(err) => { - hilog_error!("[Rust] stop_web_client lock failed {}", err); + ohrs_log_error!("[Rust] stop_web_client lock failed {}", err); return false; } }; @@ -127,7 +155,7 @@ fn stop_web_client(config_id: &str) -> bool { .delete_network_instance(tracked_ids) .map(|_| true) .unwrap_or_else(|err| { - hilog_error!( + ohrs_log_error!( "[Rust] stop config server instances failed {}: {}", config_id, err @@ -160,12 +188,12 @@ fn run_config_server_instance(config_id: &str, config: &NetworkConfig) -> bool { .next() .is_some() { - hilog_error!("[Rust] there is a running instance!"); + ohrs_log_error!("[Rust] there is a running instance!"); return false; } let Some(config_server_url) = config.public_server_url.clone() else { - hilog_error!("[Rust] public_server_url missing for config server mode"); + ohrs_log_error!("[Rust] public_server_url missing for config server mode"); return false; }; let hooks = Arc::new(TrackedWebClientHooks::default()); @@ -192,7 +220,7 @@ fn run_config_server_instance(config_id: &str, config: &NetworkConfig) -> bool { let client = match client { Ok(client) => client, Err(err) => { - hilog_error!("[Rust] start config server failed {}", err); + ohrs_log_error!("[Rust] start config server failed {}", err); return false; } }; @@ -209,7 +237,7 @@ fn run_config_server_instance(config_id: &str, config: &NetworkConfig) -> bool { true } Err(err) => { - hilog_error!("[Rust] store config server client failed {}", err); + ohrs_log_error!("[Rust] store config server client failed {}", err); false } } @@ -240,29 +268,33 @@ pub(crate) fn run_network_instance_from_json(cfg_json: &str) -> bool { let config = match serde_json::from_str::(cfg_json) { Ok(cfg) => cfg, Err(e) => { - hilog_error!("[Rust] parse config failed {}", e); + ohrs_log_error!("[Rust] parse config failed {}", e); return false; } }; if is_config_server_config(&config) { let Some(config_id) = config.instance_id.as_deref() else { - hilog_error!("[Rust] config server config missing instance id"); + ohrs_log_error!("[Rust] config server config missing instance id"); return false; }; - return run_config_server_instance(config_id, &config); + let started = run_config_server_instance(config_id, &config); + if started { + cache_runtime_config_snapshot(config_id.to_string(), config_id.to_string(), config); + } + return started; } let cfg = match config.gen_config() { Ok(toml) => toml, Err(e) => { - hilog_error!("[Rust] parse config failed {}", e); + ohrs_log_error!("[Rust] parse config failed {}", e); return false; } }; if !INSTANCE_MANAGER.list_network_instance_ids().is_empty() { - hilog_error!("[Rust] there is a running instance!"); + ohrs_log_error!("[Rust] there is a running instance!"); return false; } @@ -275,14 +307,17 @@ pub(crate) fn run_network_instance_from_json(cfg_json: &str) -> bool { .list_network_instance_ids() .contains(&inst_id) { - hilog_error!("[Rust] instance {} already exists", inst_id); + ohrs_log_error!("[Rust] instance {} already exists", inst_id); return false; } match INSTANCE_MANAGER.run_network_instance(cfg, false, ConfigFileControl::STATIC_CONFIG) { - Ok(_) => true, + Ok(_) => { + cache_runtime_config_snapshot(inst_id.to_string(), inst_id.to_string(), config); + true + } Err(err) => { - hilog_error!("[Rust] start_kernel failed for {}: {}", inst_id, err); + ohrs_log_error!("[Rust] start_kernel failed for {}: {}", inst_id, err); false } } @@ -292,7 +327,7 @@ fn parse_instance_uuid(config_id: &str) -> Option { match Uuid::parse_str(config_id) { Ok(uuid) => Some(uuid), Err(err) => { - hilog_error!("[Rust] invalid config_id {}: {}", config_id, err); + ohrs_log_error!("[Rust] invalid config_id {}: {}", config_id, err); None } } @@ -303,6 +338,11 @@ pub fn init_config_store(root_dir: String) -> bool { exports::config_api::init_config_store(root_dir) } +#[napi] +pub fn reset_config_store() -> bool { + exports::config_api::reset_config_store() +} + #[napi] pub fn list_configs() -> String { exports::config_api::list_configs() @@ -482,6 +522,11 @@ pub fn get_runtime_snapshot() -> RuntimeAggregateState { exports::runtime_api::get_runtime_snapshot() } +#[napi] +pub fn set_kernel_snapshot_enabled(enabled: bool) { + set_snapshot_broadcast_enabled(enabled); +} + pub(crate) fn get_runtime_snapshot_inner() -> RuntimeAggregateState { exports::runtime_api::get_runtime_snapshot_inner() } diff --git a/easytier-contrib/easytier-ohrs/src/platform/logging/log_manager.rs b/easytier-contrib/easytier-ohrs/src/platform/logging/log_manager.rs new file mode 100644 index 00000000..e9ab2cde --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/platform/logging/log_manager.rs @@ -0,0 +1,266 @@ +use napi_derive_ohos::napi; +use once_cell::sync::Lazy; +use std::collections::VecDeque; +use std::fs::{self, OpenOptions}; +use std::io::Write; +use std::path::{Path, PathBuf}; +use std::sync::Mutex; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::{SystemTime, UNIX_EPOCH}; + +const LOG_DIR_NAME: &str = "easytier-logs"; +const LOG_FILE_PREFIX: &str = "easytier-"; +const LOG_FILE_SUFFIX: &str = ".log"; +const MAX_LOG_FILES: usize = 10; +const MAX_MEMORY_LINES: usize = 500; + +#[derive(Clone)] +struct LogOptions { + core_log: bool, + debug_log: bool, +} + +impl Default for LogOptions { + fn default() -> Self { + Self { + core_log: false, + debug_log: false, + } + } +} + +#[derive(Default)] +struct LogManagerState { + log_dir: Option, + active_file: Option, + lines: VecDeque, + options: LogOptions, +} + +static LOG_MANAGER: Lazy> = + Lazy::new(|| Mutex::new(LogManagerState::default())); +static CORE_LOG_ENABLED: AtomicBool = AtomicBool::new(false); +static DEBUG_LOG_ENABLED: AtomicBool = AtomicBool::new(false); + +fn now_millis() -> u128 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|duration| duration.as_millis()) + .unwrap_or(0) +} + +fn sanitize_name(raw: &str) -> String { + let value = raw + .chars() + .map(|ch| { + if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' { + ch + } else { + '-' + } + }) + .collect::(); + if value.is_empty() { + "process".to_string() + } else { + value + } +} + +fn log_dir(root_dir: &str) -> PathBuf { + Path::new(root_dir).join(LOG_DIR_NAME) +} + +fn is_log_file(path: &Path) -> bool { + path.file_name() + .and_then(|name| name.to_str()) + .map(|name| name.starts_with(LOG_FILE_PREFIX) && name.ends_with(LOG_FILE_SUFFIX)) + .unwrap_or(false) +} + +fn sorted_log_files(dir: &Path) -> Vec { + let mut files = fs::read_dir(dir) + .ok() + .into_iter() + .flat_map(|entries| entries.filter_map(|entry| entry.ok())) + .map(|entry| entry.path()) + .filter(|path| is_log_file(path)) + .collect::>(); + files.sort_by(|left, right| { + left.file_name() + .and_then(|name| name.to_str()) + .unwrap_or_default() + .cmp( + right + .file_name() + .and_then(|name| name.to_str()) + .unwrap_or_default(), + ) + }); + files +} + +fn cleanup_old_logs(dir: &Path) { + let files = sorted_log_files(dir); + let overflow = files.len().saturating_sub(MAX_LOG_FILES); + for path in files.into_iter().take(overflow) { + let _ = fs::remove_file(path); + } +} + +fn push_memory_line(state: &mut LogManagerState, line: String) { + state.lines.push_back(line); + while state.lines.len() > MAX_MEMORY_LINES { + state.lines.pop_front(); + } +} + +fn append_log_file(path: &Path, line: &str) { + if let Ok(mut file) = OpenOptions::new().create(true).append(true).open(path) { + let _ = writeln!(file, "{}", line); + } +} + +fn should_record_debug(level: i32) -> bool { + level <= 3 +} + +fn format_line(level: i32, target: &str, message: &str) -> String { + format!("{}[{}] {}", level, target, message.replace('\n', "\\n")) +} + +pub(crate) fn configure(core_log: bool, debug_log: bool) { + CORE_LOG_ENABLED.store(core_log, Ordering::Relaxed); + DEBUG_LOG_ENABLED.store(debug_log, Ordering::Relaxed); + if let Ok(mut guard) = LOG_MANAGER.lock() { + guard.options.core_log = core_log; + guard.options.debug_log = debug_log; + } +} + +pub(crate) fn app_log_enabled(level: i32) -> bool { + !should_record_debug(level) || DEBUG_LOG_ENABLED.load(Ordering::Relaxed) +} + +pub(crate) fn core_log_enabled(level: i32) -> bool { + CORE_LOG_ENABLED.load(Ordering::Relaxed) && app_log_enabled(level) +} + +pub(crate) fn record_app_log(level: i32, target: &str, message: &str) { + if !app_log_enabled(level) { + return; + } + if let Ok(mut guard) = LOG_MANAGER.lock() { + let line = format_line(level, target, message); + if let Some(path) = guard.active_file.as_ref() { + append_log_file(path, &line); + } + push_memory_line(&mut guard, line); + } +} + +pub(crate) fn record_core_log(level: i32, target: &str, message: &str) { + if !core_log_enabled(level) { + return; + } + if let Ok(mut guard) = LOG_MANAGER.lock() { + let line = format_line(level, target, message); + if let Some(path) = guard.active_file.as_ref() { + append_log_file(path, &line); + } + push_memory_line(&mut guard, line); + } +} + +#[napi] +pub fn init_log_manager(root_dir: String, process_name: String) -> bool { + let dir = log_dir(&root_dir); + if fs::create_dir_all(&dir).is_err() { + return false; + } + if LOG_MANAGER + .lock() + .map(|guard| guard.active_file.is_some()) + .unwrap_or(false) + { + cleanup_old_logs(&dir); + return true; + } + + let active_file = dir.join(format!( + "{}{}-{}-{}{}", + LOG_FILE_PREFIX, + now_millis(), + std::process::id(), + sanitize_name(&process_name), + LOG_FILE_SUFFIX + )); + if OpenOptions::new() + .create(true) + .append(true) + .open(&active_file) + .is_err() + { + return false; + } + + if let Ok(mut guard) = LOG_MANAGER.lock() { + guard.log_dir = Some(dir.clone()); + guard.active_file = Some(active_file); + guard.lines.clear(); + } + cleanup_old_logs(&dir); + true +} + +#[napi] +pub fn configure_log_manager(core_log: bool, debug_log: bool) { + configure(core_log, debug_log); +} + +#[napi] +pub fn write_app_log(level: i32, target: String, message: String) { + record_app_log(level, &target, &message); +} + +#[napi] +pub fn drain_log_lines() -> Vec { + LOG_MANAGER + .lock() + .map(|mut guard| guard.lines.drain(..).collect()) + .unwrap_or_default() +} + +#[napi] +pub fn export_log_archive(target_path: String) -> bool { + let log_dir = LOG_MANAGER + .lock() + .ok() + .and_then(|guard| guard.log_dir.clone()); + let Some(log_dir) = log_dir else { + return false; + }; + + let files = sorted_log_files(&log_dir); + let mut output = match OpenOptions::new() + .create(true) + .write(true) + .truncate(true) + .open(&target_path) + { + Ok(file) => file, + Err(_) => return false, + }; + + for path in files { + let name = path + .file_name() + .and_then(|value| value.to_str()) + .unwrap_or("unknown.log"); + let _ = writeln!(output, "===== {} =====", name); + if let Ok(content) = fs::read_to_string(&path) { + let _ = writeln!(output, "{}", content); + } + } + true +} diff --git a/easytier-contrib/easytier-ohrs/src/platform/logging/mod.rs b/easytier-contrib/easytier-ohrs/src/platform/logging/mod.rs index 0b44f5b3..2ab301cb 100644 --- a/easytier-contrib/easytier-ohrs/src/platform/logging/mod.rs +++ b/easytier-contrib/easytier-ohrs/src/platform/logging/mod.rs @@ -1 +1,2 @@ +pub(crate) mod log_manager; pub(crate) mod native_log; diff --git a/easytier-contrib/easytier-ohrs/src/platform/logging/native_log.rs b/easytier-contrib/easytier-ohrs/src/platform/logging/native_log.rs index abfbc611..cc799b1e 100644 --- a/easytier-contrib/easytier-ohrs/src/platform/logging/native_log.rs +++ b/easytier-contrib/easytier-ohrs/src/platform/logging/native_log.rs @@ -1,7 +1,5 @@ +use super::log_manager; use napi_derive_ohos::napi; -use ohos_hilog_binding::{ - LogOptions, hilog_debug, hilog_error, hilog_info, hilog_warn, set_global_options, -}; use std::collections::HashMap; use std::panic; use tracing::{Event, Subscriber}; @@ -10,8 +8,9 @@ use tracing_subscriber::layer::{Context, Layer}; use tracing_subscriber::prelude::*; static INITIALIZED: std::sync::Once = std::sync::Once::new(); +static TRACING_INITIALIZED: std::sync::Once = std::sync::Once::new(); fn panic_hook(info: &panic::PanicHookInfo) { - hilog_error!("RUST PANIC: {}", info); + log_manager::record_core_log(5, "RustPanic", &format!("{}", info)); } #[napi] @@ -23,45 +22,40 @@ pub fn init_panic_hook() { #[napi] pub fn hilog_global_options(domain: u32, tag: String) { - ohos_hilog_binding::forward_stdio_to_hilog(); - set_global_options(LogOptions { - domain, - tag: Box::leak(tag.clone().into_boxed_str()), - }) + let _ = domain; + let _ = tag; } #[napi] pub fn init_tracing_subscriber() { - tracing_subscriber::registry() - .with(CallbackLayer { - callback: Box::new(tracing_callback), - }) - .init(); + TRACING_INITIALIZED.call_once(|| { + let _ = tracing_subscriber::registry() + .with(CallbackLayer { + callback: Box::new(tracing_callback), + }) + .try_init(); + }); } fn tracing_callback(event: &Event, fields: HashMap) { let metadata = event.metadata(); - #[cfg(target_env = "ohos")] - { - let loc = metadata.target().split("::").last().unwrap(); - match *metadata.level() { - Level::TRACE => { - hilog_debug!("[{}] {:?}", loc, fields.values().collect::>()); - } - Level::DEBUG => { - hilog_debug!("[{}] {:?}", loc, fields.values().collect::>()); - } - Level::INFO => { - hilog_info!("[{}] {:?}", loc, fields.values().collect::>()); - } - Level::WARN => { - hilog_warn!("[{}] {:?}", loc, fields.values().collect::>()); - } - Level::ERROR => { - hilog_error!("[{}] {:?}", loc, fields.values().collect::>()); - } - } + let loc = metadata + .target() + .split("::") + .last() + .unwrap_or(metadata.target()); + let level = match *metadata.level() { + Level::TRACE => 2, + Level::DEBUG => 3, + Level::INFO => 4, + Level::WARN => 6, + Level::ERROR => 5, + }; + if !log_manager::core_log_enabled(level) { + return; } + let values = fields.values().cloned().collect::>().join(" "); + log_manager::record_core_log(level, &format!("Rust:{}", loc), &values); } struct CallbackLayer { @@ -70,6 +64,16 @@ struct CallbackLayer { impl Layer for CallbackLayer { fn on_event(&self, event: &Event, _ctx: Context) { + let level = match *event.metadata().level() { + Level::TRACE => 2, + Level::DEBUG => 3, + Level::INFO => 4, + Level::WARN => 6, + Level::ERROR => 5, + }; + if !log_manager::core_log_enabled(level) { + return; + } // 使用 fmt::format::FmtSpan 提取字段值 let mut fields = HashMap::new(); let mut visitor = FieldCollector(&mut fields); diff --git a/easytier-contrib/easytier-ohrs/src/runtime/state/runtime_state.rs b/easytier-contrib/easytier-ohrs/src/runtime/state/runtime_state.rs index 2a1cb059..948b7b59 100644 --- a/easytier-contrib/easytier-ohrs/src/runtime/state/runtime_state.rs +++ b/easytier-contrib/easytier-ohrs/src/runtime/state/runtime_state.rs @@ -3,6 +3,7 @@ use napi_derive_ohos::napi; use serde::Serialize; use std::collections::HashSet; use std::sync::Mutex; +use url::Url; static ATTACHED_TUN_INSTANCE_IDS: once_cell::sync::Lazy>> = once_cell::sync::Lazy::new(|| Mutex::new(HashSet::new())); @@ -158,6 +159,136 @@ fn stringify_uuid(value: Option) -> Option { value.map(|v| v.to_string()) } +fn non_empty_string(value: Option) -> Option { + value.and_then(|raw| { + let trimmed = raw.trim(); + if trimmed.is_empty() { + None + } else { + Some(trimmed.to_string()) + } + }) +} + +fn config_virtual_ipv4_cidr(config: &api::manage::NetworkConfig) -> Option { + non_empty_string(config.virtual_ipv4.clone()) + .map(|ipv4| format!("{}/{}", ipv4, config.network_length.unwrap_or(24))) +} + +fn config_endpoint_urls(config: &api::manage::NetworkConfig) -> Vec { + let mut urls = Vec::new(); + let mut seen = HashSet::new(); + if let Some(url) = non_empty_string(config.public_server_url.clone()) + && seen.insert(url.clone()) + { + urls.push(url); + } + for raw in &config.peer_urls { + let trimmed = raw.trim(); + if trimmed.is_empty() { + continue; + } + let value = trimmed.to_string(); + if seen.insert(value.clone()) { + urls.push(value); + } + } + urls +} + +fn endpoint_url(url: &str) -> Option { + Url::parse(url).ok() +} + +fn endpoint_scheme(url: &str) -> Option { + endpoint_url(url) + .map(|parsed| parsed.scheme().to_string()) + .or_else(|| { + let scheme = url.split("://").next().unwrap_or("").trim(); + (!scheme.is_empty()).then_some(scheme.to_string()) + }) +} + +fn endpoint_label(url: &str) -> String { + if let Some(parsed) = endpoint_url(url) + && let Some(host) = parsed.host_str() + { + return format!("[Config] {}", host); + } + format!("[Config] {}", url) +} + +fn endpoint_remote_display(url: &str) -> String { + if let Some(parsed) = endpoint_url(url) + && let Some(host) = parsed.host_str() + { + return parsed + .port() + .map(|port| format!("{}:{}", host, port)) + .unwrap_or_else(|| host.to_string()); + } + url.to_string() +} + +fn configured_peer_id(index: usize) -> i64 { + 9_000_000 + index as i64 +} + +fn configured_route_views(endpoints: &[String], public_server_url: Option<&str>) -> Vec { + endpoints + .iter() + .enumerate() + .map(|(index, endpoint)| RouteView { + peer_id: configured_peer_id(index), + hostname: Some(endpoint_label(endpoint)), + ipv4: Some(endpoint_remote_display(endpoint)), + ipv4_cidr: None, + ipv6_cidr: None, + proxy_cidrs: Vec::new(), + next_hop_peer_id: None, + cost: Some(0), + path_latency: None, + udp_nat_type: None, + tcp_nat_type: None, + inst_id: None, + version: None, + is_public_server: public_server_url.map(|url| url == endpoint), + }) + .collect() +} + +fn configured_peer_views(endpoints: &[String]) -> Vec { + endpoints + .iter() + .enumerate() + .map(|(index, endpoint)| { + let conn_id = format!("configured-peer-{}", index); + PeerInfo { + peer_id: configured_peer_id(index), + default_conn_id: Some(conn_id.clone()), + directly_connected_conns: vec![conn_id.clone()], + conns: vec![PeerConnInfo { + conn_id, + my_peer_id: 0, + peer_id: configured_peer_id(index), + features: Vec::new(), + tunnel_type: endpoint_scheme(endpoint), + local_addr: None, + remote_addr: Some(endpoint.clone()), + resolved_remote_addr: Some(endpoint_remote_display(endpoint)), + stats: None, + loss_rate: None, + is_client: true, + network_name: None, + is_closed: false, + secure_auth_level: None, + peer_identity_type: None, + }], + } + }) + .collect() +} + fn optional_u32_to_i64(value: Option) -> Option { value.map(|v| v as i64) } @@ -291,3 +422,43 @@ pub fn runtime_instance_from_running_info( peers: info.peers.into_iter().map(peer_to_view).collect(), } } + +pub fn runtime_instance_from_config_snapshot( + config_id: String, + display_name: String, + config: api::manage::NetworkConfig, + running: bool, +) -> RuntimeInstanceState { + let tun_attached = running && is_tun_attached(&config_id); + let tun_required = + running && (config.dev_name.as_deref().unwrap_or("") != "no_tun" || tun_attached); + let endpoint_urls = config_endpoint_urls(&config); + let public_server_url = non_empty_string(config.public_server_url.clone()); + let my_node_info = MyNodeInfo { + virtual_ipv4: non_empty_string(config.virtual_ipv4.clone()), + virtual_ipv4_cidr: config_virtual_ipv4_cidr(&config), + hostname: non_empty_string(config.hostname.clone()), + version: None, + peer_id: None, + listeners: config.listener_urls.clone(), + vpn_portal_cfg: None, + udp_nat_type: None, + tcp_nat_type: None, + }; + + RuntimeInstanceState { + config_id: config_id.clone(), + instance_id: config_id, + display_name, + running, + tun_required, + tun_attached, + magic_dns_enabled: config.enable_magic_dns.unwrap_or(false), + need_exit_node: !config.exit_nodes.is_empty(), + error_message: None, + my_node_info: Some(my_node_info), + events: Vec::new(), + routes: configured_route_views(&endpoint_urls, public_server_url.as_deref()), + peers: configured_peer_views(&endpoint_urls), + } +}