Feature/bucket event notification (#365)

* add tracing instrument

* fix rebalance/decom

* modify Telemetry filter order

* feat: improve address binding and port handling mechanism (#366)

* feat: improve address binding and port handling mechanism

1. Add support for ":port" format to enable dual-stack binding (IPv4/IPv6)
2. Implement automatic port allocation when port 0 is specified
3. Optimize server startup process with unified address resolution
4. Enhance error handling and logging for address resolution
5. Improve graceful shutdown with signal listening
6. Clean up commented code in console.rs

Files:
- ecstore/src/utils/net.rs
- rustfs/src/console.rs
- rustfs/src/main.rs

Branch: feature/server-and-console-port

* improve code for console

* improve code

* improve code for console and net.rs

* Update rustfs/src/main.rs

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

* Update rustfs/src/utils/mod.rs

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

---------

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

* upgrade config file

* modify

* fix readme

Signed-off-by: junxiang Mu <1948535941@qq.com>

* improve readme.md

* improve code for readme.md
add chinese readme.md

* Implement Storage Service Event Notification System

Added event notification capability to the storage module, enabling the storage service to publish object operation events. Key changes include:

1. Created `event_notifier` module providing core functionality:
   - `create_metadata()` - Creates event metadata objects with default configuration ID
   - `send_event()` - Asynchronously sends event notifications with error handling

2. Integrated the `rustfs_event_notifier` library:
   - Supports object creation, deletion, and access events
   - Provides event metadata building and management
   - Includes proper error propagation

These changes enable the system to trigger notifications when storage operations occur, facilitating auditing, monitoring, and integration with other systems.

* fix

---------

Signed-off-by: junxiang Mu <1948535941@qq.com>
Co-authored-by: weisd <im@weisd.in>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: junxiang Mu <1948535941@qq.com>

crates/event-notifier/src/bus.rs

@@ -7,11 +7,13 @@ use std::time::{SystemTime, UNIX_EPOCH};
 use tokio::sync::mpsc;
 use tokio::time::Duration;
 use tokio_util::sync::CancellationToken;
+use tracing::instrument;
 
 /// 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.
+#[instrument(skip_all)]
 pub async fn event_bus(
     mut rx: mpsc::Receiver<Event>,
     adapters: Vec<Arc<dyn ChannelAdapter>>,

crates/event-notifier/src/event.rs

@@ -15,6 +15,18 @@ pub struct Identity {
     pub principal_id: String,
 }
 
+impl Identity {
+    /// Create a new Identity instance
+    pub fn new(principal_id: String) -> Self {
+        Self { principal_id }
+    }
+
+    /// Set the principal ID
+    pub fn set_principal_id(&mut self, principal_id: String) {
+        self.principal_id = principal_id;
+    }
+}
+
 /// A struct representing the bucket information
 #[derive(Serialize, Deserialize, Clone, Debug)]
 pub struct Bucket {
@@ -24,6 +36,32 @@ pub struct Bucket {
     pub arn: String,
 }
 
+impl Bucket {
+    /// Create a new Bucket instance
+    pub fn new(name: String, owner_identity: Identity, arn: String) -> Self {
+        Self {
+            name,
+            owner_identity,
+            arn,
+        }
+    }
+
+    /// Set the name of the bucket
+    pub fn set_name(&mut self, name: String) {
+        self.name = name;
+    }
+
+    /// Set the ARN of the bucket
+    pub fn set_arn(&mut self, arn: String) {
+        self.arn = arn;
+    }
+
+    /// Set the owner identity of the bucket
+    pub fn set_owner_identity(&mut self, owner_identity: Identity) {
+        self.owner_identity = owner_identity;
+    }
+}
+
 /// A struct representing the object information
 #[derive(Serialize, Deserialize, Clone, Debug)]
 pub struct Object {
@@ -41,6 +79,64 @@ pub struct Object {
     pub sequencer: String,
 }
 
+impl Object {
+    /// Create a new Object instance
+    pub fn new(
+        key: String,
+        size: Option<i64>,
+        etag: Option<String>,
+        content_type: Option<String>,
+        user_metadata: Option<HashMap<String, String>>,
+        version_id: Option<String>,
+        sequencer: String,
+    ) -> Self {
+        Self {
+            key,
+            size,
+            etag,
+            content_type,
+            user_metadata,
+            version_id,
+            sequencer,
+        }
+    }
+
+    /// Set the key
+    pub fn set_key(&mut self, key: String) {
+        self.key = key;
+    }
+
+    /// Set the size
+    pub fn set_size(&mut self, size: Option<i64>) {
+        self.size = size;
+    }
+
+    /// Set the etag
+    pub fn set_etag(&mut self, etag: Option<String>) {
+        self.etag = etag;
+    }
+
+    /// Set the content type
+    pub fn set_content_type(&mut self, content_type: Option<String>) {
+        self.content_type = content_type;
+    }
+
+    /// Set the user metadata
+    pub fn set_user_metadata(&mut self, user_metadata: Option<HashMap<String, String>>) {
+        self.user_metadata = user_metadata;
+    }
+
+    /// Set the version ID
+    pub fn set_version_id(&mut self, version_id: Option<String>) {
+        self.version_id = version_id;
+    }
+
+    /// Set the sequencer
+    pub fn set_sequencer(&mut self, sequencer: String) {
+        self.sequencer = sequencer;
+    }
+}
+
 /// A struct representing the metadata of the event
 #[derive(Serialize, Deserialize, Clone, Debug)]
 pub struct Metadata {
@@ -52,6 +148,57 @@ pub struct Metadata {
     pub object: Object,
 }
 
+impl Default for Metadata {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+impl Metadata {
+    /// Create a new Metadata instance
+    pub fn create(schema_version: String, configuration_id: String, bucket: Bucket, object: Object) -> Self {
+        Self {
+            schema_version,
+            configuration_id,
+            bucket,
+            object,
+        }
+    }
+
+    /// Create a new Metadata instance with default values
+    pub fn new() -> Self {
+        Self {
+            schema_version: "1.0".to_string(),
+            configuration_id: "default".to_string(),
+            bucket: Bucket::new(
+                "default".to_string(),
+                Identity::new("default".to_string()),
+                "arn:aws:s3:::default".to_string(),
+            ),
+            object: Object::new("default".to_string(), None, None, None, None, None, "default".to_string()),
+        }
+    }
+
+    /// Set the schema version
+    pub fn set_schema_version(&mut self, schema_version: String) {
+        self.schema_version = schema_version;
+    }
+
+    /// Set the configuration ID
+    pub fn set_configuration_id(&mut self, configuration_id: String) {
+        self.configuration_id = configuration_id;
+    }
+
+    /// Set the bucket
+    pub fn set_bucket(&mut self, bucket: Bucket) {
+        self.bucket = bucket;
+    }
+
+    /// Set the object
+    pub fn set_object(&mut self, object: Object) {
+        self.object = object;
+    }
+}
+
 /// A struct representing the source of the event
 #[derive(Serialize, Deserialize, Clone, Debug)]
 pub struct Source {
@@ -61,6 +208,28 @@ pub struct Source {
     pub user_agent: String,
 }
 
+impl Source {
+    /// Create a new Source instance
+    pub fn new(host: String, port: String, user_agent: String) -> Self {
+        Self { host, port, user_agent }
+    }
+
+    /// Set the host
+    pub fn set_host(&mut self, host: String) {
+        self.host = host;
+    }
+
+    /// Set the port
+    pub fn set_port(&mut self, port: String) {
+        self.port = port;
+    }
+
+    /// Set the user agent
+    pub fn set_user_agent(&mut self, user_agent: String) {
+        self.user_agent = user_agent;
+    }
+}
+
 /// Builder for creating an Event.
 ///
 /// This struct is used to build an Event object with various parameters.
@@ -301,7 +470,17 @@ pub struct Log {
     pub records: Vec<Event>,
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq, SerializeDisplay, DeserializeFromStr, Display, EnumString)]
+#[derive(
+    Debug,
+    Clone,
+    Copy,
+    PartialEq,
+    Eq,
+    SerializeDisplay,
+    DeserializeFromStr,
+    Display,
+    EnumString
+)]
 #[strum(serialize_all = "SCREAMING_SNAKE_CASE")]
 pub enum Name {
     ObjectAccessedGet,

crates/event-notifier/src/global.rs

@@ -1,6 +1,7 @@
 use crate::{create_adapters, Error, Event, NotifierConfig, NotifierSystem};
 use std::sync::{atomic, Arc};
 use tokio::sync::{Mutex, OnceCell};
+use tracing::instrument;
 
 static GLOBAL_SYSTEM: OnceCell<Arc<Mutex<NotifierSystem>>> = OnceCell::const_new();
 static INITIALIZED: atomic::AtomicBool = atomic::AtomicBool::new(false);
@@ -113,6 +114,7 @@ pub fn is_ready() -> bool {
 /// - The system is not initialized.
 /// - The system is not ready.
 /// - Sending the event fails.
+#[instrument(fields(event))]
 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"));
@@ -124,6 +126,7 @@ pub async fn send_event(event: Event) -> Result<(), Error> {
 }
 
 /// Shuts down the notification system.
+#[instrument]
 pub async fn shutdown() -> Result<(), Error> {
     if let Some(system) = GLOBAL_SYSTEM.get() {
         tracing::info!("Shutting down notification system start");

crates/event-notifier/src/notifier.rs

@@ -2,6 +2,7 @@ use crate::{event_bus, ChannelAdapter, Error, Event, EventStore, NotifierConfig}
 use std::sync::Arc;
 use tokio::sync::mpsc;
 use tokio_util::sync::CancellationToken;
+use tracing::instrument;
 
 /// The `NotificationSystem` struct represents the notification system.
 /// It manages the event bus and the adapters.
@@ -18,6 +19,7 @@ pub struct NotifierSystem {
 
 impl NotifierSystem {
     /// Creates a new `NotificationSystem` instance.
+    #[instrument(skip(config))]
     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?);
@@ -44,6 +46,7 @@ impl NotifierSystem {
 
     /// Starts the notification system.
     /// It initializes the event bus and the producer.
+    #[instrument(skip_all)]
     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");
@@ -67,6 +70,7 @@ impl NotifierSystem {
 
     /// Sends an event to the notification system.
     /// This method is used to send events to the event bus.
+    #[instrument(skip(self))]
     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() {
@@ -85,6 +89,7 @@ impl NotifierSystem {
 
     /// Shuts down the notification system.
     /// This method is used to cancel the event bus and producer tasks.
+    #[instrument(skip(self))]
     pub async fn shutdown(&mut self) -> Result<(), Error> {
         tracing::info!("Shutting down the notification system");
         self.shutdown.cancel();
@@ -112,10 +117,13 @@ impl NotifierSystem {
         self.shutdown.is_cancelled()
     }
 
-    fn handle_error(&self, context: &str, error: &Error) {
+    #[instrument(skip(self))]
+    pub 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
     }
+
+    #[instrument(skip(self))]
     fn log(&self, level: tracing::Level, context: &str, message: &str) {
         match level {
             tracing::Level::ERROR => tracing::error!("[{}] {}", context, message),

crates/event-notifier/src/store.rs

@@ -5,6 +5,7 @@ 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;
+use tracing::instrument;
 
 /// `EventStore` is a struct that manages the storage of event logs.
 pub struct EventStore {
@@ -21,6 +22,7 @@ impl EventStore {
         })
     }
 
+    #[instrument(skip(self))]
     pub async fn save_logs(&self, logs: &[Log]) -> Result<(), Error> {
         let _guard = self.lock.write().await;
         let file_path = format!(

crates/obs/src/telemetry.rs

@@ -231,7 +231,6 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard {
             .with(otel_layer)
             .with(MetricsLayer::new(meter_provider.clone()))
             .init();
-        info!("Telemetry logging enabled: {:?}", config.local_logging_enabled);
 
         if !endpoint.is_empty() {
             info!(

rustfs/src/storage/access.rs

@@ -20,6 +20,7 @@ pub(crate) struct ReqInfo {
     pub version_id: Option<String>,
 }
 
+/// Authorizes the request based on the action and credentials.
 pub async fn authorize_request<T>(req: &mut S3Request<T>, action: Action) -> S3Result<()> {
     let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
 

rustfs/src/storage/ecfs.rs

@@ -141,6 +141,7 @@ impl S3 for FS {
         Ok(S3Response::new(output))
     }
 
+    /// Copy an object from one location to another
     #[tracing::instrument(level = "debug", skip(self, req))]
     async fn copy_object(&self, req: S3Request<CopyObjectInput>) -> S3Result<S3Response<CopyObjectOutput>> {
         let CopyObjectInput {
@@ -227,6 +228,7 @@ impl S3 for FS {
         Ok(S3Response::new(output))
     }
 
+    /// Delete a bucket
     #[tracing::instrument(level = "debug", skip(self, req))]
     async fn delete_bucket(&self, req: S3Request<DeleteBucketInput>) -> S3Result<S3Response<DeleteBucketOutput>> {
         let input = req.input;
@@ -249,6 +251,7 @@ impl S3 for FS {
         Ok(S3Response::new(DeleteBucketOutput {}))
     }
 
+    /// Delete an object
     #[tracing::instrument(level = "debug", skip(self, req))]
     async fn delete_object(&self, req: S3Request<DeleteObjectInput>) -> S3Result<S3Response<DeleteObjectOutput>> {
         let DeleteObjectInput {
@@ -308,6 +311,7 @@ impl S3 for FS {
         Ok(S3Response::new(output))
     }
 
+    /// Delete multiple objects
     #[tracing::instrument(level = "debug", skip(self, req))]
     async fn delete_objects(&self, req: S3Request<DeleteObjectsInput>) -> S3Result<S3Response<DeleteObjectsOutput>> {
         // info!("delete_objects args {:?}", req.input);
@@ -367,6 +371,7 @@ impl S3 for FS {
         Ok(S3Response::new(output))
     }
 
+    /// Get bucket location
     #[tracing::instrument(level = "debug", skip(self, req))]
     async fn get_bucket_location(&self, req: S3Request<GetBucketLocationInput>) -> S3Result<S3Response<GetBucketLocationOutput>> {
         // mc get  1
@@ -385,6 +390,7 @@ impl S3 for FS {
         Ok(S3Response::new(output))
     }
 
+    /// Get bucket notification
     #[tracing::instrument(
         level = "debug",
         skip(self, req),

rustfs/src/storage/event_notifier.rs

@@ -0,0 +1,17 @@
+use rustfs_event_notifier::{Event, Metadata};
+
+/// Create a new metadata object
+#[allow(dead_code)]
+pub(crate) fn create_metadata() -> Metadata {
+    // Create a new metadata object
+    let mut metadata = Metadata::new();
+    metadata.set_configuration_id("test-config".to_string());
+    // Return the created metadata object
+    metadata
+}
+
+/// Create a new event object
+#[allow(dead_code)]
+pub(crate) async fn send_event(event: Event) -> Result<(), Box<dyn std::error::Error>> {
+    rustfs_event_notifier::send_event(event).await.map_err(|e| e.into())
+}

rustfs/src/storage/mod.rs

@@ -1,4 +1,5 @@
 pub mod access;
 pub mod ecfs;
 pub mod error;
+mod event_notifier;
 pub mod options;

rustfs/src/storage/options.rs

@@ -8,6 +8,7 @@ use lazy_static::lazy_static;
 use std::collections::HashMap;
 use uuid::Uuid;
 
+/// Creates options for deleting an object in a bucket.
 pub async fn del_opts(
     bucket: &str,
     object: &str,
@@ -56,6 +57,7 @@ pub async fn del_opts(
     Ok(opts)
 }
 
+/// Creates options for getting an object from a bucket.
 pub async fn get_opts(
     bucket: &str,
     object: &str,
@@ -105,6 +107,7 @@ pub async fn get_opts(
     Ok(opts)
 }
 
+/// Creates options for putting an object in a bucket.
 pub async fn put_opts(
     bucket: &str,
     object: &str,
@@ -151,6 +154,7 @@ pub async fn put_opts(
     Ok(opts)
 }
 
+/// Creates options for copying an object in a bucket.
 pub async fn copy_dst_opts(
     bucket: &str,
     object: &str,
@@ -172,6 +176,7 @@ pub fn put_opts_from_headers(
     get_default_opts(headers, metadata, false)
 }
 
+/// Creates default options for getting an object from a bucket.
 pub fn get_default_opts(
     _headers: &HeaderMap<HeaderValue>,
     metadata: Option<HashMap<String, String>>,
@@ -183,6 +188,7 @@ pub fn get_default_opts(
     })
 }
 
+/// Extracts metadata from headers and returns it as a HashMap.
 pub fn extract_metadata(headers: &HeaderMap<HeaderValue>) -> HashMap<String, String> {
     let mut metadata = HashMap::new();
 
@@ -191,6 +197,7 @@ pub fn extract_metadata(headers: &HeaderMap<HeaderValue>) -> HashMap<String, Str
     metadata
 }
 
+/// Extracts metadata from headers and returns it as a HashMap.
 pub fn extract_metadata_from_mime(headers: &HeaderMap<HeaderValue>, metadata: &mut HashMap<String, String>) {
     for (k, v) in headers.iter() {
         if let Some(key) = k.as_str().strip_prefix("x-amz-meta-") {
@@ -219,7 +226,9 @@ pub fn extract_metadata_from_mime(headers: &HeaderMap<HeaderValue>, metadata: &m
         metadata.insert("content-type".to_owned(), "binary/octet-stream".to_owned());
     }
 }
+
 lazy_static! {
+    /// List of supported headers.
     static ref SUPPORTED_HEADERS: Vec<&'static str> = vec![
         "content-type",
         "cache-control",