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

feat(metrics): migrate system monitoring from rustfs-obs to rustfs-metrics (#2242)

Browse Source 
Co-authored-by: houseme <housemecn@gmail.com>
This commit is contained in:
heihutu
2026-03-20 18:52:33 +08:00
committed by GitHub
parent 28f86a505e
commit 3c28f0a0ba
56 changed files with 1116 additions and 613 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/obs/src/global.rs
+4 -13
View File
@@ -12,10 +12,10 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::{AppConfig, GlobalError, OtelConfig, OtelGuard, SystemObserver, telemetry::init_telemetry};
use crate::{AppConfig, GlobalError, OtelConfig, OtelGuard, telemetry::init_telemetry};
use std::sync::{Arc, Mutex};
use tokio::sync::OnceCell;
use tracing::{error, info, warn};
use tracing::{info, warn};
/// Global guard for OpenTelemetry tracing
static GLOBAL_GUARD: OnceCell<Arc<Mutex<OtelGuard>>> = OnceCell::const_new();
@@ -116,17 +116,8 @@ pub async fn init_obs(endpoint: Option<String>) -> Result<OtelGuard, GlobalError
/// ```
pub async fn init_obs_with_config(config: &OtelConfig) -> Result<OtelGuard, GlobalError> {
let otel_guard = init_telemetry(config)?;
tokio::spawn(async move {
let obs_result = SystemObserver::init_process_observer().await;
match obs_result {
Ok(_) => {
info!(target: "rustfs::obs::system::metrics", "Process observer initialized successfully");
}
Err(e) => {
error!(target: "rustfs::obs::system::metrics", "Failed to initialize process observer: {}", e);
}
}
});
// Note: System monitoring has been migrated to rustfs-metrics
// Use rustfs_metrics::init_metrics_collectors() for system metrics
Ok(otel_guard)
}
crates/obs/src/lib.rs
+18 -15
View File
@@ -16,20 +16,12 @@
//!
//! provides tools for system and service monitoring
//!
//! ## feature mark
//! - `default`: default monitoring function
//! - `gpu`: gpu monitoring function
//! - `full`: includes all functions
//! ## Features
//!
//! to enable gpu monitoring add in cargo toml
//!
//! ```toml
//! # using gpu monitoring
//! rustfs-obs = { version = "0.1.0", features = ["gpu"] }
//!
//! # use all functions
//! rustfs-obs = { version = "0.1.0", features = ["full"] }
//! ```
//! This crate provides observability tools for RustFS:
//! - Logging with tracing
//! - Metrics collection
//! - Distributed tracing
//!
//! ## Usage
//!
@@ -53,16 +45,27 @@
//! # // Guard will be dropped here, flushing telemetry data
//! # }
//! ```
//!
//! ## System Monitoring Migration
//!
//! The system monitoring functionality has been migrated to `rustfs-metrics`.
//! Use `rustfs_metrics::init_metrics_collectors()` for system metrics collection.
//!
//! ```ignore
//! use tokio_util::sync::CancellationToken;
//! use rustfs_metrics::init_metrics_collectors;
//!
//! let token = CancellationToken::new();
//! init_metrics_collectors(token.clone());
//! ```
mod cleaner;
mod config;
mod error;
mod global;
mod system;
mod telemetry;
pub use cleaner::*;
pub use config::*;
pub use error::*;
pub use global::*;
pub use system::SystemObserver;
pub use telemetry::{OtelGuard, Recorder};
crates/obs/src/system/attributes.rs
-58
View File
@@ -1,58 +0,0 @@
// 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.
use crate::GlobalError;
use opentelemetry::KeyValue;
use sysinfo::{Pid, System};
pub(crate) const PROCESS_PID: opentelemetry::Key = opentelemetry::Key::from_static_str("process.pid");
pub(crate) const PROCESS_EXECUTABLE_NAME: opentelemetry::Key = opentelemetry::Key::from_static_str("process.executable.name");
pub(crate) const PROCESS_EXECUTABLE_PATH: opentelemetry::Key = opentelemetry::Key::from_static_str("process.executable.path");
pub(crate) const PROCESS_COMMAND: opentelemetry::Key = opentelemetry::Key::from_static_str("process.command");
/// Struct to hold process attributes
pub struct ProcessAttributes {
pub attributes: Vec<KeyValue>,
}
impl ProcessAttributes {
/// Creates a new instance of `ProcessAttributes` for the given PID.
pub fn new(pid: Pid, system: &mut System) -> Result<Self, GlobalError> {
system.refresh_processes(sysinfo::ProcessesToUpdate::Some(&[pid]), true);
let process = system
.process(pid)
.ok_or_else(|| GlobalError::ProcessNotFound(pid.as_u32()))?;
let attributes = vec![
KeyValue::new(PROCESS_PID, pid.as_u32() as i64),
KeyValue::new(PROCESS_EXECUTABLE_NAME, process.name().to_os_string().into_string().unwrap_or_default()),
KeyValue::new(
PROCESS_EXECUTABLE_PATH,
process
.exe()
.map(|path| path.to_string_lossy().into_owned())
.unwrap_or_default(),
),
KeyValue::new(
PROCESS_COMMAND,
process
.cmd()
.iter()
.fold(String::new(), |t1, t2| t1 + " " + t2.to_str().unwrap_or_default()),
),
];
Ok(ProcessAttributes { attributes })
}
}
crates/obs/src/system/collector.rs
-173
View File
@@ -1,173 +0,0 @@
// 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.
use crate::GlobalError;
use crate::system::attributes::ProcessAttributes;
use crate::system::metrics::{DIRECTION, INTERFACE, Metrics, STATUS};
use opentelemetry::KeyValue;
use std::time::SystemTime;
use sysinfo::{Networks, Pid, ProcessStatus, System};
use tokio::time::{Duration, sleep};
/// 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,
#[cfg(feature = "gpu")]
gpu_collector: crate::system::gpu::GpuCollector,
pid: Pid,
system: System,
networks: Networks,
core_count: usize,
interval_ms: u64,
}
impl Collector {
pub fn new(pid: Pid, meter: opentelemetry::metrics::Meter, interval_ms: u64) -> Result<Self, GlobalError> {
let mut system = System::new();
let attributes = ProcessAttributes::new(pid, &mut system)?;
let core_count = System::physical_core_count().ok_or(GlobalError::CoreCountError)?;
let metrics = Metrics::new(&meter);
#[cfg(feature = "gpu")]
let gpu_collector = crate::system::gpu::GpuCollector::new(pid)?;
let networks = Networks::new_with_refreshed_list();
Ok(Collector {
metrics,
attributes,
#[cfg(feature = "gpu")]
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]), false);
// refresh the network interface list and statistics
self.networks.refresh(false);
let process = self
.system
.process(self.pid)
.ok_or_else(|| GlobalError::ProcessNotFound(self.pid.as_u32()))?;
// CPU metrics
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);
// Memory metrics
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);
// Disk I/O metrics
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(),
);
// Network I/O indicators (corresponding to /system/network/internode)
let mut total_received: i64 = 0;
let mut total_transmitted: i64 = 0;
// statistics by interface
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(),
);
}
// global statistics
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(),
);
// Process status indicator (corresponding to /system/process)
let status_value = match process.status() {
ProcessStatus::Run => 0,
ProcessStatus::Sleep => 1,
ProcessStatus::Zombie => 2,
_ => 3, // other status
};
self.metrics.process_status.record(
status_value,
&[
&self.attributes.attributes[..],
&[KeyValue::new(STATUS, format!("{:?}", process.status()))],
]
.concat(),
);
// GPU Metrics (Optional) Non-MacOS
#[cfg(feature = "gpu")]
self.gpu_collector.collect(&self.metrics, &self.attributes)?;
Ok(())
}
}
crates/obs/src/system/gpu.rs
-57
View File
@@ -1,57 +0,0 @@
// 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.
use crate::GlobalError;
use crate::system::attributes::ProcessAttributes;
use crate::system::metrics::Metrics;
use nvml_wrapper::Nvml;
use nvml_wrapper::enums::device::UsedGpuMemory;
use sysinfo::Pid;
use tracing::warn;
/// `GpuCollector` is responsible for collecting GPU memory usage metrics.
pub struct GpuCollector {
nvml: Nvml,
pid: Pid,
}
impl GpuCollector {
pub fn new(pid: Pid) -> Result<Self, GlobalError> {
let nvml = Nvml::init().map_err(|e| GlobalError::GpuInitError(e.to_string()))?;
Ok(GpuCollector { nvml, pid })
}
pub fn collect(&self, metrics: &Metrics, attributes: &ProcessAttributes) -> Result<(), GlobalError> {
if let Ok(device) = self.nvml.device_by_index(0) {
if let Ok(gpu_stats) = device.running_compute_processes() {
for stat in gpu_stats.iter() {
if stat.pid == self.pid.as_u32() {
let memory_used = match stat.used_gpu_memory {
UsedGpuMemory::Used(bytes) => bytes,
UsedGpuMemory::Unavailable => 0,
};
metrics.gpu_memory_usage.record(memory_used, &attributes.attributes);
return Ok(());
}
}
} else {
warn!("Could not get GPU stats, recording 0 for GPU memory usage");
}
} else {
return Err(GlobalError::GpuDeviceError("No GPU device found".to_string()));
}
metrics.gpu_memory_usage.record(0, &attributes.attributes);
Ok(())
}
}
crates/obs/src/system/metrics.rs
-116
View File
@@ -1,116 +0,0 @@
// 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.
use opentelemetry::metrics::{Gauge, Meter};
pub(crate) const PROCESS_CPU_USAGE: &str = "process.cpu.usage";
pub(crate) const PROCESS_CPU_UTILIZATION: &str = "process.cpu.utilization";
pub(crate) const PROCESS_MEMORY_USAGE: &str = "process.memory.usage";
pub(crate) const PROCESS_MEMORY_VIRTUAL: &str = "process.memory.virtual";
pub(crate) const PROCESS_DISK_IO: &str = "process.disk.io";
pub(crate) const PROCESS_NETWORK_IO: &str = "process.network.io";
pub(crate) const PROCESS_NETWORK_IO_PER_INTERFACE: &str = "process.network.io.per_interface";
pub(crate) const PROCESS_STATUS: &str = "process.status";
#[cfg(feature = "gpu")]
pub const PROCESS_GPU_MEMORY_USAGE: &str = "process.gpu.memory.usage";
pub(crate) const DIRECTION: opentelemetry::Key = opentelemetry::Key::from_static_str("direction");
pub(crate) const STATUS: opentelemetry::Key = opentelemetry::Key::from_static_str("status");
pub(crate) 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: Gauge<f64>,
pub cpu_utilization: Gauge<f64>,
pub memory_usage: Gauge<i64>,
pub memory_virtual: Gauge<i64>,
pub disk_io: Gauge<i64>,
pub network_io: Gauge<i64>,
pub network_io_per_interface: Gauge<i64>,
pub process_status: Gauge<i64>,
#[cfg(feature = "gpu")]
pub gpu_memory_usage: Gauge<u64>,
}
impl Metrics {
pub fn new(meter: &Meter) -> Self {
let cpu_usage = meter
.f64_gauge(PROCESS_CPU_USAGE)
.with_description("The percentage of CPU in use.")
.with_unit("percent")
.build();
let cpu_utilization = meter
.f64_gauge(PROCESS_CPU_UTILIZATION)
.with_description("The amount of CPU in use.")
.with_unit("percent")
.build();
let memory_usage = meter
.i64_gauge(PROCESS_MEMORY_USAGE)
.with_description("The amount of physical memory in use.")
.with_unit("byte")
.build();
let memory_virtual = meter
.i64_gauge(PROCESS_MEMORY_VIRTUAL)
.with_description("The amount of committed virtual memory.")
.with_unit("byte")
.build();
let disk_io = meter
.i64_gauge(PROCESS_DISK_IO)
.with_description("Disk bytes transferred.")
.with_unit("byte")
.build();
let network_io = meter
.i64_gauge(PROCESS_NETWORK_IO)
.with_description("Network bytes transferred.")
.with_unit("byte")
.build();
let network_io_per_interface = meter
.i64_gauge(PROCESS_NETWORK_IO_PER_INTERFACE)
.with_description("Network bytes transferred (per interface).")
.with_unit("byte")
.build();
let process_status = meter
.i64_gauge(PROCESS_STATUS)
.with_description("Process status (0: Running, 1: Sleeping, 2: Zombie, etc.)")
.build();
#[cfg(feature = "gpu")]
let gpu_memory_usage = meter
.u64_gauge(PROCESS_GPU_MEMORY_USAGE)
.with_description("The amount of physical GPU memory in use.")
.with_unit("byte")
.build();
Metrics {
cpu_usage,
cpu_utilization,
memory_usage,
memory_virtual,
disk_io,
network_io,
network_io_per_interface,
process_status,
#[cfg(feature = "gpu")]
gpu_memory_usage,
}
}
}
crates/obs/src/system/mod.rs
-51
View File
@@ -1,51 +0,0 @@
// 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.
use crate::{GlobalError, observability_metric_enabled};
use opentelemetry::{global::meter, metrics::Meter};
use sysinfo::Pid;
mod attributes;
mod collector;
#[cfg(feature = "gpu")]
mod gpu;
mod metrics;
pub struct SystemObserver {}
impl SystemObserver {
/// 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() -> Result<(), GlobalError> {
if observability_metric_enabled() {
let meter = meter("system");
let pid = sysinfo::get_current_pid().map_err(|e| GlobalError::PidError(e.to_string()))?;
return SystemObserver::init_process_observer_for_pid(meter, pid).await;
}
Ok(())
}
/// 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: Meter, pid: Pid) -> Result<(), GlobalError> {
let interval_ms = rustfs_utils::get_env_u64(
rustfs_config::observability::ENV_OBS_METRICS_SYSTEM_INTERVAL_MS,
rustfs_config::observability::DEFAULT_METRICS_SYSTEM_INTERVAL_MS,
);
let mut collector = collector::Collector::new(pid, meter, interval_ms)?;
collector.run().await
}
}