feat: add TraceLayer for HTTP service and improve metrics

- Add TraceLayer to HTTP server for request tracing
- Implement system metrics for process monitoring
- Optimize init_telemetry method for better resource management
- Add graceful shutdown handling for telemetry components
- Fix GracefulShutdown ownership issues with Arc wrapper

Cargo.lock

@@ -525,7 +525,6 @@ version = "0.4.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "06575e6a9673580f52661c92107baabffbf41e2141373441cbcdc47cb733003c"
 dependencies = [
- "brotli",
  "bzip2",
  "flate2",
  "futures-core",

crates/obs/examples/config.toml

@@ -25,7 +25,7 @@ batch_timeout_ms = 1000 # Default is 100ms if not specified
 
 [sinks.file]
 enabled = true
-path = "logs/app.log"
+path = "deploy/logs/app.log"
 batch_size = 100
 batch_timeout_ms = 1000 # Default is 8192 bytes if not specified
 

crates/obs/examples/server.rs

@@ -1,8 +1,8 @@
 use opentelemetry::global;
-use rustfs_obs::{get_logger, init_obs, load_config, log_info, BaseLogEntry, ServerLogEntry};
+use rustfs_obs::{get_logger, init_obs, init_process_observer, load_config, log_info, BaseLogEntry, ServerLogEntry};
 use std::collections::HashMap;
 use std::time::{Duration, SystemTime};
-use tracing::{info, instrument};
+use tracing::{error, info, instrument};
 use tracing_core::Level;
 
 #[tokio::main]
@@ -38,6 +38,11 @@ async fn run(bucket: String, object: String, user: String, service_name: String)
         &[opentelemetry::KeyValue::new("operation", "run")],
     );
 
+    match init_process_observer(meter).await {
+        Ok(_) => info!("Process observer initialized successfully"),
+        Err(e) => error!("Failed to initialize process observer: {:?}", e),
+    }
+
     let base_entry = BaseLogEntry::new()
         .message(Some("run logger api_handler info".to_string()))
         .request_id(Some("request_id".to_string()))

crates/obs/src/global.rs

