diff --git a/.docker/observability/grafana/dashboards/rustfs.json b/.docker/observability/grafana/dashboards/rustfs.json index d71eaf3e6..2f76d96ee 100644 --- a/.docker/observability/grafana/dashboards/rustfs.json +++ b/.docker/observability/grafana/dashboards/rustfs.json @@ -2641,7 +2641,7 @@ }, "gridPos": { "h": 7, - "w": 24, + "w": 12, "x": 0, "y": 73 }, @@ -2670,7 +2670,7 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "sum by (job) (rate(rustfs_system_process_network_io{job=~\"$job\", direction=\"received\"}[5m]))", + "expr": "sum by (job) (rate(rustfs_system_network_host_network_io{job=~\"$job\", direction=\"received\"}[5m]))", "legendFormat": "RX - {{job}}", "range": true, "refId": "C" @@ -2681,7 +2681,7 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "sum by (job) (rate(rustfs_system_process_network_io{job=~\"$job\", direction=\"transmitted\"}[5m]))", + "expr": "sum by (job) (rate(rustfs_system_network_host_network_io{job=~\"$job\", direction=\"transmitted\"}[5m]))", "legendFormat": "TX - {{job}}", "range": true, "refId": "D" @@ -2690,6 +2690,115 @@ "title": "Network", "type": "timeseries" }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "barWidthFactor": 0.6, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "smooth", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "showValues": false, + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + } + ] + }, + "unit": "Bps" + }, + "overrides": [] + }, + "gridPos": { + "h": 7, + "w": 12, + "x": 12, + "y": 73 + }, + "id": 500, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "right", + "showLegend": true + }, + "tooltip": { + "hideZeros": false, + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "12.3.2", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by (job, interface) (rate(rustfs_system_network_host_network_io_per_interface{job=~\"$job\", direction=\"received\"}[5m]))", + "legendFormat": "RX {{interface}} - {{job}}", + "range": true, + "refId": "E" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by (job, interface) (rate(rustfs_system_network_host_network_io_per_interface{job=~\"$job\", direction=\"transmitted\"}[5m]))", + "legendFormat": "TX {{interface}} - {{job}}", + "range": true, + "refId": "F" + } + ], + "title": "Network by Interface", + "type": "timeseries" + }, { "datasource": { "type": "prometheus", diff --git a/crates/io-metrics/src/sampler/process.rs b/crates/io-metrics/src/sampler/process.rs index cfc0fa8c9..2d1b1de6e 100644 --- a/crates/io-metrics/src/sampler/process.rs +++ b/crates/io-metrics/src/sampler/process.rs @@ -12,16 +12,24 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::sync::OnceLock; -use std::time::Instant; +use std::sync::{Mutex, OnceLock}; use sysinfo::{Pid, ProcessRefreshKind, ProcessStatus, ProcessesToUpdate, System}; -/// Process start time used for computing process uptime. -static PROCESS_START: OnceLock = OnceLock::new(); +static PROCESS_SYSTEM: OnceLock> = OnceLock::new(); #[inline] -fn get_process_start() -> &'static Instant { - PROCESS_START.get_or_init(Instant::now) +fn current_pid() -> Pid { + Pid::from_u32(std::process::id()) +} + +#[inline] +fn process_system() -> &'static Mutex { + PROCESS_SYSTEM.get_or_init(|| { + let pid = current_pid(); + let mut system = System::new(); + system.refresh_processes_specifics(ProcessesToUpdate::Some(&[pid]), true, ProcessRefreshKind::everything()); + Mutex::new(system) + }) } #[derive(Debug, Clone, Copy, Default, PartialEq)] @@ -57,6 +65,8 @@ pub struct ProcessSystemSnapshot { pub locks_write_total: u64, pub cpu_total_seconds: f64, pub go_routine_total: u64, + pub disk_read_bytes: u64, + pub disk_write_bytes: u64, pub io_rchar_bytes: u64, pub io_read_bytes: u64, pub io_wchar_bytes: u64, @@ -89,17 +99,16 @@ pub fn snapshot_process_system() -> ProcessSystemSnapshot { /// Collect both resource and system snapshots in one sysinfo refresh. #[inline] pub fn snapshot_process_resource_and_system() -> (ProcessResourceSnapshot, ProcessSystemSnapshot) { - let uptime_seconds = get_process_start().elapsed().as_secs(); let platform_stats = crate::snapshot_process_platform_stats(); let lock_snapshot = crate::snapshot_process_lock_counts(); - - let mut sys = System::new(); - let pid = Pid::from_u32(std::process::id()); + let pid = current_pid(); + let mut sys = process_system().lock().unwrap_or_else(|poisoned| poisoned.into_inner()); sys.refresh_processes_specifics(ProcessesToUpdate::Some(&[pid]), true, ProcessRefreshKind::everything()); if let Some(process) = sys.process(pid) { let disk_usage = process.disk_usage(); let status = ProcessStatusSnapshot::from(process.status()); + let uptime_seconds = process.run_time(); let resource_stats = ProcessResourceSnapshot { cpu_percent: process.cpu_usage() as f64, @@ -111,6 +120,8 @@ pub fn snapshot_process_resource_and_system() -> (ProcessResourceSnapshot, Proce locks_read_total: lock_snapshot.read_locks_held, locks_write_total: lock_snapshot.write_locks_held, cpu_total_seconds: process.accumulated_cpu_time() as f64 / 1000.0, + disk_read_bytes: disk_usage.read_bytes, + disk_write_bytes: disk_usage.written_bytes, file_descriptor_limit_total: process.open_files_limit().map_or(0, |value| value as u64), file_descriptor_open_total: process.open_files().map_or(0, |value| value as u64), go_routine_total: process.tasks().map_or(0, |tasks| tasks.len() as u64), @@ -131,16 +142,7 @@ pub fn snapshot_process_resource_and_system() -> (ProcessResourceSnapshot, Proce (resource_stats, process_stats) } else { - ( - ProcessResourceSnapshot { - uptime_seconds, - ..Default::default() - }, - ProcessSystemSnapshot { - uptime_seconds, - ..Default::default() - }, - ) + (ProcessResourceSnapshot::default(), ProcessSystemSnapshot::default()) } } diff --git a/crates/obs/src/metrics/collectors/mod.rs b/crates/obs/src/metrics/collectors/mod.rs index b3b53c03b..50eb1560d 100644 --- a/crates/obs/src/metrics/collectors/mod.rs +++ b/crates/obs/src/metrics/collectors/mod.rs @@ -36,6 +36,7 @@ pub mod system_drive; pub mod system_gpu; pub mod system_memory; pub mod system_network; +pub mod system_network_host; pub mod system_process; pub use audit::{AuditTargetStats, collect_audit_metrics}; @@ -64,7 +65,8 @@ pub use system_drive::{ #[cfg(feature = "gpu")] pub use system_gpu::{GpuCollector, GpuError, GpuStats, collect_gpu_metrics}; pub use system_memory::{MemoryStats, ProcessMemoryStats, collect_memory_metrics, collect_process_memory_metrics}; -pub use system_network::{NetworkStats, ProcessNetworkStats, collect_network_metrics, collect_process_network_metrics}; +pub use system_network::{NetworkStats, collect_network_metrics}; +pub use system_network_host::{HostNetworkStats, collect_host_network_metrics}; pub use system_process::{ ProcessAttributeError, ProcessAttributes, ProcessStats, ProcessStatusType, collect_process_attributes, collect_process_metrics, diff --git a/crates/obs/src/metrics/collectors/system_network.rs b/crates/obs/src/metrics/collectors/system_network.rs index 5112bfc21..4392e398c 100644 --- a/crates/obs/src/metrics/collectors/system_network.rs +++ b/crates/obs/src/metrics/collectors/system_network.rs @@ -19,13 +19,10 @@ //! Collects internode network metrics including errors, dial times, //! and bytes sent/received. //! -//! This module provides both system-level and process-level network metrics, -//! with process-level metrics migrated from `rustfs-obs::system`. +//! This module provides system-level internode network metrics. use crate::metrics::report::PrometheusMetric; use crate::metrics::schema::system_network::*; -use crate::metrics::schema::system_process::{PROCESS_NETWORK_IO_MD, PROCESS_NETWORK_IO_PER_INTERFACE_MD}; -use std::borrow::Cow; /// Network statistics for internode communication. #[derive(Debug, Clone, Default)] @@ -42,19 +39,6 @@ pub struct NetworkStats { pub internode_recv_bytes_total: u64, } -/// Process network I/O statistics. -/// -/// Contains network I/O metrics for a specific process. -#[derive(Debug, Clone, Default)] -pub struct ProcessNetworkStats { - /// Total bytes received - pub total_received: u64, - /// Total bytes transmitted - pub total_transmitted: u64, - /// Per-interface statistics: (interface_name, received_bytes, transmitted_bytes) - pub per_interface: Vec<(String, u64, u64)>, -} - /// Collects network metrics from the given stats. /// /// Uses the metric descriptors from `metrics_type::system_network` module. @@ -69,60 +53,6 @@ pub fn collect_network_metrics(stats: &NetworkStats) -> Vec { ] } -/// Collects process network I/O metrics from the given stats. -/// -/// Returns a vector of Prometheus metrics for process network I/O statistics. -/// Each metric includes a `direction` label ("received" or "transmitted"). -/// Per-interface metrics also include an `interface` label. -/// -/// # Arguments -/// -/// * `stats` - Process network I/O statistics -/// * `labels` - Optional additional labels (e.g., process attributes) -pub fn collect_process_network_metrics( - stats: &ProcessNetworkStats, - labels: Option<&[(&'static str, Cow<'static, str>)]>, -) -> Vec { - let mut metrics = Vec::with_capacity(2 + stats.per_interface.len() * 2); - - // Total network I/O - let mut received_metric = PrometheusMetric::from_descriptor(&PROCESS_NETWORK_IO_MD, stats.total_received as f64); - let mut transmitted_metric = PrometheusMetric::from_descriptor(&PROCESS_NETWORK_IO_MD, stats.total_transmitted as f64); - - received_metric.labels.push(("direction", Cow::Borrowed("received"))); - transmitted_metric.labels.push(("direction", Cow::Borrowed("transmitted"))); - - if let Some(l) = labels { - received_metric.labels.extend(l.iter().map(|(k, v)| (*k, v.clone()))); - transmitted_metric.labels.extend(l.iter().map(|(k, v)| (*k, v.clone()))); - } - - metrics.push(received_metric); - metrics.push(transmitted_metric); - - // Per-interface network I/O - for (interface, received, transmitted) in &stats.per_interface { - let mut iface_received = PrometheusMetric::from_descriptor(&PROCESS_NETWORK_IO_PER_INTERFACE_MD, *received as f64); - let mut iface_transmitted = PrometheusMetric::from_descriptor(&PROCESS_NETWORK_IO_PER_INTERFACE_MD, *transmitted as f64); - - iface_received.labels.push(("interface", Cow::Owned(interface.clone()))); - iface_received.labels.push(("direction", Cow::Borrowed("received"))); - - iface_transmitted.labels.push(("interface", Cow::Owned(interface.clone()))); - iface_transmitted.labels.push(("direction", Cow::Borrowed("transmitted"))); - - if let Some(l) = labels { - iface_received.labels.extend(l.iter().map(|(k, v)| (*k, v.clone()))); - iface_transmitted.labels.extend(l.iter().map(|(k, v)| (*k, v.clone()))); - } - - metrics.push(iface_received); - metrics.push(iface_transmitted); - } - - metrics -} - #[cfg(test)] mod tests { use super::*; diff --git a/crates/obs/src/metrics/collectors/system_network_host.rs b/crates/obs/src/metrics/collectors/system_network_host.rs new file mode 100644 index 000000000..1bdc72d30 --- /dev/null +++ b/crates/obs/src/metrics/collectors/system_network_host.rs @@ -0,0 +1,102 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Network I/O metrics collector for host-wide interface counters. + +use crate::metrics::report::PrometheusMetric; +use crate::metrics::schema::system_network_host::{HOST_NETWORK_IO_MD, HOST_NETWORK_IO_PER_INTERFACE_MD}; +use std::borrow::Cow; + +/// Network I/O statistics. +/// +/// Contains host-wide network I/O totals and per-interface counters. +#[derive(Debug, Clone, Default)] +pub struct HostNetworkStats { + /// Total bytes received across observed host interfaces. + pub total_received: u64, + /// Total bytes transmitted across observed host interfaces. + pub total_transmitted: u64, + /// Per-interface statistics: (interface_name, received_bytes, transmitted_bytes) + pub per_interface: Vec<(String, u64, u64)>, +} + +/// Collects network I/O metrics from the given stats. +/// +/// Returns a vector of Prometheus metrics for host-wide network I/O statistics. +/// Each metric includes a `direction` label ("received" or "transmitted"). +/// Per-interface metrics also include an `interface` label. +pub fn collect_host_network_metrics( + stats: &HostNetworkStats, + labels: Option<&[(&'static str, Cow<'static, str>)]>, +) -> Vec { + let mut metrics = Vec::with_capacity(2 + stats.per_interface.len() * 2); + + let mut received_metric = PrometheusMetric::from_descriptor(&HOST_NETWORK_IO_MD, stats.total_received as f64); + let mut transmitted_metric = PrometheusMetric::from_descriptor(&HOST_NETWORK_IO_MD, stats.total_transmitted as f64); + + received_metric.labels.push(("direction", Cow::Borrowed("received"))); + transmitted_metric.labels.push(("direction", Cow::Borrowed("transmitted"))); + + if let Some(l) = labels { + received_metric.labels.extend(l.iter().map(|(k, v)| (*k, v.clone()))); + transmitted_metric.labels.extend(l.iter().map(|(k, v)| (*k, v.clone()))); + } + + metrics.push(received_metric); + metrics.push(transmitted_metric); + + for (interface, received, transmitted) in &stats.per_interface { + let mut iface_received = PrometheusMetric::from_descriptor(&HOST_NETWORK_IO_PER_INTERFACE_MD, *received as f64); + let mut iface_transmitted = PrometheusMetric::from_descriptor(&HOST_NETWORK_IO_PER_INTERFACE_MD, *transmitted as f64); + + iface_received.labels.push(("interface", Cow::Owned(interface.clone()))); + iface_received.labels.push(("direction", Cow::Borrowed("received"))); + + iface_transmitted.labels.push(("interface", Cow::Owned(interface.clone()))); + iface_transmitted.labels.push(("direction", Cow::Borrowed("transmitted"))); + + if let Some(l) = labels { + iface_received.labels.extend(l.iter().map(|(k, v)| (*k, v.clone()))); + iface_transmitted.labels.extend(l.iter().map(|(k, v)| (*k, v.clone()))); + } + + metrics.push(iface_received); + metrics.push(iface_transmitted); + } + + metrics +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn host_network_metrics_use_dedicated_network_host_prefix() { + let stats = HostNetworkStats { + total_received: 1024, + total_transmitted: 2048, + per_interface: vec![("eth0".to_string(), 512, 256)], + }; + + let metrics = collect_host_network_metrics(&stats, None); + + assert_eq!(metrics.len(), 4); + assert!( + metrics + .iter() + .all(|metric| metric.name.starts_with("rustfs_system_network_host_")) + ); + } +} diff --git a/crates/obs/src/metrics/scheduler.rs b/crates/obs/src/metrics/scheduler.rs index ff9acaed8..f2d4395e5 100644 --- a/crates/obs/src/metrics/scheduler.rs +++ b/crates/obs/src/metrics/scheduler.rs @@ -21,29 +21,30 @@ //! - Process CPU metrics //! - Process memory metrics //! - Process disk I/O metrics -//! - Process network I/O metrics +//! - Host network I/O metrics use crate::metrics::collectors::{ AuditTargetStats, NotificationStats, NotificationTargetStats, // System monitoring collectors (migrated from rustfs-obs::system) + ProcessAttributeError, ProcessCpuStats, ProcessDiskStats, ProcessMemoryStats, - ProcessNetworkStats, collect_audit_metrics, collect_bucket_metrics, collect_bucket_replication_bandwidth_metrics, collect_cluster_metrics, + collect_host_network_metrics, collect_node_metrics, collect_notification_metrics, collect_notification_target_metrics, + collect_process_attributes, collect_process_cpu_metrics, collect_process_disk_metrics, collect_process_memory_metrics, collect_process_metrics, - collect_process_network_metrics, collect_resource_metrics, }; use crate::metrics::config::{ @@ -53,17 +54,17 @@ use crate::metrics::config::{ ENV_BUCKET_REPLICATION_BANDWIDTH_METRICS_INTERVAL, ENV_CLUSTER_METRICS_INTERVAL, ENV_DEFAULT_METRICS_INTERVAL, ENV_NODE_METRICS_INTERVAL, ENV_NOTIFICATION_METRICS_INTERVAL, ENV_RESOURCE_METRICS_INTERVAL, }; -use crate::metrics::report::report_metrics; +use crate::metrics::report::{PrometheusMetric, report_metrics}; use crate::metrics::stats_collector::{ - collect_bucket_replication_bandwidth_stats, collect_bucket_stats, collect_cluster_stats, collect_disk_stats, - collect_process_resource_and_system_stats, + ProcessMetricBundle, collect_bucket_replication_bandwidth_stats, collect_bucket_stats, collect_cluster_stats, + collect_disk_stats, collect_host_network_stats, collect_process_metric_bundle, }; use rustfs_audit::audit_target_metrics; use rustfs_notify::{notification_metrics_snapshot, notification_target_metrics}; use rustfs_utils::get_env_opt_u64; use std::borrow::Cow; use std::time::Duration; -use sysinfo::{Pid, System}; +use tokio::time::Instant; use tokio_util::sync::CancellationToken; use tracing::warn; @@ -222,26 +223,6 @@ pub fn init_metrics_runtime(token: CancellationToken) { } }); - // Spawn task for resource metrics - let token_clone = token.clone(); - tokio::spawn(async move { - let mut interval = tokio::time::interval(resource_interval); - loop { - tokio::select! { - _ = interval.tick() => { - let (resource_stats, process_stats) = collect_process_resource_and_system_stats(); - let mut metrics = collect_resource_metrics(&resource_stats); - metrics.extend(collect_process_metrics(&process_stats)); - report_metrics(&metrics); - } - _ = token_clone.cancelled() => { - warn!("Metrics collection for resource stats cancelled."); - return; - } - } - } - }); - // Spawn task for audit target delivery metrics let token_clone = token.clone(); tokio::spawn(async move { @@ -315,25 +296,66 @@ pub fn init_metrics_runtime(token: CancellationToken) { let token_clone = token; tokio::spawn(async move { - // Get current process PID - let pid = match sysinfo::get_current_pid() { - Ok(p) => p, + let labels = current_process_metric_labels(); + let process_interval = resource_interval.min(system_interval); + let mut interval = tokio::time::interval(process_interval); + let now = Instant::now(); + let mut next_resource_run = now; + let mut next_system_run = now; + + #[cfg(feature = "gpu")] + let current_pid = match sysinfo::get_current_pid() { + Ok(pid) => Some(pid), Err(e) => { warn!("Failed to get current PID for system monitoring: {}", e); - return; + None } }; - let mut interval = tokio::time::interval(system_interval); loop { tokio::select! { _ = interval.tick() => { - // Collect system monitoring metrics - let metrics = collect_system_monitoring_metrics(pid); - report_metrics(&metrics); + let now = Instant::now(); + let bundle = collect_process_metric_bundle(); + + if now >= next_resource_run { + let mut metrics = collect_resource_metrics(&bundle.resource); + metrics.extend(collect_process_metrics(&bundle.process)); + report_metrics(&metrics); + advance_deadline(&mut next_resource_run, resource_interval, now); + } + + if now >= next_system_run { + #[cfg(feature = "gpu")] + let mut metrics = collect_system_monitoring_metrics(&bundle, &labels); + #[cfg(not(feature = "gpu"))] + let metrics = collect_system_monitoring_metrics(&bundle, &labels); + + #[cfg(feature = "gpu")] + if let Some(pid) = current_pid { + use crate::metrics::collectors::{GpuCollector, collect_gpu_metrics}; + + match GpuCollector::new(pid) { + Ok(collector) => match collector.collect() { + Ok(gpu_stats) => { + metrics.extend(collect_gpu_metrics(&gpu_stats, &labels)); + } + Err(e) => { + warn!("GPU metrics collection failed: {}", e); + } + }, + Err(e) => { + warn!("GPU collector initialization failed: {}", e); + } + } + } + + report_metrics(&metrics); + advance_deadline(&mut next_system_run, system_interval, now); + } } _ = token_clone.cancelled() => { - warn!("System monitoring metrics collection cancelled."); + warn!("Process metrics collection cancelled."); return; } } @@ -346,95 +368,99 @@ pub fn init_metrics_collectors(token: CancellationToken) { init_metrics_runtime(token); } -/// Collect all system monitoring metrics for a process. -/// -/// This function collects CPU, memory, disk I/O, and network I/O metrics -/// for the specified process PID. -/// -/// # Arguments -/// * `pid` - The process ID to monitor -/// -/// # Returns -/// A vector of Prometheus metrics for the process. -fn collect_system_monitoring_metrics(pid: Pid) -> Vec { - let mut metrics = Vec::new(); - let mut system = System::new(); - - // Refresh process information - system.refresh_processes(sysinfo::ProcessesToUpdate::Some(&[pid]), true); - - if let Some(process) = system.process(pid) { - // Create labels with process attributes - let labels: Vec<(&'static str, Cow<'static, str>)> = vec![ - ("process_pid", Cow::Owned(pid.as_u32().to_string())), - ("process_executable_name", Cow::Owned(process.name().to_string_lossy().to_string())), - ]; - - // Collect CPU metrics - let cpu_stats = ProcessCpuStats { - usage: process.cpu_usage() as f64, - utilization: process.cpu_usage() as f64, // Same as usage for single process - }; - metrics.extend(collect_process_cpu_metrics(&cpu_stats, Some(&labels))); - - // Collect memory metrics - let memory_stats = ProcessMemoryStats { - resident: process.memory(), - virtual_mem: process.virtual_memory(), - }; - metrics.extend(collect_process_memory_metrics(&memory_stats, Some(&labels))); - - // Collect disk I/O metrics - let disk_usage = process.disk_usage(); - let disk_stats = ProcessDiskStats { - read_bytes: disk_usage.read_bytes, - written_bytes: disk_usage.written_bytes, - }; - metrics.extend(collect_process_disk_metrics(&disk_stats, Some(&labels))); - - // Collect network I/O metrics - // Note: sysinfo 0.38.x provides network info via Networks new type - // We use Networks::new_with_refreshed_list() to get network interfaces - let networks = sysinfo::Networks::new_with_refreshed_list(); - let mut total_received = 0u64; - let mut total_transmitted = 0u64; - let mut per_interface = Vec::new(); - - for (interface_name, data) in networks.iter() { - let received = data.received(); - let transmitted = data.transmitted(); - total_received += received; - total_transmitted += transmitted; - per_interface.push((interface_name.to_string(), received, transmitted)); - } - - let network_stats = ProcessNetworkStats { - total_received, - total_transmitted, - per_interface, - }; - metrics.extend(collect_process_network_metrics(&network_stats, Some(&labels))); - - // Collect GPU metrics (if gpu feature is enabled) - #[cfg(feature = "gpu")] - { - use crate::metrics::collectors::{GpuCollector, collect_gpu_metrics}; - - match GpuCollector::new(pid) { - Ok(collector) => match collector.collect() { - Ok(gpu_stats) => { - metrics.extend(collect_gpu_metrics(&gpu_stats, &labels)); - } - Err(e) => { - warn!("GPU metrics collection failed: {}", e); - } - }, - Err(e) => { - warn!("GPU collector initialization failed: {}", e); - } - } - } +fn advance_deadline(deadline: &mut Instant, interval: Duration, now: Instant) { + if *deadline > now { + return; } + let interval_nanos = interval.as_nanos(); + if interval_nanos == 0 { + return; + } + + let elapsed = now.duration_since(*deadline); + let missed_intervals = (elapsed.as_nanos() / interval_nanos) + 1; + let mut remaining = missed_intervals; + + while remaining > 0 { + let chunk_u128 = remaining.min(u128::from(u32::MAX)); + let chunk_u32 = chunk_u128 as u32; + + if let Some(advance_by) = interval.checked_mul(chunk_u32) { + *deadline += advance_by; + remaining -= chunk_u128; + continue; + } + + *deadline += interval; + remaining -= 1; + } +} + +fn current_process_metric_labels() -> Vec<(&'static str, Cow<'static, str>)> { + match collect_process_attributes() { + Ok(attrs) => vec![ + ("process_pid", Cow::Owned(attrs.pid.to_string())), + ("process_executable_name", Cow::Owned(attrs.executable_name)), + ], + Err(err) => fallback_process_metric_labels(err), + } +} + +fn fallback_process_metric_labels(err: ProcessAttributeError) -> Vec<(&'static str, Cow<'static, str>)> { + warn!("Failed to collect process attributes for metrics labels: {}", err); + vec![ + ("process_pid", Cow::Owned(std::process::id().to_string())), + ("process_executable_name", Cow::Borrowed("unknown")), + ] +} + +fn collect_system_monitoring_metrics( + bundle: &ProcessMetricBundle, + labels: &[(&'static str, Cow<'static, str>)], +) -> Vec { + let cpu_stats = ProcessCpuStats { + usage: bundle.resource.cpu_percent, + utilization: bundle.resource.cpu_percent, + }; + let memory_stats = ProcessMemoryStats { + resident: bundle.process.resident_memory_bytes, + virtual_mem: bundle.process.virtual_memory_bytes, + }; + let disk_stats = ProcessDiskStats { + read_bytes: bundle.disk_read_bytes, + written_bytes: bundle.disk_write_bytes, + }; + let network_stats = collect_host_network_stats(); + + let mut metrics = Vec::new(); + metrics.extend(collect_process_cpu_metrics(&cpu_stats, Some(labels))); + metrics.extend(collect_process_memory_metrics(&memory_stats, Some(labels))); + metrics.extend(collect_process_disk_metrics(&disk_stats, Some(labels))); + // Interface counters are host-wide, so keep these metrics free of process labels. + metrics.extend(collect_host_network_metrics(&network_stats, None)); metrics } + +#[cfg(test)] +mod tests { + use super::advance_deadline; + use std::time::Duration; + use tokio::time::Instant; + + #[test] + fn advance_deadline_keeps_future_deadline_unchanged() { + let base = Instant::now(); + let mut deadline = base + Duration::from_secs(10); + advance_deadline(&mut deadline, Duration::from_secs(5), base); + assert_eq!(deadline, base + Duration::from_secs(10)); + } + + #[test] + fn advance_deadline_moves_to_first_tick_after_now() { + let base = Instant::now(); + let mut deadline = base; + advance_deadline(&mut deadline, Duration::from_secs(5), base + Duration::from_secs(12)); + assert_eq!(deadline, base + Duration::from_secs(15)); + } +} diff --git a/crates/obs/src/metrics/schema/entry/metric_name.rs b/crates/obs/src/metrics/schema/entry/metric_name.rs index 36b2185d8..a7cc2a812 100644 --- a/crates/obs/src/metrics/schema/entry/metric_name.rs +++ b/crates/obs/src/metrics/schema/entry/metric_name.rs @@ -355,10 +355,10 @@ pub enum MetricName { ProcessCPUUtilization, /// Process disk I/O bytes ProcessDiskIO, - /// Process network I/O bytes - ProcessNetworkIO, - /// Process network I/O bytes per interface - ProcessNetworkIOPerInterface, + /// Host network I/O bytes + HostNetworkIO, + /// Host network I/O bytes per interface + HostNetworkIOPerInterface, /// Process status (0: Running, 1: Sleeping, 2: Zombie, 3: Other) ProcessStatus, /// Process GPU memory usage in bytes @@ -688,8 +688,8 @@ impl MetricName { Self::ProcessCPUUsage => "cpu_usage".to_string(), Self::ProcessCPUUtilization => "cpu_utilization".to_string(), Self::ProcessDiskIO => "disk_io".to_string(), - Self::ProcessNetworkIO => "network_io".to_string(), - Self::ProcessNetworkIOPerInterface => "network_io_per_interface".to_string(), + Self::HostNetworkIO => "network_io".to_string(), + Self::HostNetworkIOPerInterface => "network_io_per_interface".to_string(), Self::ProcessGpuMemoryUsage => "gpu_memory_usage".to_string(), Self::ProcessStatus => "status".to_string(), diff --git a/crates/obs/src/metrics/schema/entry/subsystem.rs b/crates/obs/src/metrics/schema/entry/subsystem.rs index 2a1d33063..d0f0c611f 100644 --- a/crates/obs/src/metrics/schema/entry/subsystem.rs +++ b/crates/obs/src/metrics/schema/entry/subsystem.rs @@ -27,6 +27,7 @@ pub enum MetricSubsystem { // system related subsystems SystemNetworkInternode, + SystemNetworkHost, SystemDrive, SystemMemory, SystemCpu, @@ -68,6 +69,7 @@ impl MetricSubsystem { // system related subsystems Self::SystemNetworkInternode => "/system/network/internode", + Self::SystemNetworkHost => "/system/network/host", Self::SystemDrive => "/system/drive", Self::SystemMemory => "/system/memory", Self::SystemCpu => "/system/cpu", @@ -115,6 +117,7 @@ impl MetricSubsystem { // System-related subsystems "/system/network/internode" => Self::SystemNetworkInternode, + "/system/network/host" => Self::SystemNetworkHost, "/system/drive" => Self::SystemDrive, "/system/memory" => Self::SystemMemory, "/system/cpu" => Self::SystemCpu, @@ -183,6 +186,7 @@ pub mod subsystems { pub const BUCKET_REPLICATION: MetricSubsystem = MetricSubsystem::BucketReplication; pub const SYSTEM_GPU: MetricSubsystem = MetricSubsystem::SystemGpu; pub const SYSTEM_NETWORK_INTERNODE: MetricSubsystem = MetricSubsystem::SystemNetworkInternode; + pub const SYSTEM_NETWORK_HOST: MetricSubsystem = MetricSubsystem::SystemNetworkHost; pub const SYSTEM_DRIVE: MetricSubsystem = MetricSubsystem::SystemDrive; pub const SYSTEM_MEMORY: MetricSubsystem = MetricSubsystem::SystemMemory; pub const SYSTEM_CPU: MetricSubsystem = MetricSubsystem::SystemCpu; @@ -210,6 +214,7 @@ mod tests { fn test_metric_subsystem_formatting() { assert_eq!(MetricSubsystem::ApiRequests.as_str(), "api_requests"); assert_eq!(MetricSubsystem::SystemNetworkInternode.as_str(), "system_network_internode"); + assert_eq!(MetricSubsystem::SystemNetworkHost.as_str(), "system_network_host"); assert_eq!(MetricSubsystem::BucketApi.as_str(), "bucket_api"); assert_eq!(MetricSubsystem::ClusterHealth.as_str(), "cluster_health"); diff --git a/crates/obs/src/metrics/schema/mod.rs b/crates/obs/src/metrics/schema/mod.rs index c47f990e1..a82108700 100644 --- a/crates/obs/src/metrics/schema/mod.rs +++ b/crates/obs/src/metrics/schema/mod.rs @@ -36,6 +36,7 @@ pub mod system_drive; pub mod system_gpu; pub mod system_memory; pub mod system_network; +pub mod system_network_host; pub mod system_process; pub use entry::descriptor::MetricDescriptor; diff --git a/crates/obs/src/metrics/schema/system_network_host.rs b/crates/obs/src/metrics/schema/system_network_host.rs new file mode 100644 index 000000000..cb0881b7b --- /dev/null +++ b/crates/obs/src/metrics/schema/system_network_host.rs @@ -0,0 +1,38 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#![allow(dead_code)] + +use crate::{MetricDescriptor, MetricName, new_gauge_md, subsystems}; +use std::sync::LazyLock; + +/// Host network I/O bytes collected from system network interfaces. +pub static HOST_NETWORK_IO_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::HostNetworkIO, + "Network bytes transferred across system network interfaces", + &[], + subsystems::SYSTEM_NETWORK_HOST, + ) +}); + +/// Host network I/O bytes collected from system network interfaces, grouped per interface. +pub static HOST_NETWORK_IO_PER_INTERFACE_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::HostNetworkIOPerInterface, + "Network bytes transferred across system network interfaces (per interface)", + &[], + subsystems::SYSTEM_NETWORK_HOST, + ) +}); diff --git a/crates/obs/src/metrics/schema/system_process.rs b/crates/obs/src/metrics/schema/system_process.rs index c455faac4..34b46157c 100644 --- a/crates/obs/src/metrics/schema/system_process.rs +++ b/crates/obs/src/metrics/schema/system_process.rs @@ -221,26 +221,6 @@ pub static PROCESS_DISK_IO_MD: LazyLock = LazyLock::new(|| { ) }); -/// Process network I/O bytes -pub static PROCESS_NETWORK_IO_MD: LazyLock = LazyLock::new(|| { - new_gauge_md( - MetricName::ProcessNetworkIO, - "Network bytes transferred by the process", - &[], - subsystems::SYSTEM_PROCESS, - ) -}); - -/// Process network I/O bytes per interface -pub static PROCESS_NETWORK_IO_PER_INTERFACE_MD: LazyLock = LazyLock::new(|| { - new_gauge_md( - MetricName::ProcessNetworkIOPerInterface, - "Network bytes transferred by the process (per interface)", - &[], - subsystems::SYSTEM_PROCESS, - ) -}); - /// Process status (0: Running, 1: Sleeping, 2: Zombie, 3: Other) pub static PROCESS_STATUS_MD: LazyLock = LazyLock::new(|| { new_gauge_md( diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index 3d86c694d..d26055921 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -21,7 +21,8 @@ //! and convert them to the Stats structs used by collectors. use crate::metrics::collectors::{ - BucketReplicationBandwidthStats, BucketStats, ClusterStats, DiskStats, ProcessStats, ProcessStatusType, ResourceStats, + BucketReplicationBandwidthStats, BucketStats, ClusterStats, DiskStats, HostNetworkStats, ProcessStats, ProcessStatusType, + ResourceStats, }; use rustfs_ecstore::bucket::metadata_sys::get_quota_config; use rustfs_ecstore::data_usage::load_data_usage_from_backend; @@ -30,8 +31,17 @@ use rustfs_ecstore::pools::{get_total_usable_capacity, get_total_usable_capacity use rustfs_ecstore::store_api::{BucketOperations, BucketOptions}; use rustfs_ecstore::{StorageAPI, new_object_layer_fn}; use rustfs_io_metrics::{ProcessStatusSnapshot, snapshot_process_resource_and_system}; +use sysinfo::Networks; use tracing::{instrument, warn}; +#[derive(Debug, Clone, Default)] +pub struct ProcessMetricBundle { + pub resource: ResourceStats, + pub process: ProcessStats, + pub disk_read_bytes: u64, + pub disk_write_bytes: u64, +} + /// Collect cluster statistics from the storage layer. #[instrument] pub async fn collect_cluster_stats() -> ClusterStats { @@ -190,7 +200,7 @@ pub async fn collect_disk_stats() -> Vec { /// Collect resource and process statistics for the current process in one sysinfo refresh. #[inline] -pub fn collect_process_resource_and_system_stats() -> (ResourceStats, ProcessStats) { +pub fn collect_process_metric_bundle() -> ProcessMetricBundle { let (resource_snapshot, process_snapshot) = snapshot_process_resource_and_system(); let status = match process_snapshot.status { ProcessStatusSnapshot::Running => ProcessStatusType::Running, @@ -226,17 +236,53 @@ pub fn collect_process_resource_and_system_stats() -> (ResourceStats, ProcessSta virtual_memory_max_bytes: process_snapshot.virtual_memory_max_bytes, }; - (resource_stats, process_stats) + ProcessMetricBundle { + resource: resource_stats, + process: process_stats, + disk_read_bytes: process_snapshot.disk_read_bytes, + disk_write_bytes: process_snapshot.disk_write_bytes, + } +} + +/// Collect resource and process statistics for the current process in one sysinfo refresh. +#[inline] +pub fn collect_process_resource_and_system_stats() -> (ResourceStats, ProcessStats) { + let bundle = collect_process_metric_bundle(); + (bundle.resource, bundle.process) } /// Collect resource statistics for the current process. #[inline] pub fn collect_process_stats() -> ResourceStats { - collect_process_resource_and_system_stats().0 + collect_process_metric_bundle().resource } /// Collect process statistics for the current process. #[inline] pub fn collect_process_system_stats() -> ProcessStats { - collect_process_resource_and_system_stats().1 + collect_process_metric_bundle().process +} + +/// Collect host network statistics from the current network interface snapshot. +/// +/// These counters come from system interfaces and are host-wide, not process-scoped. +pub fn collect_host_network_stats() -> HostNetworkStats { + let networks = Networks::new_with_refreshed_list(); + let mut total_received = 0u64; + let mut total_transmitted = 0u64; + let mut per_interface = Vec::with_capacity(networks.len()); + + for (interface_name, data) in &networks { + let received = data.received(); + let transmitted = data.transmitted(); + total_received += received; + total_transmitted += transmitted; + per_interface.push((interface_name.to_string(), received, transmitted)); + } + + HostNetworkStats { + total_received, + total_transmitted, + per_interface, + } }