From 933909363838f343c3230c808ea207f46c953644 Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 10 Jun 2025 09:59:49 +0800 Subject: [PATCH] init event crate --- Cargo.lock | 28 +- Cargo.toml | 9 +- crates/config/src/notify/webhook.rs | 6 +- crates/event/Cargo.toml | 27 + crates/event/src/error.rs | 403 ++++++++++++ crates/event/src/event.rs | 617 ++++++++++++++++++ crates/event/src/lib.rs | 4 + crates/event/src/notifier.rs | 143 ++++ crates/event/src/system.rs | 80 +++ crates/notify/Cargo.toml | 8 +- crates/notify/src/adapter/kafka.rs | 180 ----- crates/notify/src/adapter/mod.rs | 10 +- crates/notify/src/adapter/webhook.rs | 3 - .../src/{config/notifier.rs => config.rs} | 62 +- crates/notify/src/config/adapter.rs | 42 -- crates/notify/src/config/kafka.rs | 44 -- crates/notify/src/config/mod.rs | 38 -- crates/notify/src/config/mqtt.rs | 46 -- crates/notify/src/config/webhook.rs | 88 --- crates/notify/src/lib.rs | 16 +- crates/notify/src/notifier.rs | 2 +- crates/notify/src/store/manager.rs | 5 - crates/notify/src/store/mod.rs | 312 +++++++++ crates/notify/src/store/queue.rs | 570 +++++----------- crates/notify/src/system.rs | 2 +- 25 files changed, 1821 insertions(+), 924 deletions(-) create mode 100644 crates/event/Cargo.toml create mode 100644 crates/event/src/error.rs create mode 100644 crates/event/src/event.rs create mode 100644 crates/event/src/lib.rs create mode 100644 crates/event/src/notifier.rs create mode 100644 crates/event/src/system.rs delete mode 100644 crates/notify/src/adapter/kafka.rs rename crates/notify/src/{config/notifier.rs => config.rs} (53%) delete mode 100644 crates/notify/src/config/adapter.rs delete mode 100644 crates/notify/src/config/kafka.rs delete mode 100644 crates/notify/src/config/mod.rs delete mode 100644 crates/notify/src/config/mqtt.rs delete mode 100644 crates/notify/src/config/webhook.rs diff --git a/Cargo.lock b/Cargo.lock index 36409c510..25298d122 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8126,6 +8126,27 @@ dependencies = [ "serde_json", ] +[[package]] +name = "rustfs-event" +version = "0.0.1" +dependencies = [ + "common", + "ecstore", + "once_cell", + "reqwest", + "rustfs-config", + "serde", + "serde_json", + "serde_with", + "smallvec", + "strum", + "thiserror 2.0.12", + "tokio", + "tokio-util", + "tracing", + "uuid", +] + [[package]] name = "rustfs-gui" version = "0.0.1" @@ -8156,11 +8177,10 @@ dependencies = [ "common", "dotenvy", "ecstore", - "lazy_static", "once_cell", - "rdkafka", "reqwest", "rumqttc", + "rustfs-config", "serde", "serde_json", "serde_with", @@ -9047,9 +9067,9 @@ dependencies = [ [[package]] name = "smallvec" -version = "1.15.0" +version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8917285742e9f3e1683f0a9c4e6b57960b7314d0b08d30d1ecd426713ee2eee9" +checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" dependencies = [ "serde", ] diff --git a/Cargo.toml b/Cargo.toml index 58995bbb9..6e7d1514f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,9 +7,11 @@ members = [ "common/protos", # Protocol buffer definitions "common/workers", # Worker thread pools and task scheduling "crates/config", # Configuration management - "crates/notify", # Event notification system + "crates/event", # Event handling and processing + "crates/notify", # Notification system for events "crates/obs", # Observability utilities "crates/utils", # Utility functions and helpers + "crates/zip", # ZIP file handling and compression "crypto", # Cryptography and security features "ecstore", # Erasure coding storage implementation "e2e_test", # End-to-end test suite @@ -18,7 +20,6 @@ members = [ "rustfs", # Core file system implementation "s3select/api", # S3 Select API interface "s3select/query", # S3 Select query engine - "crates/zip", ] resolver = "2" @@ -74,7 +75,7 @@ byteorder = "1.5.0" cfg-if = "1.0.0" chacha20poly1305 = { version = "0.10.1" } chrono = { version = "0.4.41", features = ["serde"] } -clap = { version = "4.5.39", features = ["derive", "env"] } +clap = { version = "4.5.40", features = ["derive", "env"] } const-str = { version = "0.6.2", features = ["std", "proc"] } crc32fast = "1.4.2" datafusion = "46.0.1" @@ -184,7 +185,7 @@ serde_urlencoded = "0.7.1" serde_with = "3.12.0" sha2 = "0.10.9" siphasher = "1.0.1" -smallvec = { version = "1.15.0", features = ["serde"] } +smallvec = { version = "1.15.1", features = ["serde"] } snafu = "0.8.6" snap = "1.1.1" diff --git a/crates/config/src/notify/webhook.rs b/crates/config/src/notify/webhook.rs index 5d29dbfc7..80e0b38ff 100644 --- a/crates/config/src/notify/webhook.rs +++ b/crates/config/src/notify/webhook.rs @@ -1,4 +1,5 @@ use serde::{Deserialize, Serialize}; +use std::collections::HashMap; /// WebhookArgs - Webhook target arguments. #[derive(Debug, Clone, Serialize, Deserialize)] @@ -7,6 +8,7 @@ pub struct WebhookArgs { pub endpoint: String, pub auth_token: String, #[serde(skip)] + pub custom_headers: Option>, pub queue_dir: String, pub queue_limit: u64, pub client_cert: String, @@ -20,6 +22,7 @@ impl WebhookArgs { enable: false, endpoint: "".to_string(), auth_token: "".to_string(), + custom_headers: None, queue_dir: "".to_string(), queue_limit: 0, client_cert: "".to_string(), @@ -53,13 +56,14 @@ impl Default for WebhookArgs { #[cfg(test)] mod tests { - use super::*; + use crate::notify::webhook::WebhookArgs; #[test] fn test_webhook_args_new() { let args = WebhookArgs::new(); assert_eq!(args.endpoint, ""); assert_eq!(args.auth_token, ""); + assert!(args.custom_headers.is_none()); assert_eq!(args.queue_dir, ""); assert_eq!(args.queue_limit, 0); assert_eq!(args.client_cert, ""); diff --git a/crates/event/Cargo.toml b/crates/event/Cargo.toml new file mode 100644 index 000000000..9b9ddc681 --- /dev/null +++ b/crates/event/Cargo.toml @@ -0,0 +1,27 @@ +[package] +name = "rustfs-event" +edition.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true +version.workspace = true + +[dependencies] +rustfs-config = { workspace = true, features = ["constants", "notify"] } +common = { workspace = true } +ecstore = { workspace = true } +once_cell = { workspace = true } +reqwest = { workspace = true, optional = true } +serde = { workspace = true } +serde_json = { workspace = true } +serde_with = { workspace = true } +smallvec = { workspace = true, features = ["serde"] } +strum = { workspace = true, features = ["derive"] } +tracing = { workspace = true } +thiserror = { workspace = true } +tokio = { workspace = true, features = ["sync", "net", "macros", "signal", "rt-multi-thread"] } +tokio-util = { workspace = true } +uuid = { workspace = true, features = ["v4", "serde"] } + +[lints] +workspace = true diff --git a/crates/event/src/error.rs b/crates/event/src/error.rs new file mode 100644 index 000000000..3d138298f --- /dev/null +++ b/crates/event/src/error.rs @@ -0,0 +1,403 @@ +use thiserror::Error; +use tokio::sync::mpsc::error; +use tokio::task::JoinError; + +/// The `Error` enum represents all possible errors that can occur in the application. +/// It implements the `std::error::Error` trait and provides a way to convert various error types into a single error type. +#[derive(Error, Debug)] +pub enum Error { + #[error("Join error: {0}")] + JoinError(#[from] JoinError), + #[error("IO error: {0}")] + Io(#[from] std::io::Error), + #[error("Serialization error: {0}")] + Serde(#[from] serde_json::Error), + #[error("HTTP error: {0}")] + Http(#[from] reqwest::Error), + #[cfg(all(feature = "kafka", target_os = "linux"))] + #[error("Kafka error: {0}")] + Kafka(#[from] rdkafka::error::KafkaError), + #[cfg(feature = "mqtt")] + #[error("MQTT error: {0}")] + Mqtt(#[from] rumqttc::ClientError), + #[error("Channel send error: {0}")] + ChannelSend(#[from] Box>), + #[error("Feature disabled: {0}")] + FeatureDisabled(&'static str), + #[error("Event bus already started")] + EventBusStarted, + #[error("necessary fields are missing:{0}")] + MissingField(&'static str), + #[error("field verification failed:{0}")] + ValidationError(&'static str), + #[error("Custom error: {0}")] + Custom(String), + #[error("Configuration error: {0}")] + ConfigError(String), + #[error("create adapter failed error: {0}")] + AdapterCreationFailed(String), +} + +impl Error { + pub fn custom(msg: &str) -> Error { + Self::Custom(msg.to_string()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::error::Error as StdError; + use std::io; + use tokio::sync::mpsc; + + #[test] + fn test_error_display() { + // Test error message display + let custom_error = Error::custom("test message"); + assert_eq!(custom_error.to_string(), "Custom error: test message"); + + let feature_error = Error::FeatureDisabled("test feature"); + assert_eq!(feature_error.to_string(), "Feature disabled: test feature"); + + let event_bus_error = Error::EventBusStarted; + assert_eq!(event_bus_error.to_string(), "Event bus already started"); + + let missing_field_error = Error::MissingField("required_field"); + assert_eq!(missing_field_error.to_string(), "necessary fields are missing:required_field"); + + let validation_error = Error::ValidationError("invalid format"); + assert_eq!(validation_error.to_string(), "field verification failed:invalid format"); + + let config_error = Error::ConfigError("invalid config".to_string()); + assert_eq!(config_error.to_string(), "Configuration error: invalid config"); + } + + #[test] + fn test_error_debug() { + // Test Debug trait implementation + let custom_error = Error::custom("debug test"); + let debug_str = format!("{:?}", custom_error); + assert!(debug_str.contains("Custom")); + assert!(debug_str.contains("debug test")); + + let feature_error = Error::FeatureDisabled("debug feature"); + let debug_str = format!("{:?}", feature_error); + assert!(debug_str.contains("FeatureDisabled")); + assert!(debug_str.contains("debug feature")); + } + + #[test] + fn test_custom_error_creation() { + // Test custom error creation + let error = Error::custom("test custom error"); + match error { + Error::Custom(msg) => assert_eq!(msg, "test custom error"), + _ => panic!("Expected Custom error variant"), + } + + // Test empty string + let empty_error = Error::custom(""); + match empty_error { + Error::Custom(msg) => assert_eq!(msg, ""), + _ => panic!("Expected Custom error variant"), + } + + // Test special characters + let special_error = Error::custom("Test Chinese 中文 & special chars: !@#$%"); + match special_error { + Error::Custom(msg) => assert_eq!(msg, "Test Chinese 中文 & special chars: !@#$%"), + _ => panic!("Expected Custom error variant"), + } + } + + #[test] + fn test_io_error_conversion() { + // Test IO error conversion + let io_error = io::Error::new(io::ErrorKind::NotFound, "file not found"); + let converted_error: Error = io_error.into(); + + match converted_error { + Error::Io(err) => { + assert_eq!(err.kind(), io::ErrorKind::NotFound); + assert_eq!(err.to_string(), "file not found"); + } + _ => panic!("Expected Io error variant"), + } + + // Test different types of IO errors + let permission_error = io::Error::new(io::ErrorKind::PermissionDenied, "access denied"); + let converted: Error = permission_error.into(); + assert!(matches!(converted, Error::Io(_))); + } + + #[test] + fn test_serde_error_conversion() { + // Test serialization error conversion + let invalid_json = r#"{"invalid": json}"#; + let serde_error = serde_json::from_str::(invalid_json).unwrap_err(); + let converted_error: Error = serde_error.into(); + + match converted_error { + Error::Serde(_) => { + // Verify error type is correct + assert!(converted_error.to_string().contains("Serialization error")); + } + _ => panic!("Expected Serde error variant"), + } + } + + #[tokio::test] + async fn test_channel_send_error_conversion() { + // Test channel send error conversion + let (tx, rx) = mpsc::channel::(1); + drop(rx); // Close receiver + + // Create a test event + use crate::event::{Bucket, Identity, Metadata, Name, Object, Source}; + use std::collections::HashMap; + + let identity = Identity::new("test-user".to_string()); + let bucket = Bucket::new("test-bucket".to_string(), identity.clone(), "arn:aws:s3:::test-bucket".to_string()); + let object = Object::new( + "test-key".to_string(), + Some(1024), + Some("etag123".to_string()), + Some("text/plain".to_string()), + Some(HashMap::new()), + None, + "sequencer123".to_string(), + ); + let metadata = Metadata::create("1.0".to_string(), "config1".to_string(), bucket, object); + let source = Source::new("localhost".to_string(), "8080".to_string(), "test-agent".to_string()); + + let test_event = crate::event::Event::builder() + .event_name(Name::ObjectCreatedPut) + .s3(metadata) + .source(source) + .build() + .unwrap(); + + let send_result = tx.send(test_event).await; + assert!(send_result.is_err()); + + let send_error = send_result.unwrap_err(); + let boxed_error = Box::new(send_error); + let converted_error: Error = boxed_error.into(); + + match converted_error { + Error::ChannelSend(_) => { + assert!(converted_error.to_string().contains("Channel send error")); + } + _ => panic!("Expected ChannelSend error variant"), + } + } + + #[test] + fn test_error_source_chain() { + // 测试错误源链 + let io_error = io::Error::new(io::ErrorKind::InvalidData, "invalid data"); + let converted_error: Error = io_error.into(); + + // 验证错误源 + assert!(converted_error.source().is_some()); + let source = converted_error.source().unwrap(); + assert_eq!(source.to_string(), "invalid data"); + } + + #[test] + fn test_error_variants_exhaustive() { + // 测试所有错误变体的创建 + let errors = vec![ + Error::FeatureDisabled("test"), + Error::EventBusStarted, + Error::MissingField("field"), + Error::ValidationError("validation"), + Error::Custom("custom".to_string()), + Error::ConfigError("config".to_string()), + ]; + + for error in errors { + // 验证每个错误都能正确显示 + let error_str = error.to_string(); + assert!(!error_str.is_empty()); + + // 验证每个错误都能正确调试 + let debug_str = format!("{:?}", error); + assert!(!debug_str.is_empty()); + } + } + + #[test] + fn test_error_equality_and_matching() { + // 测试错误的模式匹配 + let custom_error = Error::custom("test"); + match custom_error { + Error::Custom(msg) => assert_eq!(msg, "test"), + _ => panic!("Pattern matching failed"), + } + + let feature_error = Error::FeatureDisabled("feature"); + match feature_error { + Error::FeatureDisabled(feature) => assert_eq!(feature, "feature"), + _ => panic!("Pattern matching failed"), + } + + let event_bus_error = Error::EventBusStarted; + match event_bus_error { + Error::EventBusStarted => {} // 正确匹配 + _ => panic!("Pattern matching failed"), + } + } + + #[test] + fn test_error_message_formatting() { + // 测试错误消息格式化 + let test_cases = vec![ + (Error::FeatureDisabled("kafka"), "Feature disabled: kafka"), + (Error::MissingField("bucket_name"), "necessary fields are missing:bucket_name"), + (Error::ValidationError("invalid email"), "field verification failed:invalid email"), + (Error::ConfigError("missing file".to_string()), "Configuration error: missing file"), + ]; + + for (error, expected_message) in test_cases { + assert_eq!(error.to_string(), expected_message); + } + } + + #[test] + fn test_error_memory_efficiency() { + // 测试错误类型的内存效率 + use std::mem; + + let size = mem::size_of::(); + // 错误类型应该相对紧凑，考虑到包含多种错误类型，96 字节是合理的 + assert!(size <= 128, "Error size should be reasonable, got {} bytes", size); + + // 测试 Option的大小 + let option_size = mem::size_of::>(); + assert!(option_size <= 136, "Option should be efficient, got {} bytes", option_size); + } + + #[test] + fn test_error_thread_safety() { + // 测试错误类型的线程安全性 + fn assert_send() {} + fn assert_sync() {} + + assert_send::(); + assert_sync::(); + } + + #[test] + fn test_custom_error_edge_cases() { + // 测试自定义错误的边界情况 + let long_message = "a".repeat(1000); + let long_error = Error::custom(&long_message); + match long_error { + Error::Custom(msg) => assert_eq!(msg.len(), 1000), + _ => panic!("Expected Custom error variant"), + } + + // 测试包含换行符的消息 + let multiline_error = Error::custom("line1\nline2\nline3"); + match multiline_error { + Error::Custom(msg) => assert!(msg.contains('\n')), + _ => panic!("Expected Custom error variant"), + } + + // 测试包含 Unicode 字符的消息 + let unicode_error = Error::custom("🚀 Unicode test 测试 🎉"); + match unicode_error { + Error::Custom(msg) => assert!(msg.contains('🚀')), + _ => panic!("Expected Custom error variant"), + } + } + + #[test] + fn test_error_conversion_consistency() { + // 测试错误转换的一致性 + let original_io_error = io::Error::new(io::ErrorKind::TimedOut, "timeout"); + let error_message = original_io_error.to_string(); + let converted: Error = original_io_error.into(); + + // 验证转换后的错误包含原始错误信息 + assert!(converted.to_string().contains(&error_message)); + } + + #[test] + fn test_error_downcast() { + // 测试错误的向下转型 + let io_error = io::Error::other("test error"); + let converted: Error = io_error.into(); + + // 验证可以获取源错误 + if let Error::Io(ref inner) = converted { + assert_eq!(inner.to_string(), "test error"); + assert_eq!(inner.kind(), io::ErrorKind::Other); + } else { + panic!("Expected Io error variant"); + } + } + + #[test] + fn test_error_chain_depth() { + // 测试错误链的深度 + let root_cause = io::Error::other("root cause"); + let converted: Error = root_cause.into(); + + let mut depth = 0; + let mut current_error: &dyn StdError = &converted; + + while let Some(source) = current_error.source() { + depth += 1; + current_error = source; + // 防止无限循环 + if depth > 10 { + break; + } + } + + assert!(depth > 0, "Error should have at least one source"); + assert!(depth <= 3, "Error chain should not be too deep"); + } + + #[test] + fn test_static_str_lifetime() { + // 测试静态字符串生命周期 + fn create_feature_error() -> Error { + Error::FeatureDisabled("static_feature") + } + + let error = create_feature_error(); + match error { + Error::FeatureDisabled(feature) => assert_eq!(feature, "static_feature"), + _ => panic!("Expected FeatureDisabled error variant"), + } + } + + #[test] + fn test_error_formatting_consistency() { + // 测试错误格式化的一致性 + let errors = vec![ + Error::FeatureDisabled("test"), + Error::MissingField("field"), + Error::ValidationError("validation"), + Error::Custom("custom".to_string()), + ]; + + for error in errors { + let display_str = error.to_string(); + let debug_str = format!("{:?}", error); + + // Display 和 Debug 都不应该为空 + assert!(!display_str.is_empty()); + assert!(!debug_str.is_empty()); + + // Debug 输出通常包含更多信息，但不是绝对的 + // 这里我们只验证两者都有内容即可 + assert!(!debug_str.is_empty()); + assert!(!display_str.is_empty()); + } + } +} diff --git a/crates/event/src/event.rs b/crates/event/src/event.rs new file mode 100644 index 000000000..2a4270be4 --- /dev/null +++ b/crates/event/src/event.rs @@ -0,0 +1,617 @@ +use crate::Error; +use reqwest::dns::Name; +use serde::{Deserialize, Serialize}; +use serde_with::{DeserializeFromStr, SerializeDisplay}; +use smallvec::{smallvec, SmallVec}; +use std::borrow::Cow; +use std::collections::HashMap; +use std::time::{SystemTime, UNIX_EPOCH}; +use strum::{Display, EnumString}; +use uuid::Uuid; + +/// A struct representing the identity of the user +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct Identity { + #[serde(rename = "principalId")] + pub principal_id: String, +} + +impl Identity { + /// Create a new Identity instance + pub fn new(principal_id: String) -> Self { + Self { principal_id } + } + + /// Set the principal ID + pub fn set_principal_id(&mut self, principal_id: String) { + self.principal_id = principal_id; + } +} + +/// A struct representing the bucket information +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct Bucket { + pub name: String, + #[serde(rename = "ownerIdentity")] + pub owner_identity: Identity, + pub arn: String, +} + +impl Bucket { + /// Create a new Bucket instance + pub fn new(name: String, owner_identity: Identity, arn: String) -> Self { + Self { + name, + owner_identity, + arn, + } + } + + /// Set the name of the bucket + pub fn set_name(&mut self, name: String) { + self.name = name; + } + + /// Set the ARN of the bucket + pub fn set_arn(&mut self, arn: String) { + self.arn = arn; + } + + /// Set the owner identity of the bucket + pub fn set_owner_identity(&mut self, owner_identity: Identity) { + self.owner_identity = owner_identity; + } +} + +/// A struct representing the object information +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct Object { + pub key: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub size: Option, + #[serde(default, skip_serializing_if = "Option::is_none", rename = "eTag")] + pub etag: Option, + #[serde(default, skip_serializing_if = "Option::is_none", rename = "contentType")] + pub content_type: Option, + #[serde(default, skip_serializing_if = "Option::is_none", rename = "userMetadata")] + pub user_metadata: Option>, + #[serde(default, skip_serializing_if = "Option::is_none", rename = "versionId")] + pub version_id: Option, + pub sequencer: String, +} + +impl Object { + /// Create a new Object instance + pub fn new( + key: String, + size: Option, + etag: Option, + content_type: Option, + user_metadata: Option>, + version_id: Option, + sequencer: String, + ) -> Self { + Self { + key, + size, + etag, + content_type, + user_metadata, + version_id, + sequencer, + } + } + + /// Set the key + pub fn set_key(&mut self, key: String) { + self.key = key; + } + + /// Set the size + pub fn set_size(&mut self, size: Option) { + self.size = size; + } + + /// Set the etag + pub fn set_etag(&mut self, etag: Option) { + self.etag = etag; + } + + /// Set the content type + pub fn set_content_type(&mut self, content_type: Option) { + self.content_type = content_type; + } + + /// Set the user metadata + pub fn set_user_metadata(&mut self, user_metadata: Option>) { + self.user_metadata = user_metadata; + } + + /// Set the version ID + pub fn set_version_id(&mut self, version_id: Option) { + self.version_id = version_id; + } + + /// Set the sequencer + pub fn set_sequencer(&mut self, sequencer: String) { + self.sequencer = sequencer; + } +} + +/// A struct representing the metadata of the event +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct Metadata { + #[serde(rename = "s3SchemaVersion")] + pub schema_version: String, + #[serde(rename = "configurationId")] + pub configuration_id: String, + pub bucket: Bucket, + pub object: Object, +} + +impl Default for Metadata { + fn default() -> Self { + Self::new() + } +} +impl Metadata { + /// Create a new Metadata instance with default values + pub fn new() -> Self { + Self { + schema_version: "1.0".to_string(), + configuration_id: "default".to_string(), + bucket: Bucket::new( + "default".to_string(), + Identity::new("default".to_string()), + "arn:aws:s3:::default".to_string(), + ), + object: Object::new("default".to_string(), None, None, None, None, None, "default".to_string()), + } + } + + /// Create a new Metadata instance + pub fn create(schema_version: String, configuration_id: String, bucket: Bucket, object: Object) -> Self { + Self { + schema_version, + configuration_id, + bucket, + object, + } + } + + /// Set the schema version + pub fn set_schema_version(&mut self, schema_version: String) { + self.schema_version = schema_version; + } + + /// Set the configuration ID + pub fn set_configuration_id(&mut self, configuration_id: String) { + self.configuration_id = configuration_id; + } + + /// Set the bucket + pub fn set_bucket(&mut self, bucket: Bucket) { + self.bucket = bucket; + } + + /// Set the object + pub fn set_object(&mut self, object: Object) { + self.object = object; + } +} + +/// A struct representing the source of the event +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct Source { + pub host: String, + pub port: String, + #[serde(rename = "userAgent")] + pub user_agent: String, +} + +impl Source { + /// Create a new Source instance + pub fn new(host: String, port: String, user_agent: String) -> Self { + Self { host, port, user_agent } + } + + /// Set the host + pub fn set_host(&mut self, host: String) { + self.host = host; + } + + /// Set the port + pub fn set_port(&mut self, port: String) { + self.port = port; + } + + /// Set the user agent + pub fn set_user_agent(&mut self, user_agent: String) { + self.user_agent = user_agent; + } +} + +/// Builder for creating an Event. +/// +/// This struct is used to build an Event object with various parameters. +/// It provides methods to set each parameter and a build method to create the Event. +#[derive(Default, Clone)] +pub struct EventBuilder { + event_version: Option, + event_source: Option, + aws_region: Option, + event_time: Option, + event_name: Option, + user_identity: Option, + request_parameters: Option>, + response_elements: Option>, + s3: Option, + source: Option, + channels: Option>, +} + +impl EventBuilder { + /// create a builder that pre filled default values + pub fn new() -> Self { + Self { + event_version: Some(Cow::Borrowed("2.0").to_string()), + event_source: Some(Cow::Borrowed("aws:s3").to_string()), + aws_region: Some("us-east-1".to_string()), + event_time: Some(SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs().to_string()), + event_name: None, + user_identity: Some(Identity { + principal_id: "anonymous".to_string(), + }), + request_parameters: Some(HashMap::new()), + response_elements: Some(HashMap::new()), + s3: None, + source: None, + channels: Some(Vec::new().into()), + } + } + + /// verify and set the event version + pub fn event_version(mut self, event_version: impl Into) -> Self { + let event_version = event_version.into(); + if !event_version.is_empty() { + self.event_version = Some(event_version); + } + self + } + + /// verify and set the event source + pub fn event_source(mut self, event_source: impl Into) -> Self { + let event_source = event_source.into(); + if !event_source.is_empty() { + self.event_source = Some(event_source); + } + self + } + + /// set up aws regions + pub fn aws_region(mut self, aws_region: impl Into) -> Self { + self.aws_region = Some(aws_region.into()); + self + } + + /// set event time + pub fn event_time(mut self, event_time: impl Into) -> Self { + self.event_time = Some(event_time.into()); + self + } + + /// set event name + pub fn event_name(mut self, event_name: Name) -> Self { + self.event_name = Some(event_name); + self + } + + /// set user identity + pub fn user_identity(mut self, user_identity: Identity) -> Self { + self.user_identity = Some(user_identity); + self + } + + /// set request parameters + pub fn request_parameters(mut self, request_parameters: HashMap) -> Self { + self.request_parameters = Some(request_parameters); + self + } + + /// set response elements + pub fn response_elements(mut self, response_elements: HashMap) -> Self { + self.response_elements = Some(response_elements); + self + } + + /// setting up s3 metadata + pub fn s3(mut self, s3: Metadata) -> Self { + self.s3 = Some(s3); + self + } + + /// set event source information + pub fn source(mut self, source: Source) -> Self { + self.source = Some(source); + self + } + + /// set up the sending channel + pub fn channels(mut self, channels: Vec) -> Self { + self.channels = Some(channels.into()); + self + } + + /// Create a preconfigured builder for common object event scenarios + pub fn for_object_creation(s3: Metadata, source: Source) -> Self { + Self::new().event_name(Name::ObjectCreatedPut).s3(s3).source(source) + } + + /// Create a preconfigured builder for object deletion events + pub fn for_object_removal(s3: Metadata, source: Source) -> Self { + Self::new().event_name(Name::ObjectRemovedDelete).s3(s3).source(source) + } + + /// build event instance + /// + /// Verify the required fields and create a complete Event object + pub fn build(self) -> Result { + let event_version = self.event_version.ok_or(Error::MissingField("event_version"))?; + + let event_source = self.event_source.ok_or(Error::MissingField("event_source"))?; + + let aws_region = self.aws_region.ok_or(Error::MissingField("aws_region"))?; + + let event_time = self.event_time.ok_or(Error::MissingField("event_time"))?; + + let event_name = self.event_name.ok_or(Error::MissingField("event_name"))?; + + let user_identity = self.user_identity.ok_or(Error::MissingField("user_identity"))?; + + let request_parameters = self.request_parameters.unwrap_or_default(); + let response_elements = self.response_elements.unwrap_or_default(); + + let s3 = self.s3.ok_or(Error::MissingField("s3"))?; + + let source = self.source.ok_or(Error::MissingField("source"))?; + + let channels = self.channels.unwrap_or_else(|| smallvec![]); + + Ok(Event { + event_version, + event_source, + aws_region, + event_time, + event_name, + user_identity, + request_parameters, + response_elements, + s3, + source, + id: Uuid::new_v4(), + timestamp: SystemTime::now(), + channels, + }) + } +} + +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct Event { + #[serde(rename = "eventVersion")] + pub event_version: String, + #[serde(rename = "eventSource")] + pub event_source: String, + #[serde(rename = "awsRegion")] + pub aws_region: String, + #[serde(rename = "eventTime")] + pub event_time: String, + #[serde(rename = "eventName")] + pub event_name: Name, + #[serde(rename = "userIdentity")] + pub user_identity: Identity, + #[serde(rename = "requestParameters")] + pub request_parameters: HashMap, + #[serde(rename = "responseElements")] + pub response_elements: HashMap, + pub s3: Metadata, + pub source: Source, + pub id: Uuid, + pub timestamp: SystemTime, + pub channels: SmallVec<[String; 2]>, +} + +impl Event { + /// create a new event builder + /// + /// Returns an EventBuilder instance pre-filled with default values + pub fn builder() -> EventBuilder { + EventBuilder::new() + } + + /// Quickly create Event instances with necessary fields + /// + /// suitable for common s3 event scenarios + pub fn create(event_name: Name, s3: Metadata, source: Source, channels: Vec) -> Self { + Self::builder() + .event_name(event_name) + .s3(s3) + .source(source) + .channels(channels) + .build() + .expect("Failed to create event, missing necessary parameters") + } + + /// a convenient way to create a preconfigured builder + pub fn for_object_creation(s3: Metadata, source: Source) -> EventBuilder { + EventBuilder::for_object_creation(s3, source) + } + + /// a convenient way to create a preconfigured builder + pub fn for_object_removal(s3: Metadata, source: Source) -> EventBuilder { + EventBuilder::for_object_removal(s3, source) + } + + /// Determine whether an event belongs to a specific type + pub fn is_type(&self, event_type: Name) -> bool { + let mask = event_type.mask(); + (self.event_name.mask() & mask) != 0 + } + + /// Determine whether an event needs to be sent to a specific channel + pub fn is_for_channel(&self, channel: &str) -> bool { + self.channels.iter().any(|c| c == channel) + } +} + +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct Log { + #[serde(rename = "eventName")] + pub event_name: Name, + pub key: String, + pub records: Vec, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, SerializeDisplay, DeserializeFromStr, Display, EnumString)] +#[strum(serialize_all = "SCREAMING_SNAKE_CASE")] +pub enum Name { + ObjectAccessedGet, + ObjectAccessedGetRetention, + ObjectAccessedGetLegalHold, + ObjectAccessedHead, + ObjectAccessedAttributes, + ObjectCreatedCompleteMultipartUpload, + ObjectCreatedCopy, + ObjectCreatedPost, + ObjectCreatedPut, + ObjectCreatedPutRetention, + ObjectCreatedPutLegalHold, + ObjectCreatedPutTagging, + ObjectCreatedDeleteTagging, + ObjectRemovedDelete, + ObjectRemovedDeleteMarkerCreated, + ObjectRemovedDeleteAllVersions, + ObjectRemovedNoOp, + BucketCreated, + BucketRemoved, + ObjectReplicationFailed, + ObjectReplicationComplete, + ObjectReplicationMissedThreshold, + ObjectReplicationReplicatedAfterThreshold, + ObjectReplicationNotTracked, + ObjectRestorePost, + ObjectRestoreCompleted, + ObjectTransitionFailed, + ObjectTransitionComplete, + ObjectManyVersions, + ObjectLargeVersions, + PrefixManyFolders, + IlmDelMarkerExpirationDelete, + ObjectAccessedAll, + ObjectCreatedAll, + ObjectRemovedAll, + ObjectReplicationAll, + ObjectRestoreAll, + ObjectTransitionAll, + ObjectScannerAll, + Everything, +} + +impl Name { + pub fn expand(&self) -> Vec { + match self { + Name::ObjectAccessedAll => vec![ + Name::ObjectAccessedGet, + Name::ObjectAccessedHead, + Name::ObjectAccessedGetRetention, + Name::ObjectAccessedGetLegalHold, + Name::ObjectAccessedAttributes, + ], + Name::ObjectCreatedAll => vec![ + Name::ObjectCreatedCompleteMultipartUpload, + Name::ObjectCreatedCopy, + Name::ObjectCreatedPost, + Name::ObjectCreatedPut, + Name::ObjectCreatedPutRetention, + Name::ObjectCreatedPutLegalHold, + Name::ObjectCreatedPutTagging, + Name::ObjectCreatedDeleteTagging, + ], + Name::ObjectRemovedAll => vec![ + Name::ObjectRemovedDelete, + Name::ObjectRemovedDeleteMarkerCreated, + Name::ObjectRemovedNoOp, + Name::ObjectRemovedDeleteAllVersions, + ], + Name::ObjectReplicationAll => vec![ + Name::ObjectReplicationFailed, + Name::ObjectReplicationComplete, + Name::ObjectReplicationNotTracked, + Name::ObjectReplicationMissedThreshold, + Name::ObjectReplicationReplicatedAfterThreshold, + ], + Name::ObjectRestoreAll => vec![Name::ObjectRestorePost, Name::ObjectRestoreCompleted], + Name::ObjectTransitionAll => { + vec![Name::ObjectTransitionFailed, Name::ObjectTransitionComplete] + } + Name::ObjectScannerAll => vec![Name::ObjectManyVersions, Name::ObjectLargeVersions, Name::PrefixManyFolders], + Name::Everything => (1..=Name::IlmDelMarkerExpirationDelete as u32) + .map(|i| Name::from_repr(i).unwrap()) + .collect(), + _ => vec![*self], + } + } + + pub fn mask(&self) -> u64 { + if (*self as u32) < Name::ObjectAccessedAll as u32 { + 1 << (*self as u32 - 1) + } else { + self.expand().iter().fold(0, |acc, n| acc | (1 << (*n as u32 - 1))) + } + } + + fn from_repr(discriminant: u32) -> Option { + match discriminant { + 1 => Some(Name::ObjectAccessedGet), + 2 => Some(Name::ObjectAccessedGetRetention), + 3 => Some(Name::ObjectAccessedGetLegalHold), + 4 => Some(Name::ObjectAccessedHead), + 5 => Some(Name::ObjectAccessedAttributes), + 6 => Some(Name::ObjectCreatedCompleteMultipartUpload), + 7 => Some(Name::ObjectCreatedCopy), + 8 => Some(Name::ObjectCreatedPost), + 9 => Some(Name::ObjectCreatedPut), + 10 => Some(Name::ObjectCreatedPutRetention), + 11 => Some(Name::ObjectCreatedPutLegalHold), + 12 => Some(Name::ObjectCreatedPutTagging), + 13 => Some(Name::ObjectCreatedDeleteTagging), + 14 => Some(Name::ObjectRemovedDelete), + 15 => Some(Name::ObjectRemovedDeleteMarkerCreated), + 16 => Some(Name::ObjectRemovedDeleteAllVersions), + 17 => Some(Name::ObjectRemovedNoOp), + 18 => Some(Name::BucketCreated), + 19 => Some(Name::BucketRemoved), + 20 => Some(Name::ObjectReplicationFailed), + 21 => Some(Name::ObjectReplicationComplete), + 22 => Some(Name::ObjectReplicationMissedThreshold), + 23 => Some(Name::ObjectReplicationReplicatedAfterThreshold), + 24 => Some(Name::ObjectReplicationNotTracked), + 25 => Some(Name::ObjectRestorePost), + 26 => Some(Name::ObjectRestoreCompleted), + 27 => Some(Name::ObjectTransitionFailed), + 28 => Some(Name::ObjectTransitionComplete), + 29 => Some(Name::ObjectManyVersions), + 30 => Some(Name::ObjectLargeVersions), + 31 => Some(Name::PrefixManyFolders), + 32 => Some(Name::IlmDelMarkerExpirationDelete), + 33 => Some(Name::ObjectAccessedAll), + 34 => Some(Name::ObjectCreatedAll), + 35 => Some(Name::ObjectRemovedAll), + 36 => Some(Name::ObjectReplicationAll), + 37 => Some(Name::ObjectRestoreAll), + 38 => Some(Name::ObjectTransitionAll), + 39 => Some(Name::ObjectScannerAll), + 40 => Some(Name::Everything), + _ => None, + } + } +} diff --git a/crates/event/src/lib.rs b/crates/event/src/lib.rs new file mode 100644 index 000000000..f6e8ea019 --- /dev/null +++ b/crates/event/src/lib.rs @@ -0,0 +1,4 @@ +mod error; +mod event; +mod notifier; +mod system; diff --git a/crates/event/src/notifier.rs b/crates/event/src/notifier.rs new file mode 100644 index 000000000..7105f6563 --- /dev/null +++ b/crates/event/src/notifier.rs @@ -0,0 +1,143 @@ +use crate::config::EventNotifierConfig; +use crate::event::Event; +use common::error::{Error, Result}; +use ecstore::store::ECStore; +use std::sync::Arc; +use tokio::sync::{broadcast, mpsc}; +use tokio_util::sync::CancellationToken; +use tracing::{debug, error, info, instrument, warn}; + +/// Event Notifier +pub struct EventNotifier { + /// The event sending channel + sender: mpsc::Sender, + /// Receiver task handle + task_handle: Option>, + /// Configuration information + config: EventNotifierConfig, + /// Turn off tagging + shutdown: CancellationToken, + /// Close the notification channel + shutdown_complete_tx: Option>, +} + +impl EventNotifier { + /// Create a new event notifier + #[instrument(skip_all)] + pub async fn new(store: Arc) -> Result { + let manager = crate::store::manager::EventManager::new(store); + + let manager = Arc::new(manager.await); + + // Initialize the configuration + let config = manager.clone().init().await?; + + // Create adapters + let adapters = manager.clone().create_adapters().await?; + info!("Created {} adapters", adapters.len()); + + // Create a close marker + let shutdown = CancellationToken::new(); + let (shutdown_complete_tx, _) = broadcast::channel(1); + + // 创建事件通道 - 使用默认容量，因为每个适配器都有自己的队列 + // 这里使用较小的通道容量，因为事件会被快速分发到适配器 + let (sender, mut receiver) = mpsc::channel::(100); + + let shutdown_clone = shutdown.clone(); + let shutdown_complete_tx_clone = shutdown_complete_tx.clone(); + let adapters_clone = adapters.clone(); + + // Start the event processing task + let task_handle = tokio::spawn(async move { + debug!("The event processing task starts"); + + loop { + tokio::select! { + Some(event) = receiver.recv() => { + debug!("The event is received:{}", event.id); + + // Distribute to all adapters + for adapter in &adapters_clone { + let adapter_name = adapter.name(); + match adapter.send(&event).await { + Ok(_) => { + debug!("Event {} Successfully sent to the adapter {}", event.id, adapter_name); + } + Err(e) => { + error!("Event {} send to adapter {} failed:{}", event.id, adapter_name, e); + } + } + } + } + + _ = shutdown_clone.cancelled() => { + info!("A shutdown signal is received, and the event processing task is stopped"); + let _ = shutdown_complete_tx_clone.send(()); + break; + } + } + } + + debug!("The event processing task has been stopped"); + }); + + Ok(Self { + sender, + task_handle: Some(task_handle), + config, + shutdown, + shutdown_complete_tx: Some(shutdown_complete_tx), + }) + } + + /// Turn off the event notifier + pub async fn shutdown(&mut self) -> Result<()> { + info!("Turn off the event notifier"); + self.shutdown.cancel(); + + if let Some(shutdown_tx) = self.shutdown_complete_tx.take() { + let mut rx = shutdown_tx.subscribe(); + + // Wait for the shutdown to complete the signal or time out + tokio::select! { + _ = rx.recv() => { + debug!("A shutdown completion signal is received"); + } + _ = tokio::time::sleep(std::time::Duration::from_secs(10)) => { + warn!("Shutdown timeout and forced termination"); + } + } + } + + if let Some(handle) = self.task_handle.take() { + handle.abort(); + match handle.await { + Ok(_) => debug!("The event processing task has been terminated gracefully"), + Err(e) => { + if e.is_cancelled() { + debug!("The event processing task has been canceled"); + } else { + error!("An error occurred while waiting for the event processing task to terminate:{}", e); + } + } + } + } + + info!("The event notifier is completely turned off"); + Ok(()) + } + + /// Send events + pub async fn send(&self, event: Event) -> Result<()> { + self.sender + .send(event) + .await + .map_err(|e| Error::msg(format!("Failed to send events to channel:{}", e))) + } + + /// Get the current configuration + pub fn config(&self) -> &EventNotifierConfig { + &self.config + } +} diff --git a/crates/event/src/system.rs b/crates/event/src/system.rs new file mode 100644 index 000000000..77ebbb54b --- /dev/null +++ b/crates/event/src/system.rs @@ -0,0 +1,80 @@ +use crate::notifier::EventNotifier; +use common::error::Result; +use ecstore::store::ECStore; +use once_cell::sync::OnceCell; +use std::sync::{Arc, Mutex}; +use tracing::{debug, error, info}; + +/// Global event system +pub struct EventSystem { + /// Event Notifier + notifier: Mutex>, +} + +impl EventSystem { + /// Create a new event system + pub fn new() -> Self { + Self { + notifier: Mutex::new(None), + } + } + + /// Initialize the event system + pub async fn init(&self, store: Arc) -> Result { + info!("Initialize the event system"); + let notifier = EventNotifier::new(store).await?; + let config = notifier.config().clone(); + + let mut guard = self + .notifier + .lock() + .map_err(|e| common::error::Error::msg(format!("Failed to acquire locks:{}", e)))?; + + *guard = Some(notifier); + debug!("The event system initialization is complete"); + + Ok(config) + } + + /// Send events + pub async fn send_event(&self, event: crate::Event) -> Result<()> { + let guard = self + .notifier + .lock() + .map_err(|e| common::error::Error::msg(format!("Failed to acquire locks:{}", e)))?; + + if let Some(notifier) = &*guard { + notifier.send(event).await + } else { + error!("The event system is not initialized"); + Err(common::error::Error::msg("The event system is not initialized")) + } + } + + /// Shut down the event system + pub async fn shutdown(&self) -> Result<()> { + info!("Shut down the event system"); + let mut guard = self + .notifier + .lock() + .map_err(|e| common::error::Error::msg(format!("Failed to acquire locks:{}", e)))?; + + if let Some(ref mut notifier) = *guard { + notifier.shutdown().await?; + *guard = None; + info!("The event system is down"); + Ok(()) + } else { + debug!("The event system has been shut down"); + Ok(()) + } + } +} + +/// A global event system instance +pub static GLOBAL_EVENT_SYS: OnceCell = OnceCell::new(); + +/// Initialize the global event system +pub fn init_global_event_system() -> &'static EventSystem { + GLOBAL_EVENT_SYS.get_or_init(EventSystem::new) +} diff --git a/crates/notify/Cargo.toml b/crates/notify/Cargo.toml index 0358b1aae..15929e178 100644 --- a/crates/notify/Cargo.toml +++ b/crates/notify/Cargo.toml @@ -10,13 +10,12 @@ version.workspace = true default = ["webhook"] webhook = ["dep:reqwest"] mqtt = ["rumqttc"] -kafka = ["dep:rdkafka"] [dependencies] +rustfs-config = { workspace = true, features = ["constants", "notify"] } async-trait = { workspace = true } common = { workspace = true } ecstore = { workspace = true } -lazy_static = { workspace = true } once_cell = { workspace = true } reqwest = { workspace = true, optional = true } rumqttc = { workspace = true, optional = true } @@ -32,11 +31,6 @@ tokio-util = { workspace = true } uuid = { workspace = true, features = ["v4", "serde"] } snap = { workspace = true } - -# Only enable kafka features and related dependencies on Linux -[target.'cfg(target_os = "linux")'.dependencies] -rdkafka = { workspace = true, features = ["tokio"], optional = true } - [dev-dependencies] tokio = { workspace = true, features = ["test-util"] } tracing-subscriber = { workspace = true } diff --git a/crates/notify/src/adapter/kafka.rs b/crates/notify/src/adapter/kafka.rs deleted file mode 100644 index a774331bc..000000000 --- a/crates/notify/src/adapter/kafka.rs +++ /dev/null @@ -1,180 +0,0 @@ -use crate::config::kafka::KafkaConfig; -use crate::config::{default_queue_limit, DEFAULT_RETRY_INTERVAL, STORE_PREFIX}; -use crate::{ChannelAdapter, ChannelAdapterType}; -use crate::{Error, Event, QueueStore}; -use async_trait::async_trait; -use rdkafka::error::KafkaError; -use rdkafka::producer::{FutureProducer, FutureRecord}; -use rdkafka::types::RDKafkaErrorCode; -use rdkafka::util::Timeout; -use std::path::PathBuf; -use std::sync::Arc; -use std::time::Duration; -use tokio::time::sleep; -use ChannelAdapterType::Kafka; - -/// Kafka adapter for sending events to a Kafka topic. -pub struct KafkaAdapter { - producer: FutureProducer, - store: Option>>, - config: KafkaConfig, -} - -impl KafkaAdapter { - /// Creates a new Kafka adapter. - pub fn new(config: &KafkaConfig) -> Result { - // Create a Kafka producer with the provided configuration. - let producer = rdkafka::config::ClientConfig::new() - .set("bootstrap.servers", &config.brokers) - .set("message.timeout.ms", config.timeout.to_string()) - .create() - .map_err(|e| Error::msg(format!("Failed to create a Kafka producer: {}", e)))?; - - // create a queue store if enabled - let store = if !config.common.queue_dir.is_empty() { - let store_path = PathBuf::from(&config.common.queue_dir).join(format!( - "{}-{}-{}", - STORE_PREFIX, - Kafka.as_str(), - config.common.identifier - )); - - let queue_limit = if config.queue_limit > 0 { - config.queue_limit - } else { - default_queue_limit() - }; - let store = QueueStore::new(store_path, config.queue_limit, Some(".event".to_string())); - if let Err(e) = store.open() { - tracing::error!("Unable to open queue storage: {}", e); - None - } else { - Some(Arc::new(store)) - } - } else { - None - }; - - Ok(Self { config, producer, store }) - } - - /// Handle backlog events in storage - pub async fn process_backlog(&self) -> Result<(), Error> { - if let Some(store) = &self.store { - let keys = store.list(); - - for key in keys { - match store.get_multiple(&key) { - Ok(events) => { - for event in events { - // Use the retry interval to send events - if let Err(e) = self.send_with_retry(&event).await { - tracing::error!("Processing of backlog events failed: {}", e); - // If it still fails, we remain in the queue - break; - } - } - - // The event is deleted after it has been successfully processed - if let Err(e) = store.del(&key) { - tracing::error!("Failed to delete a handled event: {}", e); - } - } - Err(e) => { - tracing::error!("Fetch events from the queue failed: {}", e); - - // If the event cannot be read, it may be corrupted, delete it - if let Err(del_err) = store.del(&key) { - tracing::error!("Failed to delete a corrupted event: {}", del_err); - } - } - } - } - } - - Ok(()) - } - - /// Sends an event to the Kafka topic with retry logic. - async fn send_with_retry(&self, event: &Event) -> Result<(), Error> { - let retry_interval = match self.config.retry_interval { - Some(t) => Duration::from_secs(t), - None => Duration::from_secs(DEFAULT_RETRY_INTERVAL), // Default to 3 seconds if not set - }; - - for attempt in 0..self.max_retries { - match self.send_request(event).await { - Ok(_) => return Ok(()), - Err((KafkaError::MessageProduction(RDKafkaErrorCode::QueueFull), _)) => { - tracing::warn!("Kafka attempt {} failed: Queue full. Retrying...", attempt + 1); - // sleep(Duration::from_secs(2u64.pow(attempt))).await; - sleep(retry_interval).await; - } - Err((e, _)) => { - tracing::error!("Kafka send error: {}", e); - return Err(Error::Kafka(e)); - } - } - } - - Err(Error::Custom("Exceeded maximum retry attempts for Kafka message".to_string())) - } - - /// Send a single Kafka message - async fn send_request(&self, event: &Event) -> Result<(), Error> { - // Serialize events - let payload = serde_json::to_string(event).map_err(|e| Error::Custom(format!("Serialization event failed: {}", e)))?; - - // Create a Kafka record - let record = FutureRecord::to(&self.config.topic).payload(&payload).key(&event.id); // Use the event ID as the key - - // Send to Kafka - let delivery_status = self - .producer - .send(record, Duration::from_millis(self.config.timeout)) - .await - .map_err(|(e, _)| Error::Custom(format!("Failed to send to Kafka: {}", e)))?; - // Check delivery status - if let Some((err, _)) = delivery_status { - return Err(Error::Kafka(err)); - } - - Ok(()) - } - - /// Save the event to the queue - async fn save_to_queue(&self, event: &Event) -> Result<(), Error> { - if let Some(store) = &self.store { - store - .put(event.clone()) - .map_err(|e| Error::Custom(format!("Saving events to queue failed: {}", e)))?; - } - Ok(()) - } -} - -#[async_trait] -impl ChannelAdapter for KafkaAdapter { - fn name(&self) -> String { - ChannelAdapterType::Kafka.to_string() - } - - async fn send(&self, event: &Event) -> Result<(), Error> { - // Try to deal with the backlog of events first - let _ = self.process_backlog().await; - - // An attempt was made to send the current event - match self.send_with_retry(event).await { - Ok(_) => Ok(()), - Err(e) => { - // If the send fails and the queue is enabled, save to the queue - if let Some(_) = &self.store { - tracing::warn!("Failed to send events to Kafka and saved to a queue: {}", e); - self.save_to_queue(event).await?; - return Ok(()); - } - Err(e) - } - } - } -} diff --git a/crates/notify/src/adapter/mod.rs b/crates/notify/src/adapter/mod.rs index d3389f7af..7c2d9ee62 100644 --- a/crates/notify/src/adapter/mod.rs +++ b/crates/notify/src/adapter/mod.rs @@ -1,10 +1,8 @@ -use crate::config::adapter::AdapterConfig; +use crate::config::AdapterConfig; use crate::{Error, Event}; use async_trait::async_trait; use std::sync::Arc; -#[cfg(all(feature = "kafka", target_os = "linux"))] -pub(crate) mod kafka; #[cfg(feature = "mqtt")] pub(crate) mod mqtt; #[cfg(feature = "webhook")] @@ -97,10 +95,6 @@ pub fn create_adapters(configs: Vec) -> Result { - adapters.push(Arc::new(kafka::KafkaAdapter::new(kafka_config)?)); - } #[cfg(feature = "mqtt")] AdapterConfig::Mqtt(mqtt_config) => { let (mqtt, mut event_loop) = mqtt::MqttAdapter::new(mqtt_config); @@ -109,8 +103,6 @@ pub fn create_adapters(configs: Vec) -> Result return Err(Error::FeatureDisabled("webhook")), - #[cfg(any(not(feature = "kafka"), not(target_os = "linux")))] - AdapterConfig::Kafka(_) => return Err(Error::FeatureDisabled("kafka")), #[cfg(not(feature = "mqtt"))] AdapterConfig::Mqtt(_) => return Err(Error::FeatureDisabled("mqtt")), } diff --git a/crates/notify/src/adapter/webhook.rs b/crates/notify/src/adapter/webhook.rs index 2019996e2..08f2df21c 100644 --- a/crates/notify/src/adapter/webhook.rs +++ b/crates/notify/src/adapter/webhook.rs @@ -1,8 +1,5 @@ -use crate::config::webhook::WebhookConfig; use crate::config::STORE_PREFIX; -use crate::store::queue::Store; use crate::{ChannelAdapter, ChannelAdapterType}; -use crate::{Error, QueueStore}; use crate::{Event, DEFAULT_RETRY_INTERVAL}; use async_trait::async_trait; use reqwest::header::{HeaderMap, HeaderName, HeaderValue}; diff --git a/crates/notify/src/config/notifier.rs b/crates/notify/src/config.rs similarity index 53% rename from crates/notify/src/config/notifier.rs rename to crates/notify/src/config.rs index 5d0658c73..2834b78de 100644 --- a/crates/notify/src/config/notifier.rs +++ b/crates/notify/src/config.rs @@ -1,20 +1,59 @@ -use crate::config::{adapter::AdapterConfig, kafka::KafkaConfig, mqtt::MqttConfig, webhook::WebhookConfig}; +use rustfs_config::notify::mqtt::MQTTArgs; +use rustfs_config::notify::webhook::WebhookArgs; use serde::{Deserialize, Serialize}; use std::collections::HashMap; +use std::env; + +/// The default configuration file name +const DEFAULT_CONFIG_FILE: &str = "notify"; + +/// The prefix for the configuration file +pub const STORE_PREFIX: &str = "rustfs"; + +/// The default retry interval for the webhook adapter +pub const DEFAULT_RETRY_INTERVAL: u64 = 3; + +/// The default maximum retry count for the webhook adapter +pub const DEFAULT_MAX_RETRIES: u32 = 3; + +/// The default notification queue limit +pub const DEFAULT_NOTIFY_QUEUE_LIMIT: u64 = 10000; + +/// Provide temporary directories as default storage paths +pub(crate) fn default_queue_dir() -> String { + env::var("EVENT_QUEUE_DIR").unwrap_or_else(|e| { + tracing::info!("Failed to get `EVENT_QUEUE_DIR` failed err: {}", e.to_string()); + env::temp_dir().join(DEFAULT_CONFIG_FILE).to_string_lossy().to_string() + }) +} + +/// Provides the recommended default channel capacity for high concurrency systems +pub(crate) fn default_queue_limit() -> u64 { + env::var("EVENT_CHANNEL_CAPACITY") + .unwrap_or_else(|_| DEFAULT_NOTIFY_QUEUE_LIMIT.to_string()) + .parse() + .unwrap_or(DEFAULT_NOTIFY_QUEUE_LIMIT) // Default to 10000 if parsing fails +} + +/// Configuration for the adapter. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "type")] +pub enum AdapterConfig { + Webhook(WebhookArgs), + Mqtt(MQTTArgs), +} + /// Event Notifier Configuration /// This struct contains the configuration for the event notifier system, #[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct EventNotifierConfig { /// A collection of webhook configurations, with the key being a unique identifier #[serde(default)] - pub webhook: HashMap, - /// A collection of Kafka configurations, with the key being a unique identifier - #[serde(default)] - pub kafka: HashMap, + pub webhook: HashMap, ///MQTT configuration collection, with the key being a unique identifier #[serde(default)] - pub mqtt: HashMap, + pub mqtt: HashMap, } impl EventNotifierConfig { @@ -49,21 +88,14 @@ impl EventNotifierConfig { // Add all enabled webhook configurations for webhook in self.webhook.values() { - if webhook.common.enable { + if webhook.enable { adapters.push(AdapterConfig::Webhook(webhook.clone())); } } - // Add all enabled Kafka configurations - for kafka in self.kafka.values() { - if kafka.common.enable { - adapters.push(AdapterConfig::Kafka(kafka.clone())); - } - } - // Add all enabled MQTT configurations for mqtt in self.mqtt.values() { - if mqtt.common.enable { + if mqtt.enable { adapters.push(AdapterConfig::Mqtt(mqtt.clone())); } } diff --git a/crates/notify/src/config/adapter.rs b/crates/notify/src/config/adapter.rs deleted file mode 100644 index 19dc0978a..000000000 --- a/crates/notify/src/config/adapter.rs +++ /dev/null @@ -1,42 +0,0 @@ -use crate::config::kafka::KafkaConfig; -use crate::config::mqtt::MqttConfig; -use crate::config::webhook::WebhookConfig; -use crate::config::{default_queue_dir, default_queue_limit}; -use serde::{Deserialize, Serialize}; - -/// Add a common field for the adapter configuration -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct AdapterCommon { - /// Adapter identifier for unique identification - pub identifier: String, - /// Adapter description information - pub comment: String, - /// Whether to enable this adapter - #[serde(default)] - pub enable: bool, - #[serde(default = "default_queue_dir")] - pub queue_dir: String, - #[serde(default = "default_queue_limit")] - pub queue_limit: u64, -} - -impl Default for AdapterCommon { - fn default() -> Self { - Self { - identifier: String::new(), - comment: String::new(), - enable: false, - queue_dir: default_queue_dir(), - queue_limit: default_queue_limit(), - } - } -} - -/// Configuration for the adapter. -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(tag = "type")] -pub enum AdapterConfig { - Webhook(WebhookConfig), - Kafka(KafkaConfig), - Mqtt(MqttConfig), -} diff --git a/crates/notify/src/config/kafka.rs b/crates/notify/src/config/kafka.rs deleted file mode 100644 index 84a789e5c..000000000 --- a/crates/notify/src/config/kafka.rs +++ /dev/null @@ -1,44 +0,0 @@ -use crate::config::adapter::AdapterCommon; -use crate::config::{default_queue_dir, default_queue_limit}; -use serde::{Deserialize, Serialize}; - -/// Configuration for the Kafka adapter. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct KafkaConfig { - #[serde(flatten)] - pub common: AdapterCommon, - pub brokers: String, - pub topic: String, - pub max_retries: u32, - pub timeout: u64, -} - -impl Default for KafkaConfig { - fn default() -> Self { - Self { - common: AdapterCommon::default(), - brokers: String::new(), - topic: String::new(), - max_retries: 3, - timeout: 5000, - } - } -} - -impl KafkaConfig { - /// Create a new Kafka configuration - pub fn new(identifier: impl Into, brokers: impl Into, topic: impl Into) -> Self { - Self { - common: AdapterCommon { - identifier: identifier.into(), - comment: String::new(), - enable: true, - queue_dir: default_queue_dir(), - queue_limit: default_queue_limit(), - }, - brokers: brokers.into(), - topic: topic.into(), - ..Default::default() - } - } -} diff --git a/crates/notify/src/config/mod.rs b/crates/notify/src/config/mod.rs deleted file mode 100644 index fa69e5a48..000000000 --- a/crates/notify/src/config/mod.rs +++ /dev/null @@ -1,38 +0,0 @@ -use std::env; - -pub mod adapter; -pub mod kafka; -pub mod mqtt; -pub mod notifier; -pub mod webhook; - -/// The default configuration file name -const DEFAULT_CONFIG_FILE: &str = "event"; - -/// The prefix for the configuration file -pub const STORE_PREFIX: &str = "rustfs"; - -/// The default retry interval for the webhook adapter -pub const DEFAULT_RETRY_INTERVAL: u64 = 3; - -/// The default maximum retry count for the webhook adapter -pub const DEFAULT_MAX_RETRIES: u32 = 3; - -/// The default notification queue limit -pub const DEFAULT_NOTIFY_QUEUE_LIMIT: u64 = 10000; - -/// Provide temporary directories as default storage paths -pub(crate) fn default_queue_dir() -> String { - env::var("EVENT_QUEUE_DIR").unwrap_or_else(|e| { - tracing::info!("Failed to get `EVENT_QUEUE_DIR` failed err: {}", e.to_string()); - env::temp_dir().join(DEFAULT_CONFIG_FILE).to_string_lossy().to_string() - }) -} - -/// Provides the recommended default channel capacity for high concurrency systems -pub(crate) fn default_queue_limit() -> u64 { - env::var("EVENT_CHANNEL_CAPACITY") - .unwrap_or_else(|_| DEFAULT_NOTIFY_QUEUE_LIMIT.to_string()) - .parse() - .unwrap_or(DEFAULT_NOTIFY_QUEUE_LIMIT) // Default to 10000 if parsing fails -} diff --git a/crates/notify/src/config/mqtt.rs b/crates/notify/src/config/mqtt.rs deleted file mode 100644 index 5090fcc91..000000000 --- a/crates/notify/src/config/mqtt.rs +++ /dev/null @@ -1,46 +0,0 @@ -use crate::config::adapter::AdapterCommon; -use crate::config::{default_queue_dir, default_queue_limit}; -use serde::{Deserialize, Serialize}; - -/// Configuration for the MQTT adapter. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct MqttConfig { - #[serde(flatten)] - pub common: AdapterCommon, - pub broker: String, - pub port: u16, - pub client_id: String, - pub topic: String, - pub max_retries: u32, -} - -impl Default for MqttConfig { - fn default() -> Self { - Self { - common: AdapterCommon::default(), - broker: String::new(), - port: 1883, - client_id: String::new(), - topic: String::new(), - max_retries: 3, - } - } -} - -impl MqttConfig { - /// Create a new MQTT configuration - pub fn new(identifier: impl Into, broker: impl Into, topic: impl Into) -> Self { - Self { - common: AdapterCommon { - identifier: identifier.into(), - comment: String::new(), - enable: true, - queue_dir: default_queue_dir(), - queue_limit: default_queue_limit(), - }, - broker: broker.into(), - topic: topic.into(), - ..Default::default() - } - } -} diff --git a/crates/notify/src/config/webhook.rs b/crates/notify/src/config/webhook.rs deleted file mode 100644 index 79744110d..000000000 --- a/crates/notify/src/config/webhook.rs +++ /dev/null @@ -1,88 +0,0 @@ -use crate::config::adapter::AdapterCommon; -use crate::config::{default_queue_dir, default_queue_limit}; -use serde::{Deserialize, Serialize}; -use std::collections::HashMap; -use std::path::Path; -use tracing::info; - -/// Configuration for the webhook adapter. -#[derive(Debug, Clone, Serialize, Deserialize, Default)] -pub struct WebhookConfig { - #[serde(flatten)] - pub common: AdapterCommon, - pub endpoint: String, - pub auth_token: Option, - pub custom_headers: Option>, - pub max_retries: u32, - pub retry_interval: Option, - pub timeout: Option, - #[serde(default)] - pub client_cert: Option, - #[serde(default)] - pub client_key: Option, -} - -impl WebhookConfig { - /// validate the configuration for the webhook adapter - /// - /// # Returns - /// - /// - `Result<(), String>`: Ok if the configuration is valid, Err with a message if invalid. - /// - /// # Errors - /// - Returns an error if the configuration is invalid, such as empty endpoint, unreasonable timeout, or mismatched certificate and key. - pub fn validate(&self) -> Result<(), String> { - // If not enabled, the other fields are not validated - if !self.common.enable { - return Ok(()); - } - - // verify that endpoint cannot be empty - if self.endpoint.trim().is_empty() { - return Err("Webhook endpoint cannot be empty".to_string()); - } - - // verification timeout must be reasonable - if self.timeout.is_some() { - match self.timeout { - Some(timeout) if timeout > 0 => { - info!("Webhook timeout is set to {}", timeout); - } - _ => return Err("Webhook timeout must be greater than 0".to_string()), - } - } - - // Verify that the maximum number of retry is reasonable - if self.max_retries > 10 { - return Err("Maximum retry count cannot exceed 10".to_string()); - } - - // Verify the queue directory path - if !self.common.queue_dir.is_empty() && !Path::new(&self.common.queue_dir).is_absolute() { - return Err("Queue directory path should be absolute".to_string()); - } - - // The authentication certificate and key must appear in pairs - if (self.client_cert.is_some() && self.client_key.is_none()) || (self.client_cert.is_none() && self.client_key.is_some()) - { - return Err("Certificate and key must be specified as a pair".to_string()); - } - - Ok(()) - } - - /// Create a new webhook configuration - pub fn new(identifier: impl Into, endpoint: impl Into) -> Self { - Self { - common: AdapterCommon { - identifier: identifier.into(), - comment: String::new(), - enable: true, - queue_dir: default_queue_dir(), - queue_limit: default_queue_limit(), - }, - endpoint: endpoint.into(), - ..Default::default() - } - } -} diff --git a/crates/notify/src/lib.rs b/crates/notify/src/lib.rs index fc31e1727..5fdccafce 100644 --- a/crates/notify/src/lib.rs +++ b/crates/notify/src/lib.rs @@ -7,8 +7,6 @@ mod store; mod system; pub use adapter::create_adapters; -#[cfg(all(feature = "kafka", target_os = "linux"))] -pub use adapter::kafka::KafkaAdapter; #[cfg(feature = "mqtt")] pub use adapter::mqtt::MqttAdapter; #[cfg(feature = "webhook")] @@ -16,18 +14,6 @@ pub use adapter::webhook::WebhookAdapter; pub use adapter::ChannelAdapter; pub use adapter::ChannelAdapterType; -pub use config::adapter::AdapterCommon; -pub use config::adapter::AdapterConfig; -pub use config::notifier::EventNotifierConfig; -pub use config::{DEFAULT_MAX_RETRIES, DEFAULT_RETRY_INTERVAL}; +pub use config::{AdapterConfig, EventNotifierConfig, DEFAULT_MAX_RETRIES, DEFAULT_RETRY_INTERVAL}; pub use error::Error; pub use event::{Bucket, Event, EventBuilder, Identity, Log, Metadata, Name, Object, Source}; -pub use store::queue::QueueStore; - -#[cfg(all(feature = "kafka", target_os = "linux"))] -pub use config::kafka::KafkaConfig; -#[cfg(feature = "mqtt")] -pub use config::mqtt::MqttConfig; - -#[cfg(feature = "webhook")] -pub use config::webhook::WebhookConfig; diff --git a/crates/notify/src/notifier.rs b/crates/notify/src/notifier.rs index 469b6fd94..b0d16a440 100644 --- a/crates/notify/src/notifier.rs +++ b/crates/notify/src/notifier.rs @@ -1,4 +1,4 @@ -use crate::config::notifier::EventNotifierConfig; +use crate::config::EventNotifierConfig; use crate::Event; use common::error::{Error, Result}; use ecstore::store::ECStore; diff --git a/crates/notify/src/store/manager.rs b/crates/notify/src/store/manager.rs index 06841e15e..2573740b3 100644 --- a/crates/notify/src/store/manager.rs +++ b/crates/notify/src/store/manager.rs @@ -115,11 +115,6 @@ impl EventManager { merged.webhook.insert(id, config); } - // Merge Kafka configurations - for (id, config) in new.kafka { - merged.kafka.insert(id, config); - } - // Merge MQTT configurations for (id, config) in new.mqtt { merged.mqtt.insert(id, config); diff --git a/crates/notify/src/store/mod.rs b/crates/notify/src/store/mod.rs index 081e4e2e0..33ea11e0b 100644 --- a/crates/notify/src/store/mod.rs +++ b/crates/notify/src/store/mod.rs @@ -1,2 +1,314 @@ +use async_trait::async_trait; +use serde::{de::DeserializeOwned, Serialize}; +use std::error::Error; +use std::fmt; +use std::fmt::Display; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::mpsc; +use tokio::time; + pub(crate) mod manager; pub(crate) mod queue; + +// 常量定义 +pub const RETRY_INTERVAL: Duration = Duration::from_secs(3); +pub const DEFAULT_LIMIT: u64 = 100000; // 默认存储限制 +pub const DEFAULT_EXT: &str = ".unknown"; +pub const COMPRESS_EXT: &str = ".snappy"; + +// 错误类型 +#[derive(Debug)] +pub enum StoreError { + NotConnected, + LimitExceeded, + IoError(std::io::Error), + Utf8(std::str::Utf8Error), + SerdeError(serde_json::Error), + Deserialize(serde_json::Error), + UuidError(uuid::Error), + Other(String), +} + +impl Display for StoreError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + StoreError::NotConnected => write!(f, "not connected to target server/service"), + StoreError::LimitExceeded => write!(f, "the maximum store limit reached"), + StoreError::IoError(e) => write!(f, "IO error: {}", e), + StoreError::Utf8(e) => write!(f, "UTF-8 conversion error: {}", e), + StoreError::SerdeError(e) => write!(f, "serialization error: {}", e), + StoreError::Deserialize(e) => write!(f, "deserialization error: {}", e), + StoreError::UuidError(e) => write!(f, "UUID generation error: {}", e), + StoreError::Other(s) => write!(f, "{}", s), + } + } +} + +impl Error for StoreError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + match self { + StoreError::IoError(e) => Some(e), + StoreError::SerdeError(e) => Some(e), + StoreError::UuidError(e) => Some(e), + _ => None, + } + } +} + +impl From for StoreError { + fn from(e: std::io::Error) -> Self { + StoreError::IoError(e) + } +} + +impl From for StoreError { + fn from(e: serde_json::Error) -> Self { + StoreError::SerdeError(e) + } +} + +impl From for StoreError { + fn from(e: uuid::Error) -> Self { + StoreError::UuidError(e) + } +} + +pub type StoreResult = Result; + +// 日志记录器类型 +pub type Logger = fn(ctx: Option<&str>, err: StoreError, id: &str, err_kind: &[&dyn Display]); + +// Key 结构体定义 +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Key { + pub name: String, + pub compress: bool, + pub extension: String, + pub item_count: usize, +} + +impl Key { + pub fn new(name: String, extension: String) -> Self { + Self { + name, + extension, + compress: false, + item_count: 1, + } + } + + pub fn with_compression(mut self, compress: bool) -> Self { + self.compress = compress; + self + } + + pub fn with_item_count(mut self, count: usize) -> Self { + self.item_count = count; + self + } + + pub fn to_string(&self) -> String { + let mut key_str = self.name.clone(); + + if self.item_count > 1 { + key_str = format!("{}:{}", self.item_count, self.name); + } + + let ext = if self.compress { + format!("{}{}", self.extension, COMPRESS_EXT) + } else { + self.extension.clone() + }; + + format!("{}{}", key_str, ext) + } +} + +impl Display for Key { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.to_string()) + } +} + +pub fn parse_key(k: &str) -> Key { + let mut key = Key { + name: k.to_string(), + compress: false, + extension: String::new(), + item_count: 1, + }; + + // 检查压缩扩展名 + if k.ends_with(COMPRESS_EXT) { + key.compress = true; + key.name = key.name[..key.name.len() - COMPRESS_EXT.len()].to_string(); + } + + // 解析项目数量 + if let Some(colon_pos) = key.name.find(':') { + if let Ok(count) = key.name[..colon_pos].parse::() { + key.item_count = count; + key.name = key.name[colon_pos + 1..].to_string(); + } + } + + // 解析扩展名 + if let Some(dot_pos) = key.name.rfind('.') { + key.extension = key.name[dot_pos..].to_string(); + key.name = key.name[..dot_pos].to_string(); + } + + key +} + +// Target trait 定义 +#[async_trait] +pub trait Target: Send + Sync { + fn name(&self) -> String; + async fn send_from_store(&self, key: Key) -> StoreResult<()>; +} + +// Store trait 定义 +#[async_trait] +pub trait Store: Send + Sync +where + T: Serialize + DeserializeOwned + Send + Sync + 'static, +{ + async fn put(&self, item: T) -> StoreResult; + async fn put_multiple(&self, items: Vec) -> StoreResult; + async fn get(&self, key: Key) -> StoreResult; + async fn get_multiple(&self, key: Key) -> StoreResult>; + async fn get_raw(&self, key: Key) -> StoreResult>; + async fn put_raw(&self, b: Vec) -> StoreResult; + async fn len(&self) -> usize; + async fn list(&self) -> Vec; + async fn del(&self, key: Key) -> StoreResult<()>; + async fn open(&self) -> StoreResult<()>; + async fn delete(&self) -> StoreResult<()>; +} + +// 重播项目辅助函数 +pub async fn replay_items(store: Arc>, done_ch: mpsc::Receiver<()>, log: Logger, id: &str) -> mpsc::Receiver +where + T: Serialize + DeserializeOwned + Send + Sync + 'static, +{ + let (tx, rx) = mpsc::channel(100); // 合理的缓冲区大小 + let id = id.to_string(); + + tokio::spawn(async move { + let mut done_ch = done_ch; + let mut retry_interval = time::interval(RETRY_INTERVAL); + let mut retry_interval = time::interval_at(retry_interval.tick().await, RETRY_INTERVAL); + + loop { + let keys = store.list().await; + + for key in keys { + let tx = tx.clone(); + tokio::select! { + _ = tx.send(key) => { + // 成功发送下一个键 + } + _ = done_ch.recv() => { + return; + } + } + } + + tokio::select! { + _ = retry_interval.tick() => { + // 重试定时器触发，继续循环 + } + _ = done_ch.recv() => { + return; + } + } + } + }); + + rx +} + +// 发送项目辅助函数 +pub async fn send_items(target: &dyn Target, mut key_ch: mpsc::Receiver, mut done_ch: mpsc::Receiver<()>, logger: Logger) { + let mut retry_interval = time::interval(RETRY_INTERVAL); + + async fn try_send( + target: &dyn Target, + key: Key, + retry_interval: &mut time::Interval, + done_ch: &mut mpsc::Receiver<()>, + logger: Logger, + ) -> bool { + loop { + match target.send_from_store(key.clone()).await { + Ok(_) => return true, + Err(err) => { + logger(None, err, &target.name(), &[&format!("unable to send log entry to '{}'", target.name())]); + + tokio::select! { + _ = retry_interval.tick() => { + // 重试 + } + _ = done_ch.recv() => { + return false; + } + } + } + } + } + } + + loop { + tokio::select! { + maybe_key = key_ch.recv() => { + match maybe_key { + Some(key) => { + if !try_send(target, key, &mut retry_interval, &mut done_ch, logger).await { + return; + } + } + None => return, + } + } + _ = done_ch.recv() => { + return; + } + } + } +} + +// 流式传输项目 +pub async fn stream_items(store: Arc>, target: &dyn Target, done_ch: mpsc::Receiver<()>, logger: Logger) +where + T: Serialize + DeserializeOwned + Send + Sync + 'static, +{ + // 创建一个 done_ch 的克隆，以便可以将其传递给 replay_items + // let (tx, rx) = mpsc::channel::<()>(1); + + let (tx_replay, rx_replay) = mpsc::channel::<()>(1); + let (tx_send, rx_send) = mpsc::channel::<()>(1); + + let mut done_ch = done_ch; + + let key_ch = replay_items(store, rx_replay, logger, &target.name()).await; + // let key_ch = replay_items(store, rx, logger, &target.name()).await; + + let tx_replay_clone = tx_replay.clone(); + let tx_send_clone = tx_send.clone(); + + // 监听原始 done_ch，如果收到信号，则关闭我们创建的通道 + tokio::spawn(async move { + // if done_ch.recv().await.is_some() { + // let _ = tx.send(()).await; + // } + if done_ch.recv().await.is_some() { + let _ = tx_replay_clone.send(()).await; + let _ = tx_send_clone.send(()).await; + } + }); + + // send_items(target, key_ch, rx, logger).await; + send_items(target, key_ch, rx_send, logger).await; +} diff --git a/crates/notify/src/store/queue.rs b/crates/notify/src/store/queue.rs index 16a768184..b01898094 100644 --- a/crates/notify/src/store/queue.rs +++ b/crates/notify/src/store/queue.rs @@ -1,515 +1,243 @@ -use common::error::{Error, Result}; +use crate::store::{parse_key, Key, Store, StoreError, StoreResult, DEFAULT_EXT, DEFAULT_LIMIT}; +use async_trait::async_trait; use serde::{de::DeserializeOwned, Serialize}; use snap::raw::{Decoder, Encoder}; -use std::collections::HashMap; -use std::io::Read; -use std::marker::PhantomData; +use std::collections::BTreeMap; use std::path::{Path, PathBuf}; -use std::sync::{Arc, RwLock}; use std::time::{SystemTime, UNIX_EPOCH}; -use std::{fs, io}; +use tokio::fs; +use tokio::sync::RwLock; use uuid::Uuid; -/// Keys in storage -#[derive(Debug, Clone, PartialEq, Eq, Hash)] -pub struct Key { - /// Key name - pub name: String, - /// Whether to compress - pub compress: bool, - /// filename extension - pub extension: String, - /// Number of items - pub item_count: usize, -} - -impl Key { - /// Create a new key - pub fn new(name: impl Into, extension: impl Into, compress: bool) -> Self { - Self { - name: name.into(), - compress, - extension: extension.into(), - item_count: 1, - } - } - - /// Convert to string form - #[allow(clippy::inherent_to_string)] - pub fn to_string(&self) -> String { - let mut key_str = self.name.clone(); - if self.item_count > 1 { - key_str = format!("{}:{}", self.item_count, self.name); - } - - let compress_ext = if self.compress { COMPRESS_EXT } else { "" }; - format!("{}{}{}", key_str, self.extension, compress_ext) - } -} - -/// Parse key from file name -#[allow(clippy::redundant_closure)] -pub fn parse_key(filename: &str) -> Key { - let compress = filename.ends_with(COMPRESS_EXT); - let filename = if compress { - &filename[..filename.len() - 7] // 移除 ".snappy" - } else { - filename - }; - - let mut parts = filename.splitn(2, '.'); - let name_part = parts.next().unwrap_or(""); - let extension = parts - .next() - .map_or_else(|| String::new(), |ext| format!(".{}", ext)) - .to_string(); - - let mut name = name_part.to_string(); - let mut item_count = 1; - - if let Some(pos) = name_part.find(':') { - if let Ok(count) = name_part[..pos].parse::() { - item_count = count; - name = name_part[pos + 1..].to_string(); - } - } - - Key { - name, - compress, - extension, - item_count, - } -} - -/// Store the characteristics of the project -pub trait Store: Send + Sync -where - T: Serialize + DeserializeOwned + Clone + Send + Sync + 'static, -{ - /// Store a single item - fn put(&self, item: T) -> Result; - - /// Store multiple projects - fn put_multiple(&self, items: Vec) -> Result; - - /// Get a single item - fn get(&self, key: &Key) -> Result; - - /// Get multiple items - fn get_multiple(&self, key: &Key) -> Result>; - - /// Get the raw bytes - fn get_raw(&self, key: &Key) -> Result>; - - /// Stores raw bytes - fn put_raw(&self, data: &[u8]) -> Result; - - /// Gets the number of items in storage - fn len(&self) -> usize; - - /// Whether it is empty or not - fn is_empty(&self) -> bool { - self.len() == 0 - } - - /// Lists all keys - fn list(&self) -> Vec; - - /// Delete the key - fn del(&self, key: &Key) -> Result<()>; - - /// Open Storage - fn open(&self) -> Result<()>; - - /// Delete the storage - fn delete(&self) -> Result<()>; -} - -const DEFAULT_LIMIT: u64 = 100000; -const DEFAULT_EXT: &str = ".unknown"; -const COMPRESS_EXT: &str = ".snappy"; - -/// Queue storage implementation pub struct QueueStore { - /// Project Limitations entry_limit: u64, - /// Storage directory directory: PathBuf, - /// filename extension file_ext: String, - /// Item mapping: key -> modified time (Unix nanoseconds) - entries: Arc>>, - /// Type tags - _phantom: PhantomData, - /// Whether to compress - compress: bool, - /// Store name - name: String, + entries: RwLock>, + _phantom: std::marker::PhantomData, } impl QueueStore where - T: Serialize + DeserializeOwned + Clone + Send + Sync + 'static, + T: Serialize + DeserializeOwned + Send + Sync + 'static, { - /// Create a new queue store - pub fn new>(directory: P, name: String, limit: u64, ext: Option) -> Self { - let limit = if limit == 0 { DEFAULT_LIMIT } else { limit }; - let ext = ext.unwrap_or_else(|| DEFAULT_EXT.to_string()); - let mut path = PathBuf::from(directory.as_ref()); - path.push(&name); - - // Create a directory (if it doesn't exist) - if !path.exists() { - if let Err(e) = fs::create_dir_all(&path) { - tracing::error!("创建存储目录失败 {}: {}", path.display(), e); - } - } + pub fn new>(directory: P, limit: u64, ext: Option<&str>) -> Self { + let entry_limit = if limit == 0 { DEFAULT_LIMIT } else { limit }; + let ext = ext.unwrap_or(DEFAULT_EXT).to_string(); Self { directory: directory.as_ref().to_path_buf(), - name, - entry_limit: limit, + entry_limit, file_ext: ext, - compress: true, // Default to compressing - entries: Arc::new(RwLock::new(HashMap::with_capacity(limit as usize))), - _phantom: PhantomData, + entries: RwLock::new(BTreeMap::new()), + _phantom: std::marker::PhantomData, } } - /// Set the file extension - pub fn with_file_ext(mut self, file_ext: &str) -> Self { - self.file_ext = file_ext.to_string(); - self - } - - /// Set whether to compress or not - pub fn with_compression(mut self, compress: bool) -> Self { - self.compress = compress; - self - } - - /// Get the file path - fn get_file_path(&self, key: &Key) -> PathBuf { - let mut filename = key.to_string(); - filename.push_str(if self.compress { COMPRESS_EXT } else { &self.file_ext }); - self.directory.join(filename) - } - - /// Serialize the project - fn serialize_item(&self, item: &T) -> Result> { - let data = serde_json::to_vec(item).map_err(|e| Error::msg(format!("Serialization failed: {}", e)))?; - - if self.compress { - let mut encoder = Encoder::new(); - Ok(encoder - .compress_vec(&data) - .map_err(|e| Error::msg(format!("Compression failed: {}", e)))?) - } else { - Ok(data) - } - } - - /// Deserialize the project - fn deserialize_item(&self, data: &[u8], is_compressed: bool) -> Result { - let data = if is_compressed { - let mut decoder = Decoder::new(); - decoder - .decompress_vec(data) - .map_err(|e| Error::msg(format!("Unzipping failed: {}", e)))? - } else { - data.to_vec() - }; - - serde_json::from_slice(&data).map_err(|e| Error::msg(format!("Deserialization failed: {}", e))) - } - - /// Lists all files in the directory, sorted by modification time (oldest takes precedence.)） - fn list_files(&self) -> Result> { - let mut files = Vec::new(); - - for entry in fs::read_dir(&self.directory)? { - let entry = entry?; - let metadata = entry.metadata()?; - if metadata.is_file() { - files.push(entry); - } - } - - // Sort by modification time - files.sort_by(|a, b| { - let a_time = a - .metadata() - .map(|m| m.modified()) - .unwrap_or(Ok(UNIX_EPOCH)) - .unwrap_or(UNIX_EPOCH); - let b_time = b - .metadata() - .map(|m| m.modified()) - .unwrap_or(Ok(UNIX_EPOCH)) - .unwrap_or(UNIX_EPOCH); - a_time.cmp(&b_time) - }); - - Ok(files) - } - - /// Write the object to a file - fn write_object(&self, key: &Key, item: &T) -> Result<()> { - // Serialize the object - let data = serde_json::to_vec(item)?; - self.write_bytes(key, &data) - } - - /// Write multiple objects to a file - fn write_multiple_objects(&self, key: &Key, items: &[T]) -> Result<()> { - let mut data = Vec::new(); - for item in items { - let item_data = serde_json::to_vec(item)?; - data.extend_from_slice(&item_data); - data.push(b'\n'); - } - self.write_bytes(key, &data) - } - - /// Write bytes to a file - fn write_bytes(&self, key: &Key, data: &[u8]) -> Result<()> { + async fn write_bytes(&self, key: Key, data: Vec) -> StoreResult<()> { let path = self.directory.join(key.to_string()); - let file_data = if key.compress { - // Use snap to compress data + let data = if key.compress { let mut encoder = Encoder::new(); - encoder - .compress_vec(data) - .map_err(|e| Error::msg(format!("Compression failed:{}", e)))? + encoder.compress_vec(&data).map_err(|e| StoreError::Other(e.to_string()))? } else { - data.to_vec() + data }; - // Make sure the directory exists - if let Some(parent) = path.parent() { - fs::create_dir_all(parent)?; - } + fs::write(&path, &data).await?; - // Write to the file - fs::write(&path, &file_data)?; + // 更新条目映射 + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(|e| StoreError::Other(e.to_string()))? + .as_nanos() as i64; - // Update the item mapping - let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_nanos() as i64; - - let mut entries = self - .entries - .write() - .map_err(|_| Error::msg("Failed to obtain a write lock"))?; - entries.insert(key.to_string(), now); + self.entries.write().await.insert(key.to_string(), now); Ok(()) } - /// Read bytes from a file - fn read_bytes(&self, key: &Key) -> Result> { - let path = self.directory.join(key.to_string()); - let data = fs::read(&path)?; - - if data.is_empty() { - return Err(Error::msg("The file is empty")); - } - - if key.compress { - // Use Snap to extract the data - let mut decoder = Decoder::new(); - decoder - .decompress_vec(&data) - .map_err(|e| Error::msg(format!("Failed to decompress:{}", e))) - } else { - Ok(data) - } + async fn write(&self, key: Key, item: T) -> StoreResult<()> { + let data = serde_json::to_vec(&item)?; + self.write_bytes(key, data).await } - /// Check whether the storage limit is reached - fn check_entry_limit(&self) -> Result<()> { - let entries = self.entries.read().map_err(|_| Error::msg("Failed to obtain a read lock"))?; - if entries.len() as u64 >= self.entry_limit { - return Err(Error::msg("The storage limit has been reached")); + async fn multi_write(&self, key: Key, items: Vec) -> StoreResult<()> { + let mut buffer = Vec::new(); + + for item in items { + let item_data = serde_json::to_vec(&item)?; + buffer.extend_from_slice(&item_data); + buffer.push(b'\n'); // 使用换行符分隔项目 } + + self.write_bytes(key, buffer).await + } + + async fn del_internal(&self, key: &Key) -> StoreResult<()> { + let path = self.directory.join(key.to_string()); + + if let Err(e) = fs::remove_file(&path).await { + if e.kind() != std::io::ErrorKind::NotFound { + return Err(e.into()); + } + } + + self.entries.write().await.remove(&key.to_string()); + Ok(()) } } +#[async_trait] impl Store for QueueStore where - T: Serialize + DeserializeOwned + Clone + Send + Sync + 'static, + T: Serialize + DeserializeOwned + Send + Sync + 'static, { - fn put(&self, item: T) -> Result { - { - self.check_entry_limit()?; + async fn put(&self, item: T) -> StoreResult { + let entries_len = self.entries.read().await.len() as u64; + if entries_len >= self.entry_limit { + return Err(StoreError::LimitExceeded); } - // generate a new uuid + // 生成 UUID 作为键 let uuid = Uuid::new_v4(); - let key = Key::new(uuid.to_string(), &self.file_ext, true); + let key = Key::new(uuid.to_string(), self.file_ext.clone()); - self.write_object(&key, &item)?; - - Ok(key) - } - fn put_multiple(&self, items: Vec) -> Result { - if items.is_empty() { - return Err(Error::msg("The list of items is empty")); - } - - { - self.check_entry_limit()?; - } - - // Generate a new UUID - let uuid = Uuid::new_v4(); - let mut key = Key::new(uuid.to_string(), &self.file_ext, true); - key.item_count = items.len(); - - self.write_multiple_objects(&key, &items)?; + self.write(key.clone(), item).await?; Ok(key) } - fn get(&self, key: &Key) -> Result { - let items = self.get_multiple(key)?; - if items.is_empty() { - return Err(Error::msg("Item not found")); + async fn put_multiple(&self, items: Vec) -> StoreResult { + let entries_len = self.entries.read().await.len() as u64; + if entries_len >= self.entry_limit { + return Err(StoreError::LimitExceeded); } - Ok(items[0].clone()) + if items.is_empty() { + return Err(StoreError::Other("Cannot store empty item list".into())); + } + + // 生成 UUID 作为键 + let uuid = Uuid::new_v4(); + let key = Key::new(uuid.to_string(), self.file_ext.clone()) + .with_item_count(items.len()) + .with_compression(true); + + self.multi_write(key.clone(), items).await?; + + Ok(key) } - fn get_multiple(&self, key: &Key) -> Result> { - let data = self.get_raw(key)?; + async fn get(&self, key: Key) -> StoreResult { + let items = self.get_multiple(key).await?; + items + .into_iter() + .next() + .ok_or_else(|| StoreError::Other("No items found".into())) + } - let mut items = Vec::with_capacity(key.item_count); - let mut reader = io::Cursor::new(&data); + async fn get_multiple(&self, key: Key) -> StoreResult> { + let data = self.get_raw(key).await?; - // Try to read each JSON object - let mut buffer = Vec::new(); - - // if the read fails try parsing it once - if reader.read_to_end(&mut buffer).is_err() { - // Try to parse the entire data as a single object - return match serde_json::from_slice::(&data) { - Ok(item) => { - items.push(item); - Ok(items) - } - Err(_) => { - // An attempt was made to resolve to an array of objects - match serde_json::from_slice::>(&data) { - Ok(array_items) => Ok(array_items), - Err(e) => Err(Error::msg(format!("Failed to parse the data:{}", e))), - } - } - }; + // 尝试解析为 JSON 数组 + match serde_json::from_slice::>(&data) { + Ok(items) if !items.is_empty() => return Ok(items), + Ok(_) => return Err(StoreError::Other("No items deserialized".into())), + Err(_) => {} // 失败则尝试按行解析 } - - // Read JSON objects by row - for line in buffer.split(|&b| b == b'\n') { - if !line.is_empty() { - match serde_json::from_slice::(line) { - Ok(item) => items.push(item), - Err(e) => tracing::warn!("Failed to parse row data:{}", e), - } + // 如果直接解析为 Vec 失败，则尝试按行解析 + // 转换为字符串并按行解析 + let data_str = std::str::from_utf8(&data).map_err(StoreError::Utf8)?; + // 按行解析（JSON Lines） + let mut items = Vec::new(); + for line in data_str.lines() { + let line = line.trim(); + if line.is_empty() { + continue; } + let item = serde_json::from_str::(line).map_err(StoreError::Deserialize)?; + items.push(item); } if items.is_empty() { - return Err(Error::msg("Failed to resolve any items")); + return Err(StoreError::Other("Failed to deserialize items".into())); } Ok(items) } - fn get_raw(&self, key: &Key) -> Result> { - let data = self.read_bytes(key)?; + async fn get_raw(&self, key: Key) -> StoreResult> { + let path = self.directory.join(key.to_string()); + let data = fs::read(&path).await?; - // Delete the wrong file if data.is_empty() { - let _ = self.del(key); - return Err(Error::msg("the file is empty")); + return Err(StoreError::Other("Empty file".into())); } - Ok(data) + if key.compress { + let mut decoder = Decoder::new(); + decoder.decompress_vec(&data).map_err(|e| StoreError::Other(e.to_string())) + } else { + Ok(data) + } } - fn put_raw(&self, data: &[u8]) -> Result { - { - let entries = self.entries.read().map_err(|_| Error::msg("Failed to obtain a read lock"))?; - if entries.len() as u64 >= self.entry_limit { - return Err(Error::msg("the storage limit has been reached")); - } + async fn put_raw(&self, data: Vec) -> StoreResult { + let entries_len = self.entries.read().await.len() as u64; + if entries_len >= self.entry_limit { + return Err(StoreError::LimitExceeded); } - // Generate a new UUID + // 生成 UUID 作为键 let uuid = Uuid::new_v4(); - let key = Key::new(uuid.to_string(), &self.file_ext, true); + let key = Key::new(uuid.to_string(), self.file_ext.clone()); - self.write_bytes(&key, data)?; + self.write_bytes(key.clone(), data).await?; Ok(key) } - fn len(&self) -> usize { - self.entries.read().map(|e| e.len()).unwrap_or(0) + async fn len(&self) -> usize { + self.entries.read().await.len() } - fn list(&self) -> Vec { - let entries = match self.entries.read() { - Ok(guard) => guard, - Err(_) => return Vec::new(), - }; + async fn list(&self) -> Vec { + let entries = self.entries.read().await; - // Convert entries to vectors and sort by timestamp - let mut entries_vec: Vec<_> = entries.iter().collect(); - entries_vec.sort_by(|a, b| a.1.cmp(b.1)); + // 将条目转换为 (key, timestamp) 元组并排序 + let mut entries_vec: Vec<(&String, &i64)> = entries.iter().collect(); + entries_vec.sort_by_key(|(_k, &v)| v); - // Parsing key - entries_vec.iter().map(|(filename, _)| parse_key(filename)).collect() + // 将排序后的键解析为 Key 结构体 + entries_vec.into_iter().map(|(k, _)| parse_key(k)).collect() } - fn del(&self, key: &Key) -> Result<()> { - let path = self.directory.join(key.to_string()); - - // Delete the file - if let Err(e) = fs::remove_file(&path) { - if e.kind() != io::ErrorKind::NotFound { - return Err(e.into()); - } - } - - // Remove the entry from the map - let mut entries = self - .entries - .write() - .map_err(|_| Error::msg("Failed to obtain a write lock"))?; - entries.remove(&key.to_string()); - - Ok(()) + async fn del(&self, key: Key) -> StoreResult<()> { + self.del_internal(&key).await } - fn open(&self) -> Result<()> { - // Create a directory (if it doesn't exist) - fs::create_dir_all(&self.directory)?; + async fn open(&self) -> StoreResult<()> { + // 创建目录（如果不存在） + fs::create_dir_all(&self.directory).await?; - // Read existing files - let files = self.list_files()?; + // 读取已经存在的文件 + let entries = self.entries.write(); + let mut entries = entries.await; + entries.clear(); - let mut entries = self - .entries - .write() - .map_err(|_| Error::msg("Failed to obtain a write lock"))?; + let mut dir_entries = fs::read_dir(&self.directory).await?; + while let Some(entry) = dir_entries.next_entry().await? { + if let Ok(metadata) = entry.metadata().await { + if metadata.is_file() { + let modified = metadata + .modified()? + .duration_since(UNIX_EPOCH) + .map_err(|e| StoreError::Other(e.to_string()))? + .as_nanos() as i64; - for file in files { - if let Ok(meta) = file.metadata() { - if let Ok(modified) = meta.modified() { - if let Ok(since_epoch) = modified.duration_since(UNIX_EPOCH) { - entries.insert(file.file_name().to_string_lossy().to_string(), since_epoch.as_nanos() as i64); - } + entries.insert(entry.file_name().to_string_lossy().to_string(), modified); } } } @@ -517,8 +245,8 @@ where Ok(()) } - fn delete(&self) -> Result<()> { - fs::remove_dir_all(&self.directory)?; + async fn delete(&self) -> StoreResult<()> { + fs::remove_dir_all(&self.directory).await?; Ok(()) } } diff --git a/crates/notify/src/system.rs b/crates/notify/src/system.rs index 97e9c362b..c55bc686c 100644 --- a/crates/notify/src/system.rs +++ b/crates/notify/src/system.rs @@ -1,4 +1,4 @@ -use crate::config::notifier::EventNotifierConfig; +use crate::config::EventNotifierConfig; use crate::notifier::EventNotifier; use common::error::Result; use ecstore::store::ECStore;