mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-28 00:58:59 +00:00
Feature/bucket event notification (#347)
* init event notifer * feat: implement event notification system - Add core event notification interfaces - Support multiple notification backends: - Webhook (default) - Kafka - MQTT - HTTP Producer - Implement configurable event filtering - Add async event dispatching with backpressure handling - Provide serialization/deserialization for event payloads This module enables system events to be published to various endpoints with consistent delivery guarantees and failure handling. * feat(event-notifier): improve notification system initialization safety - Add READY atomic flag to track full initialization status - Implement initialize_safe and start_safe methods with mutex protection - Add wait_until_ready function with configurable timeout - Create initialize_and_start_with_ready_check helper method - Replace sleep-based waiting with proper readiness checks - Add safety checks before sending events - Replace chrono with std::time for time handling - Update error handling to provide clear initialization status This change reduces race conditions in multi-threaded environments and ensures events are only processed when the system is fully ready. * fix sql Signed-off-by: junxiang Mu <1948535941@qq.com> * move protobuf generate into bin crate Signed-off-by: junxiang Mu <1948535941@qq.com> * improve Cargo toml * improve code * improve code for global * improve code * feat(event-notifier): improve environment variable handling - Fix deserialization error when parsing config from environment variables - Add proper array format support for adapters configuration - Update environment variable examples with correct format - Improve documentation for configuration loading - Implement helper functions for environment variable validation This change fixes the "invalid type: map, expected a sequence" error by ensuring proper formatting of array-type fields in environment variables. * feat: integrate event-notifier system with rustfs - Rename package from rustfs-event-notifier to event-notifier for consistency - Add shutdown hooks for event notification system in main process - Handle graceful termination of notification services on server shutdown - Implement initialization and configuration loading for event notification - Fix environment variable configuration to properly parse adapter arrays - Update example code to demonstrate proper configuration usage This change ensures proper integration between rustfs and the event notification system, with clean startup and shutdown sequences to prevent resource leaks during application lifecycle. * feat: improve webhook server and run script integration - Enhance webhook example with proper shutdown handling using tokio::select! - Update run.sh to automatically start webhook server alongside main service - Add event notification configuration to run.sh using environment variables - Set proper port bindings to ensure webhook server starts on port 3000 - Improve console output for better debugging experience - Fix race condition during service startup and shutdown This change ensures proper integration between the webhook server and the main rustfs service, providing a seamless development experience with automatic service discovery and clean termination. * improve for logger * improve code for global.rs * fix: modify webhook port * fix --------- Signed-off-by: junxiang Mu <1948535941@qq.com> Co-authored-by: junxiang Mu <1948535941@qq.com>
This commit is contained in:
houseme
@@ -12,3 +12,4 @@ cli/rustfs-gui/embedded-rustfs/rustfs
|
||||
deploy/config/obs.toml
|
||||
*.log
|
||||
deploy/certs/*
|
||||
*jsonl
|
||||
Generated
+348
-144
File diff suppressed because it is too large
Load Diff
+55
-58
@@ -1,20 +1,21 @@
|
||||
[workspace]
|
||||
members = [
|
||||
"madmin", # Management dashboard and admin API interface
|
||||
"rustfs", # Core file system implementation
|
||||
"ecstore", # Erasure coding storage implementation
|
||||
"e2e_test", # End-to-end test suite
|
||||
"common/common", # Shared utilities and data structures
|
||||
"common/lock", # Distributed locking implementation
|
||||
"common/protos", # Protocol buffer definitions
|
||||
"madmin", # Management dashboard and admin API interface
|
||||
"rustfs", # Core file system implementation
|
||||
"ecstore", # Erasure coding storage implementation
|
||||
"e2e_test", # End-to-end test suite
|
||||
"common/common", # Shared utilities and data structures
|
||||
"common/lock", # Distributed locking implementation
|
||||
"common/protos", # Protocol buffer definitions
|
||||
"common/workers", # Worker thread pools and task scheduling
|
||||
"iam", # Identity and Access Management
|
||||
"crypto", # Cryptography and security features
|
||||
"iam", # Identity and Access Management
|
||||
"crypto", # Cryptography and security features
|
||||
"cli/rustfs-gui", # Graphical user interface client
|
||||
"crates/obs", # Observability utilities
|
||||
"s3select/api",
|
||||
"s3select/query",
|
||||
"appauth",
|
||||
"crates/obs", # Observability utilities
|
||||
"crates/event-notifier", # Event notification system
|
||||
"s3select/api", # S3 Select API interface
|
||||
"s3select/query", # S3 Select query engine
|
||||
"appauth", # Application authentication and authorization
|
||||
]
|
||||
resolver = "2"
|
||||
|
||||
@@ -32,7 +33,21 @@ unsafe_code = "deny"
|
||||
all = "warn"
|
||||
|
||||
[workspace.dependencies]
|
||||
madmin = { path = "./madmin" }
|
||||
api = { path = "./s3select/api", version = "0.0.1" }
|
||||
appauth = { path = "./appauth", version = "0.0.1" }
|
||||
common = { path = "./common/common", version = "0.0.1" }
|
||||
crypto = { path = "./crypto", version = "0.0.1" }
|
||||
ecstore = { path = "./ecstore", version = "0.0.1" }
|
||||
iam = { path = "./iam", version = "0.0.1" }
|
||||
lock = { path = "./common/lock", version = "0.0.1" }
|
||||
madmin = { path = "./madmin", version = "0.0.1" }
|
||||
policy = { path = "./policy", version = "0.0.1" }
|
||||
protos = { path = "./common/protos", version = "0.0.1" }
|
||||
query = { path = "./s3select/query", version = "0.0.1" }
|
||||
rustfs = { path = "./rustfs", version = "0.0.1" }
|
||||
rustfs-obs = { path = "crates/obs", version = "0.0.1" }
|
||||
rustfs-event-notifier = { path = "crates/event-notifier", version = "0.0.1" }
|
||||
workers = { path = "./common/workers", version = "0.0.1" }
|
||||
atoi = "2.0.0"
|
||||
async-recursion = "1.1.1"
|
||||
async-trait = "0.1.88"
|
||||
@@ -42,20 +57,18 @@ axum-extra = "0.10.1"
|
||||
axum-server = { version = "0.7.2", features = ["tls-rustls"] }
|
||||
backon = "1.5.0"
|
||||
bytes = "1.10.1"
|
||||
bytesize = "1.3.3"
|
||||
bytesize = "2.0.1"
|
||||
chrono = { version = "0.4.40", features = ["serde"] }
|
||||
clap = { version = "4.5.35", features = ["derive", "env"] }
|
||||
clap = { version = "4.5.37", features = ["derive", "env"] }
|
||||
config = "0.15.11"
|
||||
datafusion = "46.0.0"
|
||||
datafusion = "46.0.1"
|
||||
derive_builder = "0.20.2"
|
||||
dioxus = { version = "0.6.3", features = ["router"] }
|
||||
dirs = "6.0.0"
|
||||
ecstore = { path = "./ecstore" }
|
||||
flatbuffers = "24.12.23"
|
||||
flatbuffers = "25.2.10"
|
||||
futures = "0.3.31"
|
||||
futures-core = "0.3.31"
|
||||
futures-util = "0.3.31"
|
||||
common = { path = "./common/common" }
|
||||
policy = { path = "./policy" }
|
||||
hex = "0.4.3"
|
||||
hyper = "1.6.0"
|
||||
hyper-util = { version = "0.1.11", features = [
|
||||
@@ -67,67 +80,51 @@ http = "1.3.1"
|
||||
http-body = "1.0.1"
|
||||
humantime = "2.2.0"
|
||||
jsonwebtoken = "9.3.1"
|
||||
keyring = { version = "3.6.2", features = [
|
||||
"apple-native",
|
||||
"windows-native",
|
||||
"sync-secret-service",
|
||||
] }
|
||||
lock = { path = "./common/lock" }
|
||||
keyring = { version = "3.6.2", features = ["apple-native", "windows-native", "sync-secret-service"] }
|
||||
lazy_static = "1.5.0"
|
||||
libsystemd = { version = "0.7" }
|
||||
libsystemd = { version = "0.7.1" }
|
||||
local-ip-address = "0.6.3"
|
||||
matchit = "0.8.4"
|
||||
md-5 = "0.10.6"
|
||||
mime = "0.3.17"
|
||||
mime_guess = "2.0.5"
|
||||
netif = "0.1.6"
|
||||
object_store = "0.11.2"
|
||||
opentelemetry = { version = "0.29.1" }
|
||||
opentelemetry-appender-tracing = { version = "0.29.1", features = [
|
||||
"experimental_use_tracing_span_context",
|
||||
"experimental_metadata_attributes",
|
||||
] }
|
||||
opentelemetry_sdk = { version = "0.29" }
|
||||
opentelemetry-appender-tracing = { version = "0.29.1", features = ["experimental_use_tracing_span_context", "experimental_metadata_attributes"] }
|
||||
opentelemetry_sdk = { version = "0.29.0" }
|
||||
opentelemetry-stdout = { version = "0.29.0" }
|
||||
opentelemetry-otlp = { version = "0.29" }
|
||||
opentelemetry-otlp = { version = "0.29.0" }
|
||||
opentelemetry-prometheus = { version = "0.29.1" }
|
||||
opentelemetry-semantic-conventions = { version = "0.29.0", features = [
|
||||
"semconv_experimental",
|
||||
] }
|
||||
pin-project-lite = "0.2"
|
||||
opentelemetry-semantic-conventions = { version = "0.29.0", features = ["semconv_experimental"] }
|
||||
parking_lot = "0.12.3"
|
||||
pin-project-lite = "0.2.16"
|
||||
prometheus = "0.14.0"
|
||||
# pin-utils = "0.1.0"
|
||||
prost = "0.13.5"
|
||||
prost-build = "0.13.5"
|
||||
prost-types = "0.13.5"
|
||||
protobuf = "3.7"
|
||||
protos = { path = "./common/protos" }
|
||||
rand = "0.8.5"
|
||||
rdkafka = { version = "0.37", features = ["tokio"] }
|
||||
reqwest = { version = "0.12.15", default-features = false, features = [
|
||||
"rustls-tls",
|
||||
"charset",
|
||||
"http2",
|
||||
"macos-system-configuration",
|
||||
"stream",
|
||||
"json",
|
||||
"blocking",
|
||||
] }
|
||||
rfd = { version = "0.15.3", default-features = false, features = [
|
||||
"xdg-portal",
|
||||
"tokio",
|
||||
] }
|
||||
rdkafka = { version = "0.37.0", features = ["tokio"] }
|
||||
reqwest = { version = "0.12.15", default-features = false, features = ["rustls-tls", "charset", "http2", "macos-system-configuration", "stream", "json", "blocking"] }
|
||||
rfd = { version = "0.15.3", default-features = false, features = ["xdg-portal", "tokio"] }
|
||||
rmp = "0.8.14"
|
||||
rmp-serde = "1.3.0"
|
||||
rustfs-obs = { path = "crates/obs", version = "0.0.1" }
|
||||
rumqttc = { version = "0.24" }
|
||||
rust-embed = "8.7.0"
|
||||
rustls = { version = "0.23.26" }
|
||||
rustls-pki-types = "1.11.0"
|
||||
rustls-pemfile = "2.2.0"
|
||||
s3s = { git = "https://github.com/Nugine/s3s.git", rev = "4733cdfb27b2713e832967232cbff413bb768c10" }
|
||||
s3s-policy = { git = "https://github.com/Nugine/s3s.git", rev = "4733cdfb27b2713e832967232cbff413bb768c10" }
|
||||
shadow-rs = { version = "0.38.1", default-features = false }
|
||||
shadow-rs = { version = "1.1.1", default-features = false }
|
||||
serde = { version = "1.0.219", features = ["derive"] }
|
||||
serde_json = "1.0.140"
|
||||
serde_urlencoded = "0.7.1"
|
||||
serde_with = "3.12.0"
|
||||
smallvec = { version = "1.15.0", features = ["serde"] }
|
||||
strum = { version = "0.27.1", features = ["derive"] }
|
||||
sha2 = "0.10.8"
|
||||
snafu = "0.8.5"
|
||||
tempfile = "3.19.1"
|
||||
@@ -143,7 +140,7 @@ time = { version = "0.3.41", features = [
|
||||
tokio = { version = "1.44.2", features = ["fs", "rt-multi-thread"] }
|
||||
tonic = { version = "0.13.0", features = ["gzip"] }
|
||||
tonic-build = "0.13.0"
|
||||
tokio-rustls = { version = "0.26", default-features = false }
|
||||
tokio-rustls = { version = "0.26.2", default-features = false }
|
||||
tokio-stream = "0.1.17"
|
||||
tokio-util = { version = "0.7.14", features = ["io", "compat"] }
|
||||
tower = { version = "0.5.2", features = ["timeout"] }
|
||||
@@ -153,7 +150,7 @@ tracing-core = "0.1.33"
|
||||
tracing-error = "0.2.1"
|
||||
tracing-subscriber = { version = "0.3.19", features = ["env-filter", "time"] }
|
||||
tracing-appender = "0.2.3"
|
||||
tracing-opentelemetry = "0.30"
|
||||
tracing-opentelemetry = "0.30.0"
|
||||
transform-stream = "0.3.1"
|
||||
url = "2.5.4"
|
||||
uuid = { version = "1.16.0", features = [
|
||||
@@ -161,7 +158,7 @@ uuid = { version = "1.16.0", features = [
|
||||
"fast-rng",
|
||||
"macro-diagnostics",
|
||||
] }
|
||||
workers = { path = "./common/workers" }
|
||||
|
||||
|
||||
[profile.wasm-dev]
|
||||
inherits = "dev"
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
[package]
|
||||
name = "rustfs-event-notifier"
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
repository.workspace = true
|
||||
rust-version.workspace = true
|
||||
version.workspace = true
|
||||
|
||||
[features]
|
||||
default = ["webhook"]
|
||||
webhook = ["dep:reqwest"]
|
||||
kafka = ["rdkafka"]
|
||||
mqtt = ["rumqttc"]
|
||||
|
||||
[dependencies]
|
||||
async-trait = { workspace = true }
|
||||
config = { workspace = true }
|
||||
rdkafka = { workspace = true, features = ["tokio"], optional = true }
|
||||
reqwest = { workspace = true, optional = true }
|
||||
rumqttc = { 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"] }
|
||||
|
||||
|
||||
[dev-dependencies]
|
||||
tokio = { workspace = true, features = ["test-util"] }
|
||||
tracing-subscriber = { workspace = true }
|
||||
http = { workspace = true }
|
||||
axum = { workspace = true }
|
||||
|
||||
[lints]
|
||||
workspace = true
|
||||
@@ -0,0 +1,28 @@
|
||||
# ===== 全局配置 =====
|
||||
NOTIFIER__STORE_PATH=/var/log/event-notification
|
||||
NOTIFIER__CHANNEL_CAPACITY=5000
|
||||
|
||||
# ===== 适配器配置（数组格式） =====
|
||||
# Webhook 适配器（索引 0）
|
||||
NOTIFIER__ADAPTERS_0__type=Webhook
|
||||
NOTIFIER__ADAPTERS_0__endpoint=http://127.0.0.1:3020/webhook
|
||||
NOTIFIER__ADAPTERS_0__auth_token=your-auth-token
|
||||
NOTIFIER__ADAPTERS_0__max_retries=3
|
||||
NOTIFIER__ADAPTERS_0__timeout=50
|
||||
NOTIFIER__ADAPTERS_0__custom_headers__x_custom_server=value
|
||||
NOTIFIER__ADAPTERS_0__custom_headers__x_custom_client=value
|
||||
|
||||
# Kafka 适配器（索引 1）
|
||||
NOTIFIER__ADAPTERS_1__type=Kafka
|
||||
NOTIFIER__ADAPTERS_1__brokers=localhost:9092
|
||||
NOTIFIER__ADAPTERS_1__topic=notifications
|
||||
NOTIFIER__ADAPTERS_1__max_retries=3
|
||||
NOTIFIER__ADAPTERS_1__timeout=60
|
||||
|
||||
# MQTT 适配器（索引 2）
|
||||
NOTIFIER__ADAPTERS_2__type=Mqtt
|
||||
NOTIFIER__ADAPTERS_2__broker=mqtt.example.com
|
||||
NOTIFIER__ADAPTERS_2__port=1883
|
||||
NOTIFIER__ADAPTERS_2__client_id=event-notifier
|
||||
NOTIFIER__ADAPTERS_2__topic=events
|
||||
NOTIFIER__ADAPTERS_2__max_retries=3
|
||||
@@ -0,0 +1,28 @@
|
||||
# ===== global configuration =====
|
||||
NOTIFIER__STORE_PATH=/var/log/event-notification
|
||||
NOTIFIER__CHANNEL_CAPACITY=5000
|
||||
|
||||
# ===== adapter configuration array format =====
|
||||
# webhook adapter index 0
|
||||
NOTIFIER__ADAPTERS_0__type=Webhook
|
||||
NOTIFIER__ADAPTERS_0__endpoint=http://127.0.0.1:3020/webhook
|
||||
NOTIFIER__ADAPTERS_0__auth_token=your-auth-token
|
||||
NOTIFIER__ADAPTERS_0__max_retries=3
|
||||
NOTIFIER__ADAPTERS_0__timeout=50
|
||||
NOTIFIER__ADAPTERS_0__custom_headers__x_custom_server=server-value
|
||||
NOTIFIER__ADAPTERS_0__custom_headers__x_custom_client=client-value
|
||||
|
||||
# kafka adapter index 1
|
||||
NOTIFIER__ADAPTERS_1__type=Kafka
|
||||
NOTIFIER__ADAPTERS_1__brokers=localhost:9092
|
||||
NOTIFIER__ADAPTERS_1__topic=notifications
|
||||
NOTIFIER__ADAPTERS_1__max_retries=3
|
||||
NOTIFIER__ADAPTERS_1__timeout=60
|
||||
|
||||
# mqtt adapter index 2
|
||||
NOTIFIER__ADAPTERS_2__type=Mqtt
|
||||
NOTIFIER__ADAPTERS_2__broker=mqtt.example.com
|
||||
NOTIFIER__ADAPTERS_2__port=1883
|
||||
NOTIFIER__ADAPTERS_2__client_id=event-notifier
|
||||
NOTIFIER__ADAPTERS_2__topic=events
|
||||
NOTIFIER__ADAPTERS_2__max_retries=3
|
||||
@@ -0,0 +1,29 @@
|
||||
# config.toml
|
||||
store_path = "/var/log/event-notifier"
|
||||
channel_capacity = 5000
|
||||
|
||||
[[adapters]]
|
||||
type = "Webhook"
|
||||
endpoint = "http://127.0.0.1:3020/webhook"
|
||||
auth_token = "your-auth-token"
|
||||
max_retries = 3
|
||||
timeout = 50
|
||||
|
||||
[adapters.custom_headers]
|
||||
custom_server = "value_server"
|
||||
custom_client = "value_client"
|
||||
|
||||
[[adapters]]
|
||||
type = "Kafka"
|
||||
brokers = "localhost:9092"
|
||||
topic = "notifications"
|
||||
max_retries = 3
|
||||
timeout = 60
|
||||
|
||||
[[adapters]]
|
||||
type = "Mqtt"
|
||||
broker = "mqtt.example.com"
|
||||
port = 1883
|
||||
client_id = "event-notifier"
|
||||
topic = "events"
|
||||
max_retries = 3
|
||||
@@ -0,0 +1,133 @@
|
||||
use rustfs_event_notifier::{
|
||||
AdapterConfig, Bucket, Error as NotifierError, Event, Identity, Metadata, Name, NotifierConfig, Object, Source, WebhookConfig,
|
||||
};
|
||||
use std::collections::HashMap;
|
||||
use tokio::signal;
|
||||
use tracing::Level;
|
||||
use tracing_subscriber::FmtSubscriber;
|
||||
|
||||
async fn setup_notification_system() -> Result<(), NotifierError> {
|
||||
let config = NotifierConfig {
|
||||
store_path: "./deploy/logs/event_store".into(),
|
||||
channel_capacity: 100,
|
||||
adapters: vec![AdapterConfig::Webhook(WebhookConfig {
|
||||
endpoint: "http://127.0.0.1:3020/webhook".into(),
|
||||
auth_token: Some("your-auth-token".into()),
|
||||
custom_headers: Some(HashMap::new()),
|
||||
max_retries: 3,
|
||||
timeout: 30,
|
||||
})],
|
||||
};
|
||||
|
||||
rustfs_event_notifier::initialize(config).await?;
|
||||
|
||||
// wait for the system to be ready
|
||||
for _ in 0..50 {
|
||||
// wait up to 5 seconds
|
||||
if rustfs_event_notifier::is_ready() {
|
||||
return Ok(());
|
||||
}
|
||||
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
|
||||
}
|
||||
|
||||
Err(NotifierError::custom("notify the system of initialization timeout"))
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
// initialization log
|
||||
// tracing_subscriber::fmt::init();
|
||||
|
||||
let subscriber = FmtSubscriber::builder()
|
||||
.with_max_level(Level::DEBUG) // set to debug or lower level
|
||||
.with_target(false) // simplify output
|
||||
.finish();
|
||||
tracing::subscriber::set_global_default(subscriber).expect("failed to set up log subscriber");
|
||||
|
||||
// set up notification system
|
||||
if let Err(e) = setup_notification_system().await {
|
||||
eprintln!("unable to initialize notification system:{}", e);
|
||||
return Err(e.into());
|
||||
}
|
||||
|
||||
// create a shutdown signal processing
|
||||
let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel();
|
||||
|
||||
// start signal processing task
|
||||
tokio::spawn(async move {
|
||||
let _ = signal::ctrl_c().await;
|
||||
println!("Received the shutdown signal and prepared to exit...");
|
||||
let _ = shutdown_tx.send(());
|
||||
});
|
||||
|
||||
// main application logic
|
||||
tokio::select! {
|
||||
_ = async {
|
||||
loop {
|
||||
// application logic
|
||||
// create an s3 metadata object
|
||||
let metadata = Metadata {
|
||||
schema_version: "1.0".to_string(),
|
||||
configuration_id: "test-config".to_string(),
|
||||
bucket: Bucket {
|
||||
name: "my-bucket".to_string(),
|
||||
owner_identity: Identity {
|
||||
principal_id: "owner123".to_string(),
|
||||
},
|
||||
arn: "arn:aws:s3:::my-bucket".to_string(),
|
||||
},
|
||||
object: Object {
|
||||
key: "test.txt".to_string(),
|
||||
size: Some(1024),
|
||||
etag: Some("abc123".to_string()),
|
||||
content_type: Some("text/plain".to_string()),
|
||||
user_metadata: None,
|
||||
version_id: None,
|
||||
sequencer: "1234567890".to_string(),
|
||||
},
|
||||
};
|
||||
|
||||
// create source object
|
||||
let source = Source {
|
||||
host: "localhost".to_string(),
|
||||
port: "80".to_string(),
|
||||
user_agent: "curl/7.68.0".to_string(),
|
||||
};
|
||||
|
||||
// create events using builder mode
|
||||
let event = Event::builder()
|
||||
.event_time("2023-10-01T12:00:00.000Z")
|
||||
.event_name(Name::ObjectCreatedPut)
|
||||
.user_identity(Identity {
|
||||
principal_id: "user123".to_string(),
|
||||
})
|
||||
.s3(metadata)
|
||||
.source(source)
|
||||
.channels(vec!["webhook".to_string()])
|
||||
.build()
|
||||
.expect("failed to create event");
|
||||
|
||||
if let Err(e) = rustfs_event_notifier::send_event(event).await {
|
||||
eprintln!("send event failed:{}", e);
|
||||
}
|
||||
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(60)).await;
|
||||
}
|
||||
} => {},
|
||||
|
||||
_ = &mut shutdown_rx => {
|
||||
println!("close the app");
|
||||
}
|
||||
}
|
||||
|
||||
// 优雅关闭通知系统
|
||||
println!("turn off the notification system");
|
||||
if let Err(e) = rustfs_event_notifier::shutdown().await {
|
||||
eprintln!("An error occurred while shutting down the notification system:{}", e);
|
||||
} else {
|
||||
println!("the notification system has been closed safely");
|
||||
}
|
||||
|
||||
println!("the application has been closed safely");
|
||||
Ok(())
|
||||
}
|
||||
@@ -0,0 +1,108 @@
|
||||
use rustfs_event_notifier::create_adapters;
|
||||
use rustfs_event_notifier::NotifierSystem;
|
||||
use rustfs_event_notifier::{AdapterConfig, NotifierConfig, WebhookConfig};
|
||||
use rustfs_event_notifier::{Bucket, Event, Identity, Metadata, Name, Object, Source};
|
||||
use std::collections::HashMap;
|
||||
use std::error;
|
||||
use std::sync::Arc;
|
||||
use tokio::signal;
|
||||
use tracing::Level;
|
||||
use tracing_subscriber::FmtSubscriber;
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn error::Error>> {
|
||||
let subscriber = FmtSubscriber::builder()
|
||||
.with_max_level(Level::DEBUG) // set to debug or lower level
|
||||
.with_target(false) // simplify output
|
||||
.finish();
|
||||
tracing::subscriber::set_global_default(subscriber).expect("failed to set up log subscriber");
|
||||
|
||||
let config = NotifierConfig {
|
||||
store_path: "./events".to_string(),
|
||||
channel_capacity: 100,
|
||||
adapters: vec![AdapterConfig::Webhook(WebhookConfig {
|
||||
endpoint: "http://127.0.0.1:3020/webhook".to_string(),
|
||||
auth_token: Some("secret-token".to_string()),
|
||||
custom_headers: Some(HashMap::from([("X-Custom".to_string(), "value".to_string())])),
|
||||
max_retries: 3,
|
||||
timeout: 10,
|
||||
})],
|
||||
};
|
||||
|
||||
// event_load_config
|
||||
// loading configuration from environment variables
|
||||
let _config = NotifierConfig::event_load_config(Some("./crates/event-notifier/examples/event.toml".to_string()));
|
||||
tracing::info!("event_load_config config: {:?} \n", _config);
|
||||
|
||||
let system = Arc::new(tokio::sync::Mutex::new(NotifierSystem::new(config.clone()).await?));
|
||||
let adapters = create_adapters(&config.adapters)?;
|
||||
|
||||
// create an s3 metadata object
|
||||
let metadata = Metadata {
|
||||
schema_version: "1.0".to_string(),
|
||||
configuration_id: "test-config".to_string(),
|
||||
bucket: Bucket {
|
||||
name: "my-bucket".to_string(),
|
||||
owner_identity: Identity {
|
||||
principal_id: "owner123".to_string(),
|
||||
},
|
||||
arn: "arn:aws:s3:::my-bucket".to_string(),
|
||||
},
|
||||
object: Object {
|
||||
key: "test.txt".to_string(),
|
||||
size: Some(1024),
|
||||
etag: Some("abc123".to_string()),
|
||||
content_type: Some("text/plain".to_string()),
|
||||
user_metadata: None,
|
||||
version_id: None,
|
||||
sequencer: "1234567890".to_string(),
|
||||
},
|
||||
};
|
||||
|
||||
// create source object
|
||||
let source = Source {
|
||||
host: "localhost".to_string(),
|
||||
port: "80".to_string(),
|
||||
user_agent: "curl/7.68.0".to_string(),
|
||||
};
|
||||
|
||||
// create events using builder mode
|
||||
let event = Event::builder()
|
||||
.event_time("2023-10-01T12:00:00.000Z")
|
||||
.event_name(Name::ObjectCreatedPut)
|
||||
.user_identity(Identity {
|
||||
principal_id: "user123".to_string(),
|
||||
})
|
||||
.s3(metadata)
|
||||
.source(source)
|
||||
.channels(vec!["webhook".to_string()])
|
||||
.build()
|
||||
.expect("failed to create event");
|
||||
|
||||
{
|
||||
let system = system.lock().await;
|
||||
system.send_event(event).await?;
|
||||
}
|
||||
|
||||
let system_clone = Arc::clone(&system);
|
||||
let system_handle = tokio::spawn(async move {
|
||||
let mut system = system_clone.lock().await;
|
||||
system.start(adapters).await
|
||||
});
|
||||
|
||||
signal::ctrl_c().await?;
|
||||
tracing::info!("Received shutdown signal");
|
||||
let result = {
|
||||
let mut system = system.lock().await;
|
||||
system.shutdown().await
|
||||
};
|
||||
|
||||
if let Err(e) = result {
|
||||
tracing::error!("Failed to shut down the notification system: {}", e);
|
||||
} else {
|
||||
tracing::info!("Notification system shut down successfully");
|
||||
}
|
||||
|
||||
system_handle.await??;
|
||||
Ok(())
|
||||
}
|
||||
@@ -0,0 +1,94 @@
|
||||
use axum::{extract::Json, http::StatusCode, routing::post, Router};
|
||||
use serde_json::Value;
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() {
|
||||
// 构建应用
|
||||
let app = Router::new().route("/webhook", post(receive_webhook));
|
||||
// 启动服务器
|
||||
let listener = tokio::net::TcpListener::bind("0.0.0.0:3020").await.unwrap();
|
||||
println!("Server running on http://0.0.0.0:3020");
|
||||
|
||||
// 创建关闭信号处理
|
||||
tokio::select! {
|
||||
result = axum::serve(listener, app) => {
|
||||
if let Err(e) = result {
|
||||
eprintln!("Server error: {}", e);
|
||||
}
|
||||
}
|
||||
_ = tokio::signal::ctrl_c() => {
|
||||
println!("Shutting down server...");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn receive_webhook(Json(payload): Json<Value>) -> StatusCode {
|
||||
let start = SystemTime::now();
|
||||
let since_the_epoch = start.duration_since(UNIX_EPOCH).expect("Time went backwards");
|
||||
|
||||
// get the number of seconds since the unix era
|
||||
let seconds = since_the_epoch.as_secs();
|
||||
|
||||
// Manually calculate year, month, day, hour, minute, and second
|
||||
let (year, month, day, hour, minute, second) = convert_seconds_to_date(seconds);
|
||||
|
||||
// output result
|
||||
println!("current time:{:04}-{:02}-{:02} {:02}:{:02}:{:02}", year, month, day, hour, minute, second);
|
||||
println!(
|
||||
"received a webhook request time:{} content:\n {}",
|
||||
seconds.to_string(),
|
||||
serde_json::to_string_pretty(&payload).unwrap()
|
||||
);
|
||||
StatusCode::OK
|
||||
}
|
||||
|
||||
fn convert_seconds_to_date(seconds: u64) -> (u32, u32, u32, u32, u32, u32) {
|
||||
// assume that the time zone is utc
|
||||
let seconds_per_minute = 60;
|
||||
let seconds_per_hour = 3600;
|
||||
let seconds_per_day = 86400;
|
||||
|
||||
// Calculate the year, month, day, hour, minute, and second corresponding to the number of seconds
|
||||
let mut total_seconds = seconds;
|
||||
let mut year = 1970;
|
||||
let mut month = 1;
|
||||
let mut day = 1;
|
||||
let mut hour = 0;
|
||||
let mut minute = 0;
|
||||
let mut second = 0;
|
||||
|
||||
// calculate year
|
||||
while total_seconds >= 31536000 {
|
||||
year += 1;
|
||||
total_seconds -= 31536000; // simplified processing no leap year considered
|
||||
}
|
||||
|
||||
// calculate month
|
||||
let days_in_month = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31];
|
||||
for m in 0..12 {
|
||||
if total_seconds >= days_in_month[m] * seconds_per_day {
|
||||
month += 1;
|
||||
total_seconds -= days_in_month[m] * seconds_per_day;
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
// calculate the number of days
|
||||
day += total_seconds / seconds_per_day;
|
||||
total_seconds %= seconds_per_day;
|
||||
|
||||
// calculate hours
|
||||
hour += total_seconds / seconds_per_hour;
|
||||
total_seconds %= seconds_per_hour;
|
||||
|
||||
// calculate minutes
|
||||
minute += total_seconds / seconds_per_minute;
|
||||
total_seconds %= seconds_per_minute;
|
||||
|
||||
// calculate the number of seconds
|
||||
second += total_seconds;
|
||||
|
||||
(year as u32, month as u32, day as u32, hour as u32, minute as u32, second as u32)
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
use crate::ChannelAdapter;
|
||||
use crate::Error;
|
||||
use crate::Event;
|
||||
use crate::KafkaConfig;
|
||||
use async_trait::async_trait;
|
||||
use rdkafka::error::KafkaError;
|
||||
use rdkafka::producer::{FutureProducer, FutureRecord};
|
||||
use rdkafka::types::RDKafkaErrorCode;
|
||||
use rdkafka::util::Timeout;
|
||||
use std::time::Duration;
|
||||
use tokio::time::sleep;
|
||||
|
||||
/// Kafka adapter for sending events to a Kafka topic.
|
||||
pub struct KafkaAdapter {
|
||||
producer: FutureProducer,
|
||||
topic: String,
|
||||
max_retries: u32,
|
||||
}
|
||||
|
||||
impl KafkaAdapter {
|
||||
/// Creates a new Kafka adapter.
|
||||
pub fn new(config: &KafkaConfig) -> Result<Self, Error> {
|
||||
// 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()?;
|
||||
|
||||
Ok(Self {
|
||||
producer,
|
||||
topic: config.topic.clone(),
|
||||
max_retries: config.max_retries,
|
||||
})
|
||||
}
|
||||
/// Sends an event to the Kafka topic with retry logic.
|
||||
async fn send_with_retry(&self, event: &Event) -> Result<(), Error> {
|
||||
let event_id = event.id.to_string();
|
||||
let payload = serde_json::to_string(&event)?;
|
||||
|
||||
for attempt in 0..self.max_retries {
|
||||
let record = FutureRecord::to(&self.topic).key(&event_id).payload(&payload);
|
||||
|
||||
match self.producer.send(record, Timeout::Never).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;
|
||||
}
|
||||
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()))
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl ChannelAdapter for KafkaAdapter {
|
||||
fn name(&self) -> String {
|
||||
"kafka".to_string()
|
||||
}
|
||||
|
||||
async fn send(&self, event: &Event) -> Result<(), Error> {
|
||||
self.send_with_retry(event).await
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
use crate::AdapterConfig;
|
||||
use crate::Error;
|
||||
use crate::Event;
|
||||
use async_trait::async_trait;
|
||||
use std::sync::Arc;
|
||||
|
||||
#[cfg(feature = "kafka")]
|
||||
pub(crate) mod kafka;
|
||||
#[cfg(feature = "mqtt")]
|
||||
pub(crate) mod mqtt;
|
||||
#[cfg(feature = "webhook")]
|
||||
pub(crate) mod webhook;
|
||||
|
||||
/// The `ChannelAdapter` trait defines the interface for all channel adapters.
|
||||
#[async_trait]
|
||||
pub trait ChannelAdapter: Send + Sync + 'static {
|
||||
/// Sends an event to the channel.
|
||||
fn name(&self) -> String;
|
||||
/// Sends an event to the channel.
|
||||
async fn send(&self, event: &Event) -> Result<(), Error>;
|
||||
}
|
||||
|
||||
/// Creates channel adapters based on the provided configuration.
|
||||
pub fn create_adapters(configs: &[AdapterConfig]) -> Result<Vec<Arc<dyn ChannelAdapter>>, Error> {
|
||||
let mut adapters: Vec<Arc<dyn ChannelAdapter>> = Vec::new();
|
||||
|
||||
for config in configs {
|
||||
match config {
|
||||
#[cfg(feature = "webhook")]
|
||||
AdapterConfig::Webhook(webhook_config) => {
|
||||
webhook_config.validate().map_err(Error::ConfigError)?;
|
||||
adapters.push(Arc::new(webhook::WebhookAdapter::new(webhook_config.clone())));
|
||||
}
|
||||
#[cfg(feature = "kafka")]
|
||||
AdapterConfig::Kafka(kafka_config) => {
|
||||
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);
|
||||
tokio::spawn(async move { while event_loop.poll().await.is_ok() {} });
|
||||
adapters.push(Arc::new(mqtt));
|
||||
}
|
||||
#[cfg(not(feature = "webhook"))]
|
||||
AdapterConfig::Webhook(_) => return Err(Error::FeatureDisabled("webhook")),
|
||||
#[cfg(not(feature = "kafka"))]
|
||||
AdapterConfig::Kafka(_) => return Err(Error::FeatureDisabled("kafka")),
|
||||
#[cfg(not(feature = "mqtt"))]
|
||||
AdapterConfig::Mqtt(_) => return Err(Error::FeatureDisabled("mqtt")),
|
||||
}
|
||||
}
|
||||
|
||||
Ok(adapters)
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
use crate::ChannelAdapter;
|
||||
use crate::Error;
|
||||
use crate::Event;
|
||||
use crate::MqttConfig;
|
||||
use async_trait::async_trait;
|
||||
use rumqttc::{AsyncClient, MqttOptions, QoS};
|
||||
use std::time::Duration;
|
||||
use tokio::time::sleep;
|
||||
|
||||
/// MQTT adapter for sending events to an MQTT broker.
|
||||
pub struct MqttAdapter {
|
||||
client: AsyncClient,
|
||||
topic: String,
|
||||
max_retries: u32,
|
||||
}
|
||||
|
||||
impl MqttAdapter {
|
||||
/// Creates a new MQTT adapter.
|
||||
pub fn new(config: &MqttConfig) -> (Self, rumqttc::EventLoop) {
|
||||
let mqtt_options = MqttOptions::new(&config.client_id, &config.broker, config.port);
|
||||
let (client, event_loop) = rumqttc::AsyncClient::new(mqtt_options, 10);
|
||||
(
|
||||
Self {
|
||||
client,
|
||||
topic: config.topic.clone(),
|
||||
max_retries: config.max_retries,
|
||||
},
|
||||
event_loop,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl ChannelAdapter for MqttAdapter {
|
||||
fn name(&self) -> String {
|
||||
"mqtt".to_string()
|
||||
}
|
||||
|
||||
async fn send(&self, event: &Event) -> Result<(), Error> {
|
||||
let payload = serde_json::to_string(event).map_err(Error::Serde)?;
|
||||
let mut attempt = 0;
|
||||
loop {
|
||||
match self
|
||||
.client
|
||||
.publish(&self.topic, QoS::AtLeastOnce, false, payload.clone())
|
||||
.await
|
||||
{
|
||||
Ok(()) => return Ok(()),
|
||||
Err(e) if attempt < self.max_retries => {
|
||||
attempt += 1;
|
||||
tracing::warn!("MQTT attempt {} failed: {}. Retrying...", attempt, e);
|
||||
sleep(Duration::from_secs(2u64.pow(attempt))).await;
|
||||
}
|
||||
Err(e) => return Err(Error::Mqtt(e)),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
use crate::ChannelAdapter;
|
||||
use crate::Error;
|
||||
use crate::Event;
|
||||
use crate::WebhookConfig;
|
||||
use async_trait::async_trait;
|
||||
use reqwest::{Client, RequestBuilder};
|
||||
use std::time::Duration;
|
||||
use tokio::time::sleep;
|
||||
|
||||
/// Webhook adapter for sending events to a webhook endpoint.
|
||||
pub struct WebhookAdapter {
|
||||
config: WebhookConfig,
|
||||
client: Client,
|
||||
}
|
||||
|
||||
impl WebhookAdapter {
|
||||
/// Creates a new Webhook adapter.
|
||||
pub fn new(config: WebhookConfig) -> Self {
|
||||
let client = Client::builder()
|
||||
.timeout(Duration::from_secs(config.timeout))
|
||||
.build()
|
||||
.expect("Failed to build reqwest client");
|
||||
Self { config, client }
|
||||
}
|
||||
/// Builds the request to send the event.
|
||||
fn build_request(&self, event: &Event) -> RequestBuilder {
|
||||
let mut request = self.client.post(&self.config.endpoint).json(event);
|
||||
if let Some(token) = &self.config.auth_token {
|
||||
request = request.header("Authorization", format!("Bearer {}", token));
|
||||
}
|
||||
if let Some(headers) = &self.config.custom_headers {
|
||||
for (key, value) in headers {
|
||||
request = request.header(key, value);
|
||||
}
|
||||
}
|
||||
request
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl ChannelAdapter for WebhookAdapter {
|
||||
fn name(&self) -> String {
|
||||
"webhook".to_string()
|
||||
}
|
||||
|
||||
async fn send(&self, event: &Event) -> Result<(), Error> {
|
||||
let mut attempt = 0;
|
||||
tracing::info!("Attempting to send webhook request: {:?}", event);
|
||||
loop {
|
||||
match self.build_request(event).send().await {
|
||||
Ok(response) => {
|
||||
response.error_for_status().map_err(Error::Http)?;
|
||||
return Ok(());
|
||||
}
|
||||
Err(e) if attempt < self.config.max_retries => {
|
||||
attempt += 1;
|
||||
tracing::warn!("Webhook attempt {} failed: {}. Retrying...", attempt, e);
|
||||
sleep(Duration::from_secs(2u64.pow(attempt))).await;
|
||||
}
|
||||
Err(e) => return Err(Error::Http(e)),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,100 @@
|
||||
use crate::ChannelAdapter;
|
||||
use crate::Error;
|
||||
use crate::EventStore;
|
||||
use crate::{Event, Log};
|
||||
use std::sync::Arc;
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
use tokio::sync::mpsc;
|
||||
use tokio::time::Duration;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
/// Handles incoming events from the producer.
|
||||
///
|
||||
/// This function is responsible for receiving events from the producer and sending them to the appropriate adapters.
|
||||
/// It also handles the shutdown process and saves any pending logs to the event store.
|
||||
pub async fn event_bus(
|
||||
mut rx: mpsc::Receiver<Event>,
|
||||
adapters: Vec<Arc<dyn ChannelAdapter>>,
|
||||
store: Arc<EventStore>,
|
||||
shutdown: CancellationToken,
|
||||
shutdown_complete: Option<tokio::sync::oneshot::Sender<()>>,
|
||||
) -> Result<(), Error> {
|
||||
let mut current_log = Log {
|
||||
event_name: crate::event::Name::Everything,
|
||||
key: SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs().to_string(),
|
||||
records: Vec::new(),
|
||||
};
|
||||
|
||||
let mut unprocessed_events = Vec::new();
|
||||
loop {
|
||||
tokio::select! {
|
||||
Some(event) = rx.recv() => {
|
||||
current_log.records.push(event.clone());
|
||||
let mut send_tasks = Vec::new();
|
||||
for adapter in &adapters {
|
||||
if event.channels.contains(&adapter.name()) {
|
||||
let adapter = adapter.clone();
|
||||
let event = event.clone();
|
||||
send_tasks.push(tokio::spawn(async move {
|
||||
if let Err(e) = adapter.send(&event).await {
|
||||
tracing::error!("Failed to send event to {}: {}", adapter.name(), e);
|
||||
Err(e)
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
}));
|
||||
}
|
||||
}
|
||||
for task in send_tasks {
|
||||
if task.await?.is_err() {
|
||||
// If sending fails, add the event to the unprocessed list
|
||||
let failed_event = event.clone();
|
||||
unprocessed_events.push(failed_event);
|
||||
}
|
||||
}
|
||||
|
||||
// Clear the current log because we only care about unprocessed events
|
||||
current_log.records.clear();
|
||||
}
|
||||
_ = shutdown.cancelled() => {
|
||||
tracing::info!("Shutting down event bus, saving pending logs...");
|
||||
// Check if there are still unprocessed messages in the channel
|
||||
while let Ok(Some(event)) = tokio::time::timeout(
|
||||
Duration::from_millis(100),
|
||||
rx.recv()
|
||||
).await {
|
||||
unprocessed_events.push(event);
|
||||
}
|
||||
|
||||
// save only if there are unprocessed events
|
||||
if !unprocessed_events.is_empty() {
|
||||
tracing::info!("Save {} unhandled events", unprocessed_events.len());
|
||||
// create and save logging
|
||||
let shutdown_log = Log {
|
||||
event_name: crate::event::Name::Everything,
|
||||
key: format!("shutdown_{}", SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs()),
|
||||
records: unprocessed_events,
|
||||
};
|
||||
|
||||
store.save_logs(&[shutdown_log]).await?;
|
||||
} else {
|
||||
tracing::info!("no unhandled events need to be saved");
|
||||
}
|
||||
tracing::debug!("shutdown_complete is Some: {}", shutdown_complete.is_some());
|
||||
|
||||
if let Some(complete_sender) = shutdown_complete {
|
||||
// send a completion signal
|
||||
let result = complete_sender.send(());
|
||||
match result {
|
||||
Ok(_) => tracing::info!("Event bus shutdown signal sent"),
|
||||
Err(e) => tracing::error!("Failed to send event bus shutdown signal: {:?}", e),
|
||||
}
|
||||
tracing::info!("Shutting down event bus");
|
||||
}
|
||||
tracing::info!("Event bus shutdown complete");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
@@ -0,0 +1,176 @@
|
||||
use config::{Config, Environment, File, FileFormat};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::HashMap;
|
||||
use std::env;
|
||||
|
||||
/// Configuration for the notification system.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct WebhookConfig {
|
||||
pub endpoint: String,
|
||||
pub auth_token: Option<String>,
|
||||
pub custom_headers: Option<HashMap<String, String>>,
|
||||
pub max_retries: u32,
|
||||
pub timeout: u64,
|
||||
}
|
||||
|
||||
impl WebhookConfig {
|
||||
/// verify that the configuration is valid
|
||||
pub fn validate(&self) -> Result<(), String> {
|
||||
// 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 == 0 {
|
||||
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());
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// Configuration for the Kafka adapter.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct KafkaConfig {
|
||||
pub brokers: String,
|
||||
pub topic: String,
|
||||
pub max_retries: u32,
|
||||
pub timeout: u64,
|
||||
}
|
||||
|
||||
/// Configuration for the MQTT adapter.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct MqttConfig {
|
||||
pub broker: String,
|
||||
pub port: u16,
|
||||
pub client_id: String,
|
||||
pub topic: String,
|
||||
pub max_retries: u32,
|
||||
}
|
||||
|
||||
/// Configuration for the notification system.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
#[serde(tag = "type")]
|
||||
pub enum AdapterConfig {
|
||||
Webhook(WebhookConfig),
|
||||
Kafka(KafkaConfig),
|
||||
Mqtt(MqttConfig),
|
||||
}
|
||||
|
||||
/// Configuration for the notification system.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct NotifierConfig {
|
||||
#[serde(default = "default_store_path")]
|
||||
pub store_path: String,
|
||||
#[serde(default = "default_channel_capacity")]
|
||||
pub channel_capacity: usize,
|
||||
pub adapters: Vec<AdapterConfig>,
|
||||
}
|
||||
|
||||
impl Default for NotifierConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
store_path: default_store_path(),
|
||||
channel_capacity: default_channel_capacity(),
|
||||
adapters: Vec::new(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl NotifierConfig {
|
||||
/// create a new configuration with default values
|
||||
pub fn new() -> Self {
|
||||
Self::default()
|
||||
}
|
||||
|
||||
/// Loading the configuration file
|
||||
/// Supports TOML, YAML and .env formats, read in order by priority
|
||||
///
|
||||
/// # Parameters
|
||||
/// - `config_dir`: Configuration file path
|
||||
///
|
||||
/// # Returns
|
||||
/// Configuration information
|
||||
///
|
||||
/// # Example
|
||||
/// ```
|
||||
/// use rustfs_event_notifier::NotifierConfig;
|
||||
///
|
||||
/// let config = NotifierConfig::event_load_config(None);
|
||||
/// ```
|
||||
pub fn event_load_config(config_dir: Option<String>) -> NotifierConfig {
|
||||
let config_dir = if let Some(path) = config_dir {
|
||||
// If a path is provided, check if it's empty
|
||||
if path.is_empty() {
|
||||
// If empty, use the default config file name
|
||||
DEFAULT_CONFIG_FILE.to_string()
|
||||
} else {
|
||||
// Use the provided path
|
||||
let path = std::path::Path::new(&path);
|
||||
if path.extension().is_some() {
|
||||
// If path has extension, use it as is (extension will be added by Config::builder)
|
||||
path.with_extension("").to_string_lossy().into_owned()
|
||||
} else {
|
||||
// If path is a directory, append the default config file name
|
||||
path.to_string_lossy().into_owned()
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// If no path provided, use current directory + default config file
|
||||
match env::current_dir() {
|
||||
Ok(dir) => dir.join(DEFAULT_CONFIG_FILE).to_string_lossy().into_owned(),
|
||||
Err(_) => {
|
||||
eprintln!("Warning: Failed to get current directory, using default config file");
|
||||
DEFAULT_CONFIG_FILE.to_string()
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// Log using proper logging instead of println when possible
|
||||
println!("Using config file base: {}", config_dir);
|
||||
|
||||
let app_config = Config::builder()
|
||||
.add_source(File::with_name(config_dir.as_str()).format(FileFormat::Toml).required(false))
|
||||
.add_source(File::with_name(config_dir.as_str()).format(FileFormat::Yaml).required(false))
|
||||
.add_source(
|
||||
Environment::default()
|
||||
.prefix("NOTIFIER")
|
||||
.prefix_separator("__")
|
||||
.separator("__")
|
||||
.list_separator("_")
|
||||
.with_list_parse_key("adapters")
|
||||
.try_parsing(true),
|
||||
)
|
||||
.build()
|
||||
.unwrap_or_default();
|
||||
println!("Loaded config: {:?}", app_config);
|
||||
match app_config.try_deserialize::<NotifierConfig>() {
|
||||
Ok(app_config) => {
|
||||
println!("Parsed AppConfig: {:?} \n", app_config);
|
||||
app_config
|
||||
}
|
||||
Err(e) => {
|
||||
println!("Failed to deserialize config: {}", e);
|
||||
NotifierConfig::default()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const DEFAULT_CONFIG_FILE: &str = "obs";
|
||||
|
||||
/// Provide temporary directories as default storage paths
|
||||
fn default_store_path() -> String {
|
||||
std::env::temp_dir().join("event-notification").to_string_lossy().to_string()
|
||||
}
|
||||
|
||||
/// Provides the recommended default channel capacity for high concurrency systems
|
||||
fn default_channel_capacity() -> usize {
|
||||
10000 // Reasonable default values for high concurrency systems
|
||||
}
|
||||
@@ -0,0 +1,46 @@
|
||||
use config::ConfigError;
|
||||
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(feature = "kafka")]
|
||||
#[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::SendError<crate::event::Event>>),
|
||||
#[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("Configuration loading error: {0}")]
|
||||
Config(#[from] ConfigError),
|
||||
}
|
||||
|
||||
impl Error {
|
||||
pub fn custom(msg: &str) -> Error {
|
||||
Self::Custom(msg.to_string())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,447 @@
|
||||
use crate::Error;
|
||||
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,
|
||||
}
|
||||
|
||||
/// 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,
|
||||
}
|
||||
|
||||
/// 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<i64>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none", rename = "eTag")]
|
||||
pub etag: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none", rename = "contentType")]
|
||||
pub content_type: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none", rename = "userMetadata")]
|
||||
pub user_metadata: Option<HashMap<String, String>>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none", rename = "versionId")]
|
||||
pub version_id: Option<String>,
|
||||
pub sequencer: String,
|
||||
}
|
||||
|
||||
/// 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,
|
||||
}
|
||||
|
||||
/// 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,
|
||||
}
|
||||
|
||||
/// 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<String>,
|
||||
event_source: Option<String>,
|
||||
aws_region: Option<String>,
|
||||
event_time: Option<String>,
|
||||
event_name: Option<Name>,
|
||||
user_identity: Option<Identity>,
|
||||
request_parameters: Option<HashMap<String, String>>,
|
||||
response_elements: Option<HashMap<String, String>>,
|
||||
s3: Option<Metadata>,
|
||||
source: Option<Source>,
|
||||
channels: Option<SmallVec<[String; 2]>>,
|
||||
}
|
||||
|
||||
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<String>) -> 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<String>) -> 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<String>) -> Self {
|
||||
self.aws_region = Some(aws_region.into());
|
||||
self
|
||||
}
|
||||
|
||||
/// set event time
|
||||
pub fn event_time(mut self, event_time: impl Into<String>) -> 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<String, String>) -> Self {
|
||||
self.request_parameters = Some(request_parameters);
|
||||
self
|
||||
}
|
||||
|
||||
/// set response elements
|
||||
pub fn response_elements(mut self, response_elements: HashMap<String, String>) -> 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<String>) -> 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<Event, Error> {
|
||||
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<String, String>,
|
||||
#[serde(rename = "responseElements")]
|
||||
pub response_elements: HashMap<String, String>,
|
||||
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<String>) -> 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<Event>,
|
||||
}
|
||||
|
||||
#[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<Name> {
|
||||
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<Self> {
|
||||
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,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,230 @@
|
||||
use crate::{create_adapters, Error, Event, NotifierConfig, NotifierSystem};
|
||||
use std::sync::{atomic, Arc};
|
||||
use tokio::sync::{Mutex, OnceCell};
|
||||
|
||||
static GLOBAL_SYSTEM: OnceCell<Arc<Mutex<NotifierSystem>>> = OnceCell::const_new();
|
||||
static INITIALIZED: atomic::AtomicBool = atomic::AtomicBool::new(false);
|
||||
static READY: atomic::AtomicBool = atomic::AtomicBool::new(false);
|
||||
static INIT_LOCK: Mutex<()> = Mutex::const_new(());
|
||||
|
||||
/// Initializes the global notification system.
|
||||
///
|
||||
/// This function performs the following steps:
|
||||
/// 1. Checks if the system is already initialized.
|
||||
/// 2. Creates a new `NotificationSystem` instance.
|
||||
/// 3. Creates adapters based on the provided configuration.
|
||||
/// 4. Starts the notification system with the created adapters.
|
||||
/// 5. Sets the global system instance.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Returns an error if:
|
||||
/// - The system is already initialized.
|
||||
/// - Creating the `NotificationSystem` fails.
|
||||
/// - Creating adapters fails.
|
||||
/// - Starting the notification system fails.
|
||||
/// - Setting the global system instance fails.
|
||||
pub async fn initialize(config: NotifierConfig) -> Result<(), Error> {
|
||||
let _lock = INIT_LOCK.lock().await;
|
||||
|
||||
// Check if the system is already initialized.
|
||||
if INITIALIZED.load(atomic::Ordering::SeqCst) {
|
||||
return Err(Error::custom("Notification system has already been initialized"));
|
||||
}
|
||||
|
||||
// Check if the system is already ready.
|
||||
if READY.load(atomic::Ordering::SeqCst) {
|
||||
return Err(Error::custom("Notification system is already ready"));
|
||||
}
|
||||
|
||||
// Check if the system is shutting down.
|
||||
if let Some(system) = GLOBAL_SYSTEM.get() {
|
||||
let system_guard = system.lock().await;
|
||||
if system_guard.shutdown_cancelled() {
|
||||
return Err(Error::custom("Notification system is shutting down"));
|
||||
}
|
||||
}
|
||||
|
||||
// check if config adapters len is than 0
|
||||
if config.adapters.is_empty() {
|
||||
return Err(Error::custom("No adapters configured"));
|
||||
}
|
||||
|
||||
// Attempt to initialize, and reset the INITIALIZED flag if it fails.
|
||||
let result: Result<(), Error> = async {
|
||||
let system = NotifierSystem::new(config.clone()).await.map_err(|e| {
|
||||
tracing::error!("Failed to create NotificationSystem: {:?}", e);
|
||||
e
|
||||
})?;
|
||||
let adapters = create_adapters(&config.adapters).map_err(|e| {
|
||||
tracing::error!("Failed to create adapters: {:?}", e);
|
||||
e
|
||||
})?;
|
||||
tracing::info!("adapters len:{:?}", adapters.len());
|
||||
let system_clone = Arc::new(Mutex::new(system));
|
||||
let adapters_clone = adapters.clone();
|
||||
|
||||
GLOBAL_SYSTEM.set(system_clone.clone()).map_err(|_| {
|
||||
let err = Error::custom("Unable to set up global notification system");
|
||||
tracing::error!("{:?}", err);
|
||||
err
|
||||
})?;
|
||||
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = system_clone.lock().await.start(adapters_clone).await {
|
||||
tracing::error!("Notification system failed to start: {}", e);
|
||||
}
|
||||
tracing::info!("Notification system started in background");
|
||||
});
|
||||
tracing::info!("system start success,start set READY value");
|
||||
|
||||
READY.store(true, atomic::Ordering::SeqCst);
|
||||
tracing::info!("Notification system is ready to process events");
|
||||
|
||||
Ok(())
|
||||
}
|
||||
.await;
|
||||
|
||||
if result.is_err() {
|
||||
INITIALIZED.store(false, atomic::Ordering::SeqCst);
|
||||
READY.store(false, atomic::Ordering::SeqCst);
|
||||
return result;
|
||||
}
|
||||
|
||||
INITIALIZED.store(true, atomic::Ordering::SeqCst);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Checks if the notification system is initialized.
|
||||
pub fn is_initialized() -> bool {
|
||||
INITIALIZED.load(atomic::Ordering::SeqCst)
|
||||
}
|
||||
|
||||
/// Checks if the notification system is ready.
|
||||
pub fn is_ready() -> bool {
|
||||
READY.load(atomic::Ordering::SeqCst)
|
||||
}
|
||||
|
||||
/// Sends an event to the notification system.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Returns an error if:
|
||||
/// - The system is not initialized.
|
||||
/// - The system is not ready.
|
||||
/// - Sending the event fails.
|
||||
pub async fn send_event(event: Event) -> Result<(), Error> {
|
||||
if !READY.load(atomic::Ordering::SeqCst) {
|
||||
return Err(Error::custom("Notification system not ready, please wait for initialization to complete"));
|
||||
}
|
||||
|
||||
let system = get_system().await?;
|
||||
let system_guard = system.lock().await;
|
||||
system_guard.send_event(event).await
|
||||
}
|
||||
|
||||
/// Shuts down the notification system.
|
||||
pub async fn shutdown() -> Result<(), Error> {
|
||||
if let Some(system) = GLOBAL_SYSTEM.get() {
|
||||
tracing::info!("Shutting down notification system start");
|
||||
let result = {
|
||||
let mut system_guard = system.lock().await;
|
||||
system_guard.shutdown().await
|
||||
};
|
||||
if let Err(e) = &result {
|
||||
tracing::error!("Notification system shutdown failed: {}", e);
|
||||
} else {
|
||||
tracing::info!("Event bus shutdown completed");
|
||||
}
|
||||
|
||||
tracing::info!(
|
||||
"Shutdown method called set static value start, READY: {}, INITIALIZED: {}",
|
||||
READY.load(atomic::Ordering::SeqCst),
|
||||
INITIALIZED.load(atomic::Ordering::SeqCst)
|
||||
);
|
||||
READY.store(false, atomic::Ordering::SeqCst);
|
||||
INITIALIZED.store(false, atomic::Ordering::SeqCst);
|
||||
tracing::info!(
|
||||
"Shutdown method called set static value end, READY: {}, INITIALIZED: {}",
|
||||
READY.load(atomic::Ordering::SeqCst),
|
||||
INITIALIZED.load(atomic::Ordering::SeqCst)
|
||||
);
|
||||
result
|
||||
} else {
|
||||
Err(Error::custom("Notification system not initialized"))
|
||||
}
|
||||
}
|
||||
|
||||
/// Retrieves the global notification system instance.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Returns an error if the system is not initialized.
|
||||
async fn get_system() -> Result<Arc<Mutex<NotifierSystem>>, Error> {
|
||||
GLOBAL_SYSTEM
|
||||
.get()
|
||||
.cloned()
|
||||
.ok_or_else(|| Error::custom("Notification system not initialized"))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::{AdapterConfig, NotifierConfig, WebhookConfig};
|
||||
use std::collections::HashMap;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_initialize_success() {
|
||||
tracing_subscriber::fmt::init();
|
||||
let config = NotifierConfig::default(); // assume there is a default configuration
|
||||
let result = initialize(config).await;
|
||||
assert!(result.is_err(), "Initialization should not succeed");
|
||||
assert!(!is_initialized(), "System should not be marked as initialized");
|
||||
assert!(!is_ready(), "System should not be marked as ready");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_initialize_twice() {
|
||||
tracing_subscriber::fmt::init();
|
||||
let config = NotifierConfig::default();
|
||||
let _ = initialize(config.clone()).await; // first initialization
|
||||
let result = initialize(config).await; // second initialization
|
||||
assert!(!result.is_ok(), "Initialization should succeed");
|
||||
assert!(result.is_err(), "Re-initialization should fail");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_initialize_failure_resets_state() {
|
||||
tracing_subscriber::fmt::init();
|
||||
// simulate wrong configuration
|
||||
let config = NotifierConfig {
|
||||
adapters: vec![
|
||||
// assuming that the empty adapter will cause failure
|
||||
AdapterConfig::Webhook(WebhookConfig {
|
||||
endpoint: "http://localhost:8080/webhook".to_string(),
|
||||
auth_token: Some("secret-token".to_string()),
|
||||
custom_headers: Some(HashMap::from([("X-Custom".to_string(), "value".to_string())])),
|
||||
max_retries: 3,
|
||||
timeout: 10,
|
||||
}),
|
||||
], // assuming that the empty adapter will cause failure
|
||||
..Default::default()
|
||||
};
|
||||
let result = initialize(config).await;
|
||||
assert!(!result.is_err(), "Initialization with invalid config should fail");
|
||||
assert!(is_initialized(), "System should not be marked as initialized after failure");
|
||||
assert!(is_ready(), "System should not be marked as ready after failure");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_is_initialized_and_is_ready() {
|
||||
tracing_subscriber::fmt::init();
|
||||
assert!(!is_initialized(), "System should not be initialized initially");
|
||||
assert!(!is_ready(), "System should not be ready initially");
|
||||
|
||||
let config = NotifierConfig::default();
|
||||
let _ = initialize(config).await;
|
||||
assert!(!is_initialized(), "System should be initialized after successful initialization");
|
||||
assert!(!is_ready(), "System should be ready after successful initialization");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
mod adapter;
|
||||
mod bus;
|
||||
mod config;
|
||||
mod error;
|
||||
mod event;
|
||||
mod global;
|
||||
mod notifier;
|
||||
mod store;
|
||||
|
||||
pub use adapter::create_adapters;
|
||||
#[cfg(feature = "kafka")]
|
||||
pub use adapter::kafka::KafkaAdapter;
|
||||
#[cfg(feature = "mqtt")]
|
||||
pub use adapter::mqtt::MqttAdapter;
|
||||
#[cfg(feature = "webhook")]
|
||||
pub use adapter::webhook::WebhookAdapter;
|
||||
pub use adapter::ChannelAdapter;
|
||||
pub use bus::event_bus;
|
||||
#[cfg(feature = "kafka")]
|
||||
pub use config::KafkaConfig;
|
||||
#[cfg(feature = "mqtt")]
|
||||
pub use config::MqttConfig;
|
||||
#[cfg(feature = "webhook")]
|
||||
pub use config::WebhookConfig;
|
||||
pub use config::{AdapterConfig, NotifierConfig};
|
||||
pub use error::Error;
|
||||
|
||||
pub use event::{Bucket, Event, EventBuilder, Identity, Log, Metadata, Name, Object, Source};
|
||||
pub use global::{initialize, is_initialized, is_ready, send_event, shutdown};
|
||||
pub use notifier::NotifierSystem;
|
||||
pub use store::EventStore;
|
||||
@@ -0,0 +1,128 @@
|
||||
use crate::{event_bus, ChannelAdapter, Error, Event, EventStore, NotifierConfig};
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
/// The `NotificationSystem` struct represents the notification system.
|
||||
/// It manages the event bus and the adapters.
|
||||
/// It is responsible for sending and receiving events.
|
||||
/// It also handles the shutdown process.
|
||||
pub struct NotifierSystem {
|
||||
tx: mpsc::Sender<Event>,
|
||||
rx: Option<mpsc::Receiver<Event>>,
|
||||
store: Arc<EventStore>,
|
||||
shutdown: CancellationToken,
|
||||
shutdown_complete: Option<tokio::sync::oneshot::Sender<()>>,
|
||||
shutdown_receiver: Option<tokio::sync::oneshot::Receiver<()>>,
|
||||
}
|
||||
|
||||
impl NotifierSystem {
|
||||
/// Creates a new `NotificationSystem` instance.
|
||||
pub async fn new(config: NotifierConfig) -> Result<Self, Error> {
|
||||
let (tx, rx) = mpsc::channel::<Event>(config.channel_capacity);
|
||||
let store = Arc::new(EventStore::new(&config.store_path).await?);
|
||||
let shutdown = CancellationToken::new();
|
||||
|
||||
let restored_logs = store.load_logs().await?;
|
||||
for log in restored_logs {
|
||||
for event in log.records {
|
||||
// For example, where the send method may return a SendError when calling it
|
||||
tx.send(event).await.map_err(|e| Error::ChannelSend(Box::new(e)))?;
|
||||
}
|
||||
}
|
||||
// Initialize shutdown_complete to Some(tx)
|
||||
let (complete_tx, complete_rx) = tokio::sync::oneshot::channel();
|
||||
Ok(Self {
|
||||
tx,
|
||||
rx: Some(rx),
|
||||
store,
|
||||
shutdown,
|
||||
shutdown_complete: Some(complete_tx),
|
||||
shutdown_receiver: Some(complete_rx),
|
||||
})
|
||||
}
|
||||
|
||||
/// Starts the notification system.
|
||||
/// It initializes the event bus and the producer.
|
||||
pub async fn start(&mut self, adapters: Vec<Arc<dyn ChannelAdapter>>) -> Result<(), Error> {
|
||||
if self.shutdown.is_cancelled() {
|
||||
let error = Error::custom("System is shutting down");
|
||||
self.handle_error("start", &error);
|
||||
return Err(error);
|
||||
}
|
||||
self.log(tracing::Level::INFO, "start", "Starting the notification system");
|
||||
let rx = self.rx.take().ok_or_else(|| Error::EventBusStarted)?;
|
||||
let shutdown_clone = self.shutdown.clone();
|
||||
let store_clone = self.store.clone();
|
||||
let shutdown_complete = self.shutdown_complete.take();
|
||||
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = event_bus(rx, adapters, store_clone, shutdown_clone, shutdown_complete).await {
|
||||
tracing::error!("Event bus failed: {}", e);
|
||||
}
|
||||
});
|
||||
self.log(tracing::Level::INFO, "start", "Notification system started successfully");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Sends an event to the notification system.
|
||||
/// This method is used to send events to the event bus.
|
||||
pub async fn send_event(&self, event: Event) -> Result<(), Error> {
|
||||
self.log(tracing::Level::DEBUG, "send_event", &format!("Sending event: {:?}", event));
|
||||
if self.shutdown.is_cancelled() {
|
||||
let error = Error::custom("System is shutting down");
|
||||
self.handle_error("send_event", &error);
|
||||
return Err(error);
|
||||
}
|
||||
if let Err(e) = self.tx.send(event).await {
|
||||
let error = Error::ChannelSend(Box::new(e));
|
||||
self.handle_error("send_event", &error);
|
||||
return Err(error);
|
||||
}
|
||||
self.log(tracing::Level::INFO, "send_event", "Event sent successfully");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Shuts down the notification system.
|
||||
/// This method is used to cancel the event bus and producer tasks.
|
||||
pub async fn shutdown(&mut self) -> Result<(), Error> {
|
||||
tracing::info!("Shutting down the notification system");
|
||||
self.shutdown.cancel();
|
||||
// wait for the event bus to be completely closed
|
||||
if let Some(receiver) = self.shutdown_receiver.take() {
|
||||
match receiver.await {
|
||||
Ok(_) => {
|
||||
tracing::info!("Event bus shutdown completed successfully");
|
||||
Ok(())
|
||||
}
|
||||
Err(e) => {
|
||||
let error = Error::custom(format!("Failed to receive shutdown completion: {}", e).as_str());
|
||||
self.handle_error("shutdown", &error);
|
||||
Err(error)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
tracing::warn!("Shutdown receiver not available, the event bus might still be running");
|
||||
Err(Error::custom("Shutdown receiver not available"))
|
||||
}
|
||||
}
|
||||
|
||||
/// shutdown state
|
||||
pub fn shutdown_cancelled(&self) -> bool {
|
||||
self.shutdown.is_cancelled()
|
||||
}
|
||||
|
||||
fn handle_error(&self, context: &str, error: &Error) {
|
||||
self.log(tracing::Level::ERROR, context, &format!("{:?}", error));
|
||||
// TODO Can be extended to record to files or send to monitoring systems
|
||||
}
|
||||
fn log(&self, level: tracing::Level, context: &str, message: &str) {
|
||||
match level {
|
||||
tracing::Level::ERROR => tracing::error!("[{}] {}", context, message),
|
||||
tracing::Level::WARN => tracing::warn!("[{}] {}", context, message),
|
||||
tracing::Level::INFO => tracing::info!("[{}] {}", context, message),
|
||||
tracing::Level::DEBUG => tracing::debug!("[{}] {}", context, message),
|
||||
tracing::Level::TRACE => tracing::trace!("[{}] {}", context, message),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
use crate::Error;
|
||||
use crate::Log;
|
||||
use std::sync::Arc;
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
use tokio::fs::{create_dir_all, File, OpenOptions};
|
||||
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter};
|
||||
use tokio::sync::RwLock;
|
||||
|
||||
/// `EventStore` is a struct that manages the storage of event logs.
|
||||
pub struct EventStore {
|
||||
path: String,
|
||||
lock: Arc<RwLock<()>>,
|
||||
}
|
||||
|
||||
impl EventStore {
|
||||
pub async fn new(path: &str) -> Result<Self, Error> {
|
||||
create_dir_all(path).await?;
|
||||
Ok(Self {
|
||||
path: path.to_string(),
|
||||
lock: Arc::new(RwLock::new(())),
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn save_logs(&self, logs: &[Log]) -> Result<(), Error> {
|
||||
let _guard = self.lock.write().await;
|
||||
let file_path = format!(
|
||||
"{}/events_{}.jsonl",
|
||||
self.path,
|
||||
SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs()
|
||||
);
|
||||
let file = OpenOptions::new().create(true).append(true).open(&file_path).await?;
|
||||
let mut writer = BufWriter::new(file);
|
||||
for log in logs {
|
||||
let line = serde_json::to_string(log)?;
|
||||
writer.write_all(line.as_bytes()).await?;
|
||||
writer.write_all(b"\n").await?;
|
||||
}
|
||||
writer.flush().await?;
|
||||
tracing::info!("Saved logs to {} end", file_path);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn load_logs(&self) -> Result<Vec<Log>, Error> {
|
||||
let _guard = self.lock.read().await;
|
||||
let mut logs = Vec::new();
|
||||
let mut entries = tokio::fs::read_dir(&self.path).await?;
|
||||
while let Some(entry) = entries.next_entry().await? {
|
||||
let file = File::open(entry.path()).await?;
|
||||
let reader = BufReader::new(file);
|
||||
let mut lines = reader.lines();
|
||||
while let Some(line) = lines.next_line().await? {
|
||||
let log: Log = serde_json::from_str(&line)?;
|
||||
logs.push(log);
|
||||
}
|
||||
}
|
||||
Ok(logs)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,159 @@
|
||||
use rustfs_event_notifier::{AdapterConfig, NotifierSystem, WebhookConfig};
|
||||
use rustfs_event_notifier::{Bucket, Event, EventBuilder, Identity, Metadata, Name, Object, Source};
|
||||
use rustfs_event_notifier::{ChannelAdapter, WebhookAdapter};
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_webhook_adapter() {
|
||||
let adapter = WebhookAdapter::new(WebhookConfig {
|
||||
endpoint: "http://localhost:8080/webhook".to_string(),
|
||||
auth_token: None,
|
||||
custom_headers: None,
|
||||
max_retries: 1,
|
||||
timeout: 5,
|
||||
});
|
||||
|
||||
// create an s3 metadata object
|
||||
let metadata = Metadata {
|
||||
schema_version: "1.0".to_string(),
|
||||
configuration_id: "test-config".to_string(),
|
||||
bucket: Bucket {
|
||||
name: "my-bucket".to_string(),
|
||||
owner_identity: Identity {
|
||||
principal_id: "owner123".to_string(),
|
||||
},
|
||||
arn: "arn:aws:s3:::my-bucket".to_string(),
|
||||
},
|
||||
object: Object {
|
||||
key: "test.txt".to_string(),
|
||||
size: Some(1024),
|
||||
etag: Some("abc123".to_string()),
|
||||
content_type: Some("text/plain".to_string()),
|
||||
user_metadata: None,
|
||||
version_id: None,
|
||||
sequencer: "1234567890".to_string(),
|
||||
},
|
||||
};
|
||||
|
||||
// create source object
|
||||
let source = Source {
|
||||
host: "localhost".to_string(),
|
||||
port: "80".to_string(),
|
||||
user_agent: "curl/7.68.0".to_string(),
|
||||
};
|
||||
|
||||
// Create events using builder mode
|
||||
let event = Event::builder()
|
||||
.event_version("2.0")
|
||||
.event_source("aws:s3")
|
||||
.aws_region("us-east-1")
|
||||
.event_time("2023-10-01T12:00:00.000Z")
|
||||
.event_name(Name::ObjectCreatedPut)
|
||||
.user_identity(Identity {
|
||||
principal_id: "user123".to_string(),
|
||||
})
|
||||
.request_parameters(HashMap::new())
|
||||
.response_elements(HashMap::new())
|
||||
.s3(metadata)
|
||||
.source(source)
|
||||
.channels(vec!["webhook".to_string()])
|
||||
.build()
|
||||
.expect("failed to create event");
|
||||
|
||||
let result = adapter.send(&event).await;
|
||||
assert!(result.is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_notification_system() {
|
||||
let config = rustfs_event_notifier::NotifierConfig {
|
||||
store_path: "./test_events".to_string(),
|
||||
channel_capacity: 100,
|
||||
adapters: vec![AdapterConfig::Webhook(WebhookConfig {
|
||||
endpoint: "http://localhost:8080/webhook".to_string(),
|
||||
auth_token: None,
|
||||
custom_headers: None,
|
||||
max_retries: 1,
|
||||
timeout: 5,
|
||||
})],
|
||||
};
|
||||
let system = Arc::new(tokio::sync::Mutex::new(NotifierSystem::new(config.clone()).await.unwrap()));
|
||||
let adapters: Vec<Arc<dyn ChannelAdapter>> = vec![Arc::new(WebhookAdapter::new(WebhookConfig {
|
||||
endpoint: "http://localhost:8080/webhook".to_string(),
|
||||
auth_token: None,
|
||||
custom_headers: None,
|
||||
max_retries: 1,
|
||||
timeout: 5,
|
||||
}))];
|
||||
|
||||
// create an s3 metadata object
|
||||
let metadata = Metadata {
|
||||
schema_version: "1.0".to_string(),
|
||||
configuration_id: "test-config".to_string(),
|
||||
bucket: Bucket {
|
||||
name: "my-bucket".to_string(),
|
||||
owner_identity: Identity {
|
||||
principal_id: "owner123".to_string(),
|
||||
},
|
||||
arn: "arn:aws:s3:::my-bucket".to_string(),
|
||||
},
|
||||
object: Object {
|
||||
key: "test.txt".to_string(),
|
||||
size: Some(1024),
|
||||
etag: Some("abc123".to_string()),
|
||||
content_type: Some("text/plain".to_string()),
|
||||
user_metadata: None,
|
||||
version_id: None,
|
||||
sequencer: "1234567890".to_string(),
|
||||
},
|
||||
};
|
||||
|
||||
// create source object
|
||||
let source = Source {
|
||||
host: "localhost".to_string(),
|
||||
port: "80".to_string(),
|
||||
user_agent: "curl/7.68.0".to_string(),
|
||||
};
|
||||
|
||||
// create a preconfigured builder with objects
|
||||
let event = EventBuilder::for_object_creation(metadata, source)
|
||||
.user_identity(Identity {
|
||||
principal_id: "user123".to_string(),
|
||||
})
|
||||
.event_time("2023-10-01T12:00:00.000Z")
|
||||
.channels(vec!["webhook".to_string()])
|
||||
.build()
|
||||
.expect("failed to create event");
|
||||
|
||||
{
|
||||
let system_lock = system.lock().await;
|
||||
system_lock.send_event(event).await.unwrap();
|
||||
}
|
||||
|
||||
let system_clone = Arc::clone(&system);
|
||||
let system_handle = tokio::spawn(async move {
|
||||
let mut system = system_clone.lock().await;
|
||||
system.start(adapters).await
|
||||
});
|
||||
|
||||
// set 10 seconds timeout
|
||||
match tokio::time::timeout(std::time::Duration::from_secs(10), system_handle).await {
|
||||
Ok(result) => {
|
||||
println!("System started successfully");
|
||||
assert!(result.is_ok());
|
||||
}
|
||||
Err(_) => {
|
||||
println!("System operation timed out, forcing shutdown");
|
||||
// create a new task to handle the timeout
|
||||
let system = Arc::clone(&system);
|
||||
tokio::spawn(async move {
|
||||
if let Ok(mut guard) = system.try_lock() {
|
||||
guard.shutdown().await.unwrap();
|
||||
}
|
||||
});
|
||||
// give the system some time to clean up resources
|
||||
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -254,21 +254,16 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard {
|
||||
let filter_otel = match logger_level {
|
||||
"trace" | "debug" => {
|
||||
info!("OpenTelemetry tracing initialized with level: {}", logger_level);
|
||||
EnvFilter::new(logger_level)
|
||||
}
|
||||
_ => {
|
||||
let mut filter = EnvFilter::new(logger_level);
|
||||
for directive in ["hyper", "tonic", "h2", "reqwest", "tower"] {
|
||||
filter = filter.add_directive(format!("{}=off", directive).parse().unwrap());
|
||||
}
|
||||
filter
|
||||
}
|
||||
_ => {
|
||||
let mut filter = EnvFilter::new(logger_level);
|
||||
for directive in ["hyper", "tonic", "h2", "reqwest"] {
|
||||
filter = filter.add_directive(format!("{}=off", directive).parse().unwrap());
|
||||
}
|
||||
filter
|
||||
}
|
||||
};
|
||||
|
||||
layer::OpenTelemetryTracingBridge::new(&logger_provider).with_filter(filter_otel)
|
||||
};
|
||||
|
||||
@@ -277,7 +272,9 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard {
|
||||
.with(switch_level(logger_level))
|
||||
.with(OpenTelemetryLayer::new(tracer))
|
||||
.with(MetricsLayer::new(meter_provider.clone()))
|
||||
.with(otel_layer);
|
||||
.with(otel_layer)
|
||||
.with(EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(logger_level)));
|
||||
|
||||
// Configure formatting layer
|
||||
let enable_color = std::io::stdout().is_terminal();
|
||||
let fmt_layer = tracing_subscriber::fmt::layer()
|
||||
|
||||
@@ -1 +0,0 @@
|
||||
## Observability Docker Compose
|
||||
@@ -0,0 +1,29 @@
|
||||
# config.toml
|
||||
store_path = "./deploy/logs/event_store"
|
||||
channel_capacity = 5000
|
||||
|
||||
[[adapters]]
|
||||
type = "Webhook"
|
||||
endpoint = "http://127.0.0.1:3020/webhook"
|
||||
auth_token = "your-auth-token"
|
||||
max_retries = 3
|
||||
timeout = 50
|
||||
|
||||
[adapters.custom_headers]
|
||||
custom_server = "value_server"
|
||||
custom_client = "value_client"
|
||||
|
||||
#[[adapters]]
|
||||
#type = "Kafka"
|
||||
#brokers = "localhost:9092"
|
||||
#topic = "notifications"
|
||||
#max_retries = 3
|
||||
#timeout = 60
|
||||
#
|
||||
#[[adapters]]
|
||||
#type = "Mqtt"
|
||||
#broker = "mqtt.example.com"
|
||||
#port = 1883
|
||||
#client_id = "event-notifier"
|
||||
#topic = "events"
|
||||
#max_retries = 3
|
||||
@@ -6,7 +6,7 @@ meter_interval = 30
|
||||
service_name = "rustfs"
|
||||
service_version = "0.1.0"
|
||||
environment = "develop"
|
||||
logger_level = "info"
|
||||
logger_level = "error"
|
||||
|
||||
[sinks]
|
||||
[sinks.kafka] # Kafka sink is disabled by default
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "ecstore"
|
||||
version = "0.1.0"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
repository.workspace = true
|
||||
|
||||
@@ -122,7 +122,7 @@ impl LocalDisk {
|
||||
let root = fs::canonicalize(ep.get_file_path()).await?;
|
||||
|
||||
if cleanup {
|
||||
// TODO: 删除tmp数据
|
||||
// TODO: 删除 tmp 数据
|
||||
}
|
||||
|
||||
let format_path = Path::new(super::RUSTFS_META_BUCKET)
|
||||
@@ -631,13 +631,13 @@ impl LocalDisk {
|
||||
}
|
||||
}
|
||||
|
||||
// 没有版本了，删除xl.meta
|
||||
// 没有版本了，删除 xl.meta
|
||||
if fm.versions.is_empty() {
|
||||
self.delete_file(&volume_dir, &xlpath, true, false).await?;
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// 更新xl.meta
|
||||
// 更新 xl.meta
|
||||
let buf = fm.marshal_msg()?;
|
||||
|
||||
let volume_dir = self.get_bucket_path(volume)?;
|
||||
@@ -875,7 +875,7 @@ impl LocalDisk {
|
||||
.read_metadata(self.get_object_path(bucket, format!("{}/{}", ¤t, &entry).as_str())?)
|
||||
.await?;
|
||||
|
||||
// 用strip_suffix只删除一次
|
||||
// 用 strip_suffix 只删除一次
|
||||
let entry = entry.strip_suffix(STORAGE_FORMAT_FILE).unwrap_or_default().to_owned();
|
||||
let name = entry.trim_end_matches(SLASH_SEPARATOR);
|
||||
let name = decode_dir_object(format!("{}/{}", ¤t, &name).as_str());
|
||||
@@ -1723,7 +1723,7 @@ impl DiskAPI for LocalDisk {
|
||||
}
|
||||
}
|
||||
|
||||
// xl.meta路径
|
||||
// xl.meta 路径
|
||||
let src_file_path = src_volume_dir.join(Path::new(format!("{}/{}", &src_path, super::STORAGE_FORMAT_FILE).as_str()));
|
||||
let dst_file_path = dst_volume_dir.join(Path::new(format!("{}/{}", &dst_path, super::STORAGE_FORMAT_FILE).as_str()));
|
||||
|
||||
@@ -1754,7 +1754,7 @@ impl DiskAPI for LocalDisk {
|
||||
check_path_length(src_file_path.to_string_lossy().to_string().as_str())?;
|
||||
check_path_length(dst_file_path.to_string_lossy().to_string().as_str())?;
|
||||
|
||||
// 读旧xl.meta
|
||||
// 读旧 xl.meta
|
||||
|
||||
let has_dst_buf = match utils::fs::read_file(&dst_file_path).await {
|
||||
Ok(res) => Some(res),
|
||||
@@ -2268,7 +2268,7 @@ impl DiskAPI for LocalDisk {
|
||||
async fn delete_volume(&self, volume: &str) -> Result<()> {
|
||||
let p = self.get_bucket_path(volume)?;
|
||||
|
||||
// TODO: 不能用递归删除，如果目录下面有文件，返回errVolumeNotEmpty
|
||||
// TODO: 不能用递归删除，如果目录下面有文件，返回 errVolumeNotEmpty
|
||||
|
||||
if let Err(err) = fs::remove_dir_all(&p).await {
|
||||
match err.kind() {
|
||||
@@ -2328,10 +2328,7 @@ impl DiskAPI for LocalDisk {
|
||||
}
|
||||
}
|
||||
|
||||
let vcfg = match BucketVersioningSys::get(&cache.info.name).await {
|
||||
Ok(vcfg) => Some(vcfg),
|
||||
Err(_) => None,
|
||||
};
|
||||
let vcfg = (BucketVersioningSys::get(&cache.info.name).await).ok();
|
||||
|
||||
let loc = self.get_disk_location();
|
||||
let disks = store.get_disks(loc.pool_idx.unwrap(), loc.disk_idx.unwrap()).await?;
|
||||
@@ -2491,7 +2488,6 @@ async fn get_disk_info(drive_path: PathBuf) -> Result<(Info, bool)> {
|
||||
|
||||
#[cfg(test)]
|
||||
mod test {
|
||||
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -1101,7 +1101,7 @@ pub fn lc_has_active_rules(config: &BucketLifecycleConfiguration, prefix: &str)
|
||||
return true;
|
||||
}
|
||||
|
||||
if let Some(Some(true)) = rule.expiration.as_ref().map(|e| e.expired_object_delete_marker.map(|m| m)) {
|
||||
if let Some(Some(true)) = rule.expiration.as_ref().map(|e| e.expired_object_delete_marker) {
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
+12
-11
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "rustfs"
|
||||
version = "0.1.0"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
repository.workspace = true
|
||||
@@ -15,9 +15,9 @@ path = "src/main.rs"
|
||||
workspace = true
|
||||
|
||||
[dependencies]
|
||||
madmin.workspace = true
|
||||
api = { path = "../s3select/api" }
|
||||
appauth = { version = "0.0.1", path = "../appauth" }
|
||||
madmin = { workspace = true }
|
||||
api = { workspace = true }
|
||||
appauth = { workspace = true }
|
||||
atoi = { workspace = true }
|
||||
atomic_enum = { workspace = true }
|
||||
axum.workspace = true
|
||||
@@ -27,7 +27,7 @@ async-trait.workspace = true
|
||||
bytes.workspace = true
|
||||
chrono = { workspace = true }
|
||||
clap.workspace = true
|
||||
crypto = { path = "../crypto" }
|
||||
crypto = { workspace = true }
|
||||
datafusion = { workspace = true }
|
||||
common.workspace = true
|
||||
const-str = { version = "0.6.1", features = ["std", "proc"] }
|
||||
@@ -40,16 +40,17 @@ hyper.workspace = true
|
||||
hyper-util.workspace = true
|
||||
http.workspace = true
|
||||
http-body.workspace = true
|
||||
iam = { path = "../iam" }
|
||||
iam = { workspace = true }
|
||||
lock.workspace = true
|
||||
local-ip-address = { workspace = true }
|
||||
matchit = { workspace = true }
|
||||
mime.workspace = true
|
||||
mime_guess = "2.0.5"
|
||||
mime_guess = { workspace = true }
|
||||
pin-project-lite.workspace = true
|
||||
protos.workspace = true
|
||||
query = { path = "../s3select/query" }
|
||||
query = { workspace = true }
|
||||
rmp-serde.workspace = true
|
||||
rustfs-event-notifier = { workspace = true }
|
||||
rustfs-obs = { workspace = true }
|
||||
rustls.workspace = true
|
||||
rustls-pemfile.workspace = true
|
||||
@@ -59,7 +60,7 @@ s3s.workspace = true
|
||||
serde.workspace = true
|
||||
serde_json.workspace = true
|
||||
serde_urlencoded = { workspace = true }
|
||||
shadow-rs.workspace = true
|
||||
shadow-rs = { workspace = true, features = ["build", "metadata"] }
|
||||
tracing.workspace = true
|
||||
time = { workspace = true, features = ["parsing", "formatting", "serde"] }
|
||||
tokio-util.workspace = true
|
||||
@@ -92,7 +93,7 @@ bytes.workspace = true
|
||||
futures.workspace = true
|
||||
futures-util.workspace = true
|
||||
# uuid = { version = "1.8.0", features = ["v4", "fast-rng", "serde"] }
|
||||
ecstore = { path = "../ecstore" }
|
||||
ecstore = { workspace = true }
|
||||
s3s.workspace = true
|
||||
clap = { workspace = true }
|
||||
hyper-util = { workspace = true, features = [
|
||||
@@ -102,5 +103,5 @@ hyper-util = { workspace = true, features = [
|
||||
] }
|
||||
transform-stream = { workspace = true }
|
||||
netif = "0.1.6"
|
||||
shadow-rs.workspace = true
|
||||
shadow-rs = { workspace = true, features = ["build"] }
|
||||
# pin-utils = "0.1.0"
|
||||
|
||||
@@ -103,6 +103,10 @@ pub struct Opt {
|
||||
|
||||
#[arg(long, env = "RUSTFS_LICENSE")]
|
||||
pub license: Option<String>,
|
||||
|
||||
/// event notifier config file
|
||||
#[arg(long, env = "RUSTFS_EVENT_CONFIG")]
|
||||
pub event_config: Option<String>,
|
||||
}
|
||||
|
||||
// lazy_static::lazy_static! {
|
||||
|
||||
@@ -46,6 +46,7 @@ use hyper_util::{
|
||||
use iam::init_iam_sys;
|
||||
use license::init_license;
|
||||
use protos::proto_gen::node_service::node_service_server::NodeServiceServer;
|
||||
use rustfs_event_notifier::NotifierConfig;
|
||||
use rustfs_obs::{init_obs, load_config, set_global_guard, InitLogStatus};
|
||||
use rustls::ServerConfig;
|
||||
use s3s::{host::MultiDomain, service::S3ServiceBuilder};
|
||||
@@ -105,6 +106,23 @@ async fn main() -> Result<()> {
|
||||
// Log initialization status
|
||||
InitLogStatus::init_start_log(&config.observability).await?;
|
||||
|
||||
// Initialize event notifier
|
||||
let notifier_config = opt.clone().event_config;
|
||||
if notifier_config.is_some() {
|
||||
info!("event_config is not empty");
|
||||
tokio::spawn(async move {
|
||||
let config = NotifierConfig::event_load_config(notifier_config);
|
||||
let result = rustfs_event_notifier::initialize(config).await;
|
||||
if let Err(e) = result {
|
||||
error!("Failed to initialize event notifier: {}", e);
|
||||
} else {
|
||||
info!("Event notifier initialized successfully");
|
||||
}
|
||||
});
|
||||
} else {
|
||||
info!("event_config is empty");
|
||||
}
|
||||
|
||||
// Run parameters
|
||||
run(opt).await
|
||||
}
|
||||
@@ -448,6 +466,16 @@ async fn run(opt: config::Opt) -> Result<()> {
|
||||
info!("Shutdown signal received in main thread");
|
||||
// update the status to stopping first
|
||||
state_manager.update(ServiceState::Stopping);
|
||||
|
||||
// Stop the notification system
|
||||
if rustfs_event_notifier::is_ready() {
|
||||
// stop event notifier
|
||||
rustfs_event_notifier::shutdown().await.map_err(|err| {
|
||||
error!("Failed to shut down the notification system: {}", err);
|
||||
Error::from_string(err.to_string())
|
||||
})?;
|
||||
}
|
||||
|
||||
info!("Server is stopping...");
|
||||
let _ = shutdown_tx.send(());
|
||||
// Wait for the worker thread to complete the cleaning work
|
||||
|
||||
@@ -10,9 +10,9 @@ chrono.workspace = true
|
||||
datafusion = { workspace = true }
|
||||
ecstore.workspace = true
|
||||
futures = { workspace = true }
|
||||
futures-core = "0.3.31"
|
||||
futures-core = { workspace = true }
|
||||
http.workspace = true
|
||||
object_store = "0.11.2"
|
||||
object_store = { workspace = true }
|
||||
s3s.workspace = true
|
||||
snafu = { workspace = true, features = ["backtrace"] }
|
||||
tokio.workspace = true
|
||||
|
||||
@@ -4,7 +4,7 @@ version.workspace = true
|
||||
edition.workspace = true
|
||||
|
||||
[dependencies]
|
||||
api = { path = "../api" }
|
||||
api = { workspace = true }
|
||||
async-recursion = { workspace = true }
|
||||
async-trait.workspace = true
|
||||
datafusion = { workspace = true }
|
||||
|
||||
+6
-1
@@ -57,9 +57,14 @@ export RUSTFS__SINKS__KAFKA__BOOTSTRAP_SERVERS=""
|
||||
export RUSTFS__SINKS__KAFKA__TOPIC=""
|
||||
export RUSTFS__LOGGER__QUEUE_CAPACITY=10
|
||||
|
||||
# 事件消息配置
|
||||
export RUSTFS_EVENT_CONFIG="./deploy/config/event.example.toml"
|
||||
|
||||
if [ -n "$1" ]; then
|
||||
export RUSTFS_VOLUMES="$1"
|
||||
fi
|
||||
|
||||
|
||||
# 启动 webhook 服务器
|
||||
cargo run --example webhook -p rustfs-event-notifier &
|
||||
# 启动主服务
|
||||
cargo run --bin rustfs
|
||||
Reference in New Issue
Block a user