diff --git a/Cargo.lock b/Cargo.lock index fe139f857..787b4bd38 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -664,15 +664,6 @@ dependencies = [ "num-traits", ] -[[package]] -name = "atomic" -version = "0.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8d818003e740b63afc82337e3160717f4f63078720a810b7b903e70a5d1d2994" -dependencies = [ - "bytemuck", -] - [[package]] name = "atomic-waker" version = "1.1.2" @@ -1026,12 +1017,6 @@ version = "3.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1628fb46dfa0b37568d12e5edd512553eccf6a22a78e8bde00bb4aed84d5bdbf" -[[package]] -name = "bytemuck" -version = "1.22.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b6b1fc10dbac614ebc03540c9dbd60e83887fda27794998c6528f1782047d540" - [[package]] name = "byteorder" version = "1.5.0" @@ -3034,12 +3019,6 @@ dependencies = [ "const-random", ] -[[package]] -name = "dotenvy" -version = "0.15.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" - [[package]] name = "dpi" version = "0.1.1" @@ -3288,21 +3267,6 @@ dependencies = [ "rustc_version", ] -[[package]] -name = "figment" -version = "0.10.19" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8cb01cd46b0cf372153850f4c6c272d9cbea2da513e07538405148f95bd789f3" -dependencies = [ - "atomic", - "pear", - "serde", - "serde_yaml", - "toml", - "uncased", - "version_check", -] - [[package]] name = "fixedbitset" version = "0.5.7" @@ -4435,12 +4399,6 @@ dependencies = [ "cfb", ] -[[package]] -name = "inlinable_string" -version = "0.1.15" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c8fae54786f62fb2918dcfae3d568594e50eb9b5c25bf04371af6fe7516452fb" - [[package]] name = "inout" version = "0.1.4" @@ -6048,29 +6006,6 @@ dependencies = [ "hmac 0.12.1", ] -[[package]] -name = "pear" -version = "0.2.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bdeeaa00ce488657faba8ebf44ab9361f9365a97bd39ffb8a60663f57ff4b467" -dependencies = [ - "inlinable_string", - "pear_codegen", - "yansi", -] - -[[package]] -name = "pear_codegen" -version = "0.2.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4bab5b985dc082b345f812b7df84e1bef27e7207b39e448439ba8bd69c93f147" -dependencies = [ - "proc-macro2", - "proc-macro2-diagnostics", - "quote", - "syn 2.0.100", -] - [[package]] name = "pem" version = "3.0.5" @@ -6549,7 +6484,6 @@ dependencies = [ "quote", "syn 2.0.100", "version_check", - "yansi", ] [[package]] @@ -7351,8 +7285,7 @@ version = "0.0.1" dependencies = [ "async-trait", "axum", - "dotenvy", - "figment", + "config", "http", "rdkafka", "reqwest", @@ -7832,19 +7765,6 @@ dependencies = [ "syn 2.0.100", ] -[[package]] -name = "serde_yaml" -version = "0.9.34+deprecated" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6a8b1a1a2ebf674015cc02edccce75287f1a0130d394307b36743c2f5d504b47" -dependencies = [ - "indexmap 2.9.0", - "itoa 1.0.15", - "ryu", - "serde", - "unsafe-libyaml", -] - [[package]] name = "server_fn" version = "0.6.15" @@ -9175,15 +9095,6 @@ dependencies = [ "winapi", ] -[[package]] -name = "uncased" -version = "0.9.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e1b88fcfe09e89d3866a5c11019378088af2d24c3fbd4f0543f96b479ec90697" -dependencies = [ - "version_check", -] - [[package]] name = "unicase" version = "2.8.1" @@ -9224,12 +9135,6 @@ dependencies = [ "subtle", ] -[[package]] -name = "unsafe-libyaml" -version = "0.2.11" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "673aac59facbab8a9007c7f6108d11f63b603f7cabff99fabf650fea5c32b861" - [[package]] name = "untrusted" version = "0.9.0" @@ -10223,12 +10128,6 @@ dependencies = [ "hashlink", ] -[[package]] -name = "yansi" -version = "1.0.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cfe53a6657fd280eaa890a3bc59152892ffa3e30101319d168b781ed6529b049" - [[package]] name = "yoke" version = "0.7.5" diff --git a/Cargo.toml b/Cargo.toml index d69bdce23..8ba1d7198 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -65,8 +65,6 @@ datafusion = "46.0.1" derive_builder = "0.20.2" dioxus = { version = "0.6.3", features = ["router"] } dirs = "6.0.0" -dotenvy = "0.15.7" -figment = { version = "0.10.19", features = ["toml", "yaml", "env"] } flatbuffers = "25.2.10" futures = "0.3.31" futures-core = "0.3.31" diff --git a/crates/event-notifier/Cargo.toml b/crates/event-notifier/Cargo.toml index c53294946..5d1d33906 100644 --- a/crates/event-notifier/Cargo.toml +++ b/crates/event-notifier/Cargo.toml @@ -14,8 +14,7 @@ mqtt = ["rumqttc"] [dependencies] async-trait = { workspace = true } -dotenvy = { workspace = true } -figment = { workspace = true, features = ["toml", "yaml", "env"] } +config = { workspace = true } rdkafka = { workspace = true, features = ["tokio"], optional = true } reqwest = { workspace = true, optional = true } rumqttc = { workspace = true, optional = true } diff --git a/crates/event-notifier/examples/.env-zh.example b/crates/event-notifier/examples/.env-zh.example new file mode 100644 index 000000000..dffa9c6e9 --- /dev/null +++ b/crates/event-notifier/examples/.env-zh.example @@ -0,0 +1,28 @@ +# ===== 全局配置 ===== +NOTIFIER__STORE_PATH=/var/log/event-notification +NOTIFIER__CHANNEL_CAPACITY=5000 + +# ===== 适配器配置（数组格式） ===== +# Webhook 适配器（索引 0） +NOTIFIER__ADAPTERS_0__type=Webhook +NOTIFIER__ADAPTERS_0__endpoint=http://127.0.0.1:3000/webhook +NOTIFIER__ADAPTERS_0__auth_token=your-auth-token +NOTIFIER__ADAPTERS_0__max_retries=3 +NOTIFIER__ADAPTERS_0__timeout=50 +NOTIFIER__ADAPTERS_0__custom_headers__x_custom_server=value +NOTIFIER__ADAPTERS_0__custom_headers__x_custom_client=value + +# Kafka 适配器（索引 1） +NOTIFIER__ADAPTERS_1__type=Kafka +NOTIFIER__ADAPTERS_1__brokers=localhost:9092 +NOTIFIER__ADAPTERS_1__topic=notifications +NOTIFIER__ADAPTERS_1__max_retries=3 +NOTIFIER__ADAPTERS_1__timeout=60 + +# MQTT 适配器（索引 2） +NOTIFIER__ADAPTERS_2__type=Mqtt +NOTIFIER__ADAPTERS_2__broker=mqtt.example.com +NOTIFIER__ADAPTERS_2__port=1883 +NOTIFIER__ADAPTERS_2__client_id=event-notifier +NOTIFIER__ADAPTERS_2__topic=events +NOTIFIER__ADAPTERS_2__max_retries=3 \ No newline at end of file diff --git a/crates/event-notifier/examples/.env.example b/crates/event-notifier/examples/.env.example index d42bc9fc0..28b4fcf47 100644 --- a/crates/event-notifier/examples/.env.example +++ b/crates/event-notifier/examples/.env.example @@ -1,27 +1,28 @@ -# basic configuration -EVENT_NOTIF_STORE_PATH=/var/log/event-notification -EVENT_NOTIF_CHANNEL_CAPACITY=5000 +# ===== global configuration ===== +NOTIFIER__STORE_PATH=/var/log/event-notification +NOTIFIER__CHANNEL_CAPACITY=5000 -# webhook adapter configuration -EVENT_NOTIF_ADAPTERS__0__TYPE=Webhook -EVENT_NOTIF_ADAPTERS__0__ENDPOINT=https://api.example.com/webhook -EVENT_NOTIF_ADAPTERS__0__AUTH_TOKEN=your-secret-token -EVENT_NOTIF_ADAPTERS__0__MAX_RETRIES=3 -EVENT_NOTIF_ADAPTERS__0__TIMEOUT=5000 +# ===== adapter configuration array format ===== +# webhook adapter index 0 +NOTIFIER__ADAPTERS_0__type=Webhook +NOTIFIER__ADAPTERS_0__endpoint=http://127.0.0.1:3000/webhook +NOTIFIER__ADAPTERS_0__auth_token=your-auth-token +NOTIFIER__ADAPTERS_0__max_retries=3 +NOTIFIER__ADAPTERS_0__timeout=50 +NOTIFIER__ADAPTERS_0__custom_headers__x_custom_server=server-value +NOTIFIER__ADAPTERS_0__custom_headers__x_custom_client=client-value -# kafka adapter configuration -EVENT_NOTIF_ADAPTERS__1__TYPE=Kafka -EVENT_NOTIF_ADAPTERS__1__BROKERS=localhost:9092 -EVENT_NOTIF_ADAPTERS__1__TOPIC=notifications -EVENT_NOTIF_ADAPTERS__1__MAX_RETRIES=3 -EVENT_NOTIF_ADAPTERS__1__TIMEOUT=5000 +# kafka adapter index 1 +NOTIFIER__ADAPTERS_1__type=Kafka +NOTIFIER__ADAPTERS_1__brokers=localhost:9092 +NOTIFIER__ADAPTERS_1__topic=notifications +NOTIFIER__ADAPTERS_1__max_retries=3 +NOTIFIER__ADAPTERS_1__timeout=60 -# mqtt adapter configuration -EVENT_NOTIF_ADAPTERS__2__TYPE=Mqtt -EVENT_NOTIF_ADAPTERS__2__BROKER=mqtt.example.com -EVENT_NOTIF_ADAPTERS__2__PORT=1883 -EVENT_NOTIF_ADAPTERS__2__CLIENT_ID=event-notifier -EVENT_NOTIF_ADAPTERS__2__TOPIC=events -EVENT_NOTIF_ADAPTERS__2__MAX_RETRIES=3 - -EVENT_NOTIF_HTTP__PORT=8080 \ No newline at end of file +# mqtt adapter index 2 +NOTIFIER__ADAPTERS_2__type=Mqtt +NOTIFIER__ADAPTERS_2__broker=mqtt.example.com +NOTIFIER__ADAPTERS_2__port=1883 +NOTIFIER__ADAPTERS_2__client_id=event-notifier +NOTIFIER__ADAPTERS_2__topic=events +NOTIFIER__ADAPTERS_2__max_retries=3 \ No newline at end of file diff --git a/crates/event-notifier/examples/event.toml b/crates/event-notifier/examples/event.toml index dcde5abf1..de5293f21 100644 --- a/crates/event-notifier/examples/event.toml +++ b/crates/event-notifier/examples/event.toml @@ -1,20 +1,24 @@ # config.toml -store_path = "/var/log/event-notification" +store_path = "/var/log/event-notifier" channel_capacity = 5000 [[adapters]] type = "Webhook" -endpoint = "https://api.example.com/webhook" +endpoint = "http://127.0.0.1:3000/webhook" auth_token = "your-auth-token" max_retries = 3 -timeout = 5000 +timeout = 50 + +[adapters.custom_headers] +custom_server = "value_server" +custom_client = "value_client" [[adapters]] type = "Kafka" brokers = "localhost:9092" topic = "notifications" max_retries = 3 -timeout = 5000 +timeout = 60 [[adapters]] type = "Mqtt" @@ -22,7 +26,4 @@ broker = "mqtt.example.com" port = 1883 client_id = "event-notifier" topic = "events" -max_retries = 3 - -[http] -port = 8080 \ No newline at end of file +max_retries = 3 \ No newline at end of file diff --git a/crates/event-notifier/examples/full.rs b/crates/event-notifier/examples/full.rs index 5e3cd5496..14231c730 100644 --- a/crates/event-notifier/examples/full.rs +++ b/crates/event-notifier/examples/full.rs @@ -1,6 +1,5 @@ use rustfs_event_notifier::{ - AdapterConfig, Bucket, Error as NotifierError, Event, Identity, Metadata, Name, NotificationConfig, Object, Source, - WebhookConfig, + AdapterConfig, Bucket, Error as NotifierError, Event, Identity, Metadata, Name, NotifierConfig, Object, Source, WebhookConfig, }; use std::collections::HashMap; use tokio::signal; @@ -8,7 +7,7 @@ use tracing::Level; use tracing_subscriber::FmtSubscriber; async fn setup_notification_system() -> Result<(), NotifierError> { - let config = NotificationConfig { + let config = NotifierConfig { store_path: "./deploy/logs/event_store".into(), channel_capacity: 100, adapters: vec![AdapterConfig::Webhook(WebhookConfig { @@ -40,11 +39,10 @@ async fn main() -> Result<(), Box> { // tracing_subscriber::fmt::init(); let subscriber = FmtSubscriber::builder() - .with_max_level(Level::DEBUG) // set to debug or lower level - .with_target(false) // simplify output + .with_max_level(Level::DEBUG) // set to debug or lower level + .with_target(false) // simplify output .finish(); - tracing::subscriber::set_global_default(subscriber) - .expect("failed to set up log subscriber"); + tracing::subscriber::set_global_default(subscriber).expect("failed to set up log subscriber"); // set up notification system if let Err(e) = setup_notification_system().await { diff --git a/crates/event-notifier/examples/simple.rs b/crates/event-notifier/examples/simple.rs index d35156b73..815f2a0ae 100644 --- a/crates/event-notifier/examples/simple.rs +++ b/crates/event-notifier/examples/simple.rs @@ -1,21 +1,27 @@ use rustfs_event_notifier::create_adapters; -use rustfs_event_notifier::NotificationSystem; -use rustfs_event_notifier::{AdapterConfig, NotificationConfig, WebhookConfig}; +use rustfs_event_notifier::NotifierSystem; +use rustfs_event_notifier::{AdapterConfig, NotifierConfig, WebhookConfig}; use rustfs_event_notifier::{Bucket, Event, Identity, Metadata, Name, Object, Source}; use std::collections::HashMap; use std::error; use std::sync::Arc; use tokio::signal; +use tracing::Level; +use tracing_subscriber::FmtSubscriber; #[tokio::main] async fn main() -> Result<(), Box> { - tracing_subscriber::fmt::init(); + let subscriber = FmtSubscriber::builder() + .with_max_level(Level::DEBUG) // set to debug or lower level + .with_target(false) // simplify output + .finish(); + tracing::subscriber::set_global_default(subscriber).expect("failed to set up log subscriber"); - let config = NotificationConfig { + let config = NotifierConfig { store_path: "./events".to_string(), channel_capacity: 100, adapters: vec![AdapterConfig::Webhook(WebhookConfig { - endpoint: "http://localhost:8080/webhook".to_string(), + endpoint: "http://127.0.0.1:3000/webhook".to_string(), auth_token: Some("secret-token".to_string()), custom_headers: Some(HashMap::from([("X-Custom".to_string(), "value".to_string())])), max_retries: 3, @@ -23,16 +29,12 @@ async fn main() -> Result<(), Box> { })], }; - // loading configuration from specific env files - let _config = NotificationConfig::from_env_file(".env.example")?; + // load_config + // loading configuration from environment variables + let _config = NotifierConfig::load_config(Some("./crates/event-notifier/examples/event.toml".to_string())); + tracing::info!("load_config config: {:?} \n", _config); - // loading from a specific file - let _config = NotificationConfig::from_file("event.toml")?; - - // Automatically load from multiple sources (Priority: Environment Variables > YAML > TOML) - let _config = NotificationConfig::load()?; - - let system = Arc::new(tokio::sync::Mutex::new(NotificationSystem::new(config.clone()).await?)); + let system = Arc::new(tokio::sync::Mutex::new(NotifierSystem::new(config.clone()).await?)); let adapters = create_adapters(&config.adapters)?; // create an s3 metadata object @@ -90,9 +92,15 @@ async fn main() -> Result<(), Box> { signal::ctrl_c().await?; tracing::info!("Received shutdown signal"); - { - let system = system.lock().await; - system.shutdown(); + let result = { + let mut system = system.lock().await; + system.shutdown().await + }; + + if let Err(e) = result { + tracing::error!("Failed to shut down the notification system: {}", e); + } else { + tracing::info!("Notification system shut down successfully"); } system_handle.await??; diff --git a/crates/event-notifier/examples/webhook.rs b/crates/event-notifier/examples/webhook.rs index 3b9caad7e..b78af94dd 100644 --- a/crates/event-notifier/examples/webhook.rs +++ b/crates/event-notifier/examples/webhook.rs @@ -12,6 +12,71 @@ async fn main() { } async fn receive_webhook(Json(payload): Json) -> StatusCode { - println!("收到 webhook 请求 time: {}，内容：{}", SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs().to_string(), serde_json::to_string_pretty(&payload).unwrap()); + let start = SystemTime::now(); + let since_the_epoch = start.duration_since(UNIX_EPOCH).expect("Time went backwards"); + + // get the number of seconds since the unix era + let seconds = since_the_epoch.as_secs(); + + // Manually calculate year, month, day, hour, minute, and second + let (year, month, day, hour, minute, second) = convert_seconds_to_date(seconds); + + // output result + println!("current time:{:04}-{:02}-{:02} {:02}:{:02}:{:02}", year, month, day, hour, minute, second); + println!( + "received a webhook request time:{} content:\n {}", + seconds.to_string(), + serde_json::to_string_pretty(&payload).unwrap() + ); StatusCode::OK } + +fn convert_seconds_to_date(seconds: u64) -> (u32, u32, u32, u32, u32, u32) { + // assume that the time zone is utc + let seconds_per_minute = 60; + let seconds_per_hour = 3600; + let seconds_per_day = 86400; + + // Calculate the year, month, day, hour, minute, and second corresponding to the number of seconds + let mut total_seconds = seconds; + let mut year = 1970; + let mut month = 1; + let mut day = 1; + let mut hour = 0; + let mut minute = 0; + let mut second = 0; + + // calculate year + while total_seconds >= 31536000 { + year += 1; + total_seconds -= 31536000; // simplified processing no leap year considered + } + + // calculate month + let days_in_month = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]; + for m in 0..12 { + if total_seconds >= days_in_month[m] * seconds_per_day { + month += 1; + total_seconds -= days_in_month[m] * seconds_per_day; + } else { + break; + } + } + + // calculate the number of days + day += total_seconds / seconds_per_day; + total_seconds %= seconds_per_day; + + // calculate hours + hour += total_seconds / seconds_per_hour; + total_seconds %= seconds_per_hour; + + // calculate minutes + minute += total_seconds / seconds_per_minute; + total_seconds %= seconds_per_minute; + + // calculate the number of seconds + second += total_seconds; + + (year as u32, month as u32, day as u32, hour as u32, minute as u32, second as u32) +} diff --git a/crates/event-notifier/src/bus.rs b/crates/event-notifier/src/bus.rs index c04733a0a..bd4b81c5d 100644 --- a/crates/event-notifier/src/bus.rs +++ b/crates/event-notifier/src/bus.rs @@ -80,16 +80,20 @@ pub async fn event_bus( } else { tracing::info!("no unhandled events need to be saved"); } - tracing::info!("shutdown_complete is Some: {}", shutdown_complete.is_some()); - // send a completion signal + tracing::debug!("shutdown_complete is Some: {}", shutdown_complete.is_some()); + if let Some(complete_sender) = shutdown_complete { - let _ = complete_sender.send(()); + // send a completion signal + let result = complete_sender.send(()); + match result { + Ok(_) => tracing::info!("Event bus shutdown signal sent"), + Err(e) => tracing::error!("Failed to send event bus shutdown signal: {:?}", e), + } tracing::info!("Shutting down event bus"); } tracing::info!("Event bus shutdown complete"); break; } - // else => break, } } Ok(()) diff --git a/crates/event-notifier/src/config.rs b/crates/event-notifier/src/config.rs index c648ceac3..c360f2bc9 100644 --- a/crates/event-notifier/src/config.rs +++ b/crates/event-notifier/src/config.rs @@ -1,7 +1,7 @@ -use crate::Error; -use figment::providers::Format; +use config::{Config, Environment, File, FileFormat}; use serde::{Deserialize, Serialize}; use std::collections::HashMap; +use std::env; /// Configuration for the notification system. #[derive(Debug, Clone, Serialize, Deserialize)] @@ -65,7 +65,7 @@ pub enum AdapterConfig { /// Configuration for the notification system. #[derive(Debug, Clone, Serialize, Deserialize)] -pub struct NotificationConfig { +pub struct NotifierConfig { #[serde(default = "default_store_path")] pub store_path: String, #[serde(default = "default_channel_capacity")] @@ -73,7 +73,7 @@ pub struct NotificationConfig { pub adapters: Vec, } -impl Default for NotificationConfig { +impl Default for NotifierConfig { fn default() -> Self { Self { store_path: default_store_path(), @@ -83,46 +83,88 @@ impl Default for NotificationConfig { } } -impl NotificationConfig { +impl NotifierConfig { /// create a new configuration with default values pub fn new() -> Self { Self::default() } - /// create a configuration from a configuration file - pub fn from_file(path: &str) -> Result { - let config = figment::Figment::new() - .merge(figment::providers::Toml::file(path)) - .extract()?; + /// Loading the configuration file + /// Supports TOML, YAML and .env formats, read in order by priority + /// + /// # Parameters + /// - `config_dir`: Configuration file path + /// + /// # Returns + /// Configuration information + /// + /// # Example + /// ``` + /// use rustfs_event_notifier::NotifierConfig; + /// + /// let config = NotifierConfig::load_config(None); + /// ``` + pub fn load_config(config_dir: Option) -> NotifierConfig { + let config_dir = if let Some(path) = config_dir { + // If a path is provided, check if it's empty + if path.is_empty() { + // If empty, use the default config file name + DEFAULT_CONFIG_FILE.to_string() + } else { + // Use the provided path + let path = std::path::Path::new(&path); + if path.extension().is_some() { + // If path has extension, use it as is (extension will be added by Config::builder) + path.with_extension("").to_string_lossy().into_owned() + } else { + // If path is a directory, append the default config file name + path.to_string_lossy().into_owned() + } + } + } else { + // If no path provided, use current directory + default config file + match env::current_dir() { + Ok(dir) => dir.join(DEFAULT_CONFIG_FILE).to_string_lossy().into_owned(), + Err(_) => { + eprintln!("Warning: Failed to get current directory, using default config file"); + DEFAULT_CONFIG_FILE.to_string() + } + } + }; - Ok(config) - } + // Log using proper logging instead of println when possible + println!("Using config file base: {}", config_dir); - /// Read configuration from multiple sources (support TOML, YAML, .env) - pub fn load() -> Result { - let figment = figment::Figment::new() - // First try to read the config.toml of the current directory - .merge(figment::providers::Toml::file("event.toml")) - // Then try to read the config.yaml of the current directory - .merge(figment::providers::Yaml::file("event.yaml")) - // Finally read the environment variable and overwrite the previous value - .merge(figment::providers::Env::prefixed("EVENT_NOTIF_")); - - Ok(figment.extract()?) - } - - /// loading configuration from env file - pub fn from_env_file(path: &str) -> Result { - // loading env files - dotenvy::from_path(path).map_err(|e| Error::ConfigError(format!("unable to load env file: {}", e)))?; - - // Extract configuration from environment variables using figurement - let figment = figment::Figment::new().merge(figment::providers::Env::prefixed("EVENT_NOTIF_")); - - Ok(figment.extract()?) + let app_config = Config::builder() + .add_source(File::with_name(config_dir.as_str()).format(FileFormat::Toml).required(false)) + .add_source(File::with_name(config_dir.as_str()).format(FileFormat::Yaml).required(false)) + .add_source( + Environment::default() + .prefix("NOTIFIER") + .prefix_separator("__") + .separator("__") + .list_separator("_") + .with_list_parse_key("adapters") + .try_parsing(true), + ) + .build() + .unwrap_or_default(); + println!("Loaded config: {:?}", app_config); + match app_config.try_deserialize::() { + Ok(app_config) => { + println!("Parsed AppConfig: {:?} \n", app_config); + app_config + } + Err(e) => { + println!("Failed to deserialize config: {}", e); + NotifierConfig::default() + } + } } } +const DEFAULT_CONFIG_FILE: &str = "obs"; + /// Provide temporary directories as default storage paths fn default_store_path() -> String { std::env::temp_dir().join("event-notification").to_string_lossy().to_string() diff --git a/crates/event-notifier/src/error.rs b/crates/event-notifier/src/error.rs index 69a30adf8..e6c061de9 100644 --- a/crates/event-notifier/src/error.rs +++ b/crates/event-notifier/src/error.rs @@ -1,3 +1,4 @@ +use config::ConfigError; use thiserror::Error; use tokio::sync::mpsc::error; use tokio::task::JoinError; @@ -35,7 +36,7 @@ pub enum Error { #[error("Configuration error: {0}")] ConfigError(String), #[error("Configuration loading error: {0}")] - Figment(#[from] figment::Error), + Config(#[from] ConfigError), } impl Error { diff --git a/crates/event-notifier/src/global.rs b/crates/event-notifier/src/global.rs index bad6022a6..9a83e9c7a 100644 --- a/crates/event-notifier/src/global.rs +++ b/crates/event-notifier/src/global.rs @@ -1,8 +1,8 @@ -use crate::{create_adapters, Error, Event, NotificationConfig, NotificationSystem}; +use crate::{create_adapters, Error, Event, NotifierConfig, NotifierSystem}; use std::sync::{atomic, Arc}; use tokio::sync::{Mutex, OnceCell}; -static GLOBAL_SYSTEM: OnceCell>> = OnceCell::const_new(); +static GLOBAL_SYSTEM: OnceCell>> = OnceCell::const_new(); static INITIALIZED: atomic::AtomicBool = atomic::AtomicBool::new(false); static READY: atomic::AtomicBool = atomic::AtomicBool::new(false); static INIT_LOCK: Mutex<()> = Mutex::const_new(()); @@ -24,7 +24,7 @@ static INIT_LOCK: Mutex<()> = Mutex::const_new(()); /// - Creating adapters fails. /// - Starting the notification system fails. /// - Setting the global system instance fails. -pub async fn initialize(config: NotificationConfig) -> Result<(), Error> { +pub async fn initialize(config: NotifierConfig) -> Result<(), Error> { let _lock = INIT_LOCK.lock().await; // Check if the system is already initialized. @@ -52,7 +52,7 @@ pub async fn initialize(config: NotificationConfig) -> Result<(), Error> { // Attempt to initialize, and reset the INITIALIZED flag if it fails. let result: Result<(), Error> = async { - let system = NotificationSystem::new(config.clone()).await.map_err(|e| { + let system = NotifierSystem::new(config.clone()).await.map_err(|e| { tracing::error!("Failed to create NotificationSystem: {:?}", e); e })?; @@ -127,24 +127,29 @@ pub async fn send_event(event: Event) -> Result<(), Error> { pub async fn shutdown() -> Result<(), Error> { if let Some(system) = GLOBAL_SYSTEM.get() { tracing::info!("Shutting down notification system start"); - let (complete_tx, complete_rx) = tokio::sync::oneshot::channel(); - - { + let result = { let mut system_guard = system.lock().await; - // set the complete channel and trigger cancellation - system_guard.set_shutdown_complete_channel(complete_tx); - system_guard.shutdown(); - tracing::info!("Notification system shutdown triggered"); + system_guard.shutdown().await + }; + if let Err(e) = &result { + tracing::error!("Notification system shutdown failed: {}", e); + } else { + tracing::info!("Event bus shutdown completed"); } - // wait for the cleaning to be completed - let _ = complete_rx.await; - tracing::info!("Event bus shutdown completed"); - + tracing::info!( + "Shutdown method called set static value start, READY: {}, INITIALIZED: {}", + READY.load(atomic::Ordering::SeqCst), + INITIALIZED.load(atomic::Ordering::SeqCst) + ); READY.store(false, atomic::Ordering::SeqCst); INITIALIZED.store(false, atomic::Ordering::SeqCst); - tracing::info!("Notification system is ready to process events"); - Ok(()) + tracing::info!( + "Shutdown method called set static value end, READY: {}, INITIALIZED: {}", + READY.load(atomic::Ordering::SeqCst), + INITIALIZED.load(atomic::Ordering::SeqCst) + ); + result } else { Err(Error::custom("Notification system not initialized")) } @@ -155,7 +160,7 @@ pub async fn shutdown() -> Result<(), Error> { /// # Errors /// /// Returns an error if the system is not initialized. -async fn get_system() -> Result>, Error> { +async fn get_system() -> Result>, Error> { GLOBAL_SYSTEM .get() .cloned() @@ -165,13 +170,13 @@ async fn get_system() -> Result>, Error> { #[cfg(test)] mod tests { use super::*; - use crate::{AdapterConfig, NotificationConfig, WebhookConfig}; + use crate::{AdapterConfig, NotifierConfig, WebhookConfig}; use std::collections::HashMap; #[tokio::test] async fn test_initialize_success() { tracing_subscriber::fmt::init(); - let config = NotificationConfig::default(); // assume there is a default configuration + let config = NotifierConfig::default(); // assume there is a default configuration let result = initialize(config).await; assert!(!result.is_ok(), "Initialization should succeed"); assert!(!is_initialized(), "System should be marked as initialized"); @@ -181,7 +186,7 @@ mod tests { #[tokio::test] async fn test_initialize_twice() { tracing_subscriber::fmt::init(); - let config = NotificationConfig::default(); + let config = NotifierConfig::default(); let _ = initialize(config.clone()).await; // first initialization let result = initialize(config).await; // second initialization assert!(!result.is_ok(), "Initialization should succeed"); @@ -192,7 +197,7 @@ mod tests { async fn test_initialize_failure_resets_state() { tracing_subscriber::fmt::init(); // simulate wrong configuration - let config = NotificationConfig { + let config = NotifierConfig { adapters: vec![ // assuming that the empty adapter will cause failure AdapterConfig::Webhook(WebhookConfig { @@ -217,7 +222,7 @@ mod tests { assert!(!is_initialized(), "System should not be initialized initially"); assert!(!is_ready(), "System should not be ready initially"); - let config = NotificationConfig::default(); + let config = NotifierConfig::default(); let _ = initialize(config).await; assert!(!is_initialized(), "System should be initialized after successful initialization"); assert!(!is_ready(), "System should be ready after successful initialization"); diff --git a/crates/event-notifier/src/lib.rs b/crates/event-notifier/src/lib.rs index 76981822c..20ef935d2 100644 --- a/crates/event-notifier/src/lib.rs +++ b/crates/event-notifier/src/lib.rs @@ -22,10 +22,10 @@ pub use config::KafkaConfig; pub use config::MqttConfig; #[cfg(feature = "webhook")] pub use config::WebhookConfig; -pub use config::{AdapterConfig, NotificationConfig}; +pub use config::{AdapterConfig, NotifierConfig}; pub use error::Error; pub use event::{Bucket, Event, EventBuilder, Identity, Log, Metadata, Name, Object, Source}; pub use global::{initialize, is_initialized, is_ready, send_event, shutdown}; -pub use notifier::NotificationSystem; +pub use notifier::NotifierSystem; pub use store::EventStore; diff --git a/crates/event-notifier/src/notifier.rs b/crates/event-notifier/src/notifier.rs index 94953e331..0b17ddddd 100644 --- a/crates/event-notifier/src/notifier.rs +++ b/crates/event-notifier/src/notifier.rs @@ -1,4 +1,4 @@ -use crate::{event_bus, ChannelAdapter, Error, Event, EventStore, NotificationConfig}; +use crate::{event_bus, ChannelAdapter, Error, Event, EventStore, NotifierConfig}; use std::sync::Arc; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; @@ -7,17 +7,18 @@ use tokio_util::sync::CancellationToken; /// It manages the event bus and the adapters. /// It is responsible for sending and receiving events. /// It also handles the shutdown process. -pub struct NotificationSystem { +pub struct NotifierSystem { tx: mpsc::Sender, rx: Option>, store: Arc, shutdown: CancellationToken, shutdown_complete: Option>, + shutdown_receiver: Option>, } -impl NotificationSystem { +impl NotifierSystem { /// Creates a new `NotificationSystem` instance. - pub async fn new(config: NotificationConfig) -> Result { + pub async fn new(config: NotifierConfig) -> Result { let (tx, rx) = mpsc::channel::(config.channel_capacity); let store = Arc::new(EventStore::new(&config.store_path).await?); let shutdown = CancellationToken::new(); @@ -29,13 +30,15 @@ impl NotificationSystem { tx.send(event).await.map_err(|e| Error::ChannelSend(Box::new(e)))?; } } - + // Initialize shutdown_complete to Some(tx) + let (complete_tx, complete_rx) = tokio::sync::oneshot::channel(); Ok(Self { tx, rx: Some(rx), store, shutdown, - shutdown_complete: None, + shutdown_complete: Some(complete_tx), + shutdown_receiver: Some(complete_rx), }) } @@ -43,37 +46,65 @@ impl NotificationSystem { /// It initializes the event bus and the producer. pub async fn start(&mut self, adapters: Vec>) -> Result<(), Error> { if self.shutdown.is_cancelled() { - return Err(Error::custom("System is shutting down")); + let error = Error::custom("System is shutting down"); + self.handle_error("start", &error); + return Err(error); } - + self.log(tracing::Level::INFO, "start", "Starting the notification system"); let rx = self.rx.take().ok_or_else(|| Error::EventBusStarted)?; let shutdown_clone = self.shutdown.clone(); let store_clone = self.store.clone(); let shutdown_complete = self.shutdown_complete.take(); + tokio::spawn(async move { if let Err(e) = event_bus(rx, adapters, store_clone, shutdown_clone, shutdown_complete).await { tracing::error!("Event bus failed: {}", e); } }); - + self.log(tracing::Level::INFO, "start", "Notification system started successfully"); Ok(()) } /// Sends an event to the notification system. /// This method is used to send events to the event bus. pub async fn send_event(&self, event: Event) -> Result<(), Error> { + self.log(tracing::Level::DEBUG, "send_event", &format!("Sending event: {:?}", event)); if self.shutdown.is_cancelled() { - return Err(Error::custom("System is shutting down")); + let error = Error::custom("System is shutting down"); + self.handle_error("send_event", &error); + return Err(error); } - self.tx.send(event).await.map_err(|e| Error::ChannelSend(Box::new(e)))?; + if let Err(e) = self.tx.send(event).await { + let error = Error::ChannelSend(Box::new(e)); + self.handle_error("send_event", &error); + return Err(error); + } + self.log(tracing::Level::INFO, "send_event", "Event sent successfully"); Ok(()) } /// Shuts down the notification system. /// This method is used to cancel the event bus and producer tasks. - pub fn shutdown(&self) { + pub async fn shutdown(&mut self) -> Result<(), Error> { tracing::info!("Shutting down the notification system"); self.shutdown.cancel(); + // wait for the event bus to be completely closed + if let Some(receiver) = self.shutdown_receiver.take() { + match receiver.await { + Ok(_) => { + tracing::info!("Event bus shutdown completed successfully"); + Ok(()) + } + Err(e) => { + let error = Error::custom(format!("Failed to receive shutdown completion: {}", e).as_str()); + self.handle_error("shutdown", &error); + Err(error) + } + } + } else { + tracing::warn!("Shutdown receiver not available, the event bus might still be running"); + Err(Error::custom("Shutdown receiver not available")) + } } /// shutdown state @@ -81,9 +112,17 @@ impl NotificationSystem { self.shutdown.is_cancelled() } - pub fn set_shutdown_complete_channel(&mut self, tx: tokio::sync::oneshot::Sender<()>) { - // storage completion channel for use by event bus - tracing::info!("Shutting down the notification system set shutdown complete channel"); - self.shutdown_complete = Some(tx); + fn handle_error(&self, context: &str, error: &Error) { + self.log(tracing::Level::ERROR, context, &format!("{:?}", error)); + // TODO Can be extended to record to files or send to monitoring systems + } + fn log(&self, level: tracing::Level, context: &str, message: &str) { + match level { + tracing::Level::ERROR => tracing::error!("[{}] {}", context, message), + tracing::Level::WARN => tracing::warn!("[{}] {}", context, message), + tracing::Level::INFO => tracing::info!("[{}] {}", context, message), + tracing::Level::DEBUG => tracing::debug!("[{}] {}", context, message), + tracing::Level::TRACE => tracing::trace!("[{}] {}", context, message), + } } } diff --git a/crates/event-notifier/tests/integration.rs b/crates/event-notifier/tests/integration.rs index ec97a64b9..c5743605b 100644 --- a/crates/event-notifier/tests/integration.rs +++ b/crates/event-notifier/tests/integration.rs @@ -1,4 +1,4 @@ -use rustfs_event_notifier::{AdapterConfig, NotificationSystem, WebhookConfig}; +use rustfs_event_notifier::{AdapterConfig, NotifierSystem, WebhookConfig}; use rustfs_event_notifier::{Bucket, Event, EventBuilder, Identity, Metadata, Name, Object, Source}; use rustfs_event_notifier::{ChannelAdapter, WebhookAdapter}; use std::collections::HashMap; @@ -67,7 +67,7 @@ async fn test_webhook_adapter() { #[tokio::test] async fn test_notification_system() { - let config = rustfs_event_notifier::NotificationConfig { + let config = rustfs_event_notifier::NotifierConfig { store_path: "./test_events".to_string(), channel_capacity: 100, adapters: vec![AdapterConfig::Webhook(WebhookConfig { @@ -78,7 +78,7 @@ async fn test_notification_system() { timeout: 5, })], }; - let system = Arc::new(tokio::sync::Mutex::new(NotificationSystem::new(config.clone()).await.unwrap())); + let system = Arc::new(tokio::sync::Mutex::new(NotifierSystem::new(config.clone()).await.unwrap())); let adapters: Vec> = vec![Arc::new(WebhookAdapter::new(WebhookConfig { endpoint: "http://localhost:8080/webhook".to_string(), auth_token: None, @@ -148,8 +148,8 @@ async fn test_notification_system() { // create a new task to handle the timeout let system = Arc::clone(&system); tokio::spawn(async move { - if let Ok(guard) = system.try_lock() { - guard.shutdown(); + if let Ok(mut guard) = system.try_lock() { + guard.shutdown().await.unwrap(); } }); // give the system some time to clean up resources diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 24e5782fe..62e1a3ea3 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -122,7 +122,7 @@ impl LocalDisk { let root = fs::canonicalize(ep.get_file_path()).await?; if cleanup { - // TODO: 删除tmp数据 + // TODO: 删除 tmp 数据 } let format_path = Path::new(super::RUSTFS_META_BUCKET) @@ -631,13 +631,13 @@ impl LocalDisk { } } - // 没有版本了，删除xl.meta + // 没有版本了，删除 xl.meta if fm.versions.is_empty() { self.delete_file(&volume_dir, &xlpath, true, false).await?; return Ok(()); } - // 更新xl.meta + // 更新 xl.meta let buf = fm.marshal_msg()?; let volume_dir = self.get_bucket_path(volume)?; @@ -875,7 +875,7 @@ impl LocalDisk { .read_metadata(self.get_object_path(bucket, format!("{}/{}", ¤t, &entry).as_str())?) .await?; - // 用strip_suffix只删除一次 + // 用 strip_suffix 只删除一次 let entry = entry.strip_suffix(STORAGE_FORMAT_FILE).unwrap_or_default().to_owned(); let name = entry.trim_end_matches(SLASH_SEPARATOR); let name = decode_dir_object(format!("{}/{}", ¤t, &name).as_str()); @@ -1723,7 +1723,7 @@ impl DiskAPI for LocalDisk { } } - // xl.meta路径 + // xl.meta 路径 let src_file_path = src_volume_dir.join(Path::new(format!("{}/{}", &src_path, super::STORAGE_FORMAT_FILE).as_str())); let dst_file_path = dst_volume_dir.join(Path::new(format!("{}/{}", &dst_path, super::STORAGE_FORMAT_FILE).as_str())); @@ -1754,7 +1754,7 @@ impl DiskAPI for LocalDisk { check_path_length(src_file_path.to_string_lossy().to_string().as_str())?; check_path_length(dst_file_path.to_string_lossy().to_string().as_str())?; - // 读旧xl.meta + // 读旧 xl.meta let has_dst_buf = match utils::fs::read_file(&dst_file_path).await { Ok(res) => Some(res), @@ -2268,7 +2268,7 @@ impl DiskAPI for LocalDisk { async fn delete_volume(&self, volume: &str) -> Result<()> { let p = self.get_bucket_path(volume)?; - // TODO: 不能用递归删除，如果目录下面有文件，返回errVolumeNotEmpty + // TODO: 不能用递归删除，如果目录下面有文件，返回 errVolumeNotEmpty if let Err(err) = fs::remove_dir_all(&p).await { match err.kind() { @@ -2328,10 +2328,7 @@ impl DiskAPI for LocalDisk { } } - let vcfg = match BucketVersioningSys::get(&cache.info.name).await { - Ok(vcfg) => Some(vcfg), - Err(_) => None, - }; + let vcfg = (BucketVersioningSys::get(&cache.info.name).await).ok(); let loc = self.get_disk_location(); let disks = store.get_disks(loc.pool_idx.unwrap(), loc.disk_idx.unwrap()).await?; @@ -2491,7 +2488,6 @@ async fn get_disk_info(drive_path: PathBuf) -> Result<(Info, bool)> { #[cfg(test)] mod test { - use super::*; #[tokio::test] diff --git a/ecstore/src/heal/data_scanner.rs b/ecstore/src/heal/data_scanner.rs index 78c5b9f64..268af7343 100644 --- a/ecstore/src/heal/data_scanner.rs +++ b/ecstore/src/heal/data_scanner.rs @@ -1101,7 +1101,7 @@ pub fn lc_has_active_rules(config: &BucketLifecycleConfiguration, prefix: &str) return true; } - if let Some(Some(true)) = rule.expiration.as_ref().map(|e| e.expired_object_delete_marker.map(|m| m)) { + if let Some(Some(true)) = rule.expiration.as_ref().map(|e| e.expired_object_delete_marker) { return true; }