RustFS rustfs-audit Complete Implementation with Enterprise Observability (#557)

* Initial plan * Implement core audit system with multi-target fan-out and configuration management Co-authored-by: houseme <4829346+houseme@users.noreply.github.com> * Changes before error encountered Co-authored-by: houseme <4829346+houseme@users.noreply.github.com> * Complete audit system with comprehensive observability and test coverage Co-authored-by: houseme <4829346+houseme@users.noreply.github.com> * improve code * fix * improve code * fix test * fix test * fix * add `rustfs-audit` to `rustfs` * upgrade crate version * fmt * fmt * fix --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: houseme <4829346+houseme@users.noreply.github.com> Co-authored-by: houseme <housemecn@gmail.com>
2026-08-15 01:23:12 +00:00 · 2025-09-24 08:23:46 +08:00
parent 08aeca89ef
commit 29b0935be7
45 changed files with 4139 additions and 313 deletions
@@ -0,0 +1,390 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use chrono::{DateTime, Utc};
+use rustfs_targets::EventName;
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use std::collections::HashMap;
+
+/// Trait for types that can be serialized to JSON and have a timestamp
+pub trait LogRecord {
+    /// Serialize the record to a JSON string
+    fn to_json(&self) -> String;
+    /// Get the timestamp of the record
+    fn get_timestamp(&self) -> chrono::DateTime<chrono::Utc>;
+}
+
+/// ObjectVersion represents an object version with key and versionId
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
+pub struct ObjectVersion {
+    #[serde(rename = "objectName")]
+    pub object_name: String,
+    #[serde(rename = "versionId", skip_serializing_if = "Option::is_none")]
+    pub version_id: Option<String>,
+}
+
+impl ObjectVersion {
+    /// Set the object name (chainable)
+    pub fn set_object_name(&mut self, name: String) -> &mut Self {
+        self.object_name = name;
+        self
+    }
+    /// Set the version ID (chainable)
+    pub fn set_version_id(&mut self, version_id: Option<String>) -> &mut Self {
+        self.version_id = version_id;
+        self
+    }
+}
+
+/// ApiDetails contains API information for the audit entry
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+pub struct ApiDetails {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub name: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub bucket: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub object: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub objects: Option<Vec<ObjectVersion>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub status: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub status_code: Option<i32>,
+    #[serde(rename = "rx", skip_serializing_if = "Option::is_none")]
+    pub input_bytes: Option<i64>,
+    #[serde(rename = "tx", skip_serializing_if = "Option::is_none")]
+    pub output_bytes: Option<i64>,
+    #[serde(rename = "txHeaders", skip_serializing_if = "Option::is_none")]
+    pub header_bytes: Option<i64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub time_to_first_byte: Option<String>,
+    #[serde(rename = "timeToFirstByteInNS", skip_serializing_if = "Option::is_none")]
+    pub time_to_first_byte_in_ns: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub time_to_response: Option<String>,
+    #[serde(rename = "timeToResponseInNS", skip_serializing_if = "Option::is_none")]
+    pub time_to_response_in_ns: Option<String>,
+}
+
+impl ApiDetails {
+    /// Set API name (chainable)
+    pub fn set_name(&mut self, name: Option<String>) -> &mut Self {
+        self.name = name;
+        self
+    }
+    /// Set bucket name (chainable)
+    pub fn set_bucket(&mut self, bucket: Option<String>) -> &mut Self {
+        self.bucket = bucket;
+        self
+    }
+    /// Set object name (chainable)
+    pub fn set_object(&mut self, object: Option<String>) -> &mut Self {
+        self.object = object;
+        self
+    }
+    /// Set objects list (chainable)
+    pub fn set_objects(&mut self, objects: Option<Vec<ObjectVersion>>) -> &mut Self {
+        self.objects = objects;
+        self
+    }
+    /// Set status (chainable)
+    pub fn set_status(&mut self, status: Option<String>) -> &mut Self {
+        self.status = status;
+        self
+    }
+    /// Set status code (chainable)
+    pub fn set_status_code(&mut self, code: Option<i32>) -> &mut Self {
+        self.status_code = code;
+        self
+    }
+    /// Set input bytes (chainable)
+    pub fn set_input_bytes(&mut self, bytes: Option<i64>) -> &mut Self {
+        self.input_bytes = bytes;
+        self
+    }
+    /// Set output bytes (chainable)
+    pub fn set_output_bytes(&mut self, bytes: Option<i64>) -> &mut Self {
+        self.output_bytes = bytes;
+        self
+    }
+    /// Set header bytes (chainable)
+    pub fn set_header_bytes(&mut self, bytes: Option<i64>) -> &mut Self {
+        self.header_bytes = bytes;
+        self
+    }
+    /// Set time to first byte (chainable)
+    pub fn set_time_to_first_byte(&mut self, t: Option<String>) -> &mut Self {
+        self.time_to_first_byte = t;
+        self
+    }
+    /// Set time to first byte in nanoseconds (chainable)
+    pub fn set_time_to_first_byte_in_ns(&mut self, t: Option<String>) -> &mut Self {
+        self.time_to_first_byte_in_ns = t;
+        self
+    }
+    /// Set time to response (chainable)
+    pub fn set_time_to_response(&mut self, t: Option<String>) -> &mut Self {
+        self.time_to_response = t;
+        self
+    }
+    /// Set time to response in nanoseconds (chainable)
+    pub fn set_time_to_response_in_ns(&mut self, t: Option<String>) -> &mut Self {
+        self.time_to_response_in_ns = t;
+        self
+    }
+}
+
+/// AuditEntry represents an audit log entry
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+pub struct AuditEntry {
+    pub version: String,
+    #[serde(rename = "deploymentid", skip_serializing_if = "Option::is_none")]
+    pub deployment_id: Option<String>,
+    #[serde(rename = "siteName", skip_serializing_if = "Option::is_none")]
+    pub site_name: Option<String>,
+    pub time: DateTime<Utc>,
+    pub event: EventName,
+    #[serde(rename = "type", skip_serializing_if = "Option::is_none")]
+    pub entry_type: Option<String>,
+    pub trigger: String,
+    pub api: ApiDetails,
+    #[serde(rename = "remotehost", skip_serializing_if = "Option::is_none")]
+    pub remote_host: Option<String>,
+    #[serde(rename = "requestID", skip_serializing_if = "Option::is_none")]
+    pub request_id: Option<String>,
+    #[serde(rename = "userAgent", skip_serializing_if = "Option::is_none")]
+    pub user_agent: Option<String>,
+    #[serde(rename = "requestPath", skip_serializing_if = "Option::is_none")]
+    pub req_path: Option<String>,
+    #[serde(rename = "requestHost", skip_serializing_if = "Option::is_none")]
+    pub req_host: Option<String>,
+    #[serde(rename = "requestNode", skip_serializing_if = "Option::is_none")]
+    pub req_node: Option<String>,
+    #[serde(rename = "requestClaims", skip_serializing_if = "Option::is_none")]
+    pub req_claims: Option<HashMap<String, Value>>,
+    #[serde(rename = "requestQuery", skip_serializing_if = "Option::is_none")]
+    pub req_query: Option<HashMap<String, String>>,
+    #[serde(rename = "requestHeader", skip_serializing_if = "Option::is_none")]
+    pub req_header: Option<HashMap<String, String>>,
+    #[serde(rename = "responseHeader", skip_serializing_if = "Option::is_none")]
+    pub resp_header: Option<HashMap<String, String>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub tags: Option<HashMap<String, Value>>,
+    #[serde(rename = "accessKey", skip_serializing_if = "Option::is_none")]
+    pub access_key: Option<String>,
+    #[serde(rename = "parentUser", skip_serializing_if = "Option::is_none")]
+    pub parent_user: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub error: Option<String>,
+}
+
+impl AuditEntry {
+    /// Create a new AuditEntry with required fields
+    #[allow(clippy::too_many_arguments)]
+    pub fn new(
+        version: String,
+        deployment_id: Option<String>,
+        site_name: Option<String>,
+        time: DateTime<Utc>,
+        event: EventName,
+        entry_type: Option<String>,
+        trigger: String,
+        api: ApiDetails,
+    ) -> Self {
+        AuditEntry {
+            version,
+            deployment_id,
+            site_name,
+            time,
+            event,
+            entry_type,
+            trigger,
+            api,
+            remote_host: None,
+            request_id: None,
+            user_agent: None,
+            req_path: None,
+            req_host: None,
+            req_node: None,
+            req_claims: None,
+            req_query: None,
+            req_header: None,
+            resp_header: None,
+            tags: None,
+            access_key: None,
+            parent_user: None,
+            error: None,
+        }
+    }
+
+    /// Set version (chainable)
+    pub fn set_version(&mut self, version: String) -> &mut Self {
+        self.version = version;
+        self
+    }
+    /// Set deployment ID (chainable)
+    pub fn set_deployment_id(&mut self, id: Option<String>) -> &mut Self {
+        self.deployment_id = id;
+        self
+    }
+    /// Set site name (chainable)
+    pub fn set_site_name(&mut self, name: Option<String>) -> &mut Self {
+        self.site_name = name;
+        self
+    }
+    /// Set time (chainable)
+    pub fn set_time(&mut self, time: DateTime<Utc>) -> &mut Self {
+        self.time = time;
+        self
+    }
+    /// Set event (chainable)
+    pub fn set_event(&mut self, event: EventName) -> &mut Self {
+        self.event = event;
+        self
+    }
+    /// Set entry type (chainable)
+    pub fn set_entry_type(&mut self, entry_type: Option<String>) -> &mut Self {
+        self.entry_type = entry_type;
+        self
+    }
+    /// Set trigger (chainable)
+    pub fn set_trigger(&mut self, trigger: String) -> &mut Self {
+        self.trigger = trigger;
+        self
+    }
+    /// Set API details (chainable)
+    pub fn set_api(&mut self, api: ApiDetails) -> &mut Self {
+        self.api = api;
+        self
+    }
+    /// Set remote host (chainable)
+    pub fn set_remote_host(&mut self, host: Option<String>) -> &mut Self {
+        self.remote_host = host;
+        self
+    }
+    /// Set request ID (chainable)
+    pub fn set_request_id(&mut self, id: Option<String>) -> &mut Self {
+        self.request_id = id;
+        self
+    }
+    /// Set user agent (chainable)
+    pub fn set_user_agent(&mut self, agent: Option<String>) -> &mut Self {
+        self.user_agent = agent;
+        self
+    }
+    /// Set request path (chainable)
+    pub fn set_req_path(&mut self, path: Option<String>) -> &mut Self {
+        self.req_path = path;
+        self
+    }
+    /// Set request host (chainable)
+    pub fn set_req_host(&mut self, host: Option<String>) -> &mut Self {
+        self.req_host = host;
+        self
+    }
+    /// Set request node (chainable)
+    pub fn set_req_node(&mut self, node: Option<String>) -> &mut Self {
+        self.req_node = node;
+        self
+    }
+    /// Set request claims (chainable)
+    pub fn set_req_claims(&mut self, claims: Option<HashMap<String, Value>>) -> &mut Self {
+        self.req_claims = claims;
+        self
+    }
+    /// Set request query (chainable)
+    pub fn set_req_query(&mut self, query: Option<HashMap<String, String>>) -> &mut Self {
+        self.req_query = query;
+        self
+    }
+    /// Set request header (chainable)
+    pub fn set_req_header(&mut self, header: Option<HashMap<String, String>>) -> &mut Self {
+        self.req_header = header;
+        self
+    }
+    /// Set response header (chainable)
+    pub fn set_resp_header(&mut self, header: Option<HashMap<String, String>>) -> &mut Self {
+        self.resp_header = header;
+        self
+    }
+    /// Set tags (chainable)
+    pub fn set_tags(&mut self, tags: Option<HashMap<String, Value>>) -> &mut Self {
+        self.tags = tags;
+        self
+    }
+    /// Set access key (chainable)
+    pub fn set_access_key(&mut self, key: Option<String>) -> &mut Self {
+        self.access_key = key;
+        self
+    }
+    /// Set parent user (chainable)
+    pub fn set_parent_user(&mut self, user: Option<String>) -> &mut Self {
+        self.parent_user = user;
+        self
+    }
+    /// Set error message (chainable)
+    pub fn set_error(&mut self, error: Option<String>) -> &mut Self {
+        self.error = error;
+        self
+    }
+
+    /// Build AuditEntry from context or parameters (example, can be extended)
+    pub fn from_context(
+        version: String,
+        deployment_id: Option<String>,
+        time: DateTime<Utc>,
+        event: EventName,
+        trigger: String,
+        api: ApiDetails,
+        tags: Option<HashMap<String, Value>>,
+    ) -> Self {
+        AuditEntry {
+            version,
+            deployment_id,
+            site_name: None,
+            time,
+            event,
+            entry_type: None,
+            trigger,
+            api,
+            remote_host: None,
+            request_id: None,
+            user_agent: None,
+            req_path: None,
+            req_host: None,
+            req_node: None,
+            req_claims: None,
+            req_query: None,
+            req_header: None,
+            resp_header: None,
+            tags,
+            access_key: None,
+            parent_user: None,
+            error: None,
+        }
+    }
+}
+
+impl LogRecord for AuditEntry {
+    /// Serialize AuditEntry to JSON string
+    fn to_json(&self) -> String {
+        serde_json::to_string(self).unwrap_or_else(|_| String::from("{}"))
+    }
+    /// Get the timestamp of the audit entry
+    fn get_timestamp(&self) -> DateTime<Utc> {
+        self.time
+    }
+}
@@ -0,0 +1,55 @@
+//  Copyright 2024 RustFS Team
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+
+use thiserror::Error;
+
+/// Result type for audit operations
+pub type AuditResult<T> = Result<T, AuditError>;
+
+/// Errors that can occur during audit operations
+#[derive(Error, Debug)]
+pub enum AuditError {
+    #[error("Configuration error: {0}")]
+    Configuration(String),
+
+    #[error("config not loaded")]
+    ConfigNotLoaded,
+
+    #[error("Target error: {0}")]
+    Target(#[from] rustfs_targets::TargetError),
+
+    #[error("System not initialized: {0}")]
+    NotInitialized(String),
+
+    #[error("System already initialized")]
+    AlreadyInitialized,
+
+    #[error("Failed to save configuration: {0}")]
+    SaveConfig(String),
+
+    #[error("Failed to load configuration: {0}")]
+    LoadConfig(String),
+
+    #[error("Serialization error: {0}")]
+    Serialization(#[from] serde_json::Error),
+
+    #[error("I/O error: {0}")]
+    Io(#[from] std::io::Error),
+
+    #[error("Join error: {0}")]
+    Join(#[from] tokio::task::JoinError),
+
+    #[error("Server storage not initialized: {0}")]
+    ServerNotInitialized(String),
+}
@@ -0,0 +1,127 @@
+//  Copyright 2024 RustFS Team
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+
+use crate::AuditEntry;
+use crate::AuditResult;
+use crate::AuditSystem;
+use once_cell::sync::OnceCell;
+use rustfs_ecstore::config::Config;
+use std::sync::Arc;
+use tracing::{error, warn};
+
+/// Global audit system instance
+static AUDIT_SYSTEM: OnceCell<Arc<AuditSystem>> = OnceCell::new();
+
+/// Initialize the global audit system
+pub fn init_audit_system() -> Arc<AuditSystem> {
+    AUDIT_SYSTEM.get_or_init(|| Arc::new(AuditSystem::new())).clone()
+}
+
+/// Get the global audit system instance
+pub fn audit_system() -> Option<Arc<AuditSystem>> {
+    AUDIT_SYSTEM.get().cloned()
+}
+
+/// Start the global audit system with configuration
+pub async fn start_audit_system(config: Config) -> AuditResult<()> {
+    let system = init_audit_system();
+    system.start(config).await
+}
+
+/// Stop the global audit system
+pub async fn stop_audit_system() -> AuditResult<()> {
+    if let Some(system) = audit_system() {
+        system.close().await
+    } else {
+        warn!("Audit system not initialized, cannot stop");
+        Ok(())
+    }
+}
+
+/// Pause the global audit system
+pub async fn pause_audit_system() -> AuditResult<()> {
+    if let Some(system) = audit_system() {
+        system.pause().await
+    } else {
+        warn!("Audit system not initialized, cannot pause");
+        Ok(())
+    }
+}
+
+/// Resume the global audit system
+pub async fn resume_audit_system() -> AuditResult<()> {
+    if let Some(system) = audit_system() {
+        system.resume().await
+    } else {
+        warn!("Audit system not initialized, cannot resume");
+        Ok(())
+    }
+}
+
+/// Dispatch an audit log entry to all targets
+pub async fn dispatch_audit_log(entry: Arc<AuditEntry>) -> AuditResult<()> {
+    if let Some(system) = audit_system() {
+        if system.is_running().await {
+            system.dispatch(entry).await
+        } else {
+            // System not running, just drop the log entry without error
+            Ok(())
+        }
+    } else {
+        // System not initialized, just drop the log entry without error
+        Ok(())
+    }
+}
+
+/// Reload the global audit system configuration
+pub async fn reload_audit_config(config: Config) -> AuditResult<()> {
+    if let Some(system) = audit_system() {
+        system.reload_config(config).await
+    } else {
+        warn!("Audit system not initialized, cannot reload config");
+        Ok(())
+    }
+}
+
+/// Check if the global audit system is running
+pub async fn is_audit_system_running() -> bool {
+    if let Some(system) = audit_system() {
+        system.is_running().await
+    } else {
+        false
+    }
+}
+
+/// AuditLogger singleton for easy access
+pub struct AuditLogger;
+
+impl AuditLogger {
+    /// Log an audit entry
+    pub async fn log(entry: AuditEntry) {
+        if let Err(e) = dispatch_audit_log(Arc::new(entry)).await {
+            error!(error = %e, "Failed to dispatch audit log entry");
+        }
+    }
+
+    /// Check if audit logging is enabled
+    pub async fn is_enabled() -> bool {
+        is_audit_system_running().await
+    }
+
+    /// Get singleton instance
+    pub fn instance() -> &'static Self {
+        static INSTANCE: AuditLogger = AuditLogger;
+        &INSTANCE
+    }
+}
@@ -0,0 +1,33 @@
+//  Copyright 2024 RustFS Team
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+
+//! RustFS Audit System
+//!
+//! This crate provides a comprehensive audit logging system with multi-target fan-out capabilities,
+//! configuration management, and hot reload functionality. It is modeled after the notify system
+//! but specifically designed for audit logging requirements.
+
+pub mod entity;
+pub mod error;
+pub mod global;
+pub mod observability;
+pub mod registry;
+pub mod system;
+
+pub use entity::{ApiDetails, AuditEntry, LogRecord, ObjectVersion};
+pub use error::{AuditError, AuditResult};
+pub use global::*;
+pub use observability::{AuditMetrics, AuditMetricsReport, PerformanceValidation};
+pub use registry::AuditRegistry;
+pub use system::AuditSystem;
@@ -0,0 +1,368 @@
+//  Copyright 2024 RustFS Team
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+
+//! Observability and metrics for the audit system
+//!
+//! This module provides comprehensive observability features including:
+//! - Performance metrics (EPS, latency)
+//! - Target health monitoring  
+//! - Configuration change tracking
+//! - Error rate monitoring
+//! - Queue depth monitoring
+
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::{Duration, Instant};
+use tokio::sync::RwLock;
+use tracing::info;
+
+/// Metrics collector for audit system observability
+#[derive(Debug)]
+pub struct AuditMetrics {
+    // Performance metrics
+    total_events_processed: AtomicU64,
+    total_events_failed: AtomicU64,
+    total_dispatch_time_ns: AtomicU64,
+
+    // Target metrics
+    target_success_count: AtomicU64,
+    target_failure_count: AtomicU64,
+
+    // System metrics
+    config_reload_count: AtomicU64,
+    system_start_count: AtomicU64,
+
+    // Performance tracking
+    last_reset_time: Arc<RwLock<Instant>>,
+}
+
+impl Default for AuditMetrics {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl AuditMetrics {
+    /// Creates a new metrics collector
+    pub fn new() -> Self {
+        Self {
+            total_events_processed: AtomicU64::new(0),
+            total_events_failed: AtomicU64::new(0),
+            total_dispatch_time_ns: AtomicU64::new(0),
+            target_success_count: AtomicU64::new(0),
+            target_failure_count: AtomicU64::new(0),
+            config_reload_count: AtomicU64::new(0),
+            system_start_count: AtomicU64::new(0),
+            last_reset_time: Arc::new(RwLock::new(Instant::now())),
+        }
+    }
+
+    /// Records a successful event dispatch
+    pub fn record_event_success(&self, dispatch_time: Duration) {
+        self.total_events_processed.fetch_add(1, Ordering::Relaxed);
+        self.total_dispatch_time_ns
+            .fetch_add(dispatch_time.as_nanos() as u64, Ordering::Relaxed);
+    }
+
+    /// Records a failed event dispatch
+    pub fn record_event_failure(&self, dispatch_time: Duration) {
+        self.total_events_failed.fetch_add(1, Ordering::Relaxed);
+        self.total_dispatch_time_ns
+            .fetch_add(dispatch_time.as_nanos() as u64, Ordering::Relaxed);
+    }
+
+    /// Records a successful target operation
+    pub fn record_target_success(&self) {
+        self.target_success_count.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Records a failed target operation
+    pub fn record_target_failure(&self) {
+        self.target_failure_count.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Records a configuration reload
+    pub fn record_config_reload(&self) {
+        self.config_reload_count.fetch_add(1, Ordering::Relaxed);
+        info!("Audit configuration reloaded");
+    }
+
+    /// Records a system start
+    pub fn record_system_start(&self) {
+        self.system_start_count.fetch_add(1, Ordering::Relaxed);
+        info!("Audit system started");
+    }
+
+    /// Gets the current events per second (EPS)
+    pub async fn get_events_per_second(&self) -> f64 {
+        let reset_time = *self.last_reset_time.read().await;
+        let elapsed = reset_time.elapsed();
+        let total_events = self.total_events_processed.load(Ordering::Relaxed) + self.total_events_failed.load(Ordering::Relaxed);
+
+        if elapsed.as_secs_f64() > 0.0 {
+            total_events as f64 / elapsed.as_secs_f64()
+        } else {
+            0.0
+        }
+    }
+
+    /// Gets the average dispatch latency in milliseconds
+    pub fn get_average_latency_ms(&self) -> f64 {
+        let total_events = self.total_events_processed.load(Ordering::Relaxed) + self.total_events_failed.load(Ordering::Relaxed);
+        let total_time_ns = self.total_dispatch_time_ns.load(Ordering::Relaxed);
+
+        if total_events > 0 {
+            (total_time_ns as f64 / total_events as f64) / 1_000_000.0 // Convert ns to ms
+        } else {
+            0.0
+        }
+    }
+
+    /// Gets the error rate as a percentage
+    pub fn get_error_rate(&self) -> f64 {
+        let total_events = self.total_events_processed.load(Ordering::Relaxed) + self.total_events_failed.load(Ordering::Relaxed);
+        let failed_events = self.total_events_failed.load(Ordering::Relaxed);
+
+        if total_events > 0 {
+            (failed_events as f64 / total_events as f64) * 100.0
+        } else {
+            0.0
+        }
+    }
+
+    /// Gets target success rate as a percentage
+    pub fn get_target_success_rate(&self) -> f64 {
+        let total_ops = self.target_success_count.load(Ordering::Relaxed) + self.target_failure_count.load(Ordering::Relaxed);
+        let success_ops = self.target_success_count.load(Ordering::Relaxed);
+
+        if total_ops > 0 {
+            (success_ops as f64 / total_ops as f64) * 100.0
+        } else {
+            100.0 // No operations = 100% success rate
+        }
+    }
+
+    /// Resets all metrics and timing
+    pub async fn reset(&self) {
+        self.total_events_processed.store(0, Ordering::Relaxed);
+        self.total_events_failed.store(0, Ordering::Relaxed);
+        self.total_dispatch_time_ns.store(0, Ordering::Relaxed);
+        self.target_success_count.store(0, Ordering::Relaxed);
+        self.target_failure_count.store(0, Ordering::Relaxed);
+        self.config_reload_count.store(0, Ordering::Relaxed);
+        self.system_start_count.store(0, Ordering::Relaxed);
+
+        let mut reset_time = self.last_reset_time.write().await;
+        *reset_time = Instant::now();
+
+        info!("Audit metrics reset");
+    }
+
+    /// Generates a comprehensive metrics report
+    pub async fn generate_report(&self) -> AuditMetricsReport {
+        AuditMetricsReport {
+            events_per_second: self.get_events_per_second().await,
+            average_latency_ms: self.get_average_latency_ms(),
+            error_rate_percent: self.get_error_rate(),
+            target_success_rate_percent: self.get_target_success_rate(),
+            total_events_processed: self.total_events_processed.load(Ordering::Relaxed),
+            total_events_failed: self.total_events_failed.load(Ordering::Relaxed),
+            config_reload_count: self.config_reload_count.load(Ordering::Relaxed),
+            system_start_count: self.system_start_count.load(Ordering::Relaxed),
+        }
+    }
+
+    /// Validates performance requirements
+    pub async fn validate_performance_requirements(&self) -> PerformanceValidation {
+        let eps = self.get_events_per_second().await;
+        let avg_latency_ms = self.get_average_latency_ms();
+        let error_rate = self.get_error_rate();
+
+        let mut validation = PerformanceValidation {
+            meets_eps_requirement: eps >= 3000.0,
+            meets_latency_requirement: avg_latency_ms <= 30.0,
+            meets_error_rate_requirement: error_rate <= 1.0, // Less than 1% error rate
+            current_eps: eps,
+            current_latency_ms: avg_latency_ms,
+            current_error_rate: error_rate,
+            recommendations: Vec::new(),
+        };
+
+        // Generate recommendations
+        if !validation.meets_eps_requirement {
+            validation.recommendations.push(format!(
+                "EPS ({:.0}) is below requirement (3000). Consider optimizing target dispatch or adding more target instances.",
+                eps
+            ));
+        }
+
+        if !validation.meets_latency_requirement {
+            validation.recommendations.push(format!(
+                "Average latency ({:.2}ms) exceeds requirement (30ms). Consider optimizing target responses or increasing timeout values.",
+                avg_latency_ms
+            ));
+        }
+
+        if !validation.meets_error_rate_requirement {
+            validation.recommendations.push(format!(
+                "Error rate ({:.2}%) exceeds recommendation (1%). Check target connectivity and configuration.",
+                error_rate
+            ));
+        }
+
+        if validation.meets_eps_requirement && validation.meets_latency_requirement && validation.meets_error_rate_requirement {
+            validation
+                .recommendations
+                .push("All performance requirements are met.".to_string());
+        }
+
+        validation
+    }
+}
+
+/// Comprehensive metrics report
+#[derive(Debug, Clone)]
+pub struct AuditMetricsReport {
+    pub events_per_second: f64,
+    pub average_latency_ms: f64,
+    pub error_rate_percent: f64,
+    pub target_success_rate_percent: f64,
+    pub total_events_processed: u64,
+    pub total_events_failed: u64,
+    pub config_reload_count: u64,
+    pub system_start_count: u64,
+}
+
+impl AuditMetricsReport {
+    /// Formats the report as a human-readable string
+    pub fn format(&self) -> String {
+        format!(
+            "Audit System Metrics Report:\n\
+             Events per Second: {:.2}\n\
+             Average Latency: {:.2}ms\n\
+             Error Rate: {:.2}%\n\
+             Target Success Rate: {:.2}%\n\
+             Total Events Processed: {}\n\
+             Total Events Failed: {}\n\
+             Configuration Reloads: {}\n\
+             System Starts: {}",
+            self.events_per_second,
+            self.average_latency_ms,
+            self.error_rate_percent,
+            self.target_success_rate_percent,
+            self.total_events_processed,
+            self.total_events_failed,
+            self.config_reload_count,
+            self.system_start_count
+        )
+    }
+}
+
+/// Performance validation results
+#[derive(Debug, Clone)]
+pub struct PerformanceValidation {
+    pub meets_eps_requirement: bool,
+    pub meets_latency_requirement: bool,
+    pub meets_error_rate_requirement: bool,
+    pub current_eps: f64,
+    pub current_latency_ms: f64,
+    pub current_error_rate: f64,
+    pub recommendations: Vec<String>,
+}
+
+impl PerformanceValidation {
+    /// Checks if all performance requirements are met
+    pub fn all_requirements_met(&self) -> bool {
+        self.meets_eps_requirement && self.meets_latency_requirement && self.meets_error_rate_requirement
+    }
+
+    /// Formats the validation as a human-readable string
+    pub fn format(&self) -> String {
+        let status = if self.all_requirements_met() { "✅ PASS" } else { "❌ FAIL" };
+
+        let mut result = format!(
+            "Performance Requirements Validation: {}\n\
+             EPS Requirement (≥3000): {} ({:.2})\n\
+             Latency Requirement (≤30ms): {} ({:.2}ms)\n\
+             Error Rate Requirement (≤1%): {} ({:.2}%)\n\
+             \nRecommendations:",
+            status,
+            if self.meets_eps_requirement { "✅" } else { "❌" },
+            self.current_eps,
+            if self.meets_latency_requirement { "✅" } else { "❌" },
+            self.current_latency_ms,
+            if self.meets_error_rate_requirement { "✅" } else { "❌" },
+            self.current_error_rate
+        );
+
+        for rec in &self.recommendations {
+            result.push_str(&format!("\n• {}", rec));
+        }
+
+        result
+    }
+}
+
+/// Global metrics instance
+static GLOBAL_METRICS: once_cell::sync::OnceCell<Arc<AuditMetrics>> = once_cell::sync::OnceCell::new();
+
+/// Get or initialize the global metrics instance
+pub fn global_metrics() -> Arc<AuditMetrics> {
+    GLOBAL_METRICS.get_or_init(|| Arc::new(AuditMetrics::new())).clone()
+}
+
+/// Record a successful audit event dispatch
+pub fn record_audit_success(dispatch_time: Duration) {
+    global_metrics().record_event_success(dispatch_time);
+}
+
+/// Record a failed audit event dispatch
+pub fn record_audit_failure(dispatch_time: Duration) {
+    global_metrics().record_event_failure(dispatch_time);
+}
+
+/// Record a successful target operation
+pub fn record_target_success() {
+    global_metrics().record_target_success();
+}
+
+/// Record a failed target operation
+pub fn record_target_failure() {
+    global_metrics().record_target_failure();
+}
+
+/// Record a configuration reload
+pub fn record_config_reload() {
+    global_metrics().record_config_reload();
+}
+
+/// Record a system start
+pub fn record_system_start() {
+    global_metrics().record_system_start();
+}
+
+/// Get the current metrics report
+pub async fn get_metrics_report() -> AuditMetricsReport {
+    global_metrics().generate_report().await
+}
+
+/// Validate performance requirements
+pub async fn validate_performance() -> PerformanceValidation {
+    global_metrics().validate_performance_requirements().await
+}
+
+/// Reset all metrics
+pub async fn reset_metrics() {
+    global_metrics().reset().await;
+}
@@ -0,0 +1,436 @@
+//  Copyright 2024 RustFS Team
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+
+use crate::AuditEntry;
+use crate::{AuditError, AuditResult};
+use futures::StreamExt;
+use futures::stream::FuturesUnordered;
+use rustfs_config::audit::AUDIT_ROUTE_PREFIX;
+use rustfs_config::{
+    DEFAULT_DELIMITER, ENABLE_KEY, ENV_PREFIX, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD, MQTT_QOS, MQTT_QUEUE_DIR,
+    MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TOPIC, MQTT_USERNAME, WEBHOOK_AUTH_TOKEN, WEBHOOK_BATCH_SIZE,
+    WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_HTTP_TIMEOUT, WEBHOOK_MAX_RETRY, WEBHOOK_QUEUE_DIR,
+    WEBHOOK_QUEUE_LIMIT, WEBHOOK_RETRY_INTERVAL,
+};
+use rustfs_ecstore::config::{Config, KVS};
+use rustfs_targets::target::{ChannelTargetType, TargetType, mqtt::MQTTArgs, webhook::WebhookArgs};
+use rustfs_targets::{Target, TargetError};
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+use std::time::Duration;
+use tracing::{debug, error, info, warn};
+use url::Url;
+
+/// Registry for managing audit targets
+pub struct AuditRegistry {
+    /// Storage for created targets
+    targets: HashMap<String, Box<dyn Target<AuditEntry> + Send + Sync>>,
+}
+
+impl Default for AuditRegistry {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl AuditRegistry {
+    /// Creates a new AuditRegistry
+    pub fn new() -> Self {
+        Self { targets: HashMap::new() }
+    }
+
+    /// Creates all audit targets from system configuration and environment variables.
+    /// This method processes the creation of each target concurrently as follows:
+    /// 1. Iterate through supported target types (webhook, mqtt).
+    /// 2. For each type, resolve its configuration from file and environment variables.
+    /// 3. Identify all target instance IDs that need to be created.
+    /// 4. Merge configurations with precedence: ENV > file instance > file default.
+    /// 5. Create async tasks for enabled instances.
+    /// 6. Execute tasks concurrently and collect successful targets.
+    /// 7. Persist successful configurations back to system storage.
+    pub async fn create_targets_from_config(
+        &mut self,
+        config: &Config,
+    ) -> AuditResult<Vec<Box<dyn Target<AuditEntry> + Send + Sync>>> {
+        // Collect only environment variables with the relevant prefix to reduce memory usage
+        let all_env: Vec<(String, String)> = std::env::vars().filter(|(key, _)| key.starts_with(ENV_PREFIX)).collect();
+
+        // A collection of asynchronous tasks for concurrently executing target creation
+        let mut tasks = FuturesUnordered::new();
+        let mut final_config = config.clone();
+
+        // Supported target types for audit
+        let target_types = vec![ChannelTargetType::Webhook.as_str(), ChannelTargetType::Mqtt.as_str()];
+
+        // 1. Traverse all target types and process them
+        for target_type in target_types {
+            let span = tracing::Span::current();
+            span.record("target_type", target_type);
+            info!(target_type = %target_type, "Starting audit target type processing");
+
+            // 2. Prepare the configuration source
+            let section_name = format!("{AUDIT_ROUTE_PREFIX}{target_type}");
+            let file_configs = config.0.get(&section_name).cloned().unwrap_or_default();
+            let default_cfg = file_configs.get(DEFAULT_DELIMITER).cloned().unwrap_or_default();
+            debug!(?default_cfg, "Retrieved default configuration");
+
+            // Get valid fields for the target type
+            let valid_fields = match target_type {
+                "webhook" => get_webhook_valid_fields(),
+                "mqtt" => get_mqtt_valid_fields(),
+                _ => {
+                    warn!(target_type = %target_type, "Unknown target type, skipping");
+                    continue;
+                }
+            };
+            debug!(?valid_fields, "Retrieved valid configuration fields");
+
+            // 3. Resolve instance IDs and configuration overrides from environment variables
+            let mut instance_ids_from_env = HashSet::new();
+            let mut env_overrides: HashMap<String, HashMap<String, String>> = HashMap::new();
+
+            for (env_key, env_value) in &all_env {
+                let audit_prefix = format!("{ENV_PREFIX}AUDIT_{}", target_type.to_uppercase());
+                if !env_key.starts_with(&audit_prefix) {
+                    continue;
+                }
+
+                let suffix = &env_key[audit_prefix.len()..];
+                if suffix.is_empty() {
+                    continue;
+                }
+
+                // Parse field and instance from suffix (FIELD_INSTANCE or FIELD)
+                let (field_name, instance_id) = if let Some(last_underscore) = suffix.rfind('_') {
+                    let potential_field = &suffix[1..last_underscore]; // Skip leading _
+                    let potential_instance = &suffix[last_underscore + 1..];
+
+                    // Check if the part before the last underscore is a valid field
+                    if valid_fields.contains(&potential_field.to_lowercase()) {
+                        (potential_field.to_lowercase(), potential_instance.to_lowercase())
+                    } else {
+                        // Treat the entire suffix as field name with default instance
+                        (suffix[1..].to_lowercase(), DEFAULT_DELIMITER.to_string())
+                    }
+                } else {
+                    // No underscore, treat as field with default instance
+                    (suffix[1..].to_lowercase(), DEFAULT_DELIMITER.to_string())
+                };
+
+                if valid_fields.contains(&field_name) {
+                    if instance_id != DEFAULT_DELIMITER {
+                        instance_ids_from_env.insert(instance_id.clone());
+                    }
+                    env_overrides
+                        .entry(instance_id)
+                        .or_default()
+                        .insert(field_name, env_value.clone());
+                } else {
+                    debug!(
+                        env_key = %env_key,
+                        field_name = %field_name,
+                        "Ignoring environment variable field not found in valid fields for target type {}",
+                        target_type
+                    );
+                }
+            }
+            debug!(?env_overrides, "Completed environment variable analysis");
+
+            // 4. Determine all instance IDs that need to be processed
+            let mut all_instance_ids: HashSet<String> =
+                file_configs.keys().filter(|k| *k != DEFAULT_DELIMITER).cloned().collect();
+            all_instance_ids.extend(instance_ids_from_env);
+            debug!(?all_instance_ids, "Determined all instance IDs");
+
+            // 5. Merge configurations and create tasks for each instance
+            for id in all_instance_ids {
+                // 5.1. Merge configuration, priority: Environment variables > File instance > File default
+                let mut merged_config = default_cfg.clone();
+
+                // Apply file instance configuration if available
+                if let Some(file_instance_cfg) = file_configs.get(&id) {
+                    merged_config.extend(file_instance_cfg.clone());
+                }
+
+                // Apply environment variable overrides
+                if let Some(env_instance_cfg) = env_overrides.get(&id) {
+                    let mut kvs_from_env = KVS::new();
+                    for (k, v) in env_instance_cfg {
+                        kvs_from_env.insert(k.clone(), v.clone());
+                    }
+                    merged_config.extend(kvs_from_env);
+                }
+                debug!(instance_id = %id, ?merged_config, "Completed configuration merge");
+
+                // 5.2. Check if the instance is enabled
+                let enabled = merged_config
+                    .lookup(ENABLE_KEY)
+                    .map(|v| parse_enable_value(&v))
+                    .unwrap_or(false);
+
+                if enabled {
+                    info!(instance_id = %id, "Creating audit target");
+
+                    // Create task for concurrent execution
+                    let target_type_clone = target_type.to_string();
+                    let id_clone = id.clone();
+                    let merged_config_arc = Arc::new(merged_config.clone());
+                    let final_config_arc = Arc::new(final_config.clone());
+
+                    let task = tokio::spawn(async move {
+                        let result = create_audit_target(&target_type_clone, &id_clone, &merged_config_arc).await;
+                        (target_type_clone, id_clone, result, final_config_arc)
+                    });
+
+                    tasks.push(task);
+
+                    // Update final config with successful instance
+                    final_config
+                        .0
+                        .entry(section_name.clone())
+                        .or_default()
+                        .insert(id, merged_config);
+                } else {
+                    info!(instance_id = %id, "Skipping disabled audit target, will be removed from final configuration");
+                    // Remove disabled target from final configuration
+                    final_config.0.entry(section_name.clone()).or_default().remove(&id);
+                }
+            }
+        }
+
+        // 6. Concurrently execute all creation tasks and collect results
+        let mut successful_targets = Vec::new();
+
+        while let Some(task_result) = tasks.next().await {
+            match task_result {
+                Ok((target_type, id, result, _final_config)) => match result {
+                    Ok(target) => {
+                        info!(target_type = %target_type, instance_id = %id, "Created audit target successfully");
+                        successful_targets.push(target);
+                    }
+                    Err(e) => {
+                        error!(target_type = %target_type, instance_id = %id, error = %e, "Failed to create audit target");
+                    }
+                },
+                Err(e) => {
+                    error!(error = %e, "Task execution failed");
+                }
+            }
+        }
+
+        // 7. Save the new configuration to the system
+        let Some(store) = rustfs_ecstore::new_object_layer_fn() else {
+            return Err(AuditError::ServerNotInitialized(
+                "Failed to save target configuration: server storage not initialized".to_string(),
+            ));
+        };
+
+        match rustfs_ecstore::config::com::save_server_config(store, &final_config).await {
+            Ok(_) => info!("New audit configuration saved to system successfully"),
+            Err(e) => {
+                error!(error = %e, "Failed to save new audit configuration");
+                return Err(AuditError::SaveConfig(e.to_string()));
+            }
+        }
+
+        Ok(successful_targets)
+    }
+
+    /// Adds a target to the registry
+    pub fn add_target(&mut self, id: String, target: Box<dyn Target<AuditEntry> + Send + Sync>) {
+        self.targets.insert(id, target);
+    }
+
+    /// Removes a target from the registry
+    pub fn remove_target(&mut self, id: &str) -> Option<Box<dyn Target<AuditEntry> + Send + Sync>> {
+        self.targets.remove(id)
+    }
+
+    /// Gets a target from the registry
+    pub fn get_target(&self, id: &str) -> Option<&(dyn Target<AuditEntry> + Send + Sync)> {
+        self.targets.get(id).map(|t| t.as_ref())
+    }
+
+    /// Lists all target IDs
+    pub fn list_targets(&self) -> Vec<String> {
+        self.targets.keys().cloned().collect()
+    }
+
+    /// Closes all targets and clears the registry
+    pub async fn close_all(&mut self) -> AuditResult<()> {
+        let mut errors = Vec::new();
+
+        for (id, target) in self.targets.drain() {
+            if let Err(e) = target.close().await {
+                error!(target_id = %id, error = %e, "Failed to close audit target");
+                errors.push(e);
+            }
+        }
+
+        if !errors.is_empty() {
+            return Err(AuditError::Target(errors.into_iter().next().unwrap()));
+        }
+
+        Ok(())
+    }
+}
+
+/// Creates an audit target based on type and configuration
+async fn create_audit_target(
+    target_type: &str,
+    id: &str,
+    config: &KVS,
+) -> Result<Box<dyn Target<AuditEntry> + Send + Sync>, TargetError> {
+    match target_type {
+        val if val == ChannelTargetType::Webhook.as_str() => {
+            let args = parse_webhook_args(id, config)?;
+            let target = rustfs_targets::target::webhook::WebhookTarget::new(id.to_string(), args)?;
+            Ok(Box::new(target))
+        }
+        val if val == ChannelTargetType::Mqtt.as_str() => {
+            let args = parse_mqtt_args(id, config)?;
+            let target = rustfs_targets::target::mqtt::MQTTTarget::new(id.to_string(), args)?;
+            Ok(Box::new(target))
+        }
+        _ => Err(TargetError::Configuration(format!("Unknown target type: {}", target_type))),
+    }
+}
+
+/// Gets valid field names for webhook configuration
+fn get_webhook_valid_fields() -> HashSet<String> {
+    vec![
+        ENABLE_KEY.to_string(),
+        WEBHOOK_ENDPOINT.to_string(),
+        WEBHOOK_AUTH_TOKEN.to_string(),
+        WEBHOOK_CLIENT_CERT.to_string(),
+        WEBHOOK_CLIENT_KEY.to_string(),
+        WEBHOOK_BATCH_SIZE.to_string(),
+        WEBHOOK_QUEUE_LIMIT.to_string(),
+        WEBHOOK_QUEUE_DIR.to_string(),
+        WEBHOOK_MAX_RETRY.to_string(),
+        WEBHOOK_RETRY_INTERVAL.to_string(),
+        WEBHOOK_HTTP_TIMEOUT.to_string(),
+    ]
+    .into_iter()
+    .collect()
+}
+
+/// Gets valid field names for MQTT configuration
+fn get_mqtt_valid_fields() -> HashSet<String> {
+    vec![
+        ENABLE_KEY.to_string(),
+        MQTT_BROKER.to_string(),
+        MQTT_TOPIC.to_string(),
+        MQTT_USERNAME.to_string(),
+        MQTT_PASSWORD.to_string(),
+        MQTT_QOS.to_string(),
+        MQTT_KEEP_ALIVE_INTERVAL.to_string(),
+        MQTT_RECONNECT_INTERVAL.to_string(),
+        MQTT_QUEUE_DIR.to_string(),
+        MQTT_QUEUE_LIMIT.to_string(),
+    ]
+    .into_iter()
+    .collect()
+}
+
+/// Parses webhook arguments from KVS configuration
+fn parse_webhook_args(_id: &str, config: &KVS) -> Result<WebhookArgs, TargetError> {
+    let endpoint = config
+        .lookup(WEBHOOK_ENDPOINT)
+        .filter(|s| !s.is_empty())
+        .ok_or_else(|| TargetError::Configuration("webhook endpoint is required".to_string()))?;
+
+    let endpoint_url =
+        Url::parse(&endpoint).map_err(|e| TargetError::Configuration(format!("invalid webhook endpoint URL: {}", e)))?;
+
+    let args = WebhookArgs {
+        enable: true, // Already validated as enabled
+        endpoint: endpoint_url,
+        auth_token: config.lookup(WEBHOOK_AUTH_TOKEN).unwrap_or_default(),
+        queue_dir: config.lookup(WEBHOOK_QUEUE_DIR).unwrap_or_default(),
+        queue_limit: config
+            .lookup(WEBHOOK_QUEUE_LIMIT)
+            .and_then(|s| s.parse().ok())
+            .unwrap_or(100000),
+        client_cert: config.lookup(WEBHOOK_CLIENT_CERT).unwrap_or_default(),
+        client_key: config.lookup(WEBHOOK_CLIENT_KEY).unwrap_or_default(),
+        target_type: TargetType::AuditLog,
+    };
+
+    args.validate()?;
+    Ok(args)
+}
+
+/// Parses MQTT arguments from KVS configuration  
+fn parse_mqtt_args(_id: &str, config: &KVS) -> Result<MQTTArgs, TargetError> {
+    let broker = config
+        .lookup(MQTT_BROKER)
+        .filter(|s| !s.is_empty())
+        .ok_or_else(|| TargetError::Configuration("MQTT broker is required".to_string()))?;
+
+    let broker_url = Url::parse(&broker).map_err(|e| TargetError::Configuration(format!("invalid MQTT broker URL: {}", e)))?;
+
+    let topic = config
+        .lookup(MQTT_TOPIC)
+        .filter(|s| !s.is_empty())
+        .ok_or_else(|| TargetError::Configuration("MQTT topic is required".to_string()))?;
+
+    let qos = config
+        .lookup(MQTT_QOS)
+        .and_then(|s| s.parse::<u8>().ok())
+        .and_then(|q| match q {
+            0 => Some(rumqttc::QoS::AtMostOnce),
+            1 => Some(rumqttc::QoS::AtLeastOnce),
+            2 => Some(rumqttc::QoS::ExactlyOnce),
+            _ => None,
+        })
+        .unwrap_or(rumqttc::QoS::AtLeastOnce);
+
+    let args = MQTTArgs {
+        enable: true, // Already validated as enabled
+        broker: broker_url,
+        topic,
+        qos,
+        username: config.lookup(MQTT_USERNAME).unwrap_or_default(),
+        password: config.lookup(MQTT_PASSWORD).unwrap_or_default(),
+        max_reconnect_interval: parse_duration(&config.lookup(MQTT_RECONNECT_INTERVAL).unwrap_or_else(|| "5s".to_string()))
+            .unwrap_or(Duration::from_secs(5)),
+        keep_alive: parse_duration(&config.lookup(MQTT_KEEP_ALIVE_INTERVAL).unwrap_or_else(|| "60s".to_string()))
+            .unwrap_or(Duration::from_secs(60)),
+        queue_dir: config.lookup(MQTT_QUEUE_DIR).unwrap_or_default(),
+        queue_limit: config.lookup(MQTT_QUEUE_LIMIT).and_then(|s| s.parse().ok()).unwrap_or(100000),
+        target_type: TargetType::AuditLog,
+    };
+
+    args.validate()?;
+    Ok(args)
+}
+
+/// Parses enable value from string
+fn parse_enable_value(value: &str) -> bool {
+    matches!(value.to_lowercase().as_str(), "1" | "on" | "true" | "yes")
+}
+
+/// Parses duration from string (e.g., "3s", "5m")
+fn parse_duration(s: &str) -> Option<Duration> {
+    if let Some(stripped) = s.strip_suffix('s') {
+        stripped.parse::<u64>().ok().map(Duration::from_secs)
+    } else if let Some(stripped) = s.strip_suffix('m') {
+        stripped.parse::<u64>().ok().map(|m| Duration::from_secs(m * 60))
+    } else if let Some(stripped) = s.strip_suffix("ms") {
+        stripped.parse::<u64>().ok().map(Duration::from_millis)
+    } else {
+        s.parse::<u64>().ok().map(Duration::from_secs)
+    }
+}
@@ -0,0 +1,601 @@
+//  Copyright 2024 RustFS Team
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+
+use crate::AuditEntry;
+use crate::AuditRegistry;
+use crate::observability;
+use crate::{AuditError, AuditResult};
+use rustfs_ecstore::config::Config;
+use rustfs_targets::store::{Key, Store};
+use rustfs_targets::target::EntityTarget;
+use rustfs_targets::{StoreError, Target, TargetError};
+use std::sync::Arc;
+use tokio::sync::{Mutex, RwLock};
+use tracing::{error, info, warn};
+
+/// State of the audit system
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum AuditSystemState {
+    Stopped,
+    Starting,
+    Running,
+    Paused,
+    Stopping,
+}
+
+/// Main audit system that manages target lifecycle and audit log dispatch
+#[derive(Clone)]
+pub struct AuditSystem {
+    registry: Arc<Mutex<AuditRegistry>>,
+    state: Arc<RwLock<AuditSystemState>>,
+    config: Arc<RwLock<Option<Config>>>,
+}
+
+impl Default for AuditSystem {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl AuditSystem {
+    /// Creates a new audit system
+    pub fn new() -> Self {
+        Self {
+            registry: Arc::new(Mutex::new(AuditRegistry::new())),
+            state: Arc::new(RwLock::new(AuditSystemState::Stopped)),
+            config: Arc::new(RwLock::new(None)),
+        }
+    }
+
+    /// Starts the audit system with the given configuration
+    pub async fn start(&self, config: Config) -> AuditResult<()> {
+        let mut state = self.state.write().await;
+
+        match *state {
+            AuditSystemState::Running => {
+                return Err(AuditError::AlreadyInitialized);
+            }
+            AuditSystemState::Starting => {
+                warn!("Audit system is already starting");
+                return Ok(());
+            }
+            _ => {}
+        }
+
+        *state = AuditSystemState::Starting;
+        drop(state);
+
+        info!("Starting audit system");
+
+        // Record system start
+        observability::record_system_start();
+
+        // Store configuration
+        {
+            let mut config_guard = self.config.write().await;
+            *config_guard = Some(config.clone());
+        }
+
+        // Create targets from configuration
+        let mut registry = self.registry.lock().await;
+        match registry.create_targets_from_config(&config).await {
+            Ok(targets) => {
+                info!(target_count = targets.len(), "Created audit targets successfully");
+
+                // Initialize all targets
+                for target in targets {
+                    let target_id = target.id().to_string();
+                    if let Err(e) = target.init().await {
+                        error!(target_id = %target_id, error = %e, "Failed to initialize audit target");
+                    } else {
+                        // After successful initialization, if enabled and there is a store, start the send from storage task
+                        if target.is_enabled() {
+                            if let Some(store) = target.store() {
+                                info!(target_id = %target_id, "Start audit stream processing for target");
+                                let store_clone: Box<dyn Store<EntityTarget<AuditEntry>, Error = StoreError, Key = Key> + Send> =
+                                    store.boxed_clone();
+                                let target_arc: Arc<dyn Target<AuditEntry> + Send + Sync> = Arc::from(target.clone_dyn());
+                                self.start_audit_stream_with_batching(store_clone, target_arc);
+                                info!(target_id = %target_id, "Audit stream processing started");
+                            } else {
+                                info!(target_id = %target_id, "No store configured, skip audit stream processing");
+                            }
+                        } else {
+                            info!(target_id = %target_id, "Target disabled, skip audit stream processing");
+                        }
+                        registry.add_target(target_id, target);
+                    }
+                }
+
+                // Update state to running
+                let mut state = self.state.write().await;
+                *state = AuditSystemState::Running;
+                info!("Audit system started successfully");
+                Ok(())
+            }
+            Err(e) => {
+                error!(error = %e, "Failed to create audit targets");
+                let mut state = self.state.write().await;
+                *state = AuditSystemState::Stopped;
+                Err(e)
+            }
+        }
+    }
+
+    /// Pauses the audit system
+    pub async fn pause(&self) -> AuditResult<()> {
+        let mut state = self.state.write().await;
+
+        match *state {
+            AuditSystemState::Running => {
+                *state = AuditSystemState::Paused;
+                info!("Audit system paused");
+                Ok(())
+            }
+            AuditSystemState::Paused => {
+                warn!("Audit system is already paused");
+                Ok(())
+            }
+            _ => Err(AuditError::Configuration("Cannot pause audit system in current state".to_string())),
+        }
+    }
+
+    /// Resumes the audit system
+    pub async fn resume(&self) -> AuditResult<()> {
+        let mut state = self.state.write().await;
+
+        match *state {
+            AuditSystemState::Paused => {
+                *state = AuditSystemState::Running;
+                info!("Audit system resumed");
+                Ok(())
+            }
+            AuditSystemState::Running => {
+                warn!("Audit system is already running");
+                Ok(())
+            }
+            _ => Err(AuditError::Configuration("Cannot resume audit system in current state".to_string())),
+        }
+    }
+
+    /// Stops the audit system and closes all targets
+    pub async fn close(&self) -> AuditResult<()> {
+        let mut state = self.state.write().await;
+
+        match *state {
+            AuditSystemState::Stopped => {
+                warn!("Audit system is already stopped");
+                return Ok(());
+            }
+            AuditSystemState::Stopping => {
+                warn!("Audit system is already stopping");
+                return Ok(());
+            }
+            _ => {}
+        }
+
+        *state = AuditSystemState::Stopping;
+        drop(state);
+
+        info!("Stopping audit system");
+
+        // Close all targets
+        let mut registry = self.registry.lock().await;
+        if let Err(e) = registry.close_all().await {
+            error!(error = %e, "Failed to close some audit targets");
+        }
+
+        // Update state to stopped
+        let mut state = self.state.write().await;
+        *state = AuditSystemState::Stopped;
+
+        // Clear configuration
+        let mut config_guard = self.config.write().await;
+        *config_guard = None;
+
+        info!("Audit system stopped");
+        Ok(())
+    }
+
+    /// Gets the current state of the audit system
+    pub async fn get_state(&self) -> AuditSystemState {
+        self.state.read().await.clone()
+    }
+
+    /// Checks if the audit system is running
+    pub async fn is_running(&self) -> bool {
+        matches!(*self.state.read().await, AuditSystemState::Running)
+    }
+
+    /// Dispatches an audit log entry to all active targets
+    pub async fn dispatch(&self, entry: Arc<AuditEntry>) -> AuditResult<()> {
+        let start_time = std::time::Instant::now();
+
+        let state = self.state.read().await;
+
+        match *state {
+            AuditSystemState::Running => {
+                // Continue with dispatch
+                info!("Dispatching audit log entry");
+            }
+            AuditSystemState::Paused => {
+                // Skip dispatch when paused
+                return Ok(());
+            }
+            _ => {
+                // Don't dispatch when not running
+                return Err(AuditError::NotInitialized("Audit system is not running".to_string()));
+            }
+        }
+        drop(state);
+
+        let registry = self.registry.lock().await;
+        let target_ids = registry.list_targets();
+
+        if target_ids.is_empty() {
+            warn!("No audit targets configured for dispatch");
+            return Ok(());
+        }
+
+        // Dispatch to all targets concurrently
+        let mut tasks = Vec::new();
+
+        for target_id in target_ids {
+            if let Some(target) = registry.get_target(&target_id) {
+                let entry_clone = Arc::clone(&entry);
+                let target_id_clone = target_id.clone();
+
+                // Create EntityTarget for the audit log entry
+                let entity_target = rustfs_targets::target::EntityTarget {
+                    object_name: entry.api.name.clone().unwrap_or_default(),
+                    bucket_name: entry.api.bucket.clone().unwrap_or_default(),
+                    event_name: rustfs_targets::EventName::ObjectCreatedPut, // Default, should be derived from entry
+                    data: (*entry_clone).clone(),
+                };
+
+                let task = async move {
+                    let result = target.save(Arc::new(entity_target)).await;
+                    (target_id_clone, result)
+                };
+
+                tasks.push(task);
+            }
+        }
+
+        // Execute all dispatch tasks
+        let results = futures::future::join_all(tasks).await;
+
+        let mut errors = Vec::new();
+        let mut success_count = 0;
+
+        for (target_id, result) in results {
+            match result {
+                Ok(_) => {
+                    success_count += 1;
+                    observability::record_target_success();
+                }
+                Err(e) => {
+                    error!(target_id = %target_id, error = %e, "Failed to dispatch audit log to target");
+                    errors.push(e);
+                    observability::record_target_failure();
+                }
+            }
+        }
+
+        let dispatch_time = start_time.elapsed();
+
+        if errors.is_empty() {
+            observability::record_audit_success(dispatch_time);
+        } else {
+            observability::record_audit_failure(dispatch_time);
+            // Log errors but don't fail the entire dispatch
+            warn!(
+                error_count = errors.len(),
+                success_count = success_count,
+                "Some audit targets failed to receive log entry"
+            );
+        }
+
+        Ok(())
+    }
+
+    pub async fn dispatch_batch(&self, entries: Vec<Arc<AuditEntry>>) -> AuditResult<()> {
+        let start_time = std::time::Instant::now();
+
+        let state = self.state.read().await;
+        if *state != AuditSystemState::Running {
+            return Err(AuditError::NotInitialized("Audit system is not running".to_string()));
+        }
+        drop(state);
+
+        let registry = self.registry.lock().await;
+        let target_ids = registry.list_targets();
+
+        if target_ids.is_empty() {
+            warn!("No audit targets configured for batch dispatch");
+            return Ok(());
+        }
+
+        let mut tasks = Vec::new();
+        for target_id in target_ids {
+            if let Some(target) = registry.get_target(&target_id) {
+                let entries_clone: Vec<_> = entries.iter().map(Arc::clone).collect();
+                let target_id_clone = target_id.clone();
+
+                let task = async move {
+                    let mut success_count = 0;
+                    let mut errors = Vec::new();
+                    for entry in entries_clone {
+                        let entity_target = rustfs_targets::target::EntityTarget {
+                            object_name: entry.api.name.clone().unwrap_or_default(),
+                            bucket_name: entry.api.bucket.clone().unwrap_or_default(),
+                            event_name: rustfs_targets::EventName::ObjectCreatedPut,
+                            data: (*entry).clone(),
+                        };
+                        match target.save(Arc::new(entity_target)).await {
+                            Ok(_) => success_count += 1,
+                            Err(e) => errors.push(e),
+                        }
+                    }
+                    (target_id_clone, success_count, errors)
+                };
+                tasks.push(task);
+            }
+        }
+
+        let results = futures::future::join_all(tasks).await;
+        let mut total_success = 0;
+        let mut total_errors = 0;
+        for (_target_id, success_count, errors) in results {
+            total_success += success_count;
+            total_errors += errors.len();
+            for e in errors {
+                error!("Batch dispatch error: {:?}", e);
+            }
+        }
+
+        let dispatch_time = start_time.elapsed();
+        info!(
+            "Batch dispatched {} entries, success: {}, errors: {}, time: {:?}",
+            entries.len(),
+            total_success,
+            total_errors,
+            dispatch_time
+        );
+
+        Ok(())
+    }
+
+    // New: Audit flow background tasks, based on send_from_store, including retries and exponential backoffs
+    fn start_audit_stream_with_batching(
+        &self,
+        store: Box<dyn Store<EntityTarget<AuditEntry>, Error = StoreError, Key = Key> + Send>,
+        target: Arc<dyn Target<AuditEntry> + Send + Sync>,
+    ) {
+        let state = self.state.clone();
+
+        tokio::spawn(async move {
+            use std::time::Duration;
+            use tokio::time::sleep;
+
+            info!("Starting audit stream for target: {}", target.id());
+
+            const MAX_RETRIES: usize = 5;
+            const BASE_RETRY_DELAY: Duration = Duration::from_secs(2);
+
+            loop {
+                match *state.read().await {
+                    AuditSystemState::Running | AuditSystemState::Paused | AuditSystemState::Starting => {}
+                    _ => {
+                        info!("Audit stream stopped for target: {}", target.id());
+                        break;
+                    }
+                }
+
+                let keys: Vec<Key> = store.list();
+                if keys.is_empty() {
+                    sleep(Duration::from_millis(500)).await;
+                    continue;
+                }
+
+                for key in keys {
+                    let mut retries = 0usize;
+                    let mut success = false;
+
+                    while retries < MAX_RETRIES && !success {
+                        match target.send_from_store(key.clone()).await {
+                            Ok(_) => {
+                                info!("Successfully sent audit entry, target: {}, key: {}", target.id(), key.to_string());
+                                observability::record_target_success();
+                                success = true;
+                            }
+                            Err(e) => {
+                                match &e {
+                                    TargetError::NotConnected => {
+                                        warn!("Target {} not connected, retrying...", target.id());
+                                    }
+                                    TargetError::Timeout(_) => {
+                                        warn!("Timeout sending to target {}, retrying...", target.id());
+                                    }
+                                    _ => {
+                                        error!("Permanent error for target {}: {}", target.id(), e);
+                                        observability::record_target_failure();
+                                        break;
+                                    }
+                                }
+                                retries += 1;
+                                let backoff = BASE_RETRY_DELAY * (1 << retries);
+                                sleep(backoff).await;
+                            }
+                        }
+                    }
+
+                    if retries >= MAX_RETRIES && !success {
+                        warn!("Max retries exceeded for key {}, target: {}, skipping", key.to_string(), target.id());
+                        observability::record_target_failure();
+                    }
+                }
+
+                sleep(Duration::from_millis(100)).await;
+            }
+        });
+    }
+
+    /// Enables a specific target
+    pub async fn enable_target(&self, target_id: &str) -> AuditResult<()> {
+        // This would require storing enabled/disabled state per target
+        // For now, just check if target exists
+        let registry = self.registry.lock().await;
+        if registry.get_target(target_id).is_some() {
+            info!(target_id = %target_id, "Target enabled");
+            Ok(())
+        } else {
+            Err(AuditError::Configuration(format!("Target not found: {}", target_id)))
+        }
+    }
+
+    /// Disables a specific target
+    pub async fn disable_target(&self, target_id: &str) -> AuditResult<()> {
+        // This would require storing enabled/disabled state per target
+        // For now, just check if target exists
+        let registry = self.registry.lock().await;
+        if registry.get_target(target_id).is_some() {
+            info!(target_id = %target_id, "Target disabled");
+            Ok(())
+        } else {
+            Err(AuditError::Configuration(format!("Target not found: {}", target_id)))
+        }
+    }
+
+    /// Removes a target from the system
+    pub async fn remove_target(&self, target_id: &str) -> AuditResult<()> {
+        let mut registry = self.registry.lock().await;
+        if let Some(target) = registry.remove_target(target_id) {
+            if let Err(e) = target.close().await {
+                error!(target_id = %target_id, error = %e, "Failed to close removed target");
+            }
+            info!(target_id = %target_id, "Target removed");
+            Ok(())
+        } else {
+            Err(AuditError::Configuration(format!("Target not found: {}", target_id)))
+        }
+    }
+
+    /// Updates or inserts a target
+    pub async fn upsert_target(&self, target_id: String, target: Box<dyn Target<AuditEntry> + Send + Sync>) -> AuditResult<()> {
+        let mut registry = self.registry.lock().await;
+
+        // Initialize the target
+        if let Err(e) = target.init().await {
+            return Err(AuditError::Target(e));
+        }
+
+        // Remove existing target if present
+        if let Some(old_target) = registry.remove_target(&target_id) {
+            if let Err(e) = old_target.close().await {
+                error!(target_id = %target_id, error = %e, "Failed to close old target during upsert");
+            }
+        }
+
+        registry.add_target(target_id.clone(), target);
+        info!(target_id = %target_id, "Target upserted");
+        Ok(())
+    }
+
+    /// Lists all targets
+    pub async fn list_targets(&self) -> Vec<String> {
+        let registry = self.registry.lock().await;
+        registry.list_targets()
+    }
+
+    /// Gets information about a specific target
+    pub async fn get_target(&self, target_id: &str) -> Option<String> {
+        let registry = self.registry.lock().await;
+        registry.get_target(target_id).map(|target| target.id().to_string())
+    }
+
+    /// Reloads configuration and updates targets
+    pub async fn reload_config(&self, new_config: Config) -> AuditResult<()> {
+        info!("Reloading audit system configuration");
+
+        // Record config reload
+        observability::record_config_reload();
+
+        // Store new configuration
+        {
+            let mut config_guard = self.config.write().await;
+            *config_guard = Some(new_config.clone());
+        }
+
+        // Close all existing targets
+        let mut registry = self.registry.lock().await;
+        if let Err(e) = registry.close_all().await {
+            error!(error = %e, "Failed to close existing targets during reload");
+        }
+
+        // Create new targets from updated configuration
+        match registry.create_targets_from_config(&new_config).await {
+            Ok(targets) => {
+                info!(target_count = targets.len(), "Reloaded audit targets successfully");
+
+                // Initialize all new targets
+                for target in targets {
+                    let target_id = target.id().to_string();
+                    if let Err(e) = target.init().await {
+                        error!(target_id = %target_id, error = %e, "Failed to initialize reloaded audit target");
+                    } else {
+                        // Same starts the storage stream after a heavy load
+                        if target.is_enabled() {
+                            if let Some(store) = target.store() {
+                                info!(target_id = %target_id, "Start audit stream processing for target (reload)");
+                                let store_clone: Box<dyn Store<EntityTarget<AuditEntry>, Error = StoreError, Key = Key> + Send> =
+                                    store.boxed_clone();
+                                let target_arc: Arc<dyn Target<AuditEntry> + Send + Sync> = Arc::from(target.clone_dyn());
+                                self.start_audit_stream_with_batching(store_clone, target_arc);
+                                info!(target_id = %target_id, "Audit stream processing started (reload)");
+                            } else {
+                                info!(target_id = %target_id, "No store configured, skip audit stream processing (reload)");
+                            }
+                        } else {
+                            info!(target_id = %target_id, "Target disabled, skip audit stream processing (reload)");
+                        }
+                        registry.add_target(target.id().to_string(), target);
+                    }
+                }
+
+                info!("Audit configuration reloaded successfully");
+                Ok(())
+            }
+            Err(e) => {
+                error!(error = %e, "Failed to reload audit configuration");
+                Err(e)
+            }
+        }
+    }
+
+    /// Gets current audit system metrics
+    pub async fn get_metrics(&self) -> observability::AuditMetricsReport {
+        observability::get_metrics_report().await
+    }
+
+    /// Validates system performance against requirements
+    pub async fn validate_performance(&self) -> observability::PerformanceValidation {
+        observability::validate_performance().await
+    }
+
+    /// Resets all metrics
+    pub async fn reset_metrics(&self) {
+        observability::reset_metrics().await;
+    }
+}