From ef597fff31cda5df9222f60e9e027b645032b3f7 Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 24 Apr 2025 18:22:59 +0800 Subject: [PATCH] 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 | 1 - crates/obs/examples/config.toml | 2 +- crates/obs/examples/server.rs | 9 +- crates/obs/src/global.rs | 25 +- crates/obs/src/lib.rs | 40 ++- crates/obs/src/logger.rs | 102 ++---- crates/obs/src/metrics.rs | 394 ----------------------- crates/obs/src/system/attributes.rs | 44 +++ crates/obs/src/system/collector.rs | 156 +++++++++ crates/obs/src/system/gpu.rs | 70 ++++ crates/obs/src/system/metrics.rs | 100 ++++++ crates/obs/src/system/mod.rs | 24 ++ crates/obs/src/telemetry.rs | 3 +- rustfs/Cargo.toml | 2 +- rustfs/src/console.rs | 3 +- rustfs/src/main.rs | 123 +++++-- s3select/query/src/dispatcher/manager.rs | 3 +- scripts/run.sh | 4 +- tomlfmt.toml | 31 ++ 19 files changed, 600 insertions(+), 536 deletions(-) delete mode 100644 crates/obs/src/metrics.rs create mode 100644 crates/obs/src/system/attributes.rs create mode 100644 crates/obs/src/system/collector.rs create mode 100644 crates/obs/src/system/gpu.rs create mode 100644 crates/obs/src/system/metrics.rs create mode 100644 crates/obs/src/system/mod.rs create mode 100644 tomlfmt.toml diff --git a/Cargo.lock b/Cargo.lock index 2212d0896..81d6b4c3e 100644 --- a/Cargo.lock +++ b/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", diff --git a/crates/obs/examples/config.toml b/crates/obs/examples/config.toml index 929867f03..c1b3df148 100644 --- a/crates/obs/examples/config.toml +++ b/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 diff --git a/crates/obs/examples/server.rs b/crates/obs/examples/server.rs index 55977b10c..0310c158f 100644 --- a/crates/obs/examples/server.rs +++ b/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())) diff --git a/crates/obs/src/global.rs b/crates/obs/src/global.rs index 854ee0deb..8d392ad94 100644 --- a/crates/obs/src/global.rs +++ b/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}")] - GetPidError(String), - #[error("Type conversion error: {0}")] - ConversionError(String), - #[error("IO error: {0}")] - IoError(#[from] std::io::Error), - #[error("System metrics error: {0}")] - SystemMetricsError(String), - #[cfg(feature = "gpu")] - #[error("GPU metrics error: {0}")] - GpuMetricsError(String), + #[error("Failed to get current PID: {0}")] + PidError(String), + #[error("Process with PID {0} not found")] + ProcessNotFound(u32), + #[error("Failed to get physical core count")] + CoreCountError, + #[error("GPU initialization failed: {0}")] + GpuInitError(String), + #[error("GPU device not found: {0}")] + GpuDeviceError(String), + #[error("Failed to send log: {0}")] + SendFailed(&'static str), + #[error("Operation timed out: {0}")] + Timeout(&'static str), } /// Set the global guard for OpenTelemetry diff --git a/crates/obs/src/lib.rs b/crates/obs/src/lib.rs index d7bcadb2b..d41d0e66d 100644 --- a/crates/obs/src/lib.rs +++ b/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")] -pub use crate::metrics::init_gpu_metrics; - +use crate::logger::InitLogStatus; 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::{get_global_logger, init_global_logger, locked_logger, start_logger}; -pub use logger::{log_init_state, InitLogStatus}; -pub use logger::{LogError, Logger}; -pub use metrics::{init_system_metrics, init_system_metrics_for_pid}; -pub use sink::Sink; +pub use logger::Logger; +pub use logger::{get_global_logger, init_global_logger, start_logger}; +pub use logger::{log_debug, log_error, log_info, log_trace, log_warn, log_with_context}; 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}; -pub use worker::start_worker; +use tracing::{error, info}; /// Initialize the observability module /// @@ -80,6 +81,19 @@ pub async fn init_obs(config: AppConfig) -> (Arc>, 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) } diff --git a/crates/obs/src/logger.rs b/crates/obs/src/logger.rs index 1e62712cb..6329ab8fc 100644 --- a/crates/obs/src/logger.rs +++ b/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")), - Err(_) => Err(LogError::Timeout("Queue backpressure timeout")), + Ok(Err(_)) => Err(GlobalError::SendFailed("Channel closed")), + 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, user_id: Option, 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> { 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> { - 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> { /// 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, user_id: Option, 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) -> Result<(), LogError> { +async fn log_init_state(status: Option) -> Result<(), GlobalError> { let status = status.unwrap_or_default(); let base_entry = BaseLogEntry::new() diff --git a/crates/obs/src/metrics.rs b/crates/obs/src/metrics.rs deleted file mode 100644 index a6e1b2f12..000000000 --- a/crates/obs/src/metrics.rs +++ /dev/null @@ -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>>>> = OnceCell::const_new(); - -// get or initialize an nvml instance -#[cfg(feature = "gpu")] -async fn get_or_init_nvml() -> &'static Arc>>> { - 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::>().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, - 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(()) -} diff --git a/crates/obs/src/system/attributes.rs b/crates/obs/src/system/attributes.rs new file mode 100644 index 000000000..289b4b66a --- /dev/null +++ b/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, +} + +impl ProcessAttributes { + /// Creates a new instance of `ProcessAttributes` for the given PID. + pub fn new(pid: Pid, system: &mut System) -> Result { + 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 }) + } +} diff --git a/crates/obs/src/system/collector.rs b/crates/obs/src/system/collector.rs new file mode 100644 index 000000000..335c581e3 --- /dev/null +++ b/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 { + 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(()) + } +} diff --git a/crates/obs/src/system/gpu.rs b/crates/obs/src/system/gpu.rs new file mode 100644 index 000000000..ce47f2c51 --- /dev/null +++ b/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 { + 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 { + Ok(GpuCollector) + } + + pub fn collect( + &self, + _metrics: &crate::system::metrics::Metrics, + _attributes: &crate::system::attributes::ProcessAttributes, + ) -> Result<(), crate::GlobalError> { + Ok(()) + } +} diff --git a/crates/obs/src/system/metrics.rs b/crates/obs/src/system/metrics.rs new file mode 100644 index 000000000..114c285ea --- /dev/null +++ b/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, + pub cpu_utilization: opentelemetry::metrics::Gauge, + pub memory_usage: opentelemetry::metrics::Gauge, + pub memory_virtual: opentelemetry::metrics::Gauge, + pub disk_io: opentelemetry::metrics::Gauge, + pub network_io: opentelemetry::metrics::Gauge, + pub network_io_per_interface: opentelemetry::metrics::Gauge, + pub process_status: opentelemetry::metrics::Gauge, + #[cfg(feature = "gpu")] + pub gpu_memory_usage: opentelemetry::metrics::Gauge, +} + +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, + } + } +} diff --git a/crates/obs/src/system/mod.rs b/crates/obs/src/system/mod.rs new file mode 100644 index 000000000..a69bc3430 --- /dev/null +++ b/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 +} diff --git a/crates/obs/src/telemetry.rs b/crates/obs/src/telemetry.rs index b792bcda2..9ffc265dd 100644 --- a/crates/obs/src/telemetry.rs +++ b/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; diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index a4ab5744f..41c62a308 100644 --- a/rustfs/Cargo.toml +++ b/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] diff --git a/rustfs/src/console.rs b/rustfs/src/console.rs index 127d2144e..46cee6c3b 100644 --- a/rustfs/src/console.rs +++ b/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()); diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index af268611b..80b257bae 100644 --- a/rustfs/src/main.rs +++ b/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! { - () = graceful.shutdown() => { - debug!("Gracefully shutdown!"); - }, - () = tokio::time::sleep(std::time::Duration::from_secs(10)) => { - debug!("Waited 10 seconds for graceful shutdown, aborting..."); + // tokio::select! { + // () = graceful.shutdown() => { + // debug!("Gracefully shutdown!"); + // }, + // () = tokio::time::sleep(std::time::Duration::from_secs(10)) => { + // 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); diff --git a/s3select/query/src/dispatcher/manager.rs b/s3select/query/src/dispatcher/manager.rs index 05f763435..80543ed69 100644 --- a/s3select/query/src/dispatcher/manager.rs +++ b/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 { diff --git a/scripts/run.sh b/scripts/run.sh index 2cbd115a8..c31ce7ed7 100755 --- a/scripts/run.sh +++ b/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 diff --git a/tomlfmt.toml b/tomlfmt.toml new file mode 100644 index 000000000..0089ac7f8 --- /dev/null +++ b/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" +] \ No newline at end of file