diff --git a/Cargo.lock b/Cargo.lock index cb0ea66..cac6fdb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -17,6 +17,15 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" +[[package]] +name = "aho-corasick" +version = "1.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e60d3430d3a69478ad0993f19238d2df97c507009a52b3c10addcd7f6bcb916" +dependencies = [ + "memchr", +] + [[package]] name = "anstream" version = "0.6.20" @@ -517,6 +526,7 @@ dependencies = [ "thiserror 1.0.69", "tokio", "tracing", + "tracing-subscriber", ] [[package]] @@ -904,6 +914,12 @@ dependencies = [ "thiserror 2.0.16", ] +[[package]] +name = "lazy_static" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" + [[package]] name = "libc" version = "0.2.175" @@ -944,6 +960,15 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" +[[package]] +name = "matchers" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8263075bb86c5a1b1427b5ae862e8889656f126e9f77c484496e8b47cf5c5558" +dependencies = [ + "regex-automata 0.1.10", +] + [[package]] name = "memchr" version = "2.7.5" @@ -993,6 +1018,16 @@ dependencies = [ "tempfile", ] +[[package]] +name = "nu-ansi-term" +version = "0.46.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77a8165726e8236064dbb45459242600304b42a5ea24ee2948e18e023bf7ba84" +dependencies = [ + "overload", + "winapi", +] + [[package]] name = "num-bigint" version = "0.4.6" @@ -1086,6 +1121,12 @@ dependencies = [ "vcpkg", ] +[[package]] +name = "overload" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b15813163c1d831bf4a13c3610c05c0d03b39feb07f7e09fa234dac9b15aaf39" + [[package]] name = "parking_lot" version = "0.12.4" @@ -1352,6 +1393,50 @@ dependencies = [ "bitflags", ] +[[package]] +name = "regex" +version = "1.11.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23d7fd106d8c02486a8d64e778353d1cffe08ce79ac2e82f540c86d0facf6912" +dependencies = [ + "aho-corasick", + "memchr", + "regex-automata 0.4.10", + "regex-syntax 0.8.6", +] + +[[package]] +name = "regex-automata" +version = "0.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c230d73fb8d8c1b9c0b3135c5142a8acee3a0558fb8db5cf1cb65f8d7862132" +dependencies = [ + "regex-syntax 0.6.29", +] + +[[package]] +name = "regex-automata" +version = "0.4.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b9458fa0bfeeac22b5ca447c63aaf45f28439a709ccd244698632f9aa6394d6" +dependencies = [ + "aho-corasick", + "memchr", + "regex-syntax 0.8.6", +] + +[[package]] +name = "regex-syntax" +version = "0.6.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f162c6dd7b008981e4d40210aca20b4bd0f9b60ca9271061b07f78537722f2e1" + +[[package]] +name = "regex-syntax" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "caf4aa5b0f434c91fe5c7f1ecb6a5ece2130b02ad2a590589dda5146df959001" + [[package]] name = "reqwest" version = "0.12.23" @@ -1587,6 +1672,15 @@ version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbfa15b3dddfee50a0fff136974b3e1bde555604ba463834a7eb7deb6417705d" +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + [[package]] name = "shlex" version = "1.3.0" @@ -1773,6 +1867,15 @@ dependencies = [ "syn", ] +[[package]] +name = "thread_local" +version = "1.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f60246a4944f24f6e018aa17cdeffb7818b76356965d03b07d6a9886e8962185" +dependencies = [ + "cfg-if", +] + [[package]] name = "tinystr" version = "0.8.1" @@ -1966,6 +2069,36 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b9d12581f227e93f094d3af2ae690a574abb8a2b9b7a96e7cfe9647b2b617678" dependencies = [ "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8189decb5ac0fa7bc8b96b7cb9b2701d60d48805aca84a238004d665fcc4008" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex", + "sharded-slab", + "smallvec", + "thread_local", + "tracing", + "tracing-core", + "tracing-log", ] [[package]] @@ -2015,6 +2148,12 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + [[package]] name = "vcpkg" version = "0.2.15" @@ -2151,6 +2290,28 @@ dependencies = [ "rustls-pki-types", ] +[[package]] +name = "winapi" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" +dependencies = [ + "winapi-i686-pc-windows-gnu", + "winapi-x86_64-pc-windows-gnu", +] + +[[package]] +name = "winapi-i686-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" + +[[package]] +name = "winapi-x86_64-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + [[package]] name = "windows-link" version = "0.1.3" diff --git a/Cargo.toml b/Cargo.toml index 4e18fb0..8db9291 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -22,3 +22,4 @@ reqwest = { version = "0.12.7", features = ["json", "rustls-tls"] } base64 = "0.22.1" # Error derive for clean error types thiserror = "1.0.64" +tracing-subscriber = { version = "0.3.18", features = ["env-filter", "fmt"] } diff --git a/src/main.rs b/src/main.rs index fb4f3f6..90bb073 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2,6 +2,8 @@ use clap::Parser; use std::net::{IpAddr, SocketAddr}; use std::sync::Arc; +use tracing::{info, warn, error}; +use tracing_subscriber::{fmt, EnvFilter}; #[derive(Debug, Clone, Parser)] #[command( name = "herocoordinator", @@ -73,6 +75,14 @@ struct Cli { #[tokio::main] async fn main() { let cli = Cli::parse(); +// Initialize tracing subscriber (pretty formatter; controlled by RUST_LOG) +let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); +tracing_subscriber::fmt() + .with_env_filter(filter) + .pretty() + .with_target(true) + .with_level(true) + .init(); let http_addr = SocketAddr::new(cli.api_http_ip, cli.api_http_port); let ws_addr = SocketAddr::new(cli.api_ws_ip, cli.api_ws_port); @@ -107,10 +117,7 @@ async fn main() { let http_module = herocoordinator::rpc::build_module(state.clone()); let ws_module = herocoordinator::rpc::build_module(state.clone()); - println!( - "Starting JSON-RPC servers: HTTP http://{} | WS ws://{} | redis_addr={}", - http_addr, ws_addr, cli.redis_addr - ); + info!(%http_addr, %ws_addr, redis_addr=%cli.redis_addr, "Starting JSON-RPC servers"); // Start servers let _http_handle = herocoordinator::rpc::start_http(http_addr, http_module) @@ -122,7 +129,7 @@ async fn main() { // Wait for Ctrl+C to terminate if let Err(e) = tokio::signal::ctrl_c().await { - eprintln!("Failed to listen for shutdown signal: {e}"); + error!(error=%e, "Failed to listen for shutdown signal"); } - println!("Shutdown signal received, exiting."); + info!("Shutdown signal received, exiting."); } diff --git a/src/router.rs b/src/router.rs index 7a671a9..aa7a6f2 100644 --- a/src/router.rs +++ b/src/router.rs @@ -3,6 +3,7 @@ use std::{collections::HashSet, sync::Arc}; use serde_json::{Value, json}; use tokio::sync::Semaphore; +use tracing::{info, warn, error}; use crate::{ clients::{Destination, SupervisorClient, MyceliumClient}, models::{Job, JobStatus, Message, MessageStatus, ScriptType, TransportStatus}, @@ -40,7 +41,7 @@ pub fn start_router(service: AppService, cfg: RouterConfig) -> Vec break Arc::new(c), Err(e) => { - eprintln!("[router ctx={}] MyceliumClient init error: {}", ctx_id, e); + error!(context_id=ctx_id, error=%e, "MyceliumClient init error"); tokio::time::sleep(std::time::Duration::from_secs(1)).await; } } @@ -72,10 +73,7 @@ pub fn start_router(service: AppService, cfg: RouterConfig) -> Vec Vec { - eprintln!("[router ctx={}] brpop error: {}", ctx_id, e); + error!(context_id=ctx_id, error=%e, "BRPOP error"); // small backoff to avoid busy-loop on persistent errors tokio::time::sleep(std::time::Duration::from_millis(200)).await; } @@ -454,12 +452,12 @@ pub fn start_router_auto(service: AppService, cfg: RouterConfig) -> tokio::task: }; let _ = start_router(service.clone(), cfg_ctx); active.insert(ctx_id); - eprintln!("[router] started loop for context {}", ctx_id); + info!(context_id=ctx_id, "Started loop for context"); } } } Err(e) => { - eprintln!("[router] list_context_ids error: {}", e); + error!(error=%e, "list_context_ids error"); } } tokio::time::sleep(std::time::Duration::from_secs(5)).await; diff --git a/src/storage/redis.rs b/src/storage/redis.rs index 0e99dbb..4e2a5f0 100644 --- a/src/storage/redis.rs +++ b/src/storage/redis.rs @@ -6,6 +6,7 @@ use serde::de::DeserializeOwned; use serde_json::{Map as JsonMap, Value}; use tokio::sync::Mutex; +use tracing::{error, warn, info, debug, trace}; use crate::models::{ Actor, Context, Flow, FlowStatus, Job, JobStatus, Message, MessageStatus, Runner, TransportStatus, }; @@ -52,8 +53,14 @@ impl RedisDriver { // Slow path: create a new manager and cache it let url = format!("{}/{}", self.base_addr.trim_end_matches('/'), db); - let client = redis::Client::open(url.as_str())?; - let cm = client.get_connection_manager().await?; + let client = redis::Client::open(url.as_str()).map_err(|e| { + error!(%url, db=%db, error=%e, "Redis client open failed"); + e + })?; + let cm = client.get_connection_manager().await.map_err(|e| { + error!(%url, db=%db, error=%e, "Redis connection manager init failed"); + e + })?; let mut guard = self.managers.lock().await; let entry = guard.entry(db).or_insert(cm); @@ -104,21 +111,37 @@ impl RedisDriver { async fn hset_model(&self, db: u32, key: &str, model: &T) -> Result<()> { let mut cm = self.manager_for_db(db).await?; - let pairs = Self::struct_to_hset_pairs(model)?; + let pairs = Self::struct_to_hset_pairs(model).map_err(|e| { + error!(db=%db, key=%key, error=%e, "Serialize model to HSET pairs failed"); + e + })?; // Ensure no stale fields - let _: u64 = cm.del(key).await.unwrap_or(0); + let del_res: redis::RedisResult = cm.del(key).await; + if let Err(e) = del_res { + warn!(db=%db, key=%key, error=%e, "DEL before HSET failed"); + } // Write all fields - let _: usize = cm.hset_multiple(key, &pairs).await?; + let _: usize = cm.hset_multiple(key, &pairs).await.map_err(|e| { + error!(db=%db, key=%key, error=%e, "HSET multiple failed"); + e + })?; Ok(()) } async fn hget_model(&self, db: u32, key: &str) -> Result { let mut cm = self.manager_for_db(db).await?; - let map: StdHashMap = cm.hgetall(key).await?; + let map: StdHashMap = cm.hgetall(key).await.map_err(|e| { + error!(db=%db, key=%key, error=%e, "HGETALL failed"); + e + })?; if map.is_empty() { + // NotFound is expected in some flows; don't log as error return Err(format!("Key not found: {}", key).into()); } - Self::hmap_to_struct(map) + Self::hmap_to_struct(map).map_err(|e| { + error!(db=%db, key=%key, error=%e, "Deserialize model from HGETALL failed"); + e + }) } // ----------------------------- @@ -299,7 +322,10 @@ impl RedisDriver { ("status".to_string(), status_str), ("updated_at".to_string(), ts.to_string()), ]; - let _: usize = cm.hset_multiple(key, &pairs).await?; + let _: usize = cm.hset_multiple(&key, &pairs).await.map_err(|e| { + error!(db=%db, key=%key, error=%e, "HSET update_job_status failed"); + e + })?; Ok(()) } @@ -345,7 +371,10 @@ impl RedisDriver { ("status".to_string(), status_str), ("updated_at".to_string(), ts.to_string()), ]; - let _: usize = cm.hset_multiple(key, &pairs).await?; + let _: usize = cm.hset_multiple(&key, &pairs).await.map_err(|e| { + error!(db=%db, key=%key, error=%e, "HSET update_flow_status failed"); + e + })?; Ok(()) } @@ -370,7 +399,10 @@ impl RedisDriver { ("status".to_string(), status_str), ("updated_at".to_string(), ts.to_string()), ]; - let _: usize = cm.hset_multiple(key, &pairs).await?; + let _: usize = cm.hset_multiple(&key, &pairs).await.map_err(|e| { + error!(db=%db, key=%key, error=%e, "HSET update_message_status failed"); + e + })?; Ok(()) } @@ -404,7 +436,10 @@ impl RedisDriver { let ts = crate::time::current_timestamp(); pairs.push(("updated_at".to_string(), ts.to_string())); - let _: usize = cm.hset_multiple(key, &pairs).await?; + let _: usize = cm.hset_multiple(&key, &pairs).await.map_err(|e| { + error!(db=%db, key=%key, error=%e, "HSET update_message_transport failed"); + e + })?; Ok(()) } @@ -437,7 +472,10 @@ impl RedisDriver { ("env_vars".to_string(), env_vars_str), ("updated_at".to_string(), ts.to_string()), ]; - let _: usize = cm.hset_multiple(key, &pairs).await?; + let _: usize = cm.hset_multiple(&key, &pairs).await.map_err(|e| { + error!(db=%db, key=%key, error=%e, "HSET update_flow_env_vars_merge failed"); + e + })?; Ok(()) } @@ -470,7 +508,10 @@ impl RedisDriver { ("result".to_string(), result_str), ("updated_at".to_string(), ts.to_string()), ]; - let _: usize = cm.hset_multiple(key, &pairs).await?; + let _: usize = cm.hset_multiple(&key, &pairs).await.map_err(|e| { + error!(db=%db, key=%key, error=%e, "HSET update_flow_result_merge failed"); + e + })?; Ok(()) } @@ -504,7 +545,10 @@ impl RedisDriver { ("env_vars".to_string(), env_vars_str), ("updated_at".to_string(), ts.to_string()), ]; - let _: usize = cm.hset_multiple(key, &pairs).await?; + let _: usize = cm.hset_multiple(&key, &pairs).await.map_err(|e| { + error!(db=%db, key=%key, error=%e, "HSET update_job_env_vars_merge failed"); + e + })?; Ok(()) } @@ -538,7 +582,10 @@ impl RedisDriver { ("result".to_string(), result_str), ("updated_at".to_string(), ts.to_string()), ]; - let _: usize = cm.hset_multiple(key, &pairs).await?; + let _: usize = cm.hset_multiple(&key, &pairs).await.map_err(|e| { + error!(db=%db, key=%key, error=%e, "HSET update_job_result_merge failed"); + e + })?; Ok(()) } @@ -553,7 +600,10 @@ impl RedisDriver { ("jobs".to_string(), jobs_str), ("updated_at".to_string(), ts.to_string()), ]; - let _: usize = cm.hset_multiple(key, &pairs).await?; + let _: usize = cm.hset_multiple(&key, &pairs).await.map_err(|e| { + error!(db=%db, key=%key, error=%e, "HSET update_flow_jobs_set failed"); + e + })?; Ok(()) } @@ -584,7 +634,10 @@ impl RedisDriver { ("logs".to_string(), logs_str), ("updated_at".to_string(), ts.to_string()), ]; - let _: usize = cm.hset_multiple(key, &pairs).await?; + let _: usize = cm.hset_multiple(&key, &pairs).await.map_err(|e| { + error!(db=%db, key=%key, error=%e, "HSET append_message_logs failed"); + e + })?; Ok(()) } @@ -595,7 +648,10 @@ impl RedisDriver { /// Push a value onto a Redis list using LPUSH in the given DB. pub async fn lpush_list(&self, db: u32, list: &str, value: &str) -> Result<()> { let mut cm = self.manager_for_db(db).await?; - let _: i64 = cm.lpush(list, value).await?; + let _: i64 = cm.lpush(list, value).await.map_err(|e| { + error!(db=%db, list=%list, value=%value, error=%e, "LPUSH failed"); + e + })?; Ok(()) } @@ -615,7 +671,11 @@ impl RedisDriver { .arg("msg_out") .arg(timeout_secs) .query_async(&mut cm) - .await?; + .await + .map_err(|e| { + error!(db=%db, timeout_secs=%timeout_secs, error=%e, "BRPOP failed"); + e + })?; Ok(res.map(|(_, v)| v)) } @@ -632,7 +692,11 @@ impl RedisDriver { .arg("COUNT") .arg(100) .query_async(&mut cm) - .await?; + .await + .map_err(|e| { + error!(db=%db, cursor=%cursor, error=%e, "SCAN failed"); + e + })?; for k in keys { if let Ok(r) = self.hget_model::(db, &k).await { out.push(r); @@ -653,7 +717,10 @@ impl RedisDriver { /// Register a context id in the global set "contexts" stored in DB 0. pub async fn register_context_id(&self, id: u32) -> Result<()> { let mut cm = self.manager_for_db(0).await?; - let _: i64 = redis::cmd("SADD").arg("contexts").arg(id).query_async(&mut cm).await?; + let _: i64 = redis::cmd("SADD").arg("contexts").arg(id).query_async(&mut cm).await.map_err(|e| { + error!(db=0, context_id=%id, error=%e, "SADD contexts failed"); + e + })?; Ok(()) } @@ -661,7 +728,10 @@ impl RedisDriver { pub async fn list_context_ids(&self) -> Result> { let mut cm = self.manager_for_db(0).await?; // Using SMEMBERS and parsing into u32 - let vals: Vec = redis::cmd("SMEMBERS").arg("contexts").query_async(&mut cm).await?; + let vals: Vec = redis::cmd("SMEMBERS").arg("contexts").query_async(&mut cm).await.map_err(|e| { + error!(db=0, error=%e, "SMEMBERS contexts failed"); + e + })?; let mut out = Vec::with_capacity(vals.len()); for v in vals { if let Ok(n) = v.parse::() {