@@ -23,17 +23,20 @@ pub enum GlobalError {
     NotInitialized,
     #[error("Global system metrics err: {0}")]
     MetricsError(String),
-    #[error("Failed to get process ID: {0}")]
+    #[error("Failed to get current PID: {0}")]
-    GetPidError(String),
+    PidError(String),
-    #[error("Type conversion error: {0}")]
+    #[error("Process with PID {0} not found")]
-    ConversionError(String),
+    ProcessNotFound(u32),
-    #[error("IO error: {0}")]
+    #[error("Failed to get physical core count")]
-    IoError(#[from] std::io::Error),
+    CoreCountError,
-    #[error("System metrics error: {0}")]
+    #[error("GPU initialization failed: {0}")]
-    SystemMetricsError(String),
+    GpuInitError(String),
-    #[cfg(feature = "gpu")]
+    #[error("GPU device not found: {0}")]
-    #[error("GPU metrics error: {0}")]
+    GpuDeviceError(String),
-    GpuMetricsError(String),
+    #[error("Failed to send log: {0}")]
+    SendFailed(&'static str),
+    #[error("Operation timed out: {0}")]
+    Timeout(&'static str),
 }
 
 /// Set the global guard for OpenTelemetry

crates/obs/src/lib.rs

@@ -32,34 +32,35 @@ mod config;
 mod entry;
 mod global;
 mod logger;
-mod metrics;
 mod sink;
+mod system;
 mod telemetry;
 mod utils;
 mod worker;
 
-#[cfg(feature = "gpu")]
+use crate::logger::InitLogStatus;
-pub use crate::metrics::init_gpu_metrics;
-
 pub use config::load_config;
-pub use config::{AppConfig, OtelConfig};
+#[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};
 pub use entry::base::BaseLogEntry;
 pub use entry::unified::{ConsoleLogEntry, ServerLogEntry, UnifiedLogEntry};
 pub use entry::{LogKind, LogRecord, ObjectVersion, SerializableLevel};
 pub use global::{get_global_guard, set_global_guard, try_get_global_guard, GlobalError};
-pub use logger::{ensure_logger_initialized, log_debug, log_error, log_info, log_trace, log_warn, log_with_context};
+pub use logger::Logger;
-pub use logger::{get_global_logger, init_global_logger, locked_logger, start_logger};
+pub use logger::{get_global_logger, init_global_logger, start_logger};
-pub use logger::{log_init_state, InitLogStatus};
+pub use logger::{log_debug, log_error, log_info, log_trace, log_warn, log_with_context};
-pub use logger::{LogError, Logger};
-pub use metrics::{init_system_metrics, init_system_metrics_for_pid};
-pub use sink::Sink;
 use std::sync::Arc;
+pub use system::{init_process_observer, init_process_observer_for_pid};
 pub use telemetry::init_telemetry;
 use tokio::sync::Mutex;
-pub use utils::{get_local_ip, get_local_ip_with_default};
+use tracing::{error, info};
-pub use worker::start_worker;
 
 /// Initialize the observability module
 ///
@@ -80,6 +81,19 @@ pub async fn init_obs(config: AppConfig) -> (Arc<Mutex<Logger>>, telemetry::Otel
     let guard = init_telemetry(&config.observability);
     let sinks = sink::create_sinks(&config).await;
     let logger = init_global_logger(&config, sinks).await;
+    let obs_config = config.observability.clone();
+    tokio::spawn(async move {
+        let result = InitLogStatus::init_start_log(&obs_config).await;
+        match result {
+            Ok(_) => {
+                info!("Logger initialized successfully");
+            }
+            Err(e) => {
+                error!("Failed to initialize logger: {}", e);
+            }
+        }
+    });
+
     (logger, guard)
 }
 

crates/obs/src/logger.rs

@@ -1,5 +1,6 @@
 use crate::global::{ENVIRONMENT, SERVICE_NAME, SERVICE_VERSION};
-use crate::{AppConfig, AuditLogEntry, BaseLogEntry, ConsoleLogEntry, OtelConfig, ServerLogEntry, Sink, UnifiedLogEntry};
+use crate::sink::Sink;
+use crate::{AppConfig, AuditLogEntry, BaseLogEntry, ConsoleLogEntry, GlobalError, OtelConfig, ServerLogEntry, UnifiedLogEntry};
 use std::sync::Arc;
 use std::time::SystemTime;
 use tokio::sync::mpsc::{self, Receiver, Sender};
@@ -43,26 +44,26 @@ impl Logger {
 
     /// Log a server entry
     #[tracing::instrument(skip(self), fields(log_source = "logger_server"))]
-    pub async fn log_server_entry(&self, entry: ServerLogEntry) -> Result<(), LogError> {
+    pub async fn log_server_entry(&self, entry: ServerLogEntry) -> Result<(), GlobalError> {
         self.log_entry(UnifiedLogEntry::Server(entry)).await
     }
 
     /// Log an audit entry
     #[tracing::instrument(skip(self), fields(log_source = "logger_audit"))]
-    pub async fn log_audit_entry(&self, entry: AuditLogEntry) -> Result<(), LogError> {
+    pub async fn log_audit_entry(&self, entry: AuditLogEntry) -> Result<(), GlobalError> {
         self.log_entry(UnifiedLogEntry::Audit(Box::new(entry))).await
     }
 
     /// Log a console entry
     #[tracing::instrument(skip(self), fields(log_source = "logger_console"))]
-    pub async fn log_console_entry(&self, entry: ConsoleLogEntry) -> Result<(), LogError> {
+    pub async fn log_console_entry(&self, entry: ConsoleLogEntry) -> Result<(), GlobalError> {
         self.log_entry(UnifiedLogEntry::Console(entry)).await
     }
 
     /// Asynchronous logging of unified log entries
     #[tracing::instrument(skip(self), fields(log_source = "logger"))]
     #[tracing::instrument(level = "error", skip_all)]
-    pub async fn log_entry(&self, entry: UnifiedLogEntry) -> Result<(), LogError> {
+    pub async fn log_entry(&self, entry: UnifiedLogEntry) -> Result<(), GlobalError> {
         // Extract information for tracing based on entry type
         match &entry {
             UnifiedLogEntry::Server(server) => {
@@ -123,11 +124,11 @@ impl Logger {
                 tracing::warn!("Log queue full, applying backpressure");
                 match tokio::time::timeout(std::time::Duration::from_millis(500), self.sender.send(entry)).await {
                     Ok(Ok(_)) => Ok(()),
-                    Ok(Err(_)) => Err(LogError::SendFailed("Channel closed")),
+                    Ok(Err(_)) => Err(GlobalError::SendFailed("Channel closed")),
-                    Err(_) => Err(LogError::Timeout("Queue backpressure timeout")),
+                    Err(_) => Err(GlobalError::Timeout("Queue backpressure timeout")),
                 }
             }
-            Err(mpsc::error::TrySendError::Closed(_)) => Err(LogError::SendFailed("Logger channel closed")),
+            Err(mpsc::error::TrySendError::Closed(_)) => Err(GlobalError::SendFailed("Logger channel closed")),
         }
     }
 
@@ -160,7 +161,7 @@ impl Logger {
         request_id: Option<String>,
         user_id: Option<String>,
         fields: Vec<(String, String)>,
-    ) -> Result<(), LogError> {
+    ) -> Result<(), GlobalError> {
         let base = BaseLogEntry::new().message(Some(message.to_string())).request_id(request_id);
 
         let server_entry = ServerLogEntry::new(level, source.to_string())
@@ -190,7 +191,7 @@ impl Logger {
     ///   let _ = logger.write("This is an information message", "example", Level::INFO).await;
     /// }
     /// ```
-    pub async fn write(&self, message: &str, source: &str, level: Level) -> Result<(), LogError> {
+    pub async fn write(&self, message: &str, source: &str, level: Level) -> Result<(), GlobalError> {
         self.write_with_context(message, source, level, None, None, Vec::new()).await
     }
 
@@ -208,31 +209,12 @@ impl Logger {
     ///  let _ = logger.shutdown().await;
     /// }
     /// ```
-    pub async fn shutdown(self) -> Result<(), LogError> {
+    pub async fn shutdown(self) -> Result<(), GlobalError> {
         drop(self.sender); //Close the sending end so that the receiver knows that there is no new message
         Ok(())
     }
 }
 
-/// Log error type
-/// This enum defines the error types that can occur when logging.
-/// It is used to provide more detailed error information.
-/// # Example
-/// ```
-/// use rustfs_obs::LogError;
-/// use thiserror::Error;
-///
-/// LogError::SendFailed("Failed to send log");
-/// LogError::Timeout("Operation timed out");
-/// ```
-#[derive(Debug, thiserror::Error)]
-pub enum LogError {
-    #[error("Failed to send log: {0}")]
-    SendFailed(&'static str),
-    #[error("Operation timed out: {0}")]
-    Timeout(&'static str),
-}
-
 /// Start the log module
 /// This function starts the log module.
 /// It initializes the logger and starts the worker to process logs.
@@ -297,48 +279,6 @@ pub fn get_global_logger() -> &'static Arc<Mutex<Logger>> {
     GLOBAL_LOGGER.get().expect("Logger not initialized")
 }
 
-/// Get the global logger instance with a lock
-/// This function returns a reference to the global logger instance with a lock.
-/// It is used to ensure that the logger is thread-safe.
-///
-/// # Returns
-/// A reference to the global logger instance with a lock
-///
-/// # Example
-/// ```
-/// use rustfs_obs::locked_logger;
-///
-/// async fn example() {
-///     let logger = locked_logger().await;
-/// }
-/// ```
-pub async fn locked_logger() -> tokio::sync::MutexGuard<'static, Logger> {
-    get_global_logger().lock().await
-}
-
-/// Initialize with default empty logger if needed (optional)
-/// This function initializes the logger with a default empty logger if needed.
-/// It is used to ensure that the logger is initialized before logging.
-///
-/// # Returns
-/// A reference to the global logger instance
-///
-/// # Example
-/// ```
-/// use rustfs_obs::ensure_logger_initialized;
-///
-/// let logger = ensure_logger_initialized();
-/// ```
-pub fn ensure_logger_initialized() -> &'static Arc<Mutex<Logger>> {
-    if GLOBAL_LOGGER.get().is_none() {
-        let config = AppConfig::default();
-        let sinks = vec![];
-        let logger = Arc::new(Mutex::new(start_logger(&config, sinks)));
-        let _ = GLOBAL_LOGGER.set(logger);
-    }
-    GLOBAL_LOGGER.get().unwrap()
-}
-
 /// Log information
 /// This function logs information messages.
 ///
@@ -357,7 +297,7 @@ pub fn ensure_logger_initialized() -> &'static Arc<Mutex<Logger>> {
 ///    let _ = log_info("This is an information message", "example").await;
 /// }
 /// ```
-pub async fn log_info(message: &str, source: &str) -> Result<(), LogError> {
+pub async fn log_info(message: &str, source: &str) -> Result<(), GlobalError> {
     get_global_logger().lock().await.write(message, source, Level::INFO).await
 }
 
@@ -375,7 +315,7 @@ pub async fn log_info(message: &str, source: &str) -> Result<(), LogError> {
 /// async fn example() {
 ///     let _ = log_error("This is an error message", "example").await;
 /// }
-pub async fn log_error(message: &str, source: &str) -> Result<(), LogError> {
+pub async fn log_error(message: &str, source: &str) -> Result<(), GlobalError> {
     get_global_logger().lock().await.write(message, source, Level::ERROR).await
 }
 
@@ -395,7 +335,7 @@ pub async fn log_error(message: &str, source: &str) -> Result<(), LogError> {
 ///     let _ = log_warn("This is a warning message", "example").await;
 /// }
 /// ```
-pub async fn log_warn(message: &str, source: &str) -> Result<(), LogError> {
+pub async fn log_warn(message: &str, source: &str) -> Result<(), GlobalError> {
     get_global_logger().lock().await.write(message, source, Level::WARN).await
 }
 
@@ -415,7 +355,7 @@ pub async fn log_warn(message: &str, source: &str) -> Result<(), LogError> {
 ///     let _ = log_debug("This is a debug message", "example").await;
 /// }
 /// ```
-pub async fn log_debug(message: &str, source: &str) -> Result<(), LogError> {
+pub async fn log_debug(message: &str, source: &str) -> Result<(), GlobalError> {
     get_global_logger().lock().await.write(message, source, Level::DEBUG).await
 }
 
@@ -436,7 +376,7 @@ pub async fn log_debug(message: &str, source: &str) -> Result<(), LogError> {
 ///    let _ = log_trace("This is a trace message", "example").await;
 /// }
 /// ```
-pub async fn log_trace(message: &str, source: &str) -> Result<(), LogError> {
+pub async fn log_trace(message: &str, source: &str) -> Result<(), GlobalError> {
     get_global_logger().lock().await.write(message, source, Level::TRACE).await
 }
 
@@ -467,7 +407,7 @@ pub async fn log_with_context(
     request_id: Option<String>,
     user_id: Option<String>,
     fields: Vec<(String, String)>,
-) -> Result<(), LogError> {
+) -> Result<(), GlobalError> {
     get_global_logger()
         .lock()
         .await
@@ -477,7 +417,7 @@ pub async fn log_with_context(
 
 /// Log initialization status
 #[derive(Debug)]
-pub struct InitLogStatus {
+pub(crate) struct InitLogStatus {
     pub timestamp: SystemTime,
     pub service_name: String,
     pub version: String,
@@ -508,14 +448,14 @@ impl InitLogStatus {
         }
     }
 
-    pub async fn init_start_log(config: &OtelConfig) -> Result<(), LogError> {
+    pub async fn init_start_log(config: &OtelConfig) -> Result<(), GlobalError> {
         let status = Self::new_config(config);
         log_init_state(Some(status)).await
     }
 }
 
 /// Log initialization details during system startup
-pub async fn log_init_state(status: Option<InitLogStatus>) -> Result<(), LogError> {
+async fn log_init_state(status: Option<InitLogStatus>) -> Result<(), GlobalError> {
     let status = status.unwrap_or_default();
 
     let base_entry = BaseLogEntry::new()

crates/obs/src/metrics.rs

@@ -1,394 +0,0 @@
-//! # Metrics
-//! This file is part of the RustFS project
-//! Current metrics observed are:
-//! - CPU
-//! - Memory
-//! - Disk
-//! - Network
-//!
-//! # Getting started
-//!
-//! ```
-//! use opentelemetry::global;
-//! use rustfs_obs::init_system_metrics;
-//!
-//! #[tokio::main]
-//! async fn main() {
-//!     let meter = global::meter("rustfs-system-meter");
-//!     let result = init_system_metrics(meter);
-//! }
-//! ```
-//!
-
-use crate::GlobalError;
-#[cfg(feature = "gpu")]
-use nvml_wrapper::enums::device::UsedGpuMemory;
-#[cfg(feature = "gpu")]
-use nvml_wrapper::Nvml;
-use opentelemetry::metrics::Meter;
-use opentelemetry::Key;
-use opentelemetry::KeyValue;
-use std::time::Duration;
-use sysinfo::{get_current_pid, System};
-use tokio::time::sleep;
-use tracing::warn;
-
-const PROCESS_PID: Key = Key::from_static_str("process.pid");
-const PROCESS_EXECUTABLE_NAME: Key = Key::from_static_str("process.executable.name");
-const PROCESS_EXECUTABLE_PATH: Key = Key::from_static_str("process.executable.path");
-const PROCESS_COMMAND: Key = Key::from_static_str("process.command");
-const PROCESS_CPU_USAGE: &str = "process.cpu.usage";
-const PROCESS_CPU_UTILIZATION: &str = "process.cpu.utilization";
-const PROCESS_MEMORY_USAGE: &str = "process.memory.usage";
-const PROCESS_MEMORY_VIRTUAL: &str = "process.memory.virtual";
-const PROCESS_DISK_IO: &str = "process.disk.io";
-const DIRECTION: Key = Key::from_static_str("direction");
-const PROCESS_GPU_MEMORY_USAGE: &str = "process.gpu.memory.usage";
-
-// add static variables that delay initialize nvml
-#[cfg(feature = "gpu")]
-static NVML_INSTANCE: OnceCell<Arc<Mutex<Option<Result<Nvml, nvml_wrapper::error::NvmlError>>>>> = OnceCell::const_new();
-
-// get or initialize an nvml instance
-#[cfg(feature = "gpu")]
-async fn get_or_init_nvml() -> &'static Arc<Mutex<Option<Result<Nvml, nvml_wrapper::error::NvmlError>>>> {
-    NVML_INSTANCE
-        .get_or_init(|| async { Arc::new(Mutex::new(Some(Nvml::init()))) })
-        .await
-}
-
-/// Record asynchronously information about the current process.
-/// This function is useful for monitoring the current process.
-///
-/// # Arguments
-/// * `meter` - The OpenTelemetry meter to use.
-///
-/// # Returns
-/// * `Ok(())` if successful
-/// * `Err(GlobalError)` if an error occurs
-///
-/// # Example
-/// ```
-/// use opentelemetry::global;
-/// use rustfs_obs::init_system_metrics;
-///
-/// #[tokio::main]
-/// async fn main() {
-///     let meter = global::meter("rustfs-system-meter");
-///     let result = init_system_metrics(meter);
-/// }
-/// ```
-pub async fn init_system_metrics(meter: Meter) -> Result<(), GlobalError> {
-    let pid = get_current_pid().map_err(|err| GlobalError::MetricsError(err.to_string()))?;
-    register_system_metrics(meter, pid).await
-}
-
-/// Record asynchronously information about a specific process by its PID.
-/// This function is useful for monitoring processes other than the current one.
-///
-/// # Arguments
-/// * `meter` - The OpenTelemetry meter to use.
-/// * `pid` - The PID of the process to monitor.
-///
-/// # Returns
-/// * `Ok(())` if successful
-/// * `Err(GlobalError)` if an error occurs
-///
-/// # Example
-/// ```
-/// use opentelemetry::global;
-/// use rustfs_obs::init_system_metrics_for_pid;
-///
-/// #[tokio::main]
-/// async fn main() {
-///     let meter = global::meter("rustfs-system-meter");
-///     // replace with the actual PID
-///     let pid = 1234;
-///     let result = init_system_metrics_for_pid(meter, pid).await;
-/// }
-/// ```
-///
-pub async fn init_system_metrics_for_pid(meter: Meter, pid: u32) -> Result<(), GlobalError> {
-    let pid = sysinfo::Pid::from_u32(pid);
-    register_system_metrics(meter, pid).await
-}
-
-/// Register system metrics for the current process.
-/// This function is useful for monitoring the current process.
-///
-/// # Arguments
-/// * `meter` - The OpenTelemetry meter to use.
-/// * `pid` - The PID of the process to monitor.
-///
-/// # Returns
-/// * `Ok(())` if successful
-/// * `Err(GlobalError)` if an error occurs
-///
-async fn register_system_metrics(meter: Meter, pid: sysinfo::Pid) -> Result<(), GlobalError> {
-    // cache core counts to avoid repeated calculations
-    let core_count = System::physical_core_count()
-        .ok_or_else(|| GlobalError::SystemMetricsError("Could not get physical core count".to_string()))?;
-    let core_count_f32 = core_count as f32;
-
-    // create metric meter
-    let (
-        process_cpu_utilization,
-        process_cpu_usage,
-        process_memory_usage,
-        process_memory_virtual,
-        process_disk_io,
-        process_gpu_memory_usage,
-    ) = create_metrics(&meter);
-
-    // initialize system object
-    let mut sys = System::new_all();
-    sys.refresh_all();
-
-    // Prepare public properties to avoid repeated construction in loops
-    let common_attributes = prepare_common_attributes(&sys, pid)?;
-
-    // get the metric export interval
-    let interval = get_export_interval();
-
-    // Use asynchronous tasks to process CPU, memory, and disk metrics to avoid blocking the main asynchronous tasks
-    let cpu_mem_task = tokio::spawn(async move {
-        loop {
-            sleep(Duration::from_millis(interval)).await;
-            if let Err(e) = update_process_metrics(
-                &mut sys,
-                pid,
-                &process_cpu_usage,
-                &process_cpu_utilization,
-                &process_memory_usage,
-                &process_memory_virtual,
-                &process_disk_io,
-                &common_attributes,
-                core_count_f32,
-            ) {
-                warn!("Failed to update process metrics: {}", e);
-            }
-        }
-    });
-
-    // Use another asynchronous task to handle GPU metrics
-    #[cfg(feature = "gpu")]
-    let gpu_task = tokio::spawn(async move {
-        loop {
-            sleep(Duration::from_millis(interval)).await;
-
-            // delayed initialization nvml
-            let nvml_arc = get_or_init_nvml().await;
-            let nvml_option = nvml_arc.lock().unwrap();
-
-            if let Err(e) = update_gpu_metrics(&nvml, pid, &process_gpu_memory_usage, &common_attributes) {
-                warn!("Failed to update GPU metrics: {}", e);
-            }
-        }
-    });
-
-    // record empty values when non gpu function
-    #[cfg(not(feature = "gpu"))]
-    let gpu_task = tokio::spawn(async move {
-        loop {
-            sleep(Duration::from_millis(interval)).await;
-            process_gpu_memory_usage.record(0, &common_attributes);
-        }
-    });
-
-    // Wait for the two tasks to complete (actually they will run forever)
-    let _ = tokio::join!(cpu_mem_task, gpu_task);
-
-    Ok(())
-}
-
-fn create_metrics(meter: &Meter) -> (F64Gauge, F64Gauge, I64Gauge, I64Gauge, I64Gauge, U64Gauge) {
-    let process_cpu_utilization = meter
-        .f64_gauge(PROCESS_CPU_USAGE)
-        .with_description("The percentage of CPU in use.")
-        .with_unit("percent")
-        .build();
-
-    let process_cpu_usage = meter
-        .f64_gauge(PROCESS_CPU_UTILIZATION)
-        .with_description("The amount of CPU in use.")
-        .with_unit("percent")
-        .build();
-
-    let process_memory_usage = meter
-        .i64_gauge(PROCESS_MEMORY_USAGE)
-        .with_description("The amount of physical memory in use.")
-        .with_unit("byte")
-        .build();
-
-    let process_memory_virtual = meter
-        .i64_gauge(PROCESS_MEMORY_VIRTUAL)
-        .with_description("The amount of committed virtual memory.")
-        .with_unit("byte")
-        .build();
-
-    let process_disk_io = meter
-        .i64_gauge(PROCESS_DISK_IO)
-        .with_description("Disk bytes transferred.")
-        .with_unit("byte")
-        .build();
-
-    let process_gpu_memory_usage = meter
-        .u64_gauge(PROCESS_GPU_MEMORY_USAGE)
-        .with_description("The amount of physical GPU memory in use.")
-        .with_unit("byte")
-        .build();
-
-    (
-        process_cpu_utilization,
-        process_cpu_usage,
-        process_memory_usage,
-        process_memory_virtual,
-        process_disk_io,
-        process_gpu_memory_usage,
-    )
-}
-
-fn prepare_common_attributes(sys: &System, pid: sysinfo::Pid) -> Result<[KeyValue; 4], GlobalError> {
-    let process = sys
-        .process(pid)
-        .ok_or_else(|| GlobalError::SystemMetricsError(format!("Process with PID {} not found", pid.as_u32())))?;
-
-    // optimize string operations and reduce allocation
-    let cmd = process.cmd().iter().filter_map(|s| s.to_str()).collect::<Vec<_>>().join(" ");
-
-    let executable_path = process
-        .exe()
-        .map(|path| path.to_string_lossy().into_owned())
-        .unwrap_or_default();
-
-    let name = process.name().to_os_string().into_string().unwrap_or_default();
-
-    Ok([
-        KeyValue::new(PROCESS_PID, pid.as_u32() as i64),
-        KeyValue::new(PROCESS_EXECUTABLE_NAME, name),
-        KeyValue::new(PROCESS_EXECUTABLE_PATH, executable_path),
-        KeyValue::new(PROCESS_COMMAND, cmd),
-    ])
-}
-
-fn get_export_interval() -> u64 {
-    std::env::var("OTEL_METRIC_EXPORT_INTERVAL")
-        .ok()
-        .and_then(|s| s.parse().ok())
-        .unwrap_or(30000)
-}
-
-fn update_process_metrics(
-    sys: &mut System,
-    pid: sysinfo::Pid,
-    process_cpu_usage: &F64Gauge,
-    process_cpu_utilization: &F64Gauge,
-    process_memory_usage: &I64Gauge,
-    process_memory_virtual: &I64Gauge,
-    process_disk_io: &I64Gauge,
-    common_attributes: &[KeyValue; 4],
-    core_count: f32,
-) -> Result<(), GlobalError> {
-    // Only refresh the data of the required process to reduce system call overhead
-    sys.refresh_processes(sysinfo::ProcessesToUpdate::Some(&[pid]), true);
-
-    let process = match sys.process(pid) {
-        Some(p) => p,
-        None => {
-            return Err(GlobalError::SystemMetricsError(format!(
-                "Process with PID {} no longer exists",
-                pid.as_u32()
-            )))
-        }
-    };
-
-    // collect data in batches and record it again
-    let cpu_usage = process.cpu_usage();
-    process_cpu_usage.record(cpu_usage.into(), &[]);
-    process_cpu_utilization.record((cpu_usage / core_count).into(), &common_attributes);
-
-    // safe type conversion
-    let memory = process.memory();
-    let virtual_memory = process.virtual_memory();
-
-    // Avoid multiple error checks and use .map_err to handle errors in chain
-    let memory_i64 =
-        i64::try_from(memory).map_err(|_| GlobalError::ConversionError("Failed to convert memory usage to i64".to_string()))?;
-
-    let virtual_memory_i64 = i64::try_from(virtual_memory)
-        .map_err(|_| GlobalError::ConversionError("Failed to convert virtual memory to i64".to_string()))?;
-
-    process_memory_usage.record(memory_i64, common_attributes);
-    process_memory_virtual.record(virtual_memory_i64, common_attributes);
-
-    // process disk io metrics
-    let disk_io = process.disk_usage();
-
-    // batch conversion to reduce duplicate code
-    let read_bytes_i64 = i64::try_from(disk_io.read_bytes)
-        .map_err(|_| GlobalError::ConversionError("Failed to convert read bytes to i64".to_string()))?;
-
-    let written_bytes_i64 = i64::try_from(disk_io.written_bytes)
-        .map_err(|_| GlobalError::ConversionError("Failed to convert written bytes to i64".to_string()))?;
-
-    // Optimize attribute array stitching to reduce heap allocation
-    let mut read_attributes = [KeyValue::new(DIRECTION, "read")];
-    let read_attrs = [common_attributes, &read_attributes].concat();
-
-    let mut write_attributes = [KeyValue::new(DIRECTION, "write")];
-    let write_attrs = [common_attributes, &write_attributes].concat();
-
-    process_disk_io.record(read_bytes_i64, &read_attrs);
-    process_disk_io.record(written_bytes_i64, &write_attrs);
-
-    Ok(())
-}
-
-// GPU metric update function, conditional compilation based on feature flags
-#[cfg(feature = "gpu")]
-fn update_gpu_metrics(
-    nvml: &Result<Nvml, nvml_wrapper::error::NvmlError>,
-    pid: sysinfo::Pid,
-    process_gpu_memory_usage: &U64Gauge,
-    common_attributes: &[KeyValue; 4],
-) -> Result<(), GlobalError> {
-    match nvml {
-        Ok(nvml) => {
-            if let Ok(device) = nvml.device_by_index(0) {
-                if let Ok(gpu_stats) = device.running_compute_processes() {
-                    for stat in gpu_stats.iter() {
-                        if stat.pid == pid.as_u32() {
-                            let memory_used = match stat.used_gpu_memory {
-                                UsedGpuMemory::Used(bytes) => bytes,
-                                UsedGpuMemory::Unavailable => 0,
-                            };
-
-                            process_gpu_memory_usage.record(memory_used, common_attributes);
-                            return Ok(());
-                        }
-                    }
-                }
-            }
-            // If no GPU usage record of the process is found, the record is 0
-            process_gpu_memory_usage.record(0, common_attributes);
-            Ok(())
-        }
-        Err(e) => {
-            warn!("Could not get NVML, recording 0 for GPU memory usage: {}", e);
-            process_gpu_memory_usage.record(0, common_attributes);
-            Ok(())
-        }
-    }
-}
-
-#[cfg(not(feature = "gpu"))]
-fn update_gpu_metrics(
-    _: &(), // blank placeholder parameters
-    _: sysinfo::Pid,
-    process_gpu_memory_usage: &U64Gauge,
-    common_attributes: &[KeyValue; 4],
-) -> Result<(), GlobalError> {
-    // always logged when non gpu function 0
-    process_gpu_memory_usage.record(0, common_attributes);
-    Ok(())
-}

crates/obs/src/system/attributes.rs

@@ -0,0 +1,44 @@
+use crate::GlobalError;
+use opentelemetry::KeyValue;
+use sysinfo::{Pid, System};
+
+pub const PROCESS_PID: opentelemetry::Key = opentelemetry::Key::from_static_str("process.pid");
+pub const PROCESS_EXECUTABLE_NAME: opentelemetry::Key = opentelemetry::Key::from_static_str("process.executable.name");
+pub const PROCESS_EXECUTABLE_PATH: opentelemetry::Key = opentelemetry::Key::from_static_str("process.executable.path");
+pub const PROCESS_COMMAND: opentelemetry::Key = opentelemetry::Key::from_static_str("process.command");
+
+/// Struct to hold process attributes
+pub struct ProcessAttributes {
+    pub attributes: Vec<KeyValue>,
+}
+
+impl ProcessAttributes {
+    /// Creates a new instance of `ProcessAttributes` for the given PID.
+    pub fn new(pid: Pid, system: &mut System) -> Result<Self, GlobalError> {
+        system.refresh_processes(sysinfo::ProcessesToUpdate::Some(&[pid]), true);
+        let process = system
+            .process(pid)
+            .ok_or_else(|| GlobalError::ProcessNotFound(pid.as_u32()))?;
+
+        let attributes = vec![
+            KeyValue::new(PROCESS_PID, pid.as_u32() as i64),
+            KeyValue::new(PROCESS_EXECUTABLE_NAME, process.name().to_os_string().into_string().unwrap_or_default()),
+            KeyValue::new(
+                PROCESS_EXECUTABLE_PATH,
+                process
+                    .exe()
+                    .map(|path| path.to_string_lossy().into_owned())
+                    .unwrap_or_default(),
+            ),
+            KeyValue::new(
+                PROCESS_COMMAND,
+                process
+                    .cmd()
+                    .iter()
+                    .fold(String::new(), |t1, t2| t1 + " " + t2.to_str().unwrap_or_default()),
+            ),
+        ];
+
+        Ok(ProcessAttributes { attributes })
+    }
+}

crates/obs/src/system/collector.rs

@@ -0,0 +1,156 @@
+use crate::system::attributes::ProcessAttributes;
+use crate::system::gpu::GpuCollector;
+use crate::system::metrics::{Metrics, DIRECTION, INTERFACE, STATUS};
+use crate::GlobalError;
+use opentelemetry::KeyValue;
+use std::time::SystemTime;
+use sysinfo::{Networks, Pid, ProcessStatus, System};
+use tokio::time::{sleep, Duration};
+
+/// Collector is responsible for collecting system metrics and attributes.
+/// It uses the sysinfo crate to gather information about the system and processes.
+/// It also uses OpenTelemetry to record metrics.
+pub struct Collector {
+    metrics: Metrics,
+    attributes: ProcessAttributes,
+    gpu_collector: GpuCollector,
+    pid: Pid,
+    system: System,
+    networks: Networks,
+    core_count: usize,
+    interval_ms: u64,
+}
+
+impl Collector {
+    pub fn new(pid: Pid, meter: opentelemetry::metrics::Meter, interval_ms: u64) -> Result<Self, GlobalError> {
+        let mut system = System::new_all();
+        let attributes = ProcessAttributes::new(pid, &mut system)?;
+        let core_count = System::physical_core_count().ok_or(GlobalError::CoreCountError)?;
+        let metrics = Metrics::new(&meter);
+        let gpu_collector = GpuCollector::new(pid)?;
+        let networks = Networks::new_with_refreshed_list();
+
+        Ok(Collector {
+            metrics,
+            attributes,
+            gpu_collector,
+            pid,
+            system,
+            networks,
+            core_count,
+            interval_ms,
+        })
+    }
+
+    pub async fn run(&mut self) -> Result<(), GlobalError> {
+        loop {
+            self.collect()?;
+            tracing::debug!("Collected metrics for PID: {} ,time: {:?}", self.pid, SystemTime::now());
+            sleep(Duration::from_millis(self.interval_ms)).await;
+        }
+    }
+
+    fn collect(&mut self) -> Result<(), GlobalError> {
+        self.system
+            .refresh_processes(sysinfo::ProcessesToUpdate::Some(&[self.pid]), true);
+
+        // 刷新网络接口列表和统计数据
+        self.networks.refresh(false); // 刷新网络统计数据
+
+        let process = self
+            .system
+            .process(self.pid)
+            .ok_or_else(|| GlobalError::ProcessNotFound(self.pid.as_u32()))?;
+
+        // CPU 指标
+        let cpu_usage = process.cpu_usage();
+        self.metrics.cpu_usage.record(cpu_usage as f64, &[]);
+        self.metrics
+            .cpu_utilization
+            .record((cpu_usage / self.core_count as f32) as f64, &self.attributes.attributes);
+
+        // 内存指标
+        self.metrics
+            .memory_usage
+            .record(process.memory() as i64, &self.attributes.attributes);
+        self.metrics
+            .memory_virtual
+            .record(process.virtual_memory() as i64, &self.attributes.attributes);
+
+        // 磁盘I/O指标
+        let disk_io = process.disk_usage();
+        self.metrics.disk_io.record(
+            disk_io.read_bytes as i64,
+            &[&self.attributes.attributes[..], &[KeyValue::new(DIRECTION, "read")]].concat(),
+        );
+        self.metrics.disk_io.record(
+            disk_io.written_bytes as i64,
+            &[&self.attributes.attributes[..], &[KeyValue::new(DIRECTION, "write")]].concat(),
+        );
+
+        // 网络I/O指标（对应 /system/network/internode）
+        let mut total_received: i64 = 0;
+        let mut total_transmitted: i64 = 0;
+
+        // 按接口统计
+        for (interface_name, data) in self.networks.iter() {
+            total_received += data.total_received() as i64;
+            total_transmitted += data.total_transmitted() as i64;
+
+            let received = data.received() as i64;
+            let transmitted = data.transmitted() as i64;
+            self.metrics.network_io_per_interface.record(
+                received,
+                &[
+                    &self.attributes.attributes[..],
+                    &[
+                        KeyValue::new(INTERFACE, interface_name.to_string()),
+                        KeyValue::new(DIRECTION, "received"),
+                    ],
+                ]
+                .concat(),
+            );
+            self.metrics.network_io_per_interface.record(
+                transmitted,
+                &[
+                    &self.attributes.attributes[..],
+                    &[
+                        KeyValue::new(INTERFACE, interface_name.to_string()),
+                        KeyValue::new(DIRECTION, "transmitted"),
+                    ],
+                ]
+                .concat(),
+            );
+        }
+        // 全局统计
+        self.metrics.network_io.record(
+            total_received,
+            &[&self.attributes.attributes[..], &[KeyValue::new(DIRECTION, "received")]].concat(),
+        );
+        self.metrics.network_io.record(
+            total_transmitted,
+            &[&self.attributes.attributes[..], &[KeyValue::new(DIRECTION, "transmitted")]].concat(),
+        );
+
+        // 进程状态指标（对应 /system/process）
+        let status_value = match process.status() {
+            ProcessStatus::Run => 0,
+            ProcessStatus::Sleep => 1,
+            ProcessStatus::Zombie => 2,
+            _ => 3, // 其他状态
+        };
+        self.metrics.process_status.record(
+            status_value,
+            &[
+                &self.attributes.attributes[..],
+                &[KeyValue::new(STATUS, format!("{:?}", process.status()))],
+            ]
+            .concat(),
+        );
+
+        // GPU 指标（可选）
+        self.gpu_collector.collect(&self.metrics, &self.attributes)?;
+
+        Ok(())
+    }
+}

crates/obs/src/system/gpu.rs

@@ -0,0 +1,70 @@
+#[cfg(feature = "gpu")]
+use crate::system::attributes::ProcessAttributes;
+#[cfg(feature = "gpu")]
+use crate::system::metrics::Metrics;
+#[cfg(feature = "gpu")]
+use crate::GlobalError;
+#[cfg(feature = "gpu")]
+use nvml_wrapper::enums::device::UsedGpuMemory;
+#[cfg(feature = "gpu")]
+use nvml_wrapper::Nvml;
+#[cfg(feature = "gpu")]
+use sysinfo::Pid;
+#[cfg(feature = "gpu")]
+use tracing::warn;
+
+/// `GpuCollector` is responsible for collecting GPU memory usage metrics.
+#[cfg(feature = "gpu")]
+pub struct GpuCollector {
+    nvml: Nvml,
+    pid: Pid,
+}
+
+#[cfg(feature = "gpu")]
+impl GpuCollector {
+    pub fn new(pid: Pid) -> Result<Self, GlobalError> {
+        let nvml = Nvml::init().map_err(|e| GlobalError::GpuInitError(e.to_string()))?;
+        Ok(GpuCollector { nvml, pid })
+    }
+
+    pub fn collect(&self, metrics: &Metrics, attributes: &ProcessAttributes) -> Result<(), GlobalError> {
+        if let Ok(device) = self.nvml.device_by_index(0) {
+            if let Ok(gpu_stats) = device.running_compute_processes() {
+                for stat in gpu_stats.iter() {
+                    if stat.pid == self.pid.as_u32() {
+                        let memory_used = match stat.used_gpu_memory {
+                            UsedGpuMemory::Used(bytes) => bytes,
+                            UsedGpuMemory::Unavailable => 0,
+                        };
+                        metrics.gpu_memory_usage.record(memory_used, &attributes.attributes);
+                        return Ok(());
+                    }
+                }
+            } else {
+                warn!("Could not get GPU stats, recording 0 for GPU memory usage");
+            }
+        } else {
+            return Err(GlobalError::GpuDeviceError("No GPU device found".to_string()));
+        }
+        metrics.gpu_memory_usage.record(0, &attributes.attributes);
+        Ok(())
+    }
+}
+
+#[cfg(not(feature = "gpu"))]
+pub struct GpuCollector;
+
+#[cfg(not(feature = "gpu"))]
+impl GpuCollector {
+    pub fn new(_pid: sysinfo::Pid) -> Result<Self, crate::GlobalError> {
+        Ok(GpuCollector)
+    }
+
+    pub fn collect(
+        &self,
+        _metrics: &crate::system::metrics::Metrics,
+        _attributes: &crate::system::attributes::ProcessAttributes,
+    ) -> Result<(), crate::GlobalError> {
+        Ok(())
+    }
+}

crates/obs/src/system/metrics.rs

@@ -0,0 +1,100 @@
+pub const PROCESS_CPU_USAGE: &str = "process.cpu.usage";
+pub const PROCESS_CPU_UTILIZATION: &str = "process.cpu.utilization";
+pub const PROCESS_MEMORY_USAGE: &str = "process.memory.usage";
+pub const PROCESS_MEMORY_VIRTUAL: &str = "process.memory.virtual";
+pub const PROCESS_DISK_IO: &str = "process.disk.io";
+pub const PROCESS_NETWORK_IO: &str = "process.network.io";
+pub const PROCESS_NETWORK_IO_PER_INTERFACE: &str = "process.network.io.per_interface";
+pub const PROCESS_STATUS: &str = "process.status";
+#[cfg(feature = "gpu")]
+pub const PROCESS_GPU_MEMORY_USAGE: &str = "process.gpu.memory.usage";
+pub const DIRECTION: opentelemetry::Key = opentelemetry::Key::from_static_str("direction");
+pub const STATUS: opentelemetry::Key = opentelemetry::Key::from_static_str("status");
+pub const INTERFACE: opentelemetry::Key = opentelemetry::Key::from_static_str("interface");
+
+/// `Metrics` struct holds the OpenTelemetry metrics for process monitoring.
+/// It contains various metrics such as CPU usage, memory usage,
+/// disk I/O, network I/O, and process status.
+///
+/// The `Metrics` struct is designed to be used with OpenTelemetry's
+/// metrics API to record and export these metrics.
+///
+/// The `new` method initializes the metrics using the provided
+/// `opentelemetry::metrics::Meter`.
+pub struct Metrics {
+    pub cpu_usage: opentelemetry::metrics::Gauge<f64>,
+    pub cpu_utilization: opentelemetry::metrics::Gauge<f64>,
+    pub memory_usage: opentelemetry::metrics::Gauge<i64>,
+    pub memory_virtual: opentelemetry::metrics::Gauge<i64>,
+    pub disk_io: opentelemetry::metrics::Gauge<i64>,
+    pub network_io: opentelemetry::metrics::Gauge<i64>,
+    pub network_io_per_interface: opentelemetry::metrics::Gauge<i64>,
+    pub process_status: opentelemetry::metrics::Gauge<i64>,
+    #[cfg(feature = "gpu")]
+    pub gpu_memory_usage: opentelemetry::metrics::Gauge<u64>,
+}
+
+impl Metrics {
+    pub fn new(meter: &opentelemetry::metrics::Meter) -> Self {
+        let cpu_usage = meter
+            .f64_gauge(PROCESS_CPU_USAGE)
+            .with_description("The percentage of CPU in use.")
+            .with_unit("percent")
+            .build();
+        let cpu_utilization = meter
+            .f64_gauge(PROCESS_CPU_UTILIZATION)
+            .with_description("The amount of CPU in use.")
+            .with_unit("percent")
+            .build();
+        let memory_usage = meter
+            .i64_gauge(PROCESS_MEMORY_USAGE)
+            .with_description("The amount of physical memory in use.")
+            .with_unit("byte")
+            .build();
+        let memory_virtual = meter
+            .i64_gauge(PROCESS_MEMORY_VIRTUAL)
+            .with_description("The amount of committed virtual memory.")
+            .with_unit("byte")
+            .build();
+        let disk_io = meter
+            .i64_gauge(PROCESS_DISK_IO)
+            .with_description("Disk bytes transferred.")
+            .with_unit("byte")
+            .build();
+        let network_io = meter
+            .i64_gauge(PROCESS_NETWORK_IO)
+            .with_description("Network bytes transferred.")
+            .with_unit("byte")
+            .build();
+        let network_io_per_interface = meter
+            .i64_gauge(PROCESS_NETWORK_IO_PER_INTERFACE)
+            .with_description("Network bytes transferred (per interface).")
+            .with_unit("byte")
+            .build();
+
+        let process_status = meter
+            .i64_gauge(PROCESS_STATUS)
+            .with_description("Process status (0: Running, 1: Sleeping, 2: Zombie, etc.)")
+            .build();
+
+        #[cfg(feature = "gpu")]
+        let gpu_memory_usage = meter
+            .u64_gauge(PROCESS_GPU_MEMORY_USAGE)
+            .with_description("The amount of physical GPU memory in use.")
+            .with_unit("byte")
+            .build();
+
+        Metrics {
+            cpu_usage,
+            cpu_utilization,
+            memory_usage,
+            memory_virtual,
+            disk_io,
+            network_io,
+            network_io_per_interface,
+            process_status,
+            #[cfg(feature = "gpu")]
+            gpu_memory_usage,
+        }
+    }
+}

crates/obs/src/system/mod.rs

@@ -0,0 +1,24 @@
+use crate::GlobalError;
+
+pub(crate) mod attributes;
+mod collector;
+pub(crate) mod gpu;
+pub(crate) mod metrics;
+
+/// Initialize the indicator collector for the current process
+/// This function will create a new `Collector` instance and start collecting metrics.
+/// It will run indefinitely until the process is terminated.
+pub async fn init_process_observer(meter: opentelemetry::metrics::Meter) -> Result<(), GlobalError> {
+    let pid = sysinfo::get_current_pid().map_err(|e| GlobalError::PidError(e.to_string()))?;
+    let mut collector = collector::Collector::new(pid, meter, 30000)?;
+    collector.run().await
+}
+
+/// Initialize the metric collector for the specified PID process
+/// This function will create a new `Collector` instance and start collecting metrics.
+/// It will run indefinitely until the process is terminated.
+pub async fn init_process_observer_for_pid(meter: opentelemetry::metrics::Meter, pid: u32) -> Result<(), GlobalError> {
+    let pid = sysinfo::Pid::from_u32(pid);
+    let mut collector = collector::Collector::new(pid, meter, 30000)?;
+    collector.run().await
+}

crates/obs/src/telemetry.rs

@@ -1,5 +1,6 @@
 use crate::global::{ENVIRONMENT, LOGGER_LEVEL, METER_INTERVAL, SAMPLE_RATIO, SERVICE_NAME, SERVICE_VERSION, USE_STDOUT};
-use crate::{get_local_ip_with_default, OtelConfig};
+use crate::utils::get_local_ip_with_default;
+use crate::OtelConfig;
 use opentelemetry::trace::TracerProvider;
 use opentelemetry::{global, KeyValue};
 use opentelemetry_appender_tracing::layer;

rustfs/Cargo.toml

@@ -77,7 +77,7 @@ tokio-stream.workspace = true
 tonic = { workspace = true }
 tower.workspace = true
 transform-stream.workspace = true
-tower-http = { workspace = true, features = ["trace", "compression-full", "cors"] }
+tower-http = { workspace = true, features = ["trace", "compression-deflate", "compression-gzip", "cors"] }
 uuid = { workspace = true }
 
 [target.'cfg(target_os = "linux")'.dependencies]

rustfs/src/console.rs

@@ -271,7 +271,8 @@ pub async fn start_static_file_server(
         .route("/config.json", get(config_handler))
         .fallback_service(get(static_handler))
         .layer(cors)
-        .layer(tower_http::compression::CompressionLayer::new())
+        .layer(tower_http::compression::CompressionLayer::new().gzip(true))
+        .layer(tower_http::compression::CompressionLayer::new().deflate(true))
         .layer(TraceLayer::new_for_http());
     let local_addr: SocketAddr = addrs.parse().expect("Failed to parse socket address");
     info!("WebUI: http://{}:{} http://127.0.0.1:{}", local_ip, local_addr.port(), local_addr.port());

rustfs/src/main.rs

@@ -14,6 +14,7 @@ use crate::console::{init_console_cfg, CONSOLE_CONFIG};
 // Ensure the correct path for parse_license is imported
 use crate::server::{wait_for_shutdown, ServiceState, ServiceStateManager, ShutdownSignal, SHUTDOWN_TIMEOUT};
 use crate::utils::error;
+use bytes::Bytes;
 use chrono::Datelike;
 use clap::Parser;
 use common::{
@@ -37,6 +38,7 @@ use ecstore::{
 };
 use ecstore::{global::set_global_rustfs_port, notification_sys::new_global_notification_sys};
 use grpc::make_server;
+use http::{HeaderMap, Request as HttpRequest, Response};
 use hyper_util::server::graceful::GracefulShutdown;
 use hyper_util::{
     rt::{TokioExecutor, TokioIo},
@@ -47,17 +49,20 @@ use iam::init_iam_sys;
 use license::init_license;
 use protos::proto_gen::node_service::node_service_server::NodeServiceServer;
 use rustfs_event_notifier::NotifierConfig;
-use rustfs_obs::{init_obs, init_system_metrics, load_config, set_global_guard, InitLogStatus};
+use rustfs_obs::{init_obs, init_process_observer, load_config, set_global_guard};
 use rustls::ServerConfig;
 use s3s::{host::MultiDomain, service::S3ServiceBuilder};
 use service::hybrid;
 use std::net::SocketAddr;
 use std::sync::Arc;
+use std::time::Duration;
 use tokio::net::TcpListener;
 use tokio::signal::unix::{signal, SignalKind};
 use tokio_rustls::TlsAcceptor;
 use tonic::{metadata::MetadataValue, Request, Status};
 use tower_http::cors::CorsLayer;
+use tower_http::trace::TraceLayer;
+use tracing::Span;
 use tracing::{debug, error, info, info_span, warn};
 
 #[cfg(all(target_os = "linux", target_env = "gnu"))]
@@ -103,9 +108,6 @@ async fn main() -> Result<()> {
     // Store in global storage
     set_global_guard(guard)?;
 
-    // Log initialization status
-    InitLogStatus::init_start_log(&config.observability).await?;
-
     // Initialize event notifier
     let notifier_config = opt.clone().event_config;
     if notifier_config.is_some() {
@@ -123,16 +125,6 @@ async fn main() -> Result<()> {
         info!("event_config is empty");
     }
 
-    let meter = opentelemetry::global::meter("system");
-    let _ = init_system_metrics(meter).await;
-
-    // If GPU function is enabled, specific functions can be used
-    #[cfg(feature = "gpu")]
-    {
-        let gpu_meter = opentelemetry::global::meter("system.gpu");
-        let _ = rustfs_obs::init_gpu_metrics(gpu_meter).await;
-    }
-
     // Run parameters
     run(opt).await
 }
@@ -257,6 +249,18 @@ async fn run(opt: config::Opt) -> Result<()> {
         b.build()
     };
 
+    // Record the PID-related metrics of the current process
+    let meter = opentelemetry::global::meter("system");
+    let obs_result = init_process_observer(meter).await;
+    match obs_result {
+        Ok(_) => {
+            info!("Process observer initialized successfully");
+        }
+        Err(e) => {
+            error!("Failed to initialize process observer: {}", e);
+        }
+    }
+
     let rpc_service = NodeServiceServer::with_interceptor(make_server(), check_auth);
 
     let tls_path = opt.tls_path.clone().unwrap_or_default();
@@ -310,18 +314,53 @@ async fn run(opt: config::Opt) -> Result<()> {
         let mut sigint_inner = sigint_inner;
         let hybrid_service = TowerToHyperService::new(
             tower::ServiceBuilder::new()
+                .layer(
+                    TraceLayer::new_for_http()
+                        .make_span_with(|request: &HttpRequest<_>| {
+                            let span = tracing::debug_span!("http-request",
+                                status_code = tracing::field::Empty,
+                                method = %request.method(),
+                                uri = %request.uri(),
+                                version = ?request.version(),
+                            );
+                            for (header_name, header_value) in request.headers() {
+                                if header_name == "user-agent" || header_name == "content-type" || header_name == "content-length"
+                                {
+                                    span.record(header_name.as_str(), header_value.to_str().unwrap_or("invalid"));
+                                }
+                            }
+
+                            span
+                        })
+                        .on_request(|request: &HttpRequest<_>, _span: &Span| {
+                            debug!("started method: {}, url path: {}", request.method(), request.uri().path())
+                        })
+                        .on_response(|response: &Response<_>, latency: Duration, _span: &Span| {
+                            _span.record("status_code", tracing::field::display(response.status()));
+                            debug!("response generated in {:?}", latency)
+                        })
+                        .on_body_chunk(|chunk: &Bytes, latency: Duration, _span: &Span| {
+                            debug!("sending {} bytes in {:?}", chunk.len(), latency)
+                        })
+                        .on_eos(|_trailers: Option<&HeaderMap>, stream_duration: Duration, _span: &Span| {
+                            debug!("stream closed after {:?}", stream_duration)
+                        })
+                        .on_failure(|_error, latency: Duration, _span: &Span| {
+                            debug!("request error: {:?} in {:?}", _error, latency)
+                        }),
+                )
                 .layer(CorsLayer::permissive())
                 .service(hybrid(s3_service, rpc_service)),
         );
 
-        let http_server = ConnBuilder::new(TokioExecutor::new());
+        let http_server = Arc::new(ConnBuilder::new(TokioExecutor::new()));
         let mut ctrl_c = std::pin::pin!(tokio::signal::ctrl_c());
-        let graceful = GracefulShutdown::new();
+        let graceful = Arc::new(GracefulShutdown::new());
         debug!("graceful initiated");
 
         // 服务准备就绪
         worker_state_manager.update(ServiceState::Ready);
-
+        let value = hybrid_service.clone();
         loop {
             debug!("waiting for SIGINT or SIGTERM has_tls_certs: {}", has_tls_certs);
             // Wait for a connection
@@ -376,12 +415,17 @@ async fn run(opt: config::Opt) -> Result<()> {
                         continue;
                     }
                 };
-                let conn = http_server.serve_connection(TokioIo::new(tls_socket), hybrid_service.clone());
+
-                let conn = graceful.watch(conn.into_owned());
+                let http_server_clone = http_server.clone();
+                let value_clone = value.clone();
+                let graceful_clone = graceful.clone();
+
                 tokio::task::spawn_blocking(move || {
                     tokio::runtime::Runtime::new()
                         .expect("Failed to create runtime")
                         .block_on(async move {
+                            let conn = http_server_clone.serve_connection(TokioIo::new(tls_socket), value_clone);
+                            let conn = graceful_clone.watch(conn);
                             if let Err(err) = conn.await {
                                 error!("Https Connection error: {}", err);
                             }
@@ -390,9 +434,13 @@ async fn run(opt: config::Opt) -> Result<()> {
                 debug!("TLS handshake success");
             } else {
                 debug!("Http handshake start");
-                let conn = http_server.serve_connection(TokioIo::new(socket), hybrid_service.clone());
+
-                let conn = graceful.watch(conn.into_owned());
+                let http_server_clone = http_server.clone();
+                let value_clone = value.clone();
+                let graceful_clone = graceful.clone();
                 tokio::spawn(async move {
+                    let conn = http_server_clone.serve_connection(TokioIo::new(socket), value_clone);
+                    let conn = graceful_clone.watch(conn);
                     if let Err(err) = conn.await {
                         error!("Http Connection error: {}", err);
                     }
@@ -401,12 +449,33 @@ async fn run(opt: config::Opt) -> Result<()> {
             }
         }
         worker_state_manager.update(ServiceState::Stopping);
-        tokio::select! {
+        // tokio::select! {
-            () = graceful.shutdown() => {
+        //     () = graceful.shutdown() => {
-                 debug!("Gracefully shutdown!");
+        //          debug!("Gracefully shutdown!");
-            },
+        //     },
-            () = tokio::time::sleep(std::time::Duration::from_secs(10)) => {
+        //     () = tokio::time::sleep(std::time::Duration::from_secs(10)) => {
-                 debug!("Waited 10 seconds for graceful shutdown, aborting...");
+        //          debug!("Waited 10 seconds for graceful shutdown, aborting...");
+        //     }
+        // }
+        match Arc::try_unwrap(graceful) {
+            Ok(g) => {
+                // 成功获取唯一所有权，可以调用 shutdown
+                tokio::select! {
+                    () = g.shutdown() => {
+                        debug!("Gracefully shutdown!");
+                    },
+                    () = tokio::time::sleep(Duration::from_secs(10)) => {
+                        debug!("Waited 10 seconds for graceful shutdown, aborting...");
+                    }
+                }
+            }
+            Err(arc_graceful) => {
+                // 还有其他引用存在，无法获取唯一所有权
+                debug!("Cannot perform graceful shutdown, other references exist");
+                error!("Cannot perform graceful shutdown, other references exist err: {:?}", arc_graceful);
+                // 在这种情况下，我们只能等待超时
+                tokio::time::sleep(Duration::from_secs(10)).await;
+                debug!("Timeout reached, forcing shutdown");
             }
         }
         worker_state_manager.update(ServiceState::Stopped);

s3select/query/src/dispatcher/manager.rs

@@ -166,7 +166,8 @@ impl SimpleQueryDispatcher {
                 if let Some(delimiter) = csv.field_delimiter.as_ref() {
                     file_format = file_format.with_delimiter(delimiter.as_bytes().first().copied().unwrap_or_default());
                 }
-                if csv.file_header_info.is_some() {}
+                // TODO waiting for processing @junxiang Mu
+                // if csv.file_header_info.is_some() {}
                 match csv.file_header_info.as_ref() {
                     Some(info) => {
                         if *info == *NONE {

scripts/run.sh

@@ -40,13 +40,13 @@ export RUSTFS_OBS_CONFIG="./deploy/config/obs.example.toml"
 
 # 如下变量需要必须参数都有值才可以，以及会覆盖配置文件中的值
 export RUSTFS__OBSERVABILITY__ENDPOINT=http://localhost:4317
-export RUSTFS__OBSERVABILITY__USE_STDOUT=true
+export RUSTFS__OBSERVABILITY__USE_STDOUT=false
 export RUSTFS__OBSERVABILITY__SAMPLE_RATIO=2.0
 export RUSTFS__OBSERVABILITY__METER_INTERVAL=30
 export RUSTFS__OBSERVABILITY__SERVICE_NAME=rustfs
 export RUSTFS__OBSERVABILITY__SERVICE_VERSION=0.1.0
 export RUSTFS__OBSERVABILITY__ENVIRONMENT=develop
-export RUSTFS__OBSERVABILITY__LOGGER_LEVEL=info
+export RUSTFS__OBSERVABILITY__LOGGER_LEVEL=debug
 export RUSTFS__SINKS__FILE__ENABLED=true
 export RUSTFS__SINKS__FILE__PATH="./deploy/logs/rustfs.log"
 export RUSTFS__SINKS__WEBHOOK__ENABLED=false

tomlfmt.toml

@@ -0,0 +1,31 @@
+# trailing comma in arrays
+always_trailing_comma = false
+# trailing comma when multi-line
+multiline_trailing_comma = true
+# the maximum length in bytes of the string of an array object
+max_array_line_len = 80
+# number of spaces to indent
+indent_count = 4
+# space around equal sign
+space_around_eq = true
+# remove all the spacing inside the array
+compact_arrays = false
+# remove all the spacing inside the object
+compact_inline_tables = false
+trailing_newline = true
+# is it ok to have blank lines inside a table
+# this option needs to be true for the --grouped flag
+key_value_newlines = true
+allowed_blank_lines = 1
+# windows style line endings
+crlf = false
+# The user specified ordering of tables in a document.
+# All unspecified tables will come after these.
+table_order = [
+    "package",
+    "features",
+    "dependencies",
+    "build-dependencies",
+    "dev-dependencies",
+    "workspace"
+]