add

Cargo.lock

@@ -7495,8 +7495,12 @@ version = "0.0.1"
 dependencies = [
  "async-trait",
  "axum",
+ "common",
  "config",
  "dotenvy",
+ "ecstore",
+ "http",
+ "lazy_static",
  "rdkafka",
  "reqwest",
  "rumqttc",

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 }

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 {

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<NotifierConfig> {
+        let m: NotifierConfig = serde_json::from_slice(data)?;
+        Ok(m)
+    }
+
+    /// marshal the configuration to a byte array
+    pub fn marshal(&self) -> common::error::Result<Vec<u8>> {
+        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

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<NotifierConfig> = 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<ECStore>) -> 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<S: StorageAPI>(api: Arc<S>) -> Result<NotifierConfig> {
+    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<S: StorageAPI>(api: Arc<S>, file: &str, data: Vec<u8>, 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<S: StorageAPI>(api: Arc<S>) -> Result<NotifierConfig> {
+    let cfg = new_server_config();
+    save_server_config(api, &cfg).await?;
+
+    Ok(cfg)
+}
+
+async fn read_server_config<S: StorageAPI>(api: Arc<S>, data: &[u8]) -> Result<NotifierConfig> {
+    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<S: StorageAPI>(api: Arc<S>, 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<S: StorageAPI>(api: Arc<S>, file: &str, data: Vec<u8>) -> 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<S: StorageAPI>(api: Arc<S>, 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<S: StorageAPI>(api: Arc<S>, file: &str) -> Result<Vec<u8>> {
+    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<S: StorageAPI>(
+    api: Arc<S>,
+    file: &str,
+    opts: &ObjectOptions,
+) -> Result<(Vec<u8>, 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))
+}

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 _ = initialize(&config.clone()).await; // first initialization
-        let result = initialize(config).await; // second 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");
     }

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;

ecstore/src/config/com.rs

@@ -168,47 +168,45 @@ async fn new_and_save_server_config<S: StorageAPI>(api: Arc<S>) -> Result<Config
     Ok(cfg)
 }
 
-/// * read_config_without_migrate
+fn get_config_file() -> String {
+    format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE)
+}
+
+/// * read config without migrate
 pub async fn read_config_without_migrate<S: StorageAPI>(api: Arc<S>) -> Result<Config> {
-    let config_file = format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE);
+    let config_file = get_config_file();
-    let data = match read_config(api.clone(), config_file.as_str()).await {
+    let data = match read_config_and_init_if_not_found(api.clone(), config_file.as_str()).await? {
-        Ok(res) => res,
+        Some(data) => data,
-        Err(err) => {
+        None => return Ok(new_and_save_server_config(api).await?),
-            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
 }
 
+/// read config and init if not found
+async fn read_config_and_init_if_not_found<S: StorageAPI>(api: Arc<S>, config_file: &str) -> Result<Option<Vec<u8>>> {
+    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<S: StorageAPI>(api: Arc<S>, data: &[u8]) -> Result<Config> {
     let cfg = {
         if data.is_empty() {
-            let config_file = format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE);
+            let config_file = get_config_file();
-            let cfg_data = match read_config(api.clone(), config_file.as_str()).await {
+            let cfg_data = match read_config_and_init_if_not_found(api.clone(), config_file.as_str()).await? {
-                Ok(res) => res,
+                Some(data) => data,
-                Err(err) => {
+                None => return Ok(new_and_save_server_config(api).await?),
-                    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
-
             Config::unmarshal(cfg_data.as_slice())?
         } else {
             Config::unmarshal(data)?
@@ -221,7 +219,7 @@ async fn read_server_config<S: StorageAPI>(api: Arc<S>, data: &[u8]) -> Result<C
 async fn save_server_config<S: StorageAPI>(api: Arc<S>, 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
 }

rustfs/src/event.rs

@@ -8,7 +8,7 @@ pub(crate) async fn init_event_notifier(notifier_config: Option<String>) {
         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<String>) {
         });
     } 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");
+            }
+        });
     }
 }

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,