gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 08:18:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(metrics): unify process sampling and split network IO (#2590)

Browse Source
This commit is contained in:
houseme
2026-04-18 23:30:44 +08:00
committed by GitHub
parent f9b5ad17a9
commit 116db4f5d9
12 changed files with 491 additions and 250 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.docker/observability/grafana/dashboards/rustfs.json
+112 -3
View File
@@ -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",
crates/io-metrics/src/sampler/process.rs
+22 -20
View File
@@ -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<Instant> = OnceLock::new();
static PROCESS_SYSTEM: OnceLock<Mutex<System>> = 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<System> {
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())
}
}
crates/obs/src/metrics/collectors/mod.rs
+3 -1
View File
@@ -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,
crates/obs/src/metrics/collectors/system_network.rs
+1 -71
View File
@@ -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<PrometheusMetric> {
]
}
/// 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<PrometheusMetric> {
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::*;
crates/obs/src/metrics/collectors/system_network_host.rs
+102
View File
@@ -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<PrometheusMetric> {
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_"))
);
}
}
crates/obs/src/metrics/scheduler.rs
+150 -124
View File
@@ -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<crate::metrics::report::PrometheusMetric> {
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<PrometheusMetric> {
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));
}
}
crates/obs/src/metrics/schema/entry/metric_name.rs
+6 -6
View File
@@ -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(),
crates/obs/src/metrics/schema/entry/subsystem.rs
+5
View File
@@ -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");
crates/obs/src/metrics/schema/mod.rs
+1
View File
@@ -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;
crates/obs/src/metrics/schema/system_network_host.rs
+38
View File
@@ -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<MetricDescriptor> = 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<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::HostNetworkIOPerInterface,
"Network bytes transferred across system network interfaces (per interface)",
&[],
subsystems::SYSTEM_NETWORK_HOST,
)
});
crates/obs/src/metrics/schema/system_process.rs
-20
View File
@@ -221,26 +221,6 @@ pub static PROCESS_DISK_IO_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
)
});
/// Process network I/O bytes
pub static PROCESS_NETWORK_IO_MD: LazyLock<MetricDescriptor> = 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<MetricDescriptor> = 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<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
crates/obs/src/metrics/stats_collector.rs
+51 -5
View File
@@ -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<DiskStats> {
/// 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,
}
}