diff --git a/Cargo.lock b/Cargo.lock index 14e66b9e7..a55e8ea60 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7495,8 +7495,12 @@ version = "0.0.1" dependencies = [ "async-trait", "axum", + "common", "config", "dotenvy", + "ecstore", + "http", + "lazy_static", "rdkafka", "reqwest", "rumqttc", diff --git a/crates/event/Cargo.toml b/crates/event/Cargo.toml index 6d4331e01..8aa232761 100644 --- a/crates/event/Cargo.toml +++ b/crates/event/Cargo.toml @@ -15,6 +15,10 @@ kafka = ["dep:rdkafka"] [dependencies] async-trait = { workspace = true } config = { workspace = true } +common = { workspace = true } +ecstore = { workspace = true } +http = { workspace = true } +lazy_static = { workspace = true } reqwest = { workspace = true, optional = true } rumqttc = { workspace = true, optional = true } serde = { workspace = true } diff --git a/crates/event/examples/full.rs b/crates/event/examples/full.rs index fae56cd8e..128dfe2d8 100644 --- a/crates/event/examples/full.rs +++ b/crates/event/examples/full.rs @@ -19,7 +19,7 @@ async fn setup_notification_system() -> Result<(), NotifierError> { })], }; - rustfs_event::initialize(config).await?; + rustfs_event::initialize(&config).await?; // wait for the system to be ready for _ in 0..50 { diff --git a/crates/event/src/config.rs b/crates/event/src/config.rs index e564f7264..12b2c7a03 100644 --- a/crates/event/src/config.rs +++ b/crates/event/src/config.rs @@ -3,7 +3,7 @@ use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::env; -/// Configuration for the notification system. +/// Configuration for the webhook adapter. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct WebhookConfig { pub endpoint: String, @@ -14,7 +14,26 @@ pub struct WebhookConfig { } impl WebhookConfig { - /// verify that the configuration is valid + /// validate the configuration for the webhook adapter + /// + /// # Returns + /// + /// - `Result<(), String>`: Ok if the configuration is valid, Err with a message if invalid. + /// + /// # Example + /// + /// ``` + /// use rustfs_event::WebhookConfig; + /// + /// let config = WebhookConfig { + /// endpoint: "http://example.com/webhook".to_string(), + /// auth_token: Some("my_token".to_string()), + /// custom_headers: None, + /// max_retries: 3, + /// timeout: 5000, + /// }; + /// + /// assert!(config.validate().is_ok()); pub fn validate(&self) -> Result<(), String> { // verify that endpoint cannot be empty if self.endpoint.trim().is_empty() { @@ -54,7 +73,7 @@ pub struct MqttConfig { pub max_retries: u32, } -/// Configuration for the notification system. +/// Configuration for the adapter. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "type")] pub enum AdapterConfig { @@ -64,6 +83,23 @@ pub enum AdapterConfig { } /// Configuration for the notification system. +/// +/// This struct contains the configuration for the notification system, including +/// the storage path, channel capacity, and a list of adapters. +/// +/// # Fields +/// +/// - `store_path`: The path to the storage directory. +/// - `channel_capacity`: The capacity of the notification channel. +/// - `adapters`: A list of adapters to be used for notifications. +/// +/// # Example +/// +/// ``` +/// use rustfs_event::NotifierConfig; +/// +/// let config = NotifierConfig::new(); +/// ``` #[derive(Debug, Clone, Serialize, Deserialize)] pub struct NotifierConfig { #[serde(default = "default_store_path")] @@ -151,13 +187,33 @@ impl NotifierConfig { } } } + + /// unmarshal the configuration from a byte array + pub fn unmarshal(data: &[u8]) -> common::error::Result { + let m: NotifierConfig = serde_json::from_slice(data)?; + Ok(m) + } + + /// marshal the configuration to a byte array + pub fn marshal(&self) -> common::error::Result> { + let data = serde_json::to_vec(&self)?; + Ok(data) + } + + /// merge the configuration with default values + pub fn merge(&self) -> NotifierConfig { + self.clone() + } } const DEFAULT_CONFIG_FILE: &str = "event"; /// Provide temporary directories as default storage paths fn default_store_path() -> String { - std::env::temp_dir().join("event-notification").to_string_lossy().to_string() + env::var("EVENT_STORE_PATH").unwrap_or_else(|e| { + tracing::error!("Failed to get EVENT_STORE_PATH: {}", e); + env::temp_dir().join(DEFAULT_CONFIG_FILE).to_string_lossy().to_string() + }) } /// Provides the recommended default channel capacity for high concurrency systems diff --git a/crates/event/src/event_sys/mod.rs b/crates/event/src/event_sys/mod.rs new file mode 100644 index 000000000..41f76e39e --- /dev/null +++ b/crates/event/src/event_sys/mod.rs @@ -0,0 +1,267 @@ +use crate::NotifierConfig; +use common::error::{Error, Result}; +use ecstore::config::com::CONFIG_PREFIX; +use ecstore::config::error::{is_err_config_not_found, ConfigError}; +use ecstore::disk::RUSTFS_META_BUCKET; +use ecstore::store::ECStore; +use ecstore::store_api::{ObjectInfo, ObjectOptions, PutObjReader}; +use ecstore::store_err::is_err_object_not_found; +use ecstore::utils::path::SLASH_SEPARATOR; +use ecstore::StorageAPI; +use http::HeaderMap; +use lazy_static::lazy_static; +use std::io::Cursor; +use std::sync::{Arc, OnceLock}; +use tracing::{error, instrument, warn}; + +lazy_static! { + pub static ref GLOBAL_EventSys: EventSys = EventSys::new(); + pub static ref GLOBAL_EventSysConfig: OnceLock = OnceLock::new(); +} +/// * config file +const CONFIG_FILE: &str = "event.json"; + +/// event sys config +const EVENT: &str = "event"; + +#[derive(Debug)] +pub struct EventSys {} + +impl Default for EventSys { + fn default() -> Self { + Self::new() + } +} + +impl EventSys { + pub fn new() -> Self { + Self {} + } + #[instrument(skip_all)] + pub async fn init(&self, api: Arc) -> Result<()> { + tracing::info!("event sys config init start"); + let cfg = read_config_without_migrate(api.clone().clone()).await?; + let _ = GLOBAL_EventSysConfig.set(cfg); + tracing::info!("event sys config init done"); + Ok(()) + } +} + +/// get event sys config file +/// +/// # Returns +/// NotifierConfig +pub fn get_event_notifier_config() -> &'static NotifierConfig { + GLOBAL_EventSysConfig.get_or_init(NotifierConfig::default) +} + +fn get_event_sys_file() -> String { + format!("{}{}{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, EVENT, SLASH_SEPARATOR, CONFIG_FILE) +} + +/// read config without migrate +/// +/// # Parameters +/// - `api`: StorageAPI +/// +/// # Returns +/// Configuration information +pub async fn read_config_without_migrate(api: Arc) -> Result { + let config_file = get_event_sys_file(); + let data = match read_config(api.clone(), config_file.as_str()).await { + Ok(res) => res, + Err(err) => { + return if is_err_config_not_found(&err) { + warn!("config not found, start to init"); + let cfg = new_and_save_server_config(api).await?; + warn!("config init done"); + Ok(cfg) + } else { + error!("read config err {:?}", &err); + Err(err) + } + } + }; + + read_server_config(api, data.as_slice()).await +} + +/// save config with options +/// +/// # Parameters +/// - `api`: StorageAPI +/// - `file`: file name +/// - `data`: data to save +/// - `opts`: object options +/// +/// # Returns +/// Result +pub async fn save_config_with_opts(api: Arc, file: &str, data: Vec, opts: &ObjectOptions) -> Result<()> { + let size = data.len(); + let _ = api + .put_object(RUSTFS_META_BUCKET, file, &mut PutObjReader::new(Box::new(Cursor::new(data)), size), opts) + .await?; + Ok(()) +} + +/// new server config +/// +/// # Returns +/// NotifierConfig +fn new_server_config() -> NotifierConfig { + NotifierConfig::new() +} + +async fn new_and_save_server_config(api: Arc) -> Result { + let cfg = new_server_config(); + save_server_config(api, &cfg).await?; + + Ok(cfg) +} + +async fn read_server_config(api: Arc, data: &[u8]) -> Result { + let cfg = { + if data.is_empty() { + let config_file = get_event_sys_file(); + let cfg_data = match read_config(api.clone(), config_file.as_str()).await { + Ok(res) => res, + Err(err) => { + return if is_err_config_not_found(&err) { + warn!("config not found init start"); + let cfg = new_and_save_server_config(api).await?; + warn!("config not found init done"); + Ok(cfg) + } else { + error!("read config err {:?}", &err); + Err(err) + } + } + }; + // TODO: decrypt + + NotifierConfig::unmarshal(cfg_data.as_slice())? + } else { + NotifierConfig::unmarshal(data)? + } + }; + + Ok(cfg.merge()) +} + +/// save server config +/// +/// # Parameters +/// - `api`: StorageAPI +/// - `cfg`: configuration to save +/// +/// # Returns +/// Result +async fn save_server_config(api: Arc, cfg: &NotifierConfig) -> Result<()> { + let data = cfg.marshal()?; + + let config_file = get_event_sys_file(); + + save_config(api, &config_file, data).await +} + +/// save config +/// +/// # Parameters +/// - `api`: StorageAPI +/// - `file`: file name +/// - `data`: data to save +/// +/// # Returns +/// Result +pub async fn save_config(api: Arc, file: &str, data: Vec) -> Result<()> { + save_config_with_opts( + api, + file, + data, + &ObjectOptions { + max_parity: true, + ..Default::default() + }, + ) + .await +} + +/// delete config +/// +/// # Parameters +/// - `api`: StorageAPI +/// - `file`: file name +/// +/// # Returns +/// Result +pub async fn delete_config(api: Arc, file: &str) -> Result<()> { + match api + .delete_object( + RUSTFS_META_BUCKET, + file, + ObjectOptions { + delete_prefix: true, + delete_prefix_object: true, + ..Default::default() + }, + ) + .await + { + Ok(_) => Ok(()), + Err(err) => { + if is_err_object_not_found(&err) { + Err(Error::new(ConfigError::NotFound)) + } else { + Err(err) + } + } + } +} + +/// read config +/// +/// # Parameters +/// - `api`: StorageAPI +/// - `file`: file name +/// +/// # Returns +/// Configuration data +pub async fn read_config(api: Arc, file: &str) -> Result> { + let (data, _obj) = read_config_with_metadata(api, file, &ObjectOptions::default()).await?; + Ok(data) +} + +/// read config with metadata +/// +/// # Parameters +/// - `api`: StorageAPI +/// - `file`: file name +/// - `opts`: object options +/// +/// # Returns +/// Configuration data and object info +pub async fn read_config_with_metadata( + api: Arc, + file: &str, + opts: &ObjectOptions, +) -> Result<(Vec, ObjectInfo)> { + let h = HeaderMap::new(); + let mut rd = api + .get_object_reader(RUSTFS_META_BUCKET, file, None, h, opts) + .await + .map_err(|err| { + if is_err_object_not_found(&err) { + Error::new(ConfigError::NotFound) + } else { + err + } + })?; + + let data = rd.read_all().await?; + + if data.is_empty() { + return Err(Error::new(ConfigError::NotFound)); + } + + Ok((data, rd.object_info)) +} diff --git a/crates/event/src/global.rs b/crates/event/src/global.rs index 0ffcb8b00..ab5b727db 100644 --- a/crates/event/src/global.rs +++ b/crates/event/src/global.rs @@ -25,7 +25,8 @@ 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: NotifierConfig) -> Result<(), Error> { +#[instrument] +pub async fn initialize(config: &NotifierConfig) -> Result<(), Error> { let _lock = INIT_LOCK.lock().await; // Check if the system is already initialized. @@ -180,7 +181,7 @@ mod tests { async fn test_initialize_success() { tracing_subscriber::fmt::init(); let config = NotifierConfig::default(); // assume there is a default configuration - let result = initialize(config).await; + let result = initialize(&config).await; assert!(result.is_err(), "Initialization should not succeed"); assert!(!is_initialized(), "System should not be marked as initialized"); assert!(!is_ready(), "System should not be marked as ready"); @@ -190,8 +191,8 @@ mod tests { async fn test_initialize_twice() { tracing_subscriber::fmt::init(); let config = NotifierConfig::default(); - let _ = initialize(config.clone()).await; // first initialization - let result = initialize(config).await; // second initialization + let _ = initialize(&config.clone()).await; // first initialization + let result = initialize(&config).await; // second initialization assert!(result.is_err(), "Initialization should succeed"); assert!(result.is_err(), "Re-initialization should fail"); } @@ -213,7 +214,7 @@ mod tests { ], // assuming that the empty adapter will cause failure ..Default::default() }; - let result = initialize(config).await; + let result = initialize(&config).await; assert!(result.is_ok(), "Initialization with invalid config should fail"); assert!(is_initialized(), "System should not be marked as initialized after failure"); assert!(is_ready(), "System should not be marked as ready after failure"); @@ -226,7 +227,7 @@ mod tests { assert!(!is_ready(), "System should not be ready initially"); let config = NotifierConfig::default(); - let _ = initialize(config).await; + 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/src/lib.rs b/crates/event/src/lib.rs index fe2e5e3da..3b5de3fb1 100644 --- a/crates/event/src/lib.rs +++ b/crates/event/src/lib.rs @@ -3,6 +3,7 @@ mod bus; mod config; mod error; mod event; +mod event_sys; mod global; mod notifier; mod store; @@ -29,3 +30,11 @@ pub use event::{Bucket, Event, EventBuilder, Identity, Log, Metadata, Name, Obje pub use global::{initialize, is_initialized, is_ready, send_event, shutdown}; pub use notifier::NotifierSystem; pub use store::EventStore; + +pub use event_sys::delete_config; +pub use event_sys::get_event_notifier_config; +pub use event_sys::read_config; +pub use event_sys::save_config; + +pub use event_sys::EventSys; +pub use event_sys::GLOBAL_EventSys; diff --git a/ecstore/src/config/com.rs b/ecstore/src/config/com.rs index d08b0da27..ab34ebee5 100644 --- a/ecstore/src/config/com.rs +++ b/ecstore/src/config/com.rs @@ -168,47 +168,45 @@ async fn new_and_save_server_config(api: Arc) -> Result String { + format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE) +} + +/// * read config without migrate pub async fn read_config_without_migrate(api: Arc) -> Result { - let config_file = format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE); - let data = match read_config(api.clone(), config_file.as_str()).await { - Ok(res) => res, - Err(err) => { - return if is_err_config_not_found(&err) { - warn!("config not found, start to init"); - let cfg = new_and_save_server_config(api).await?; - warn!("config init done"); - Ok(cfg) - } else { - error!("read config err {:?}", &err); - Err(err) - } - } + let config_file = get_config_file(); + let data = match read_config_and_init_if_not_found(api.clone(), config_file.as_str()).await? { + Some(data) => data, + None => return Ok(new_and_save_server_config(api).await?), }; read_server_config(api, data.as_slice()).await } +/// read config and init if not found +async fn read_config_and_init_if_not_found(api: Arc, config_file: &str) -> Result>> { + match read_config(api.clone(), config_file).await { + Ok(data) => Ok(Some(data)), + Err(err) => { + if is_err_config_not_found(&err) { + warn!("config not found, start to init"); + return Ok(None); + } + error!("read config err {:?}", &err); + Err(err) + } + } +} + async fn read_server_config(api: Arc, data: &[u8]) -> Result { let cfg = { if data.is_empty() { - let config_file = format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE); - let cfg_data = match read_config(api.clone(), config_file.as_str()).await { - Ok(res) => res, - Err(err) => { - return if is_err_config_not_found(&err) { - warn!("config not found init start"); - let cfg = new_and_save_server_config(api).await?; - warn!("config not found init done"); - Ok(cfg) - } else { - error!("read config err {:?}", &err); - Err(err) - } - } + let config_file = get_config_file(); + let cfg_data = match read_config_and_init_if_not_found(api.clone(), config_file.as_str()).await? { + Some(data) => data, + None => return Ok(new_and_save_server_config(api).await?), }; // TODO: decrypt - Config::unmarshal(cfg_data.as_slice())? } else { Config::unmarshal(data)? @@ -221,7 +219,7 @@ async fn read_server_config(api: Arc, data: &[u8]) -> Result(api: Arc, cfg: &Config) -> Result<()> { let data = cfg.marshal()?; - let config_file = format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE); + let config_file = get_config_file(); save_config(api, &config_file, data).await } diff --git a/rustfs/src/event.rs b/rustfs/src/event.rs index a6819b4cd..e64f12822 100644 --- a/rustfs/src/event.rs +++ b/rustfs/src/event.rs @@ -8,7 +8,7 @@ pub(crate) async fn init_event_notifier(notifier_config: Option) { info!("event_config is not empty"); tokio::spawn(async move { let config = NotifierConfig::event_load_config(notifier_config); - let result = rustfs_event::initialize(config).await; + let result = rustfs_event::initialize(&config).await; if let Err(e) = result { error!("Failed to initialize event notifier: {}", e); } else { @@ -17,5 +17,15 @@ pub(crate) async fn init_event_notifier(notifier_config: Option) { }); } else { info!("event_config is empty"); + tokio::spawn(async move { + let config = rustfs_event::get_event_notifier_config(); + info!("event_config is {:?}", config); + let result = rustfs_event::initialize(config).await; + if let Err(e) = result { + error!("Failed to initialize event notifier: {}", e); + } else { + info!("Event notifier initialized successfully"); + } + }); } } diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index 01d500ed8..e3b867385 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -48,6 +48,7 @@ use iam::init_iam_sys; use license::init_license; use protos::proto_gen::node_service::node_service_server::NodeServiceServer; use rustfs_config::{DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY, RUSTFS_TLS_CERT, RUSTFS_TLS_KEY}; +use rustfs_event::GLOBAL_EventSys; use rustfs_obs::{init_obs, init_process_observer, load_config, set_global_guard}; use rustls::ServerConfig; use s3s::{host::MultiDomain, service::S3ServiceBuilder}; @@ -118,7 +119,7 @@ async fn run(opt: config::Opt) -> Result<()> { debug!("opt: {:?}", &opt); // Initialize event notifier - event::init_event_notifier(opt.event_config).await; + // event::init_event_notifier(opt.event_config).await; let server_addr = net::parse_and_resolve_address(opt.address.as_str())?; let server_port = server_addr.port(); @@ -507,6 +508,11 @@ async fn run(opt: config::Opt) -> Result<()> { ecconfig::init(); GLOBAL_ConfigSys.init(store.clone()).await?; + GLOBAL_EventSys.init(store.clone()).await?; + + // Initialize event notifier + event::init_event_notifier(opt.event_config).await; + let buckets_list = store .list_bucket(&BucketOptions { no_metadata: true,