refactor: replace lazy_static with LazyLock and notify crate registry create_targets_from_config (#311)

* improve code for notify * improve code for logger and fix typo (#272) * Add GNU to build.yml (#275) * fix unzip error * fix url change error fix url change error * Simplify user experience and integrate console and endpoint Simplify user experience and integrate console and endpoint * Add gnu to build.yml * upgrade version * feat: add `cargo clippy --fix --allow-dirty` to pre-commit command (#282) Resolves #277 - Add --fix flag to automatically fix clippy warnings - Add --allow-dirty flag to run on dirty Git trees - Improves code quality in pre-commit workflow * fix: the issue where preview fails when the path length exceeds 255 characters (#280) * fix * fix: improve Windows build support and CI/CD workflow (#283) - Fix Windows zip command issue by using PowerShell Compress-Archive - Add Windows support for OSS upload with ossutil - Replace Chinese comments with English in build.yml - Fix bash syntax error in package_zip function - Improve code formatting and consistency - Update various configuration files for better cross-platform support Resolves Windows build failures in GitHub Actions. * fix: update link in README.md leading to a 404 error (#285) * add rustfs.spec for rustfs (#103) add support on loongarch64 * improve cargo.lock * build(deps): bump the dependencies group with 5 updates (#289) Bumps the dependencies group with 5 updates: | Package | From | To | | --- | --- | --- | | [hyper-util](https://github.com/hyperium/hyper-util) | `0.1.15` | `0.1.16` | | [rand](https://github.com/rust-random/rand) | `0.9.1` | `0.9.2` | | [serde_json](https://github.com/serde-rs/json) | `1.0.140` | `1.0.141` | | [strum](https://github.com/Peternator7/strum) | `0.27.1` | `0.27.2` | | [sysinfo](https://github.com/GuillaumeGomez/sysinfo) | `0.36.0` | `0.36.1` | Updates `hyper-util` from 0.1.15 to 0.1.16 - [Release notes](https://github.com/hyperium/hyper-util/releases) - [Changelog](https://github.com/hyperium/hyper-util/blob/master/CHANGELOG.md) - [Commits](https://github.com/hyperium/hyper-util/compare/v0.1.15...v0.1.16) Updates `rand` from 0.9.1 to 0.9.2 - [Release notes](https://github.com/rust-random/rand/releases) - [Changelog](https://github.com/rust-random/rand/blob/master/CHANGELOG.md) - [Commits](https://github.com/rust-random/rand/compare/rand_core-0.9.1...rand_core-0.9.2) Updates `serde_json` from 1.0.140 to 1.0.141 - [Release notes](https://github.com/serde-rs/json/releases) - [Commits](https://github.com/serde-rs/json/compare/v1.0.140...v1.0.141) Updates `strum` from 0.27.1 to 0.27.2 - [Release notes](https://github.com/Peternator7/strum/releases) - [Changelog](https://github.com/Peternator7/strum/blob/master/CHANGELOG.md) - [Commits](https://github.com/Peternator7/strum/compare/v0.27.1...v0.27.2) Updates `sysinfo` from 0.36.0 to 0.36.1 - [Changelog](https://github.com/GuillaumeGomez/sysinfo/blob/master/CHANGELOG.md) - [Commits](https://github.com/GuillaumeGomez/sysinfo/compare/v0.36.0...v0.36.1) --- updated-dependencies: - dependency-name: hyper-util dependency-version: 0.1.16 dependency-type: direct:production update-type: version-update:semver-patch dependency-group: dependencies - dependency-name: rand dependency-version: 0.9.2 dependency-type: direct:production update-type: version-update:semver-patch dependency-group: dependencies - dependency-name: serde_json dependency-version: 1.0.141 dependency-type: direct:production update-type: version-update:semver-patch dependency-group: dependencies - dependency-name: strum dependency-version: 0.27.2 dependency-type: direct:production update-type: version-update:semver-patch dependency-group: dependencies - dependency-name: sysinfo dependency-version: 0.36.1 dependency-type: direct:production update-type: version-update:semver-patch dependency-group: dependencies ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * improve code for logger * improve * upgrade * refactor: 优化构建工作流，统一 latest 文件处理和简化制品上传 (#293) * Refactor: DatabaseManagerSystem as global Signed-off-by: junxiang Mu <1948535941@qq.com> * fix: fmt Signed-off-by: junxiang Mu <1948535941@qq.com> * Test: add e2e_test for s3select Signed-off-by: junxiang Mu <1948535941@qq.com> * Test: add test script for e2e Signed-off-by: junxiang Mu <1948535941@qq.com> * improve code for registry and intergation * improve code for registry `create_targets_from_config` * fix * Feature up/ilm (#305) * fix * fix * fix * fix delete-marker expiration. add api_restore. * fix * time retry object upload * lock file * make fmt * fix * restore object * fix * fix * serde-rs-xml -> quick-xml * fix * checksum * fix * fix * fix * fix * fix * fix * fix * transfer lang to english * upgrade clap version from 4.5.41 to 4.5.42 * refactor: replace `lazy_static` with `LazyLock` * add router * fix: modify comment * improve code * fix typos * fix * fix: modify name and fmt * improve code for registry * fix test --------- Signed-off-by: dependabot[bot] <support@github.com> Signed-off-by: junxiang Mu <1948535941@qq.com> Co-authored-by: loverustfs <155562731+loverustfs@users.noreply.github.com> Co-authored-by: 安正超 <anzhengchao@gmail.com> Co-authored-by: shiro.lee <69624924+shiroleeee@users.noreply.github.com> Co-authored-by: Marco Orlandin <mipnamic@mipnamic.net> Co-authored-by: zhangwenlong <zhangwenlong@loongson.cn> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: junxiang Mu <1948535941@qq.com> Co-authored-by: likewu <likewu@126.com>
2026-08-08 06:13:14 +00:00 · 2025-07-30 19:02:10 +08:00
parent 4f1770d3fe
commit c5df1f92c2
49 changed files with 1575 additions and 1435 deletions
@@ -82,6 +82,15 @@ pub enum TargetError {

    #[error("Target is disabled")]
    Disabled,
+
+    #[error("Configuration parsing error: {0}")]
+    ParseError(String),
+
+    #[error("Failed to save configuration: {0}")]
+    SaveConfig(String),
+
+    #[error("Server not initialized: {0}")]
+    ServerNotInitialized(String),
 }

 /// Error types for the notification system
@@ -112,7 +121,7 @@ pub enum NotificationError {
    AlreadyInitialized,

    #[error("I/O error: {0}")]
-    Io(std::io::Error),
+    Io(io::Error),

    #[error("Failed to read configuration: {0}")]
    ReadConfig(String),
@@ -19,40 +19,17 @@ use crate::{
 use async_trait::async_trait;
 use rumqttc::QoS;
 use rustfs_config::notify::{
-    DEFAULT_DIR, DEFAULT_LIMIT, ENV_MQTT_BROKER, ENV_MQTT_ENABLE, ENV_MQTT_KEEP_ALIVE_INTERVAL, ENV_MQTT_PASSWORD, ENV_MQTT_QOS,
-    ENV_MQTT_QUEUE_DIR, ENV_MQTT_QUEUE_LIMIT, ENV_MQTT_RECONNECT_INTERVAL, ENV_MQTT_TOPIC, ENV_MQTT_USERNAME,
-    ENV_WEBHOOK_AUTH_TOKEN, ENV_WEBHOOK_CLIENT_CERT, ENV_WEBHOOK_CLIENT_KEY, ENV_WEBHOOK_ENABLE, ENV_WEBHOOK_ENDPOINT,
-    ENV_WEBHOOK_QUEUE_DIR, ENV_WEBHOOK_QUEUE_LIMIT, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD, MQTT_QOS,
-    MQTT_QUEUE_DIR, MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TOPIC, MQTT_USERNAME, WEBHOOK_AUTH_TOKEN,
-    WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_QUEUE_DIR, WEBHOOK_QUEUE_LIMIT,
+    DEFAULT_DIR, DEFAULT_LIMIT, ENV_NOTIFY_MQTT_KEYS, ENV_NOTIFY_WEBHOOK_KEYS, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL,
+    MQTT_PASSWORD, MQTT_QOS, MQTT_QUEUE_DIR, MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TOPIC, MQTT_USERNAME,
+    NOTIFY_MQTT_KEYS, NOTIFY_WEBHOOK_KEYS, WEBHOOK_AUTH_TOKEN, WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT,
+    WEBHOOK_QUEUE_DIR, WEBHOOK_QUEUE_LIMIT,
 };
-use rustfs_config::{DEFAULT_DELIMITER, ENV_WORD_DELIMITER_DASH};
-use rustfs_ecstore::config::{ENABLE_KEY, ENABLE_ON, KVS};
+use rustfs_ecstore::config::KVS;
+use std::collections::HashSet;
 use std::time::Duration;
 use tracing::{debug, warn};
 use url::Url;

-/// Helper function to get values from environment variables or KVS configurations.
-///
-/// It will give priority to reading from environment variables such as `BASE_ENV_KEY_ID` and fall back to the KVS configuration if it fails.
-fn get_config_value(id: &str, base_env_key: &str, config_key: &str, config: &KVS) -> Option<String> {
-    let env_key = if id != DEFAULT_DELIMITER {
-        format!(
-            "{}{}{}",
-            base_env_key,
-            DEFAULT_DELIMITER,
-            id.to_uppercase().replace(ENV_WORD_DELIMITER_DASH, DEFAULT_DELIMITER)
-        )
-    } else {
-        base_env_key.to_string()
-    };
-
-    match std::env::var(&env_key) {
-        Ok(val) => Some(val),
-        Err(_) => config.lookup(config_key),
-    }
-}
-
 /// Trait for creating targets from configuration
 #[async_trait]
 pub trait TargetFactory: Send + Sync {
@@ -61,6 +38,14 @@ pub trait TargetFactory: Send + Sync {

    /// Validates target configuration
    fn validate_config(&self, id: &str, config: &KVS) -> Result<(), TargetError>;
+
+    /// Returns a set of valid configuration field names for this target type.
+    /// This is used to filter environment variables.
+    fn get_valid_fields(&self) -> HashSet<String>;
+
+    /// Returns a set of valid configuration env field names for this target type.
+    /// This is used to filter environment variables.
+    fn get_valid_env_fields(&self) -> HashSet<String>;
 }

 /// Factory for creating Webhook targets
@@ -69,65 +54,42 @@ pub struct WebhookTargetFactory;
 #[async_trait]
 impl TargetFactory for WebhookTargetFactory {
    async fn create_target(&self, id: String, config: &KVS) -> Result<Box<dyn Target + Send + Sync>, TargetError> {
-        let get = |base_env_key: &str, config_key: &str| get_config_value(&id, base_env_key, config_key, config);
-
-        let enable = get(ENV_WEBHOOK_ENABLE, ENABLE_KEY)
-            .map(|v| v.eq_ignore_ascii_case(ENABLE_ON) || v.eq_ignore_ascii_case("true"))
-            .unwrap_or(false);
-
-        if !enable {
-            return Err(TargetError::Configuration("Target is disabled".to_string()));
-        }
-
-        let endpoint = get(ENV_WEBHOOK_ENDPOINT, WEBHOOK_ENDPOINT)
+        // All config values are now read directly from the merged `config` KVS.
+        let endpoint = config
+            .lookup(WEBHOOK_ENDPOINT)
            .ok_or_else(|| TargetError::Configuration("Missing webhook endpoint".to_string()))?;
        let endpoint_url = Url::parse(&endpoint)
            .map_err(|e| TargetError::Configuration(format!("Invalid endpoint URL: {e} (value: '{endpoint}')")))?;

-        let auth_token = get(ENV_WEBHOOK_AUTH_TOKEN, WEBHOOK_AUTH_TOKEN).unwrap_or_default();
-        let queue_dir = get(ENV_WEBHOOK_QUEUE_DIR, WEBHOOK_QUEUE_DIR).unwrap_or(DEFAULT_DIR.to_string());
-
-        let queue_limit = get(ENV_WEBHOOK_QUEUE_LIMIT, WEBHOOK_QUEUE_LIMIT)
-            .and_then(|v| v.parse::<u64>().ok())
-            .unwrap_or(DEFAULT_LIMIT);
-
-        let client_cert = get(ENV_WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_CERT).unwrap_or_default();
-        let client_key = get(ENV_WEBHOOK_CLIENT_KEY, WEBHOOK_CLIENT_KEY).unwrap_or_default();
-
        let args = WebhookArgs {
-            enable,
+            enable: true, // If we are here, it's already enabled.
            endpoint: endpoint_url,
-            auth_token,
-            queue_dir,
-            queue_limit,
-            client_cert,
-            client_key,
+            auth_token: config.lookup(WEBHOOK_AUTH_TOKEN).unwrap_or_default(),
+            queue_dir: config.lookup(WEBHOOK_QUEUE_DIR).unwrap_or(DEFAULT_DIR.to_string()),
+            queue_limit: config
+                .lookup(WEBHOOK_QUEUE_LIMIT)
+                .and_then(|v| v.parse::<u64>().ok())
+                .unwrap_or(DEFAULT_LIMIT),
+            client_cert: config.lookup(WEBHOOK_CLIENT_CERT).unwrap_or_default(),
+            client_key: config.lookup(WEBHOOK_CLIENT_KEY).unwrap_or_default(),
        };

        let target = crate::target::webhook::WebhookTarget::new(id, args)?;
        Ok(Box::new(target))
    }

-    fn validate_config(&self, id: &str, config: &KVS) -> Result<(), TargetError> {
-        let get = |base_env_key: &str, config_key: &str| get_config_value(id, base_env_key, config_key, config);
-
-        let enable = get(ENV_WEBHOOK_ENABLE, ENABLE_KEY)
-            .map(|v| v.eq_ignore_ascii_case(ENABLE_ON) || v.eq_ignore_ascii_case("true"))
-            .unwrap_or(false);
-
-        if !enable {
-            return Ok(());
-        }
-
-        let endpoint = get(ENV_WEBHOOK_ENDPOINT, WEBHOOK_ENDPOINT)
+    fn validate_config(&self, _id: &str, config: &KVS) -> Result<(), TargetError> {
+        // Validation also uses the merged `config` KVS directly.
+        let endpoint = config
+            .lookup(WEBHOOK_ENDPOINT)
            .ok_or_else(|| TargetError::Configuration("Missing webhook endpoint".to_string()))?;
        debug!("endpoint: {}", endpoint);
        let parsed_endpoint = endpoint.trim();
        Url::parse(parsed_endpoint)
            .map_err(|e| TargetError::Configuration(format!("Invalid endpoint URL: {e} (value: '{parsed_endpoint}')")))?;

-        let client_cert = get(ENV_WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_CERT).unwrap_or_default();
-        let client_key = get(ENV_WEBHOOK_CLIENT_KEY, WEBHOOK_CLIENT_KEY).unwrap_or_default();
+        let client_cert = config.lookup(WEBHOOK_CLIENT_CERT).unwrap_or_default();
+        let client_key = config.lookup(WEBHOOK_CLIENT_KEY).unwrap_or_default();

        if client_cert.is_empty() != client_key.is_empty() {
            return Err(TargetError::Configuration(
@@ -135,15 +97,21 @@ impl TargetFactory for WebhookTargetFactory {
            ));
        }

-        let queue_dir = get(ENV_WEBHOOK_QUEUE_DIR, WEBHOOK_QUEUE_DIR)
-            .and_then(|v| v.parse::<String>().ok())
-            .unwrap_or(DEFAULT_DIR.to_string());
+        let queue_dir = config.lookup(WEBHOOK_QUEUE_DIR).unwrap_or(DEFAULT_DIR.to_string());
        if !queue_dir.is_empty() && !std::path::Path::new(&queue_dir).is_absolute() {
            return Err(TargetError::Configuration("Webhook queue directory must be an absolute path".to_string()));
        }

        Ok(())
    }
+
+    fn get_valid_fields(&self) -> HashSet<String> {
+        NOTIFY_WEBHOOK_KEYS.iter().map(|s| s.to_string()).collect()
+    }
+
+    fn get_valid_env_fields(&self) -> HashSet<String> {
+        ENV_NOTIFY_WEBHOOK_KEYS.iter().map(|s| s.to_string()).collect()
+    }
 }

 /// Factory for creating MQTT targets
@@ -152,84 +120,57 @@ pub struct MQTTTargetFactory;
 #[async_trait]
 impl TargetFactory for MQTTTargetFactory {
    async fn create_target(&self, id: String, config: &KVS) -> Result<Box<dyn Target + Send + Sync>, TargetError> {
-        let get = |base_env_key: &str, config_key: &str| get_config_value(&id, base_env_key, config_key, config);
-
-        let enable = get(ENV_MQTT_ENABLE, ENABLE_KEY)
-            .map(|v| v.eq_ignore_ascii_case(ENABLE_ON) || v.eq_ignore_ascii_case("true"))
-            .unwrap_or(false);
-
-        if !enable {
-            return Err(TargetError::Configuration("Target is disabled".to_string()));
-        }
-
-        let broker =
-            get(ENV_MQTT_BROKER, MQTT_BROKER).ok_or_else(|| TargetError::Configuration("Missing MQTT broker".to_string()))?;
+        let broker = config
+            .lookup(MQTT_BROKER)
+            .ok_or_else(|| TargetError::Configuration("Missing MQTT broker".to_string()))?;
        let broker_url = Url::parse(&broker)
            .map_err(|e| TargetError::Configuration(format!("Invalid broker URL: {e} (value: '{broker}')")))?;

-        let topic =
-            get(ENV_MQTT_TOPIC, MQTT_TOPIC).ok_or_else(|| TargetError::Configuration("Missing MQTT topic".to_string()))?;
-
-        let qos = get(ENV_MQTT_QOS, MQTT_QOS)
-            .and_then(|v| v.parse::<u8>().ok())
-            .map(|q| match q {
-                0 => QoS::AtMostOnce,
-                1 => QoS::AtLeastOnce,
-                2 => QoS::ExactlyOnce,
-                _ => QoS::AtLeastOnce,
-            })
-            .unwrap_or(QoS::AtLeastOnce);
-
-        let username = get(ENV_MQTT_USERNAME, MQTT_USERNAME).unwrap_or_default();
-        let password = get(ENV_MQTT_PASSWORD, MQTT_PASSWORD).unwrap_or_default();
-
-        let reconnect_interval = get(ENV_MQTT_RECONNECT_INTERVAL, MQTT_RECONNECT_INTERVAL)
-            .and_then(|v| v.parse::<u64>().ok())
-            .map(Duration::from_secs)
-            .unwrap_or_else(|| Duration::from_secs(5));
-
-        let keep_alive = get(ENV_MQTT_KEEP_ALIVE_INTERVAL, MQTT_KEEP_ALIVE_INTERVAL)
-            .and_then(|v| v.parse::<u64>().ok())
-            .map(Duration::from_secs)
-            .unwrap_or_else(|| Duration::from_secs(30));
-
-        let queue_dir = get(ENV_MQTT_QUEUE_DIR, MQTT_QUEUE_DIR)
-            .and_then(|v| v.parse::<String>().ok())
-            .unwrap_or(DEFAULT_DIR.to_string());
-        let queue_limit = get(ENV_MQTT_QUEUE_LIMIT, MQTT_QUEUE_LIMIT)
-            .and_then(|v| v.parse::<u64>().ok())
-            .unwrap_or(DEFAULT_LIMIT);
+        let topic = config
+            .lookup(MQTT_TOPIC)
+            .ok_or_else(|| TargetError::Configuration("Missing MQTT topic".to_string()))?;

        let args = MQTTArgs {
-            enable,
+            enable: true, // Assumed enabled.
            broker: broker_url,
            topic,
-            qos,
-            username,
-            password,
-            max_reconnect_interval: reconnect_interval,
-            keep_alive,
-            queue_dir,
-            queue_limit,
+            qos: config
+                .lookup(MQTT_QOS)
+                .and_then(|v| v.parse::<u8>().ok())
+                .map(|q| match q {
+                    0 => QoS::AtMostOnce,
+                    1 => QoS::AtLeastOnce,
+                    2 => QoS::ExactlyOnce,
+                    _ => QoS::AtLeastOnce,
+                })
+                .unwrap_or(QoS::AtLeastOnce),
+            username: config.lookup(MQTT_USERNAME).unwrap_or_default(),
+            password: config.lookup(MQTT_PASSWORD).unwrap_or_default(),
+            max_reconnect_interval: config
+                .lookup(MQTT_RECONNECT_INTERVAL)
+                .and_then(|v| v.parse::<u64>().ok())
+                .map(Duration::from_secs)
+                .unwrap_or_else(|| Duration::from_secs(5)),
+            keep_alive: config
+                .lookup(MQTT_KEEP_ALIVE_INTERVAL)
+                .and_then(|v| v.parse::<u64>().ok())
+                .map(Duration::from_secs)
+                .unwrap_or_else(|| Duration::from_secs(30)),
+            queue_dir: config.lookup(MQTT_QUEUE_DIR).unwrap_or(DEFAULT_DIR.to_string()),
+            queue_limit: config
+                .lookup(MQTT_QUEUE_LIMIT)
+                .and_then(|v| v.parse::<u64>().ok())
+                .unwrap_or(DEFAULT_LIMIT),
        };

        let target = crate::target::mqtt::MQTTTarget::new(id, args)?;
        Ok(Box::new(target))
    }

-    fn validate_config(&self, id: &str, config: &KVS) -> Result<(), TargetError> {
-        let get = |base_env_key: &str, config_key: &str| get_config_value(id, base_env_key, config_key, config);
-
-        let enable = get(ENV_MQTT_ENABLE, ENABLE_KEY)
-            .map(|v| v.eq_ignore_ascii_case(ENABLE_ON) || v.eq_ignore_ascii_case("true"))
-            .unwrap_or(false);
-
-        if !enable {
-            return Ok(());
-        }
-
-        let broker =
-            get(ENV_MQTT_BROKER, MQTT_BROKER).ok_or_else(|| TargetError::Configuration("Missing MQTT broker".to_string()))?;
+    fn validate_config(&self, _id: &str, config: &KVS) -> Result<(), TargetError> {
+        let broker = config
+            .lookup(MQTT_BROKER)
+            .ok_or_else(|| TargetError::Configuration("Missing MQTT broker".to_string()))?;
        let url = Url::parse(&broker)
            .map_err(|e| TargetError::Configuration(format!("Invalid broker URL: {e} (value: '{broker}')")))?;

@@ -240,11 +181,11 @@ impl TargetFactory for MQTTTargetFactory {
            }
        }

-        if get(ENV_MQTT_TOPIC, MQTT_TOPIC).is_none() {
+        if config.lookup(MQTT_TOPIC).is_none() {
            return Err(TargetError::Configuration("Missing MQTT topic".to_string()));
        }

-        if let Some(qos_str) = get(ENV_MQTT_QOS, MQTT_QOS) {
+        if let Some(qos_str) = config.lookup(MQTT_QOS) {
            let qos = qos_str
                .parse::<u8>()
                .map_err(|_| TargetError::Configuration("Invalid QoS value".to_string()))?;
@@ -253,14 +194,12 @@ impl TargetFactory for MQTTTargetFactory {
            }
        }

-        let queue_dir = get(ENV_MQTT_QUEUE_DIR, MQTT_QUEUE_DIR)
-            .and_then(|v| v.parse::<String>().ok())
-            .unwrap_or(DEFAULT_DIR.to_string());
+        let queue_dir = config.lookup(MQTT_QUEUE_DIR).unwrap_or_default();
        if !queue_dir.is_empty() {
            if !std::path::Path::new(&queue_dir).is_absolute() {
                return Err(TargetError::Configuration("MQTT queue directory must be an absolute path".to_string()));
            }
-            if let Some(qos_str) = get(ENV_MQTT_QOS, MQTT_QOS) {
+            if let Some(qos_str) = config.lookup(MQTT_QOS) {
                if qos_str == "0" {
                    warn!("Using queue_dir with QoS 0 may result in event loss");
                }
@@ -269,4 +208,12 @@ impl TargetFactory for MQTTTargetFactory {

        Ok(())
    }
+
+    fn get_valid_fields(&self) -> HashSet<String> {
+        NOTIFY_MQTT_KEYS.iter().map(|s| s.to_string()).collect()
+    }
+
+    fn get_valid_env_fields(&self) -> HashSet<String> {
+        ENV_NOTIFY_MQTT_KEYS.iter().map(|s| s.to_string()).collect()
+    }
 }
@@ -210,10 +210,10 @@ impl NotificationSystem {
            return Ok(());
        }

-        if let Err(e) = rustfs_ecstore::config::com::save_server_config(store, &new_config).await {
-            error!("Failed to save config: {}", e);
-            return Err(NotificationError::SaveConfig(e.to_string()));
-        }
+        // if let Err(e) = rustfs_ecstore::config::com::save_server_config(store, &new_config).await {
+        //     error!("Failed to save config: {}", e);
+        //     return Err(NotificationError::SaveConfig(e.to_string()));
+        // }

        info!("Configuration updated. Reloading system...");
        self.reload_config(new_config).await
@@ -323,7 +323,6 @@ impl NotificationSystem {
        metrics: Arc<NotificationMetrics>,
        semaphore: Arc<Semaphore>,
    ) -> mpsc::Sender<()> {
-        // Event Stream Processing Using Batch Version
        stream::start_event_stream_with_batching(store, target, metrics, semaphore)
    }

@@ -348,6 +347,7 @@ impl NotificationSystem {
        self.update_config(new_config.clone()).await;

        // Create a new target from configuration
+        // This function will now be responsible for merging env, creating and persisting the final configuration.
        let targets: Vec<Box<dyn Target + Send + Sync>> = self
            .registry
            .create_targets_from_config(&new_config)
@@ -18,11 +18,12 @@ use crate::{
    factory::{MQTTTargetFactory, TargetFactory, WebhookTargetFactory},
    target::Target,
 };
-use rustfs_config::notify::NOTIFY_ROUTE_PREFIX;
+use futures::stream::{FuturesUnordered, StreamExt};
+use rustfs_config::notify::{ENABLE_KEY, ENABLE_ON, NOTIFY_ROUTE_PREFIX};
 use rustfs_config::{DEFAULT_DELIMITER, ENV_PREFIX};
-use rustfs_ecstore::config::{Config, ENABLE_KEY, ENABLE_OFF, ENABLE_ON, KVS};
-use std::collections::HashMap;
-use tracing::{debug, error, info};
+use rustfs_ecstore::config::{Config, KVS};
+use std::collections::{HashMap, HashSet};
+use tracing::{debug, error, info, warn};

 /// Registry for managing target factories
 pub struct TargetRegistry {
@@ -74,77 +75,204 @@ impl TargetRegistry {
    }

    /// Creates all targets from a configuration
+    /// Create all notification targets from system configuration and environment variables.
+    /// This method processes the creation of each target concurrently as follows:
+    /// 1. Iterate through all registered target types (e.g. webhooks, mqtt).
+    /// 2. For each type, resolve its configuration in the configuration file and environment variables.
+    /// 3. Identify all target instance IDs that need to be created.
+    /// 4. Combine the default configuration, file configuration, and environment variable configuration for each instance.
+    /// 5. If the instance is enabled, create an asynchronous task for it to instantiate.
+    /// 6. Concurrency executes all creation tasks and collects results.
    pub async fn create_targets_from_config(&self, config: &Config) -> Result<Vec<Box<dyn Target + Send + Sync>>, TargetError> {
-        let mut targets: Vec<Box<dyn Target + Send + Sync>> = Vec::new();
+        // Collect only environment variables with the relevant prefix to reduce memory usage
+        let all_env: Vec<(String, String)> = std::env::vars().filter(|(key, _)| key.starts_with(ENV_PREFIX)).collect();
+        // A collection of asynchronous tasks for concurrently executing target creation
+        let mut tasks = FuturesUnordered::new();
+        let mut final_config = config.clone(); // Clone a configuration for aggregating the final result
+        // 1. Traverse all registered plants and process them by target type
+        for (target_type, factory) in &self.factories {
+            tracing::Span::current().record("target_type", target_type.as_str());
+            info!("Start working on target types...");

-        // Iterate through configuration sections
-        for (section, subsections) in &config.0 {
-            // Only process notification sections
-            if !section.starts_with(NOTIFY_ROUTE_PREFIX) {
-                continue;
+            // 2. Prepare the configuration source
+            // 2.1. Get the configuration segment in the file, e.g. 'notify_webhook'
+            let section_name = format!("{NOTIFY_ROUTE_PREFIX}{target_type}");
+            let file_configs = config.0.get(&section_name).cloned().unwrap_or_default();
+            // 2.2. Get the default configuration for that type
+            let default_cfg = file_configs.get(DEFAULT_DELIMITER).cloned().unwrap_or_default();
+            debug!(?default_cfg, "Get the default configuration");
+
+            // *** Optimization point 1: Get all legitimate fields of the current target type ***
+            let valid_fields = factory.get_valid_fields();
+            debug!(?valid_fields, "Get the legitimate configuration fields");
+
+            // 3. Resolve instance IDs and configuration overrides from environment variables
+            let mut instance_ids_from_env = HashSet::new();
+            // 3.1. Instance discovery: Based on the '..._ENABLE_INSTANCEID' format
+            let enable_prefix = format!("{ENV_PREFIX}{NOTIFY_ROUTE_PREFIX}{target_type}_{ENABLE_KEY}_").to_uppercase();
+            for (key, value) in &all_env {
+                if value.eq_ignore_ascii_case(ENABLE_ON)
+                    || value.eq_ignore_ascii_case("true")
+                    || value.eq_ignore_ascii_case("1")
+                    || value.eq_ignore_ascii_case("yes")
+                {
+                    if let Some(id) = key.strip_prefix(&enable_prefix) {
+                        if !id.is_empty() {
+                            instance_ids_from_env.insert(id.to_lowercase());
+                        }
+                    }
+                }
            }

-            // Extract target type from section name
-            let target_type = section.trim_start_matches(NOTIFY_ROUTE_PREFIX);
+            // 3.2. Parse all relevant environment variable configurations
+            // 3.2.1. Build environment variable prefixes such as 'RUSTFS_NOTIFY_WEBHOOK_'
+            let env_prefix = format!("{ENV_PREFIX}{NOTIFY_ROUTE_PREFIX}{target_type}_").to_uppercase();
+            // 3.2.2. 'env_overrides' is used to store configurations parsed from environment variables in the format: {instance id -> {field -> value}}
+            let mut env_overrides: HashMap<String, HashMap<String, String>> = HashMap::new();
+            for (key, value) in &all_env {
+                if let Some(rest) = key.strip_prefix(&env_prefix) {
+                    // Use rsplitn to split from the right side to properly extract the INSTANCE_ID at the end
+                    // Format: <FIELD_NAME>_<INSTANCE_ID> or <FIELD_NAME>
+                    let mut parts = rest.rsplitn(2, '_');

-            // Iterate through subsections (each representing a target instance)
-            for (target_id, target_config) in subsections {
-                // Skip disabled targets
+                    // The first part from the right is INSTANCE_ID
+                    let instance_id_part = parts.next().unwrap_or(DEFAULT_DELIMITER);
+                    // The remaining part is FIELD_NAME
+                    let field_name_part = parts.next();

-                let enable_from_config = target_config.lookup(ENABLE_KEY).unwrap_or_else(|| ENABLE_OFF.to_string());
-                debug!("Target enablement from config: {}/{}: {}", target_type, target_id, enable_from_config);
-                // Check environment variable for target enablement example: RUSTFS_NOTIFY_WEBHOOK_ENABLE|RUSTFS_NOTIFY_WEBHOOK_ENABLE_[TARGET_ID]
-                let env_key = if target_id == DEFAULT_DELIMITER {
-                    // If no specific target ID, use the base target type, example: RUSTFS_NOTIFY_WEBHOOK_ENABLE
-                    format!(
-                        "{}{}{}{}{}",
-                        ENV_PREFIX,
-                        NOTIFY_ROUTE_PREFIX,
-                        target_type.to_uppercase(),
-                        DEFAULT_DELIMITER,
-                        ENABLE_KEY
-                    )
-                } else {
-                    // If specific target ID, append it to the key, example: RUSTFS_NOTIFY_WEBHOOK_ENABLE_[TARGET_ID]
-                    format!(
-                        "{}{}{}{}{}{}{}",
-                        ENV_PREFIX,
-                        NOTIFY_ROUTE_PREFIX,
-                        target_type.to_uppercase(),
-                        DEFAULT_DELIMITER,
-                        ENABLE_KEY,
-                        DEFAULT_DELIMITER,
-                        target_id.to_uppercase()
-                    )
+                    let (field_name, instance_id) = match field_name_part {
+                        // Case 1: The format is <FIELD_NAME>_<INSTANCE_ID>
+                        // e.g., rest = "ENDPOINT_PRIMARY" -> field_name="ENDPOINT", instance_id="PRIMARY"
+                        Some(field) => (field.to_lowercase(), instance_id_part.to_lowercase()),
+                        // Case 2: The format is <FIELD_NAME> (无 INSTANCE_ID)
+                        // e.g., rest = "ENABLE" -> field_name="ENABLE", instance_id="" (Universal configuration `_ DEFAULT_DELIMITER`)
+                        None => (instance_id_part.to_lowercase(), DEFAULT_DELIMITER.to_string()),
+                    };
+
+                    // *** Optimization point 2: Verify whether the parsed field_name is legal ***
+                    if !field_name.is_empty() && valid_fields.contains(&field_name) {
+                        debug!(
+                            instance_id = %if instance_id.is_empty() { DEFAULT_DELIMITER } else { &instance_id },
+                            %field_name,
+                            %value,
+                            "Parsing to environment variables"
+                        );
+                        env_overrides
+                            .entry(instance_id)
+                            .or_default()
+                            .insert(field_name, value.clone());
+                    } else {
+                        // Ignore illegal field names
+                        warn!(
+                            field_name = %field_name,
+                            "Ignore environment variable fields, not found in the list of valid fields for target type {}",
+                            target_type
+                        );
+                    }
                }
-                .to_uppercase();
-                debug!("Target env key: {},Target id: {}", env_key, target_id);
-                let enable_from_env = std::env::var(&env_key)
-                    .map(|v| v.eq_ignore_ascii_case(ENABLE_ON) || v.eq_ignore_ascii_case("true"))
+            }
+            debug!(?env_overrides, "Complete the environment variable analysis");
+
+            // 4. Determine all instance IDs that need to be processed
+            let mut all_instance_ids: HashSet<String> =
+                file_configs.keys().filter(|k| *k != DEFAULT_DELIMITER).cloned().collect();
+            all_instance_ids.extend(instance_ids_from_env);
+            debug!(?all_instance_ids, "Determine all instance IDs");
+
+            // 5. Merge configurations and create tasks for each instance
+            for id in all_instance_ids {
+                // 5.1. Merge configuration, priority: Environment variables > File instance configuration > File default configuration
+                let mut merged_config = default_cfg.clone();
+                // Instance-specific configuration in application files
+                if let Some(file_instance_cfg) = file_configs.get(&id) {
+                    merged_config.extend(file_instance_cfg.clone());
+                }
+                // Application instance-specific environment variable configuration
+                if let Some(env_instance_cfg) = env_overrides.get(&id) {
+                    // Convert HashMap<String, String> to KVS
+                    let mut kvs_from_env = KVS::new();
+                    for (k, v) in env_instance_cfg {
+                        kvs_from_env.insert(k.clone(), v.clone());
+                    }
+                    merged_config.extend(kvs_from_env);
+                }
+                debug!(instance_id = %id, ?merged_config, "Complete configuration merge");
+
+                // 5.2. Check if the instance is enabled
+                let enabled = merged_config
+                    .lookup(ENABLE_KEY)
+                    .map(|v| {
+                        v.eq_ignore_ascii_case(ENABLE_ON)
+                            || v.eq_ignore_ascii_case("true")
+                            || v.eq_ignore_ascii_case("1")
+                            || v.eq_ignore_ascii_case("yes")
+                    })
                    .unwrap_or(false);
-                debug!("Target env value: {},key: {},Target id: {}", enable_from_env, env_key, target_id);
-                debug!(
-                    "Target enablement from env: {}/{}: result: {}",
-                    target_type, target_id, enable_from_config
-                );
-                if enable_from_config != ENABLE_ON && !enable_from_env {
-                    info!("Skipping disabled target: {}/{}", target_type, target_id);
-                    continue;
-                }
-                debug!("create target: {}/{} start", target_type, target_id);
-                // Create target
-                match self.create_target(target_type, target_id.clone(), target_config).await {
-                    Ok(target) => {
-                        info!("Created target: {}/{}", target_type, target_id);
-                        targets.push(target);
-                    }
-                    Err(e) => {
-                        error!("Failed to create target {}/{}: reason: {}", target_type, target_id, e);
-                    }
+
+                if enabled {
+                    info!(instance_id = %id, "Target is enabled, ready to create a task");
+                    // 5.3. Create asynchronous tasks for enabled instances
+                    let target_type_clone = target_type.clone();
+                    let tid = id.clone();
+                    let merged_config_arc = std::sync::Arc::new(merged_config);
+                    tasks.push(async move {
+                        let result = factory.create_target(tid.clone(), &merged_config_arc).await;
+                        (target_type_clone, tid, result, std::sync::Arc::clone(&merged_config_arc))
+                    });
+                } else {
+                    info!(instance_id = %id, "Skip the disabled target and will be removed from the final configuration");
+                    // Remove disabled target from final configuration
+                    final_config.0.entry(section_name.clone()).or_default().remove(&id);
                }
            }
        }

-        Ok(targets)
+        // 6. Concurrently execute all creation tasks and collect results
+        let mut successful_targets = Vec::new();
+        let mut successful_configs = Vec::new();
+        while let Some((target_type, id, result, final_config)) = tasks.next().await {
+            match result {
+                Ok(target) => {
+                    info!(target_type = %target_type, instance_id = %id, "Create a target successfully");
+                    successful_targets.push(target);
+                    successful_configs.push((target_type, id, final_config));
+                }
+                Err(e) => {
+                    error!(target_type = %target_type, instance_id = %id, error = %e, "Failed to create a target");
+                }
+            }
+        }
+
+        // 7. Aggregate new configuration and write back to system configuration
+        if !successful_configs.is_empty() {
+            info!(
+                "Prepare to update {} successfully created target configurations to the system configuration...",
+                successful_configs.len()
+            );
+            let mut new_config = config.clone();
+            for (target_type, id, kvs) in successful_configs {
+                let section_name = format!("{NOTIFY_ROUTE_PREFIX}{target_type}").to_lowercase();
+                new_config.0.entry(section_name).or_default().insert(id, (*kvs).clone());
+            }
+
+            let Some(store) = rustfs_ecstore::global::new_object_layer_fn() else {
+                return Err(TargetError::ServerNotInitialized(
+                    "Failed to save target configuration: server storage not initialized".to_string(),
+                ));
+            };
+
+            match rustfs_ecstore::config::com::save_server_config(store, &new_config).await {
+                Ok(_) => {
+                    info!("The new configuration was saved to the system successfully.")
+                }
+                Err(e) => {
+                    error!("Failed to save the new configuration: {}", e);
+                    return Err(TargetError::SaveConfig(e.to_string()));
+                }
+            }
+        }
+
+        info!(count = successful_targets.len(), "All target processing completed");
+        Ok(successful_targets)
    }
 }
@@ -109,3 +109,11 @@ impl std::fmt::Display for ChannelTargetType {
        }
    }
 }
+
+pub fn parse_bool(value: &str) -> Result<bool, TargetError> {
+    match value.to_lowercase().as_str() {
+        "true" | "on" | "yes" | "1" => Ok(true),
+        "false" | "off" | "no" | "0" => Ok(false),
+        _ => Err(TargetError::ParseError(format!("Unable to parse boolean: {value}"))),
+    }
+}