Merge branch 'main' of github.com:rustfs/s3-rustfs into feature/observability-metrics

# Conflicts:
#	Cargo.toml
#	crates/obs/examples/config.toml
#	crates/obs/src/telemetry.rs

crates/obs/Cargo.toml

@@ -13,11 +13,11 @@ workspace = true
 default = ["file"]
 file = []
 gpu = ["dep:nvml-wrapper"]
-kafka = ["dep:rdkafka"]
 webhook = ["dep:reqwest"]
-full = ["file", "gpu", "kafka", "webhook"]
+kafka = ["dep:rdkafka"]
 
 [dependencies]
+rustfs-config = { workspace = true }
 async-trait = { workspace = true }
 chrono = { workspace = true }
 config = { workspace = true }
@@ -29,6 +29,7 @@ opentelemetry_sdk = { workspace = true, features = ["rt-tokio"] }
 opentelemetry-stdout = { workspace = true }
 opentelemetry-otlp = { workspace = true, features = ["grpc-tonic", "gzip-tonic"] }
 opentelemetry-semantic-conventions = { workspace = true, features = ["semconv_experimental"] }
+rustfs-utils = { workspace = true, features = ["ip"] }
 serde = { workspace = true }
 smallvec = { workspace = true, features = ["serde"] }
 tracing = { workspace = true, features = ["std", "attributes"] }
@@ -37,12 +38,14 @@ tracing-error = { workspace = true }
 tracing-opentelemetry = { workspace = true }
 tracing-subscriber = { workspace = true, features = ["registry", "std", "fmt", "env-filter", "tracing-log", "time", "local-time", "json"] }
 tokio = { workspace = true, features = ["sync", "fs", "rt-multi-thread", "rt", "time", "macros"] }
-rdkafka = { workspace = true, features = ["tokio"], optional = true }
 reqwest = { workspace = true, optional = true, default-features = false }
 serde_json = { workspace = true }
 sysinfo = { workspace = true }
 thiserror = { workspace = true }
 
+# Only enable kafka features and related dependencies on Linux
+[target.'cfg(target_os = "linux")'.dependencies]
+rdkafka = { workspace = true, features = ["tokio"], optional = true }
 
 
 [dev-dependencies]

crates/obs/examples/config.toml

@@ -9,26 +9,27 @@ environments = "develop"
 logger_level = "debug"
 local_logging_enabled = true # Default is false if not specified
 
-[sinks]
-[sinks.kafka]
-enabled = false
-bootstrap_servers = "localhost:9092"
-topic = "logs"
-batch_size = 100 # Default is 100 if not specified
-batch_timeout_ms = 1000 # Default is 1000ms if not specified
 
-[sinks.webhook]
-enabled = false
-endpoint = "http://localhost:8080/webhook"
-auth_token = ""
-batch_size = 100 # Default is 3 if not specified
-batch_timeout_ms = 1000 # Default is 100ms if not specified
+#[[sinks]]
+#type = "Kafka"
+#bootstrap_servers = "localhost:9092"
+#topic = "logs"
+#batch_size = 100 # Default is 100 if not specified
+#batch_timeout_ms = 100 # Default is 1000ms if not specified
+#
+#[[sinks]]
+#type = "Webhook"
+#endpoint = "http://localhost:8080/webhook"
+#auth_token = ""
+#batch_size = 100 # Default is 3 if not specified
+#batch_timeout_ms = 100 # Default is 100ms if not specified
 
-[sinks.file]
-enabled = true
-path = "deploy/logs/app.log"
-batch_size = 100
-batch_timeout_ms = 1000 # Default is 8192 bytes if not specified
+[[sinks]]
+type = "File"
+path = "deploy/logs/rustfs.log"
+buffer_size = 102 # Default is 8192 bytes if not specified
+flush_interval_ms = 1000
+flush_threshold = 100
 
 [logger]
 queue_capacity = 10000

crates/obs/src/config.rs

@@ -1,6 +1,6 @@
-use crate::global::{ENVIRONMENT, LOGGER_LEVEL, METER_INTERVAL, SAMPLE_RATIO, SERVICE_NAME, SERVICE_VERSION, USE_STDOUT};
-use config::{Config, Environment, File, FileFormat};
-use serde::Deserialize;
+use config::{Config, File, FileFormat};
+use rustfs_config::{APP_NAME, DEFAULT_LOG_LEVEL, ENVIRONMENT, METER_INTERVAL, SAMPLE_RATIO, SERVICE_VERSION, USE_STDOUT};
+use serde::{Deserialize, Serialize};
 use std::env;
 
 /// OpenTelemetry Configuration
@@ -11,7 +11,7 @@ use std::env;
 /// Add use_stdout for output to stdout
 /// Add logger level for log level
 /// Add local_logging_enabled for local logging enabled
