From 18a3902d2cd075b70a29fe1321dcc78e3be30825 Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 1 Jul 2025 12:34:24 +0800 Subject: [PATCH] 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 Co-authored-by: lygn128 Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- crates/notify/src/notifier.rs | 5 + crates/notify/src/rules/config.rs | 4 +- crates/notify/src/rules/pattern_rules.rs | 3 +- crates/notify/src/rules/rules_map.rs | 3 +- ecstore/src/disk/endpoint.rs | 31 ++- ecstore/src/disk/local.rs | 15 +- rustfs/src/admin/handlers/event.rs | 264 +++++++++++++++++++++++ 7 files changed, 311 insertions(+), 14 deletions(-) diff --git a/crates/notify/src/notifier.rs b/crates/notify/src/notifier.rs index 4447c305b..ed44d7a7d 100644 --- a/crates/notify/src/notifier.rs +++ b/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 { + 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); diff --git a/crates/notify/src/rules/config.rs b/crates/notify/src/rules/config.rs index 11b684af8..c2c6cda4a 100644 --- a/crates/notify/src/rules/config.rs +++ b/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 diff --git a/crates/notify/src/rules/pattern_rules.rs b/crates/notify/src/rules/pattern_rules.rs index b720e0f7c..9d64dbcf3 100644 --- a/crates/notify/src/rules/pattern_rules.rs +++ b/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, } diff --git a/crates/notify/src/rules/rules_map.rs b/crates/notify/src/rules/rules_map.rs index 86ac21727..bf3d990ff 100644 --- a/crates/notify/src/rules/rules_map.rs +++ b/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, /// A bitmask that represents the union of all event types in this map. diff --git a/ecstore/src/disk/endpoint.rs b/ecstore/src/disk/endpoint.rs index 38b7b6df7..6c38aec0d 100644 --- a/ecstore/src/disk/endpoint.rs +++ b/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 } } diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index a21dd732f..6aee14e64 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -130,7 +130,16 @@ impl Debug for LocalDisk { impl LocalDisk { pub async fn new(ep: &Endpoint, cleanup: bool) -> Result { - 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) } diff --git a/rustfs/src/admin/handlers/event.rs b/rustfs/src/admin/handlers/event.rs index 8b1378917..886e51fba 100644 --- a/rustfs/src/admin/handlers/event.rs +++ b/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, _params: Params<'_, '_>) -> S3Result> { + // 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 = 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, _params: Params<'_, '_>) -> S3Result> { + // 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, _params: Params<'_, '_>) -> S3Result> { + // 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, _params: Params<'_, '_>) -> S3Result> { + // 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, +} +pub struct GetBucketNotification {} +#[async_trait::async_trait] +impl Operation for GetBucketNotification { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + 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, _params: Params<'_, '_>) -> S3Result> { + 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)) + } +}