feat(admin): enhance notification event handlers (#530)

* add log

* add logs

* test

* cargo fmt

* add event admin api

* feat: add ImportIam handle

* refact bucket replication

* Update ecstore/src/disk/endpoint.rs

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

---------

Co-authored-by: weisd <weishidavip@163.com>
Co-authored-by: lygn128 <lygn128@163.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

crates/notify/src/notifier.rs

@@ -66,6 +66,11 @@ impl EventNotifier {
         info!("Added rules for bucket: {}", bucket_name);
     }
 
+    /// Gets the rules map for a specific bucket.
+    pub fn get_rules_map(&self, bucket_name: &str) -> Option<RulesMap> {
+        self.bucket_rules_map.get(bucket_name).map(|r| r.clone())
+    }
+
     /// Removes notification rules for a bucket
     pub async fn remove_notification(&self, bucket_name: &str) {
         self.bucket_rules_map.remove(bucket_name);

crates/notify/src/rules/config.rs

@@ -1,17 +1,17 @@
 use super::rules_map::RulesMap;
-// Keep for existing structure if any, or remove if not used
 use super::xml_config::ParseConfigError as BucketNotificationConfigError;
 use crate::EventName;
 use crate::arn::TargetID;
 use crate::rules::NotificationConfiguration;
 use crate::rules::pattern_rules;
 use crate::rules::target_id_set;
+use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::io::Read;
 
 /// Configuration for bucket notifications.
 /// This struct now holds the parsed and validated rules in the new RulesMap format.
-#[derive(Debug, Clone, Default)]
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct BucketNotificationConfig {
     pub region: String,  // Region where this config is applicable
     pub rules: RulesMap, // The new, more detailed RulesMap

crates/notify/src/rules/pattern_rules.rs

@@ -2,10 +2,11 @@ use super::pattern;
 use super::target_id_set::TargetIdSet;
 use crate::arn::TargetID;
 use std::collections::HashMap;
+use serde::{Deserialize, Serialize};
 
 /// PatternRules - Event rule that maps object name patterns to TargetID collections.
 /// `event.Rules` (map[string]TargetIDSet) in the Go code
-#[derive(Debug, Clone, Default)]
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct PatternRules {
     pub(crate) rules: HashMap<String, TargetIdSet>,
 }

crates/notify/src/rules/rules_map.rs

@@ -2,11 +2,12 @@ use super::pattern_rules::PatternRules;
 use super::target_id_set::TargetIdSet;
 use crate::arn::TargetID;
 use crate::event::EventName;
+use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 
 /// RulesMap - Rule mapping organized by event name。
 /// `event.RulesMap` (map[Name]Rules) in the corresponding Go code
-#[derive(Debug, Clone, Default)]
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct RulesMap {
     map: HashMap<EventName, PatternRules>,
     /// A bitmask that represents the union of all event types in this map.

ecstore/src/disk/endpoint.rs

@@ -2,6 +2,7 @@ use super::error::{Error, Result};
 use path_absolutize::Absolutize;
 use rustfs_utils::{is_local_host, is_socket_addr};
 use std::{fmt::Display, path::Path};
+use tracing::debug;
 use url::{ParseError, Url};
 
 /// enum for endpoint type.
@@ -75,6 +76,8 @@ impl TryFrom<&str> for Endpoint {
                 #[cfg(windows)]
                 let path = Path::new(&path[1..]).absolutize()?;
 
+                debug!("endpoint try_from: path={}", path.display());
+
                 if path.parent().is_none() || Path::new("").eq(&path) {
                     return Err(Error::other("empty or root path is not supported in URL endpoint"));
                 }
@@ -155,26 +158,40 @@ impl Endpoint {
     /// returns the host to be used for grid connections.
     pub fn grid_host(&self) -> String {
         match (self.url.host(), self.url.port()) {
-            (Some(host), Some(port)) => format!("{}://{}:{}", self.url.scheme(), host, port),
-            (Some(host), None) => format!("{}://{}", self.url.scheme(), host),
+            (Some(host), Some(port)) => {
+                debug!("grid_host scheme={}: host={}, port={}", self.url.scheme(), host, port);
+                format!("{}://{}:{}", self.url.scheme(), host, port)
+            }
+            (Some(host), None) => {
+                debug!("grid_host scheme={}: host={}", self.url.scheme(), host);
+                format!("{}://{}", self.url.scheme(), host)
+            }
             _ => String::new(),
         }
     }
 
     pub fn host_port(&self) -> String {
         match (self.url.host(), self.url.port()) {
-            (Some(host), Some(port)) => format!("{host}:{port}"),
-            (Some(host), None) => format!("{host}"),
+            (Some(host), Some(port)) => {
+                debug!("host_port host={}, port={}", host, port);
+                format!("{host}:{port}")
+            }
+            (Some(host), None) => {
+                debug!("host_port host={}, port={}", host, self.url.port().unwrap_or(0));
+                format!("{host}")
+            }
             _ => String::new(),
         }
     }
 
     pub fn get_file_path(&self) -> &str {
         let path = self.url.path();
-
         #[cfg(windows)]
-        let path = &path[1..];
-
+        if self.url.scheme() == "file" {
+            let stripped = path.strip_prefix('/').unwrap_or(path);
+            debug!("get_file_path windows: path={}", stripped);
+            return stripped;
+        }
         path
     }
 }

ecstore/src/disk/local.rs

@@ -130,7 +130,16 @@ impl Debug for LocalDisk {
 
 impl LocalDisk {
     pub async fn new(ep: &Endpoint, cleanup: bool) -> Result<Self> {
-        let root = fs::canonicalize(ep.get_file_path()).await?;
+        debug!("Creating local disk");
+        let root = match fs::canonicalize(ep.get_file_path()).await {
+            Ok(path) => path,
+            Err(e) => {
+                if e.kind() == ErrorKind::NotFound {
+                    return Err(DiskError::VolumeNotFound.into());
+                }
+                return Err(to_file_error(e).into());
+            }
+        };
 
         if cleanup {
             // TODO: 删除 tmp 数据
@@ -140,7 +149,7 @@ impl LocalDisk {
             .join(Path::new(super::FORMAT_CONFIG_FILE))
             .absolutize_virtually(&root)?
             .into_owned();
-
+        debug!("format_path: {:?}", format_path);
         let (format_data, format_meta) = read_file_exists(&format_path).await?;
 
         let mut id = None;
@@ -245,7 +254,7 @@ impl LocalDisk {
 
         let root = disk.root.clone();
         tokio::spawn(Self::cleanup_deleted_objects_loop(root, exit_rx));
-
+        debug!("LocalDisk created: {:?}", disk);
         Ok(disk)
     }
 

rustfs/src/admin/handlers/event.rs

@@ -1 +1,265 @@
+use crate::admin::router::Operation;
+use crate::auth::{check_key_valid, get_session_token};
+use http::{HeaderMap, StatusCode};
+use matchit::Params;
+use rustfs_config::notify::{NOTIFY_MQTT_SUB_SYS, NOTIFY_WEBHOOK_SUB_SYS};
+use rustfs_notify::rules::{BucketNotificationConfig, PatternRules};
+use rustfs_notify::EventName;
+use s3s::{header::CONTENT_TYPE, s3_error, Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result};
+use serde::{Deserialize, Serialize};
+use serde_urlencoded::from_bytes;
+use std::collections::HashMap;
+use tracing::{error, info, warn};
 
+#[derive(Debug, Deserialize)]
+struct TargetQuery {
+    #[serde(rename = "targetType")]
+    target_type: String,
+    #[serde(rename = "targetName")]
+    target_name: String,
+}
+
+#[derive(Debug, Deserialize)]
+struct BucketQuery {
+    #[serde(rename = "bucketName")]
+    bucket_name: String,
+}
+
+/// Set (create or update) a notification target
+pub struct SetNotificationTarget {}
+#[async_trait::async_trait]
+impl Operation for SetNotificationTarget {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        // 1. Analyze query parameters
+        let query: TargetQuery = from_bytes(req.uri.query().unwrap_or("").as_bytes())
+            .map_err(|e| s3_error!(InvalidArgument, "invalid query parameters: {}", e))?;
+
+        let target_type = query.target_type.to_lowercase();
+        if target_type != NOTIFY_WEBHOOK_SUB_SYS.to_string() && target_type != NOTIFY_MQTT_SUB_SYS.to_string() {
+            return Err(s3_error!(InvalidArgument, "unsupported target type: {}", query.target_type));
+        }
+
+        // 2. Permission verification
+        let Some(input_cred) = &req.credentials else {
+            return Err(s3_error!(InvalidRequest, "credentials not found"));
+        };
+        let (_cred, _owner) =
+            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+
+        // 3. Get notification system instance
+        let Some(ns) = rustfs_notify::global::notification_system() else {
+            return Err(s3_error!(InternalError, "notification system not initialized"));
+        };
+
+        // 4. The parsing request body is KVS (Key-Value Store)
+        let mut input = req.input;
+        let body = input.store_all_unlimited().await.map_err(|e| {
+            warn!("failed to read request body: {:?}", e);
+            s3_error!(InvalidRequest, "failed to read request body")
+        })?;
+        let mut kvs_map: HashMap<String, String> = serde_json::from_slice(&body)
+            .map_err(|e| s3_error!(InvalidArgument, "invalid json body for target config: {}", e))?;
+        // If there is an enable key, add an enable key value to "on"
+        if !kvs_map.contains_key(ecstore::config::ENABLE_KEY) {
+            kvs_map.insert(ecstore::config::ENABLE_KEY.to_string(), ecstore::config::ENABLE_ON.to_string());
+        }
+
+        let kvs = ecstore::config::KVS(
+            kvs_map
+                .into_iter()
+                .map(|(key, value)| ecstore::config::KV {
+                    key,
+                    value,
+                    hidden_if_empty: false, // Set a default value
+                })
+                .collect(),
+        );
+
+        // 5. Call notification system to set target configuration
+        info!("Setting target config for type '{}', name '{}'", &query.target_type, &query.target_name);
+        ns.set_target_config(&query.target_type, &query.target_name, kvs)
+            .await
+            .map_err(|e| {
+                error!("failed to set target config: {}", e);
+                S3Error::with_message(S3ErrorCode::InternalError, format!("failed to set target config: {}", e))
+            })?;
+
+        let mut header = HeaderMap::new();
+        header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
+        Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
+    }
+}
+
+/// Get a list of notification targets for all activities
+pub struct ListNotificationTargets {}
+#[async_trait::async_trait]
+impl Operation for ListNotificationTargets {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        // 1. Permission verification
+        let Some(input_cred) = &req.credentials else {
+            return Err(s3_error!(InvalidRequest, "credentials not found"));
+        };
+        let (_cred, _owner) =
+            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+
+        // 2. Get notification system instance
+        let Some(ns) = rustfs_notify::global::notification_system() else {
+            return Err(s3_error!(InternalError, "notification system not initialized"));
+        };
+
+        // 3. Get the list of activity targets
+        let active_targets = ns.get_active_targets().await;
+
+        // 4. Serialize and return the result
+        let data = serde_json::to_vec(&active_targets)
+            .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("failed to serialize targets: {}", e)))?;
+
+        let mut header = HeaderMap::new();
+        header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
+        Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header))
+    }
+}
+
+/// Delete a specified notification target
+pub struct RemoveNotificationTarget {}
+#[async_trait::async_trait]
+impl Operation for RemoveNotificationTarget {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        // 1. Analyze query parameters
+        let query: TargetQuery = from_bytes(req.uri.query().unwrap_or("").as_bytes())
+            .map_err(|e| s3_error!(InvalidArgument, "invalid query parameters: {}", e))?;
+
+        // 2. Permission verification
+        let Some(input_cred) = &req.credentials else {
+            return Err(s3_error!(InvalidRequest, "credentials not found"));
+        };
+        let (_cred, _owner) =
+            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+
+        // 3. Get notification system instance
+        let Some(ns) = rustfs_notify::global::notification_system() else {
+            return Err(s3_error!(InternalError, "notification system not initialized"));
+        };
+
+        // 4. Call notification system to remove target configuration
+        info!("Removing target config for type '{}', name '{}'", &query.target_type, &query.target_name);
+        ns.remove_target_config(&query.target_type, &query.target_name)
+            .await
+            .map_err(|e| {
+                error!("failed to remove target config: {}", e);
+                S3Error::with_message(S3ErrorCode::InternalError, format!("failed to remove target config: {}", e))
+            })?;
+
+        let mut header = HeaderMap::new();
+        header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
+        Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
+    }
+}
+
+/// Set notification rules for buckets
+pub struct SetBucketNotification {}
+#[async_trait::async_trait]
+impl Operation for SetBucketNotification {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        // 1. Analyze query parameters
+        let query: BucketQuery = from_bytes(req.uri.query().unwrap_or("").as_bytes())
+            .map_err(|e| s3_error!(InvalidArgument, "invalid query parameters: {}", e))?;
+
+        // 2. Permission verification
+        let Some(input_cred) = &req.credentials else {
+            return Err(s3_error!(InvalidRequest, "credentials not found"));
+        };
+        let (_cred, _owner) =
+            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+
+        // 3. Get notification system instance
+        let Some(ns) = rustfs_notify::global::notification_system() else {
+            return Err(s3_error!(InternalError, "notification system not initialized"));
+        };
+
+        // 4. The parsing request body is BucketNotificationConfig
+        let mut input = req.input;
+        let body = input.store_all_unlimited().await.map_err(|e| {
+            warn!("failed to read request body: {:?}", e);
+            s3_error!(InvalidRequest, "failed to read request body")
+        })?;
+        let config: BucketNotificationConfig = serde_json::from_slice(&body)
+            .map_err(|e| s3_error!(InvalidArgument, "invalid json body for bucket notification config: {}", e))?;
+
+        // 5. Load bucket notification configuration
+        info!("Loading notification config for bucket '{}'", &query.bucket_name);
+        ns.load_bucket_notification_config(&query.bucket_name, &config)
+            .await
+            .map_err(|e| {
+                error!("failed to load bucket notification config: {}", e);
+                S3Error::with_message(S3ErrorCode::InternalError, format!("failed to load bucket notification config: {}", e))
+            })?;
+
+        let mut header = HeaderMap::new();
+        header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
+        Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
+    }
+}
+
+/// Get notification rules for buckets
+#[derive(Serialize)]
+struct BucketRulesResponse {
+    rules: HashMap<EventName, PatternRules>,
+}
+pub struct GetBucketNotification {}
+#[async_trait::async_trait]
+impl Operation for GetBucketNotification {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let query: BucketQuery = from_bytes(req.uri.query().unwrap_or("").as_bytes())
+            .map_err(|e| s3_error!(InvalidArgument, "invalid query parameters: {}", e))?;
+
+        let Some(input_cred) = &req.credentials else {
+            return Err(s3_error!(InvalidRequest, "credentials not found"));
+        };
+        let (_cred, _owner) =
+            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+
+        let Some(ns) = rustfs_notify::global::notification_system() else {
+            return Err(s3_error!(InternalError, "notification system not initialized"));
+        };
+
+        let rules_map = ns.notifier.get_rules_map(&query.bucket_name);
+        let response = BucketRulesResponse {
+            rules: rules_map.unwrap_or_default().inner().clone(),
+        };
+
+        let data = serde_json::to_vec(&response)
+            .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("failed to serialize rules: {}", e)))?;
+
+        let mut header = HeaderMap::new();
+        header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
+        Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header))
+    }
+}
+
+/// 删除存储桶的所有通知规则
+pub struct RemoveBucketNotification {}
+#[async_trait::async_trait]
+impl Operation for RemoveBucketNotification {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let query: BucketQuery = from_bytes(req.uri.query().unwrap_or("").as_bytes())
+            .map_err(|e| s3_error!(InvalidArgument, "invalid query parameters: {}", e))?;
+
+        let Some(input_cred) = &req.credentials else {
+            return Err(s3_error!(InvalidRequest, "credentials not found"));
+        };
+        let (_cred, _owner) =
+            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+
+        let Some(ns) = rustfs_notify::global::notification_system() else {
+            return Err(s3_error!(InternalError, "notification system not initialized"));
+        };
+
+        info!("Removing notification config for bucket '{}'", &query.bucket_name);
+        ns.remove_bucket_notification_config(&query.bucket_name).await;
+
+        let mut header = HeaderMap::new();
+        header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
+        Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
+    }
+}