-#[derive(Debug, Deserialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone)]
 pub struct OtelConfig {
     pub endpoint: String,                    // Endpoint for metric collection
     pub use_stdout: Option<bool>,            // Output to stdout
@@ -24,7 +24,7 @@ pub struct OtelConfig {
     pub local_logging_enabled: Option<bool>, // Local logging enabled
 }
 
-// Helper function: Extract observable configuration from environment variables
+/// Helper function: Extract observable configuration from environment variables
 fn extract_otel_config_from_env() -> OtelConfig {
     OtelConfig {
         endpoint: env::var("RUSTFS_OBSERVABILITY_ENDPOINT").unwrap_or_else(|_| "".to_string()),
@@ -43,7 +43,7 @@ fn extract_otel_config_from_env() -> OtelConfig {
         service_name: env::var("RUSTFS_OBSERVABILITY_SERVICE_NAME")
             .ok()
             .and_then(|v| v.parse().ok())
-            .or(Some(SERVICE_NAME.to_string())),
+            .or(Some(APP_NAME.to_string())),
         service_version: env::var("RUSTFS_OBSERVABILITY_SERVICE_VERSION")
             .ok()
             .and_then(|v| v.parse().ok())
@@ -55,7 +55,7 @@ fn extract_otel_config_from_env() -> OtelConfig {
         logger_level: env::var("RUSTFS_OBSERVABILITY_LOGGER_LEVEL")
             .ok()
             .and_then(|v| v.parse().ok())
-            .or(Some(LOGGER_LEVEL.to_string())),
+            .or(Some(DEFAULT_LOG_LEVEL.to_string())),
         local_logging_enabled: env::var("RUSTFS_OBSERVABILITY_LOCAL_LOGGING_ENABLED")
             .ok()
             .and_then(|v| v.parse().ok())
@@ -63,36 +63,89 @@ fn extract_otel_config_from_env() -> OtelConfig {
     }
 }
 
-impl Default for OtelConfig {
-    fn default() -> Self {
+impl OtelConfig {
+    /// Create a new instance of OtelConfig with default values
+    ///
+    /// # Returns
+    /// A new instance of OtelConfig
+    pub fn new() -> Self {
         extract_otel_config_from_env()
     }
 }
 
+impl Default for OtelConfig {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 /// Kafka Sink Configuration - Add batch parameters
-#[derive(Debug, Deserialize, Clone, Default)]
+#[derive(Debug, Deserialize, Serialize, Clone)]
 pub struct KafkaSinkConfig {
-    pub enabled: bool,
-    pub bootstrap_servers: String,
+    pub brokers: String,
     pub topic: String,
     pub batch_size: Option<usize>,     // Batch size, default 100
     pub batch_timeout_ms: Option<u64>, // Batch timeout time, default 1000ms
 }
 
+impl KafkaSinkConfig {
+    pub fn new() -> Self {
+        Self {
+            brokers: env::var("RUSTFS_SINKS_KAFKA_BROKERS")
+                .ok()
+                .filter(|s| !s.trim().is_empty())
+                .unwrap_or_else(|| "localhost:9092".to_string()),
+            topic: env::var("RUSTFS_SINKS_KAFKA_TOPIC")
+                .ok()
+                .filter(|s| !s.trim().is_empty())
+                .unwrap_or_else(|| "default_topic".to_string()),
+            batch_size: Some(100),
+            batch_timeout_ms: Some(1000),
+        }
+    }
+}
+
+impl Default for KafkaSinkConfig {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 /// Webhook Sink Configuration - Add Retry Parameters
-#[derive(Debug, Deserialize, Clone, Default)]
+#[derive(Debug, Deserialize, Serialize, Clone)]
 pub struct WebhookSinkConfig {
-    pub enabled: bool,
     pub endpoint: String,
     pub auth_token: String,
     pub max_retries: Option<usize>,  // Maximum number of retry times, default 3
     pub retry_delay_ms: Option<u64>, // Retry the delay cardinality, default 100ms
 }
 
+impl WebhookSinkConfig {
+    pub fn new() -> Self {
+        Self {
+            endpoint: env::var("RUSTFS_SINKS_WEBHOOK_ENDPOINT")
+                .ok()
+                .filter(|s| !s.trim().is_empty())
+                .unwrap_or_else(|| "http://localhost:8080".to_string()),
+            auth_token: env::var("RUSTFS_SINKS_WEBHOOK_AUTH_TOKEN")
+                .ok()
+                .filter(|s| !s.trim().is_empty())
+                .unwrap_or_else(|| "default_token".to_string()),
+            max_retries: Some(3),
+            retry_delay_ms: Some(100),
+        }
+    }
+}
+
+impl Default for WebhookSinkConfig {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 /// File Sink Configuration - Add buffering parameters
-#[derive(Debug, Deserialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone)]
 pub struct FileSinkConfig {
-    pub enabled: bool,
     pub path: String,
     pub buffer_size: Option<usize>,     // Write buffer size, default 8192
     pub flush_interval_ms: Option<u64>, // Refresh interval time, default 1000ms
@@ -107,19 +160,15 @@ impl FileSinkConfig {
             eprintln!("Failed to create log directory: {}", e);
             return "rustfs/rustfs.log".to_string();
         }
-
+        println!("Using log directory: {:?}", temp_dir);
         temp_dir
             .join("rustfs.log")
             .to_str()
             .unwrap_or("rustfs/rustfs.log")
             .to_string()
     }
-}
-
-impl Default for FileSinkConfig {
-    fn default() -> Self {
-        FileSinkConfig {
-            enabled: true,
+    pub fn new() -> Self {
+        Self {
             path: env::var("RUSTFS_SINKS_FILE_PATH")
                 .ok()
                 .filter(|s| !s.trim().is_empty())
@@ -131,38 +180,53 @@ impl Default for FileSinkConfig {
     }
 }
 
+impl Default for FileSinkConfig {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 /// Sink configuration collection
-#[derive(Debug, Deserialize, Clone)]
-pub struct SinkConfig {
-    pub kafka: Option<KafkaSinkConfig>,
-    pub webhook: Option<WebhookSinkConfig>,
-    pub file: Option<FileSinkConfig>,
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(tag = "type")]
+pub enum SinkConfig {
+    File(FileSinkConfig),
+    Kafka(KafkaSinkConfig),
+    Webhook(WebhookSinkConfig),
+}
+
+impl SinkConfig {
+    pub fn new() -> Self {
+        Self::File(FileSinkConfig::new())
+    }
 }
 
 impl Default for SinkConfig {
     fn default() -> Self {
-        SinkConfig {
-            kafka: None,
-            webhook: None,
-            file: Some(FileSinkConfig::default()),
-        }
+        Self::new()
     }
 }
 
 ///Logger Configuration
-#[derive(Debug, Deserialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone)]
 pub struct LoggerConfig {
     pub queue_capacity: Option<usize>,
 }
 
-impl Default for LoggerConfig {
-    fn default() -> Self {
-        LoggerConfig {
+impl LoggerConfig {
+    pub fn new() -> Self {
+        Self {
             queue_capacity: Some(10000),
         }
     }
 }
 
+impl Default for LoggerConfig {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 /// Overall application configuration
 /// Add observability, sinks, and logger configuration
 ///
@@ -180,7 +244,7 @@ impl Default for LoggerConfig {
 #[derive(Debug, Deserialize, Clone)]
 pub struct AppConfig {
     pub observability: OtelConfig,
-    pub sinks: SinkConfig,
+    pub sinks: Vec<SinkConfig>,
     pub logger: Option<LoggerConfig>,
 }
 
@@ -192,7 +256,7 @@ impl AppConfig {
     pub fn new() -> Self {
         Self {
             observability: OtelConfig::default(),
-            sinks: SinkConfig::default(),
+            sinks: vec![SinkConfig::default()],
             logger: Some(LoggerConfig::default()),
         }
     }
@@ -258,14 +322,6 @@ pub fn load_config(config_dir: Option<String>) -> AppConfig {
     let app_config = Config::builder()
         .add_source(File::with_name(config_dir.as_str()).format(FileFormat::Toml).required(false))
         .add_source(File::with_name(config_dir.as_str()).format(FileFormat::Yaml).required(false))
-        .add_source(
-            Environment::default()
-                .prefix("RUSTFS")
-                .prefix_separator("__")
-                .separator("__")
-                .with_list_parse_key("volumes")
-                .try_parsing(true),
-        )
         .build()
         .unwrap_or_default();
 

crates/obs/src/global.rs

@@ -3,14 +3,6 @@ use std::sync::{Arc, Mutex};
 use tokio::sync::{OnceCell, SetError};
 use tracing::{error, info};
 
-pub(crate) const USE_STDOUT: bool = true;
-pub(crate) const SERVICE_NAME: &str = "RustFS";
-pub(crate) const SAMPLE_RATIO: f64 = 1.0;
-pub(crate) const METER_INTERVAL: u64 = 60;
-pub(crate) const SERVICE_VERSION: &str = "0.1.0";
-pub(crate) const ENVIRONMENT: &str = "production";
-pub(crate) const LOGGER_LEVEL: &str = "info";
-
 /// Global guard for OpenTelemetry tracing
 static GLOBAL_GUARD: OnceCell<Arc<Mutex<OtelGuard>>> = OnceCell::const_new();
 

crates/obs/src/lib.rs

@@ -32,20 +32,13 @@ mod config;
 mod entry;
 mod global;
 mod logger;
-mod sink;
+mod sinks;
 mod system;
 mod telemetry;
-mod utils;
 mod worker;
 
 use crate::logger::InitLogStatus;
 pub use config::load_config;
-#[cfg(feature = "file")]
-pub use config::FileSinkConfig;
-#[cfg(feature = "kafka")]
-pub use config::KafkaSinkConfig;
-#[cfg(feature = "webhook")]
-pub use config::WebhookSinkConfig;
 pub use config::{AppConfig, LoggerConfig, OtelConfig, SinkConfig};
 pub use entry::args::Args;
 pub use entry::audit::{ApiDetails, AuditLogEntry};
@@ -79,7 +72,7 @@ use tracing::{error, info};
 /// ```
 pub async fn init_obs(config: AppConfig) -> (Arc<Mutex<Logger>>, telemetry::OtelGuard) {
     let guard = init_telemetry(&config.observability);
-    let sinks = sink::create_sinks(&config).await;
+    let sinks = sinks::create_sinks(&config).await;
     let logger = init_global_logger(&config, sinks).await;
     let obs_config = config.observability.clone();
     tokio::spawn(async move {

crates/obs/src/logger.rs

@@ -1,6 +1,6 @@
-use crate::global::{ENVIRONMENT, SERVICE_NAME, SERVICE_VERSION};
-use crate::sink::Sink;
+use crate::sinks::Sink;
 use crate::{AppConfig, AuditLogEntry, BaseLogEntry, ConsoleLogEntry, GlobalError, OtelConfig, ServerLogEntry, UnifiedLogEntry};
+use rustfs_config::{APP_NAME, ENVIRONMENT, SERVICE_VERSION};
 use std::sync::Arc;
 use std::time::SystemTime;
 use tokio::sync::mpsc::{self, Receiver, Sender};
@@ -428,7 +428,7 @@ impl Default for InitLogStatus {
     fn default() -> Self {
         Self {
             timestamp: SystemTime::now(),
-            service_name: String::from(SERVICE_NAME),
+            service_name: String::from(APP_NAME),
             version: SERVICE_VERSION.to_string(),
             environment: ENVIRONMENT.to_string(),
         }
@@ -442,7 +442,7 @@ impl InitLogStatus {
         let version = config.service_version.unwrap_or(SERVICE_VERSION.to_string());
         Self {
             timestamp: SystemTime::now(),
-            service_name: String::from(SERVICE_NAME),
+            service_name: String::from(APP_NAME),
             version,
             environment,
         }

crates/obs/src/sink.rs

@@ -1,497 +0,0 @@
-use crate::{AppConfig, LogRecord, UnifiedLogEntry};
-use async_trait::async_trait;
-use std::sync::Arc;
-use tokio::fs::OpenOptions;
-use tokio::io;
-use tokio::io::AsyncWriteExt;
-
-/// Sink Trait definition, asynchronously write logs
-#[async_trait]
-pub trait Sink: Send + Sync {
-    async fn write(&self, entry: &UnifiedLogEntry);
-}
-
-#[cfg(feature = "kafka")]
-/// Kafka Sink Implementation
-pub struct KafkaSink {
-    producer: rdkafka::producer::FutureProducer,
-    topic: String,
-    batch_size: usize,
-    batch_timeout_ms: u64,
-    entries: Arc<tokio::sync::Mutex<Vec<UnifiedLogEntry>>>,
-    last_flush: Arc<std::sync::atomic::AtomicU64>,
-}
-
-#[cfg(feature = "kafka")]
-impl KafkaSink {
-    /// Create a new KafkaSink instance
-    pub fn new(producer: rdkafka::producer::FutureProducer, topic: String, batch_size: usize, batch_timeout_ms: u64) -> Self {
-        // Create Arc-wrapped values first
-        let entries = Arc::new(tokio::sync::Mutex::new(Vec::with_capacity(batch_size)));
-        let last_flush = Arc::new(std::sync::atomic::AtomicU64::new(
-            std::time::SystemTime::now()
-                .duration_since(std::time::UNIX_EPOCH)
-                .unwrap()
-                .as_millis() as u64,
-        ));
-        let sink = KafkaSink {
-            producer: producer.clone(),
-            topic: topic.clone(),
-            batch_size,
-            batch_timeout_ms,
-            entries: entries.clone(),
-            last_flush: last_flush.clone(),
-        };
-
-        // Start background flusher
-        tokio::spawn(Self::periodic_flush(producer, topic, entries, last_flush, batch_timeout_ms));
-
-        sink
-    }
-
-    /// Add a getter method to read the batch_timeout_ms field
-    #[allow(dead_code)]
-    pub fn batch_timeout(&self) -> u64 {
-        self.batch_timeout_ms
-    }
-
-    /// Add a method to dynamically adjust the timeout if needed
-    #[allow(dead_code)]
-    pub fn set_batch_timeout(&mut self, new_timeout_ms: u64) {
-        self.batch_timeout_ms = new_timeout_ms;
-    }
-
-    async fn periodic_flush(
-        producer: rdkafka::producer::FutureProducer,
-        topic: String,
-        entries: Arc<tokio::sync::Mutex<Vec<UnifiedLogEntry>>>,
-        last_flush: Arc<std::sync::atomic::AtomicU64>,
-        timeout_ms: u64,
-    ) {
-        loop {
-            tokio::time::sleep(tokio::time::Duration::from_millis(timeout_ms / 2)).await;
-
-            let now = std::time::SystemTime::now()
-                .duration_since(std::time::UNIX_EPOCH)
-                .unwrap()
-                .as_millis() as u64;
-
-            let last = last_flush.load(std::sync::atomic::Ordering::Relaxed);
-
-            if now - last >= timeout_ms {
-                let mut batch = entries.lock().await;
-                if !batch.is_empty() {
-                    Self::send_batch(&producer, &topic, batch.drain(..).collect()).await;
-                    last_flush.store(now, std::sync::atomic::Ordering::Relaxed);
-                }
-            }
-        }
-    }
-
-    async fn send_batch(producer: &rdkafka::producer::FutureProducer, topic: &str, entries: Vec<UnifiedLogEntry>) {
-        for entry in entries {
-            let payload = match serde_json::to_string(&entry) {
-                Ok(p) => p,
-                Err(e) => {
-                    eprintln!("Failed to serialize log entry: {}", e);
-                    continue;
-                }
-            };
-
-            let span_id = entry.get_timestamp().to_rfc3339();
-
-            let _ = producer
-                .send(
-                    rdkafka::producer::FutureRecord::to(topic).payload(&payload).key(&span_id),
-                    std::time::Duration::from_secs(5),
-                )
-                .await;
-        }
-    }
-}
-
-#[cfg(feature = "kafka")]
-#[async_trait]
-impl Sink for KafkaSink {
-    async fn write(&self, entry: &UnifiedLogEntry) {
-        let mut batch = self.entries.lock().await;
-        batch.push(entry.clone());
-
-        let should_flush_by_size = batch.len() >= self.batch_size;
-        let should_flush_by_time = {
-            let now = std::time::SystemTime::now()
-                .duration_since(std::time::UNIX_EPOCH)
-                .unwrap()
-                .as_millis() as u64;
-            let last = self.last_flush.load(std::sync::atomic::Ordering::Relaxed);
-            now - last >= self.batch_timeout_ms
-        };
-
-        if should_flush_by_size || should_flush_by_time {
-            // Existing flush logic
-            let entries_to_send: Vec<UnifiedLogEntry> = batch.drain(..).collect();
-            let producer = self.producer.clone();
-            let topic = self.topic.clone();
-
-            self.last_flush.store(
-                std::time::SystemTime::now()
-                    .duration_since(std::time::UNIX_EPOCH)
-                    .unwrap()
-                    .as_millis() as u64,
-                std::sync::atomic::Ordering::Relaxed,
-            );
-
-            tokio::spawn(async move {
-                KafkaSink::send_batch(&producer, &topic, entries_to_send).await;
-            });
-        }
-    }
-}
-
-#[cfg(feature = "kafka")]
-impl Drop for KafkaSink {
-    fn drop(&mut self) {
-        // Perform any necessary cleanup here
-        // For example, you might want to flush any remaining entries
-        let producer = self.producer.clone();
-        let topic = self.topic.clone();
-        let entries = self.entries.clone();
-        let last_flush = self.last_flush.clone();
-
-        tokio::spawn(async move {
-            let mut batch = entries.lock().await;
-            if !batch.is_empty() {
-                KafkaSink::send_batch(&producer, &topic, batch.drain(..).collect()).await;
-                last_flush.store(
-                    std::time::SystemTime::now()
-                        .duration_since(std::time::UNIX_EPOCH)
-                        .unwrap()
-                        .as_millis() as u64,
-                    std::sync::atomic::Ordering::Relaxed,
-                );
-            }
-        });
-
-        eprintln!("Dropping KafkaSink with topic: {}", self.topic);
-    }
-}
-
-#[cfg(feature = "webhook")]
-/// Webhook Sink Implementation
-pub struct WebhookSink {
-    endpoint: String,
-    auth_token: String,
-    client: reqwest::Client,
-    max_retries: usize,
-    retry_delay_ms: u64,
-}
-
-#[cfg(feature = "webhook")]
-impl WebhookSink {
-    pub fn new(endpoint: String, auth_token: String, max_retries: usize, retry_delay_ms: u64) -> Self {
-        WebhookSink {
-            endpoint,
-            auth_token,
-            client: reqwest::Client::builder()
-                .timeout(std::time::Duration::from_secs(10))
-                .build()
-                .unwrap_or_else(|_| reqwest::Client::new()),
-            max_retries,
-            retry_delay_ms,
-        }
-    }
-}
-
-#[cfg(feature = "webhook")]
-#[async_trait]
-impl Sink for WebhookSink {
-    async fn write(&self, entry: &UnifiedLogEntry) {
-        let mut retries = 0;
-        let url = self.endpoint.clone();
-        let entry_clone = entry.clone();
-        let auth_value = reqwest::header::HeaderValue::from_str(format!("Bearer {}", self.auth_token.clone()).as_str()).unwrap();
-        while retries < self.max_retries {
-            match self
-                .client
-                .post(&url)
-                .header(reqwest::header::AUTHORIZATION, auth_value.clone())
-                .json(&entry_clone)
-                .send()
-                .await
-            {
-                Ok(response) if response.status().is_success() => {
-                    return;
-                }
-                _ => {
-                    retries += 1;
-                    if retries < self.max_retries {
-                        tokio::time::sleep(tokio::time::Duration::from_millis(
-                            self.retry_delay_ms * (1 << retries), // Exponential backoff
-                        ))
-                        .await;
-                    }
-                }
-            }
-        }
-
-        eprintln!("Failed to send log to webhook after {} retries", self.max_retries);
-    }
-}
-
-#[cfg(feature = "webhook")]
-impl Drop for WebhookSink {
-    fn drop(&mut self) {
-        // Perform any necessary cleanup here
-        // For example, you might want to log that the sink is being dropped
-        eprintln!("Dropping WebhookSink with URL: {}", self.endpoint);
-    }
-}
-
-#[cfg(feature = "file")]
-/// File Sink Implementation
-pub struct FileSink {
-    path: String,
-    buffer_size: usize,
-    writer: Arc<tokio::sync::Mutex<io::BufWriter<tokio::fs::File>>>,
-    entry_count: std::sync::atomic::AtomicUsize,
-    last_flush: std::sync::atomic::AtomicU64,
-    flush_interval_ms: u64, // Time between flushes
-    flush_threshold: usize, // Number of entries before flush
-}
-
-#[cfg(feature = "file")]
-impl FileSink {
-    /// Create a new FileSink instance
-    pub async fn new(
-        path: String,
-        buffer_size: usize,
-        flush_interval_ms: u64,
-        flush_threshold: usize,
-    ) -> Result<Self, io::Error> {
-        // check if the file exists
-        let file_exists = tokio::fs::metadata(&path).await.is_ok();
-        // if the file not exists, create it
-        if !file_exists {
-            tokio::fs::create_dir_all(std::path::Path::new(&path).parent().unwrap()).await?;
-            tracing::debug!("File does not exist, creating it. Path: {:?}", path)
-        }
-        let file = if file_exists {
-            // If the file exists, open it in append mode
-            tracing::debug!("FileSink: File exists, opening in append mode.");
-            OpenOptions::new().append(true).create(true).open(&path).await?
-        } else {
-            // If the file does not exist, create it
-            tracing::debug!("FileSink: File does not exist, creating a new file.");
-            // Create the file and write a header or initial content if needed
-            OpenOptions::new().create(true).truncate(true).write(true).open(&path).await?
-        };
-        let writer = io::BufWriter::with_capacity(buffer_size, file);
-        let now = std::time::SystemTime::now()
-            .duration_since(std::time::UNIX_EPOCH)
-            .unwrap()
-            .as_millis() as u64;
-        Ok(FileSink {
-            path,
-            buffer_size,
-            writer: Arc::new(tokio::sync::Mutex::new(writer)),
-            entry_count: std::sync::atomic::AtomicUsize::new(0),
-            last_flush: std::sync::atomic::AtomicU64::new(now),
-            flush_interval_ms,
-            flush_threshold,
-        })
-    }
-
-    #[allow(dead_code)]
-    async fn initialize_writer(&mut self) -> io::Result<()> {
-        let file = tokio::fs::File::create(&self.path).await?;
-
-        // Use buffer_size to create a buffer writer with a specified capacity
-        let buf_writer = io::BufWriter::with_capacity(self.buffer_size, file);
-
-        // Replace the original writer with the new Mutex
-        self.writer = Arc::new(tokio::sync::Mutex::new(buf_writer));
-        Ok(())
-    }
-
-    // Get the current buffer size
-    #[allow(dead_code)]
-    pub fn buffer_size(&self) -> usize {
-        self.buffer_size
-    }
-
-    // How to dynamically adjust the buffer size
-    #[allow(dead_code)]
-    pub async fn set_buffer_size(&mut self, new_size: usize) -> io::Result<()> {
-        if self.buffer_size != new_size {
-            self.buffer_size = new_size;
-            // Reinitialize the writer directly, without checking is_some()
-            self.initialize_writer().await?;
-        }
-        Ok(())
-    }
-
-    // Check if flushing is needed based on count or time
-    fn should_flush(&self) -> bool {
-        // Check entry count threshold
-        if self.entry_count.load(std::sync::atomic::Ordering::Relaxed) >= self.flush_threshold {
-            return true;
-        }
-
-        // Check time threshold
-        let now = std::time::SystemTime::now()
-            .duration_since(std::time::UNIX_EPOCH)
-            .unwrap()
-            .as_millis() as u64;
-
-        let last = self.last_flush.load(std::sync::atomic::Ordering::Relaxed);
-        now - last >= self.flush_interval_ms
-    }
-}
-
-#[cfg(feature = "file")]
-#[async_trait]
-impl Sink for FileSink {
-    async fn write(&self, entry: &UnifiedLogEntry) {
-        let line = format!("{:?}\n", entry);
-        let mut writer = self.writer.lock().await;
-
-        if let Err(e) = writer.write_all(line.as_bytes()).await {
-            eprintln!(
-                "Failed to write log to file {}: {},entry timestamp:{:?}",
-                self.path,
-                e,
-                entry.get_timestamp()
-            );
-            return;
-        }
-
-        // Only flush periodically to improve performance
-        // Logic to determine when to flush could be added here
-        // Increment the entry count
-        self.entry_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-
-        // Check if we should flush
-        if self.should_flush() {
-            if let Err(e) = writer.flush().await {
-                eprintln!("Failed to flush log file {}: {}", self.path, e);
-                return;
-            }
-
-            // Reset counters
-            self.entry_count.store(0, std::sync::atomic::Ordering::Relaxed);
-
-            let now = std::time::SystemTime::now()
-                .duration_since(std::time::UNIX_EPOCH)
-                .unwrap()
-                .as_millis() as u64;
-
-            self.last_flush.store(now, std::sync::atomic::Ordering::Relaxed);
-        }
-    }
-}
-
-#[cfg(feature = "file")]
-impl Drop for FileSink {
-    fn drop(&mut self) {
-        let writer = self.writer.clone();
-        let path = self.path.clone();
-
-        tokio::task::spawn_blocking(move || {
-            let rt = tokio::runtime::Runtime::new().unwrap();
-            rt.block_on(async {
-                let mut writer = writer.lock().await;
-                if let Err(e) = writer.flush().await {
-                    eprintln!("Failed to flush log file {}: {}", path, e);
-                }
-            });
-        });
-    }
-}
-
-/// Create a list of Sink instances
-pub async fn create_sinks(config: &AppConfig) -> Vec<Arc<dyn Sink>> {
-    let mut sinks: Vec<Arc<dyn Sink>> = Vec::new();
-
-    #[cfg(feature = "kafka")]
-    {
-        match &config.sinks.kafka {
-            Some(sink_kafka) => {
-                if sink_kafka.enabled {
-                    match rdkafka::config::ClientConfig::new()
-                        .set("bootstrap.servers", &sink_kafka.bootstrap_servers)
-                        .set("message.timeout.ms", "5000")
-                        .create()
-                    {
-                        Ok(producer) => {
-                            sinks.push(Arc::new(KafkaSink::new(
-                                producer,
-                                sink_kafka.topic.clone(),
-                                sink_kafka.batch_size.unwrap_or(100),
-                                sink_kafka.batch_timeout_ms.unwrap_or(1000),
-                            )));
-                        }
-                        Err(e) => {
-                            tracing::error!("Failed to create Kafka producer: {}", e);
-                        }
-                    }
-                } else {
-                    tracing::info!("Kafka sink is disabled in the configuration");
-                }
-            }
-            _ => {
-                tracing::info!("Kafka sink is not configured or disabled");
-            }
-        }
-    }
-    #[cfg(feature = "webhook")]
-    {
-        match &config.sinks.webhook {
-            Some(sink_webhook) => {
-                if sink_webhook.enabled {
-                    sinks.push(Arc::new(WebhookSink::new(
-                        sink_webhook.endpoint.clone(),
-                        sink_webhook.auth_token.clone(),
-                        sink_webhook.max_retries.unwrap_or(3),
-                        sink_webhook.retry_delay_ms.unwrap_or(100),
-                    )));
-                } else {
-                    tracing::info!("Webhook sink is disabled in the configuration");
-                }
-            }
-            _ => {
-                tracing::info!("Webhook sink is not configured or disabled");
-            }
-        }
-    }
-
-    #[cfg(feature = "file")]
-    {
-        // let config = config.clone();
-        match &config.sinks.file {
-            Some(sink_file) => {
-                tracing::info!("File sink is enabled in the configuration");
-                let path = if sink_file.enabled {
-                    sink_file.path.clone()
-                } else {
-                    "rustfs.log".to_string()
-                };
-                tracing::debug!("FileSink: Using path: {}", path);
-                sinks.push(Arc::new(
-                    FileSink::new(
-                        path.clone(),
-                        sink_file.buffer_size.unwrap_or(8192),
-                        sink_file.flush_interval_ms.unwrap_or(1000),
-                        sink_file.flush_threshold.unwrap_or(100),
-                    )
-                    .await
-                    .unwrap(),
-                ));
-            }
-            _ => {
-                tracing::info!("File sink is not configured or disabled");
-            }
-        }
-    }
-
-    sinks
-}

crates/obs/src/sinks/file.rs

@@ -0,0 +1,164 @@
+use crate::sinks::Sink;
+use crate::{LogRecord, UnifiedLogEntry};
+use async_trait::async_trait;
+use std::sync::Arc;
+use tokio::fs::OpenOptions;
+use tokio::io;
+use tokio::io::AsyncWriteExt;
+
+/// File Sink Implementation
+pub struct FileSink {
+    path: String,
+    buffer_size: usize,
+    writer: Arc<tokio::sync::Mutex<io::BufWriter<tokio::fs::File>>>,
+    entry_count: std::sync::atomic::AtomicUsize,
+    last_flush: std::sync::atomic::AtomicU64,
+    flush_interval_ms: u64, // Time between flushes
+    flush_threshold: usize, // Number of entries before flush
+}
+
+impl FileSink {
+    /// Create a new FileSink instance
+    pub async fn new(
+        path: String,
+        buffer_size: usize,
+        flush_interval_ms: u64,
+        flush_threshold: usize,
+    ) -> Result<Self, io::Error> {
+        // check if the file exists
+        let file_exists = tokio::fs::metadata(&path).await.is_ok();
+        // if the file not exists, create it
+        if !file_exists {
+            tokio::fs::create_dir_all(std::path::Path::new(&path).parent().unwrap()).await?;
+            tracing::debug!("File does not exist, creating it. Path: {:?}", path)
+        }
+        let file = if file_exists {
+            // If the file exists, open it in append mode
+            tracing::debug!("FileSink: File exists, opening in append mode.");
+            OpenOptions::new().append(true).create(true).open(&path).await?
+        } else {
+            // If the file does not exist, create it
+            tracing::debug!("FileSink: File does not exist, creating a new file.");
+            // Create the file and write a header or initial content if needed
+            OpenOptions::new().create(true).truncate(true).write(true).open(&path).await?
+        };
+        let writer = io::BufWriter::with_capacity(buffer_size, file);
+        let now = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_millis() as u64;
+        Ok(FileSink {
+            path,
+            buffer_size,
+            writer: Arc::new(tokio::sync::Mutex::new(writer)),
+            entry_count: std::sync::atomic::AtomicUsize::new(0),
+            last_flush: std::sync::atomic::AtomicU64::new(now),
+            flush_interval_ms,
+            flush_threshold,
+        })
+    }
+
+    #[allow(dead_code)]
+    async fn initialize_writer(&mut self) -> io::Result<()> {
+        let file = tokio::fs::File::create(&self.path).await?;
+
+        // Use buffer_size to create a buffer writer with a specified capacity
+        let buf_writer = io::BufWriter::with_capacity(self.buffer_size, file);
+
+        // Replace the original writer with the new Mutex
+        self.writer = Arc::new(tokio::sync::Mutex::new(buf_writer));
+        Ok(())
+    }
+
+    // Get the current buffer size
+    #[allow(dead_code)]
+    pub fn buffer_size(&self) -> usize {
+        self.buffer_size
+    }
+
+    // How to dynamically adjust the buffer size
+    #[allow(dead_code)]
+    pub async fn set_buffer_size(&mut self, new_size: usize) -> io::Result<()> {
+        if self.buffer_size != new_size {
+            self.buffer_size = new_size;
+            // Reinitialize the writer directly, without checking is_some()
+            self.initialize_writer().await?;
+        }
+        Ok(())
+    }
+
+    // Check if flushing is needed based on count or time
+    fn should_flush(&self) -> bool {
+        // Check entry count threshold
+        if self.entry_count.load(std::sync::atomic::Ordering::Relaxed) >= self.flush_threshold {
+            return true;
+        }
+
+        // Check time threshold
+        let now = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_millis() as u64;
+
+        let last = self.last_flush.load(std::sync::atomic::Ordering::Relaxed);
+        now - last >= self.flush_interval_ms
+    }
+}
+
+#[async_trait]
+impl Sink for FileSink {
+    async fn write(&self, entry: &UnifiedLogEntry) {
+        let line = format!("{:?}\n", entry);
+        let mut writer = self.writer.lock().await;
+
+        if let Err(e) = writer.write_all(line.as_bytes()).await {
+            eprintln!(
+                "Failed to write log to file {}: {},entry timestamp:{:?}",
+                self.path,
+                e,
+                entry.get_timestamp()
+            );
+            return;
+        }
+
+        // Only flush periodically to improve performance
+        // Logic to determine when to flush could be added here
+        // Increment the entry count
+        self.entry_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+
+        // Check if we should flush
+        if self.should_flush() {
+            if let Err(e) = writer.flush().await {
+                eprintln!("Failed to flush log file {}: {}", self.path, e);
+                return;
+            }
+
+            // Reset counters
+            self.entry_count.store(0, std::sync::atomic::Ordering::Relaxed);
+
+            let now = std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .unwrap()
+                .as_millis() as u64;
+
+            self.last_flush.store(now, std::sync::atomic::Ordering::Relaxed);
+        }
+    }
+}
+
+impl Drop for FileSink {
+    fn drop(&mut self) {
+        let writer = self.writer.clone();
+        let path = self.path.clone();
+
+        tokio::task::spawn_blocking(move || {
+            let rt = tokio::runtime::Runtime::new().unwrap();
+            rt.block_on(async {
+                let mut writer = writer.lock().await;
+                if let Err(e) = writer.flush().await {
+                    eprintln!("Failed to flush log file {}: {}", path, e);
+                }
+            });
+        });
+    }
+}

crates/obs/src/sinks/kafka.rs

@@ -0,0 +1,165 @@
+use crate::sinks::Sink;
+use crate::{LogRecord, UnifiedLogEntry};
+use async_trait::async_trait;
+use std::sync::Arc;
+
+/// Kafka Sink Implementation
+pub struct KafkaSink {
+    producer: rdkafka::producer::FutureProducer,
+    topic: String,
+    batch_size: usize,
+    batch_timeout_ms: u64,
+    entries: Arc<tokio::sync::Mutex<Vec<UnifiedLogEntry>>>,
+    last_flush: Arc<std::sync::atomic::AtomicU64>,
+}
+
+impl KafkaSink {
+    /// Create a new KafkaSink instance
+    pub fn new(producer: rdkafka::producer::FutureProducer, topic: String, batch_size: usize, batch_timeout_ms: u64) -> Self {
+        // Create Arc-wrapped values first
+        let entries = Arc::new(tokio::sync::Mutex::new(Vec::with_capacity(batch_size)));
+        let last_flush = Arc::new(std::sync::atomic::AtomicU64::new(
+            std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .unwrap()
+                .as_millis() as u64,
+        ));
+        let sink = KafkaSink {
+            producer: producer.clone(),
+            topic: topic.clone(),
+            batch_size,
+            batch_timeout_ms,
+            entries: entries.clone(),
+            last_flush: last_flush.clone(),
+        };
+
+        // Start background flusher
+        tokio::spawn(Self::periodic_flush(producer, topic, entries, last_flush, batch_timeout_ms));
+
+        sink
+    }
+
+    /// Add a getter method to read the batch_timeout_ms field
+    #[allow(dead_code)]
+    pub fn batch_timeout(&self) -> u64 {
+        self.batch_timeout_ms
+    }
+
+    /// Add a method to dynamically adjust the timeout if needed
+    #[allow(dead_code)]
+    pub fn set_batch_timeout(&mut self, new_timeout_ms: u64) {
+        self.batch_timeout_ms = new_timeout_ms;
+    }
+
+    async fn periodic_flush(
+        producer: rdkafka::producer::FutureProducer,
+        topic: String,
+        entries: Arc<tokio::sync::Mutex<Vec<UnifiedLogEntry>>>,
+        last_flush: Arc<std::sync::atomic::AtomicU64>,
+        timeout_ms: u64,
+    ) {
+        loop {
+            tokio::time::sleep(tokio::time::Duration::from_millis(timeout_ms / 2)).await;
+
+            let now = std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .unwrap()
+                .as_millis() as u64;
+
+            let last = last_flush.load(std::sync::atomic::Ordering::Relaxed);
+
+            if now - last >= timeout_ms {
+                let mut batch = entries.lock().await;
+                if !batch.is_empty() {
+                    Self::send_batch(&producer, &topic, batch.drain(..).collect()).await;
+                    last_flush.store(now, std::sync::atomic::Ordering::Relaxed);
+                }
+            }
+        }
+    }
+
+    async fn send_batch(producer: &rdkafka::producer::FutureProducer, topic: &str, entries: Vec<UnifiedLogEntry>) {
+        for entry in entries {
+            let payload = match serde_json::to_string(&entry) {
+                Ok(p) => p,
+                Err(e) => {
+                    eprintln!("Failed to serialize log entry: {}", e);
+                    continue;
+                }
+            };
+
+            let span_id = entry.get_timestamp().to_rfc3339();
+
+            let _ = producer
+                .send(
+                    rdkafka::producer::FutureRecord::to(topic).payload(&payload).key(&span_id),
+                    std::time::Duration::from_secs(5),
+                )
+                .await;
+        }
+    }
+}
+
+#[async_trait]
+impl Sink for KafkaSink {
+    async fn write(&self, entry: &UnifiedLogEntry) {
+        let mut batch = self.entries.lock().await;
+        batch.push(entry.clone());
+
+        let should_flush_by_size = batch.len() >= self.batch_size;
+        let should_flush_by_time = {
+            let now = std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .unwrap()
+                .as_millis() as u64;
+            let last = self.last_flush.load(std::sync::atomic::Ordering::Relaxed);
+            now - last >= self.batch_timeout_ms
+        };
+
+        if should_flush_by_size || should_flush_by_time {
+            // Existing flush logic
+            let entries_to_send: Vec<UnifiedLogEntry> = batch.drain(..).collect();
+            let producer = self.producer.clone();
+            let topic = self.topic.clone();
+
+            self.last_flush.store(
+                std::time::SystemTime::now()
+                    .duration_since(std::time::UNIX_EPOCH)
+                    .unwrap()
+                    .as_millis() as u64,
+                std::sync::atomic::Ordering::Relaxed,
+            );
+
+            tokio::spawn(async move {
+                KafkaSink::send_batch(&producer, &topic, entries_to_send).await;
+            });
+        }
+    }
+}
+
+impl Drop for KafkaSink {
+    fn drop(&mut self) {
+        // Perform any necessary cleanup here
+        // For example, you might want to flush any remaining entries
+        let producer = self.producer.clone();
+        let topic = self.topic.clone();
+        let entries = self.entries.clone();
+        let last_flush = self.last_flush.clone();
+
+        tokio::spawn(async move {
+            let mut batch = entries.lock().await;
+            if !batch.is_empty() {
+                KafkaSink::send_batch(&producer, &topic, batch.drain(..).collect()).await;
+                last_flush.store(
+                    std::time::SystemTime::now()
+                        .duration_since(std::time::UNIX_EPOCH)
+                        .unwrap()
+                        .as_millis() as u64,
+                    std::sync::atomic::Ordering::Relaxed,
+                );
+            }
+        });
+
+        eprintln!("Dropping KafkaSink with topic: {}", self.topic);
+    }
+}

crates/obs/src/sinks/mod.rs

@@ -0,0 +1,92 @@
+use crate::{AppConfig, SinkConfig, UnifiedLogEntry};
+use async_trait::async_trait;
+use std::sync::Arc;
+
+#[cfg(feature = "file")]
+mod file;
+#[cfg(all(feature = "kafka", target_os = "linux"))]
+mod kafka;
+#[cfg(feature = "webhook")]
+mod webhook;
+
+/// Sink Trait definition, asynchronously write logs
+#[async_trait]
+pub trait Sink: Send + Sync {
+    async fn write(&self, entry: &UnifiedLogEntry);
+}
+
+/// Create a list of Sink instances
+pub async fn create_sinks(config: &AppConfig) -> Vec<Arc<dyn Sink>> {
+    let mut sinks: Vec<Arc<dyn Sink>> = Vec::new();
+
+    for sink_config in &config.sinks {
+        match sink_config {
+            #[cfg(all(feature = "kafka", target_os = "linux"))]
+            SinkConfig::Kafka(kafka_config) => {
+                match rdkafka::config::ClientConfig::new()
+                    .set("bootstrap.servers", &kafka_config.brokers)
+                    .set("message.timeout.ms", "5000")
+                    .create()
+                {
+                    Ok(producer) => {
+                        sinks.push(Arc::new(kafka::KafkaSink::new(
+                            producer,
+                            kafka_config.topic.clone(),
+                            kafka_config.batch_size.unwrap_or(100),
+                            kafka_config.batch_timeout_ms.unwrap_or(1000),
+                        )));
+                        tracing::info!("Kafka sink created for topic: {}", kafka_config.topic);
+                    }
+                    Err(e) => {
+                        tracing::error!("Failed to create Kafka producer: {}", e);
+                    }
+                }
+            }
+            #[cfg(feature = "webhook")]
+            SinkConfig::Webhook(webhook_config) => {
+                sinks.push(Arc::new(webhook::WebhookSink::new(
+                    webhook_config.endpoint.clone(),
+                    webhook_config.auth_token.clone(),
+                    webhook_config.max_retries.unwrap_or(3),
+                    webhook_config.retry_delay_ms.unwrap_or(100),
+                )));
+                tracing::info!("Webhook sink created for endpoint: {}", webhook_config.endpoint);
+            }
+
+            #[cfg(feature = "file")]
+            SinkConfig::File(file_config) => {
+                tracing::debug!("FileSink: Using path: {}", file_config.path);
+                match file::FileSink::new(
+                    file_config.path.clone(),
+                    file_config.buffer_size.unwrap_or(8192),
+                    file_config.flush_interval_ms.unwrap_or(1000),
+                    file_config.flush_threshold.unwrap_or(100),
+                )
+                .await
+                {
+                    Ok(sink) => {
+                        sinks.push(Arc::new(sink));
+                        tracing::info!("File sink created for path: {}", file_config.path);
+                    }
+                    Err(e) => {
+                        tracing::error!("Failed to create File sink: {}", e);
+                    }
+                }
+            }
+            #[cfg(any(not(feature = "kafka"), not(target_os = "linux")))]
+            SinkConfig::Kafka(_) => {
+                tracing::warn!("Kafka sink is configured but the 'kafka' feature is not enabled");
+            }
+            #[cfg(not(feature = "webhook"))]
+            SinkConfig::Webhook(_) => {
+                tracing::warn!("Webhook sink is configured but the 'webhook' feature is not enabled");
+            }
+            #[cfg(not(feature = "file"))]
+            SinkConfig::File(_) => {
+                tracing::warn!("File sink is configured but the 'file' feature is not enabled");
+            }
+        }
+    }
+
+    sinks
+}

crates/obs/src/sinks/webhook.rs

@@ -0,0 +1,70 @@
+use crate::sinks::Sink;
+use crate::UnifiedLogEntry;
+use async_trait::async_trait;
+
+/// Webhook Sink Implementation
+pub struct WebhookSink {
+    endpoint: String,
+    auth_token: String,
+    client: reqwest::Client,
+    max_retries: usize,
+    retry_delay_ms: u64,
+}
+
+impl WebhookSink {
+    pub fn new(endpoint: String, auth_token: String, max_retries: usize, retry_delay_ms: u64) -> Self {
+        WebhookSink {
+            endpoint,
+            auth_token,
+            client: reqwest::Client::builder()
+                .timeout(std::time::Duration::from_secs(10))
+                .build()
+                .unwrap_or_else(|_| reqwest::Client::new()),
+            max_retries,
+            retry_delay_ms,
+        }
+    }
+}
+
+#[async_trait]
+impl Sink for WebhookSink {
+    async fn write(&self, entry: &UnifiedLogEntry) {
+        let mut retries = 0;
+        let url = self.endpoint.clone();
+        let entry_clone = entry.clone();
+        let auth_value = reqwest::header::HeaderValue::from_str(format!("Bearer {}", self.auth_token.clone()).as_str()).unwrap();
+        while retries < self.max_retries {
+            match self
+                .client
+                .post(&url)
+                .header(reqwest::header::AUTHORIZATION, auth_value.clone())
+                .json(&entry_clone)
+                .send()
+                .await
+            {
+                Ok(response) if response.status().is_success() => {
+                    return;
+                }
+                _ => {
+                    retries += 1;
+                    if retries < self.max_retries {
+                        tokio::time::sleep(tokio::time::Duration::from_millis(
+                            self.retry_delay_ms * (1 << retries), // Exponential backoff
+                        ))
+                        .await;
+                    }
+                }
+            }
+        }
+
+        eprintln!("Failed to send log to webhook after {} retries", self.max_retries);
+    }
+}
+
+impl Drop for WebhookSink {
+    fn drop(&mut self) {
+        // Perform any necessary cleanup here
+        // For example, you might want to log that the sink is being dropped
+        eprintln!("Dropping WebhookSink with URL: {}", self.endpoint);
+    }
+}

crates/obs/src/telemetry.rs

@@ -1,5 +1,3 @@
-use crate::global::{ENVIRONMENT, LOGGER_LEVEL, METER_INTERVAL, SAMPLE_RATIO, SERVICE_NAME, SERVICE_VERSION, USE_STDOUT};
-use crate::utils::get_local_ip_with_default;
 use crate::OtelConfig;
 use opentelemetry::trace::TracerProvider;
 use opentelemetry::{global, KeyValue};
@@ -15,6 +13,8 @@ use opentelemetry_semantic_conventions::{
     attribute::{DEPLOYMENT_ENVIRONMENT_NAME, NETWORK_LOCAL_ADDRESS, SERVICE_VERSION as OTEL_SERVICE_VERSION},
     SCHEMA_URL,
 };
+use rustfs_config::{APP_NAME, DEFAULT_LOG_LEVEL, ENVIRONMENT, METER_INTERVAL, SAMPLE_RATIO, SERVICE_VERSION, USE_STDOUT};
+use rustfs_utils::get_local_ip_with_default;
 use smallvec::SmallVec;
 use std::borrow::Cow;
 use std::io::IsTerminal;
@@ -70,7 +70,7 @@ impl Drop for OtelGuard {
 /// create OpenTelemetry Resource
 fn resource(config: &OtelConfig) -> Resource {
     Resource::builder()
-        .with_service_name(Cow::Borrowed(config.service_name.as_deref().unwrap_or(SERVICE_NAME)).to_string())
+        .with_service_name(Cow::Borrowed(config.service_name.as_deref().unwrap_or(APP_NAME)).to_string())
         .with_schema_url(
             [
                 KeyValue::new(
@@ -101,8 +101,8 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard {
     let endpoint = &config.endpoint;
     let use_stdout = config.use_stdout.unwrap_or(USE_STDOUT);
     let meter_interval = config.meter_interval.unwrap_or(METER_INTERVAL);
-    let logger_level = config.logger_level.as_deref().unwrap_or(LOGGER_LEVEL);
-    let service_name = config.service_name.as_deref().unwrap_or(SERVICE_NAME);
+    let logger_level = config.logger_level.as_deref().unwrap_or(DEFAULT_LOG_LEVEL);
+    let service_name = config.service_name.as_deref().unwrap_or(APP_NAME);
 
     // Pre-create resource objects to avoid repeated construction
     let res = resource(config);
@@ -210,7 +210,8 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard {
             .with_thread_names(true)
             .with_thread_ids(true)
             .with_file(true)
-            .with_line_number(true);
+            .with_line_number(true)
+            .with_filter(build_env_filter(logger_level, None));
 
         let filter = build_env_filter(logger_level, None);
         let otel_filter = build_env_filter(logger_level, None);
@@ -218,7 +219,7 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard {
         let tracer = tracer_provider.tracer(Cow::Borrowed(service_name).to_string());
 
         // Configure registry to avoid repeated calls to filter methods
-        let _registry = tracing_subscriber::registry()
+        tracing_subscriber::registry()
             .with(filter)
             .with(ErrorLayer::default())
             .with(if config.local_logging_enabled.unwrap_or(false) {
@@ -230,10 +231,13 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard {
             .with(otel_layer)
             .with(MetricsLayer::new(meter_provider.clone()))
             .init();
+
         if !endpoint.is_empty() {
             info!(
-                "OpenTelemetry telemetry initialized with OTLP endpoint: {}, logger_level: {}",
-                endpoint, logger_level
+                "OpenTelemetry telemetry initialized with OTLP endpoint: {}, logger_level: {},RUST_LOG env: {}",
+                endpoint,
+                logger_level,
+                std::env::var("RUST_LOG").unwrap_or_else(|_| "未设置".to_string())
             );
         }
     }
@@ -248,7 +252,6 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard {
 fn build_env_filter(logger_level: &str, default_level: Option<&str>) -> EnvFilter {
     let level = default_level.unwrap_or(logger_level);
     let mut filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(level));
-
     if !matches!(logger_level, "trace" | "debug") {
         let directives: SmallVec<[&str; 5]> = smallvec::smallvec!["hyper", "tonic", "h2", "reqwest", "tower"];
         for directive in directives {

crates/obs/src/utils.rs

@@ -1,42 +0,0 @@
-use local_ip_address::{local_ip, local_ipv6};
-use std::net::{IpAddr, Ipv4Addr};
-
-/// Get the IP address of the machine
-///
-/// Priority is given to trying to get the IPv4 address, and if it fails, try to get the IPv6 address.
-/// If both fail to retrieve, None is returned.
-///
-/// # Returns
-///
-/// * `Some(IpAddr)` - Native IP address (IPv4 or IPv6)
-/// * `None` - Unable to obtain any native IP address
-pub fn get_local_ip() -> Option<IpAddr> {
-    local_ip().ok().or_else(|| local_ipv6().ok())
-}
-
-/// Get the IP address of the machine as a string
-///
-/// If the IP address cannot be obtained, returns "127.0.0.1" as the default value.
-///
-/// # Returns
-///
-/// * `String` - Native IP address (IPv4 or IPv6) as a string, or the default value
-pub fn get_local_ip_with_default() -> String {
-    get_local_ip()
-        .unwrap_or_else(|| IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1))) // Provide a safe default value
-        .to_string()
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_get_local_ip() {
-        match get_local_ip() {
-            Some(ip) => println!("the ip address of this machine:{}", ip),
-            None => println!("Unable to obtain the IP address of the machine"),
-        }
-        assert!(get_local_ip().is_some());
-    }
-}

crates/obs/src/worker.rs

@@ -1,9 +1,9 @@
-use crate::{sink::Sink, UnifiedLogEntry};
+use crate::{sinks::Sink, UnifiedLogEntry};
 use std::sync::Arc;
 use tokio::sync::mpsc::Receiver;
 
 /// Start the log processing worker thread
-pub async fn start_worker(receiver: Receiver<UnifiedLogEntry>, sinks: Vec<Arc<dyn Sink>>) {
+pub(crate) async fn start_worker(receiver: Receiver<UnifiedLogEntry>, sinks: Vec<Arc<dyn Sink>>) {
     let mut receiver = receiver;
     while let Some(entry) = receiver.recv().await {
         for sink in &sinks {