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

fix(obs): label node-local metrics by server (#5465)

Browse Source 
Add stable server labels to node-local Prometheus metrics and OTLP resource attributes so dashboards can distinguish per-node CPU, memory, host network, and internode traffic series.

Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-07-30 12:57:22 +08:00
committed by GitHub
parent ad7663afd1
commit 30dc04c94b
21 changed files with 640 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+2
View File
@@ -9353,7 +9353,9 @@ dependencies = [
"metrics",
"metrics-util",
"num_cpus",
"rustfs-common",
"rustfs-s3-ops",
"rustfs-utils",
"sysinfo",
"thiserror 2.0.19",
"tokio",
crates/io-metrics/Cargo.toml
+2
View File
@@ -30,7 +30,9 @@ harness = false
[dependencies]
metrics = { workspace = true }
rustfs-common = { workspace = true }
rustfs-s3-ops = { workspace = true }
rustfs-utils = { workspace = true, features = ["ip"] }
num_cpus = { workspace = true }
thiserror = { workspace = true }
tokio = { workspace = true, features = ["sync", "fs", "rt-multi-thread"] }
crates/io-metrics/src/internode_metrics.rs
+199 -54
View File
@@ -15,7 +15,7 @@
use metrics::{counter, gauge};
use std::collections::HashMap;
use std::sync::{
Arc, LazyLock, RwLock,
Arc, LazyLock, OnceLock, RwLock,
atomic::{AtomicU64, Ordering},
};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
@@ -40,6 +40,7 @@ pub const INTERNODE_MSGPACK_CODEC_JSON: &str = "json";
const OPERATION_LABEL: &str = "operation";
const BACKEND_LABEL: &str = "backend";
const SERVER_LABEL: &str = "server";
const CLASSIFICATION_LABEL: &str = "classification";
const STAGE_LABEL: &str = "stage";
const DOMINANT_ERROR_LABEL: &str = "dominant_error";
@@ -77,74 +78,93 @@ pub struct InternodeOperationMetricDescriptor {
pub labels: &'static [&'static str],
}
const OPERATION_BACKEND_LABELS: &[&str] = &[OPERATION_LABEL, BACKEND_LABEL];
const OPERATION_BACKEND_CLASSIFICATION_LABELS: &[&str] = &[OPERATION_LABEL, BACKEND_LABEL, CLASSIFICATION_LABEL];
const OPERATION_BACKEND_HTTP_VERSION_LABELS: &[&str] = &[OPERATION_LABEL, BACKEND_LABEL, HTTP_VERSION_LABEL];
const QUORUM_FAILURE_LABELS: &[&str] = &[STAGE_LABEL, DOMINANT_ERROR_LABEL];
const SERVER_OPERATION_BACKEND_LABELS: &[&str] = &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL];
const SERVER_OPERATION_BACKEND_CLASSIFICATION_LABELS: &[&str] =
&[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, CLASSIFICATION_LABEL];
const SERVER_OPERATION_BACKEND_HTTP_VERSION_LABELS: &[&str] = &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, HTTP_VERSION_LABEL];
const SERVER_QUORUM_FAILURE_LABELS: &[&str] = &[SERVER_LABEL, STAGE_LABEL, DOMINANT_ERROR_LABEL];
pub const INTERNODE_OPERATION_METRICS: &[InternodeOperationMetricDescriptor] = &[
InternodeOperationMetricDescriptor {
name: INTERNODE_OPERATION_SENT_BYTES_TOTAL,
labels: OPERATION_BACKEND_LABELS,
labels: SERVER_OPERATION_BACKEND_LABELS,
},
InternodeOperationMetricDescriptor {
name: INTERNODE_OPERATION_RECV_BYTES_TOTAL,
labels: OPERATION_BACKEND_LABELS,
labels: SERVER_OPERATION_BACKEND_LABELS,
},
InternodeOperationMetricDescriptor {
name: INTERNODE_OPERATION_REQUESTS_OUTGOING_TOTAL,
labels: OPERATION_BACKEND_LABELS,
labels: SERVER_OPERATION_BACKEND_LABELS,
},
InternodeOperationMetricDescriptor {
name: INTERNODE_OPERATION_REQUESTS_INCOMING_TOTAL,
labels: OPERATION_BACKEND_LABELS,
labels: SERVER_OPERATION_BACKEND_LABELS,
},
InternodeOperationMetricDescriptor {
name: INTERNODE_OPERATION_ERRORS_TOTAL,
labels: OPERATION_BACKEND_LABELS,
labels: SERVER_OPERATION_BACKEND_LABELS,
},
InternodeOperationMetricDescriptor {
name: INTERNODE_OPERATION_DURATION_MS,
labels: OPERATION_BACKEND_LABELS,
labels: SERVER_OPERATION_BACKEND_LABELS,
},
InternodeOperationMetricDescriptor {
name: INTERNODE_OPERATION_CLASSIFIED_ERRORS_TOTAL,
labels: OPERATION_BACKEND_CLASSIFICATION_LABELS,
labels: SERVER_OPERATION_BACKEND_CLASSIFICATION_LABELS,
},
InternodeOperationMetricDescriptor {
name: INTERNODE_OPERATION_RETRIES_TOTAL,
labels: OPERATION_BACKEND_CLASSIFICATION_LABELS,
labels: SERVER_OPERATION_BACKEND_CLASSIFICATION_LABELS,
},
InternodeOperationMetricDescriptor {
name: INTERNODE_OPERATION_RETRY_SUCCESSES_TOTAL,
labels: OPERATION_BACKEND_CLASSIFICATION_LABELS,
labels: SERVER_OPERATION_BACKEND_CLASSIFICATION_LABELS,
},
InternodeOperationMetricDescriptor {
name: INTERNODE_OPERATION_HTTP_VERSIONS_TOTAL,
labels: OPERATION_BACKEND_HTTP_VERSION_LABELS,
labels: SERVER_OPERATION_BACKEND_HTTP_VERSION_LABELS,
},
InternodeOperationMetricDescriptor {
name: INTERNODE_OPERATION_STALL_TIMEOUTS_TOTAL,
labels: OPERATION_BACKEND_LABELS,
labels: SERVER_OPERATION_BACKEND_LABELS,
},
InternodeOperationMetricDescriptor {
name: INTERNODE_OPERATION_WRITE_SHUTDOWN_ERRORS_TOTAL,
labels: OPERATION_BACKEND_LABELS,
labels: SERVER_OPERATION_BACKEND_LABELS,
},
InternodeOperationMetricDescriptor {
name: ERASURE_WRITE_QUORUM_FAILURES_TOTAL,
labels: QUORUM_FAILURE_LABELS,
labels: SERVER_QUORUM_FAILURE_LABELS,
},
InternodeOperationMetricDescriptor {
name: INTERNODE_OPERATION_PAYLOAD_BYTES,
labels: OPERATION_BACKEND_LABELS,
labels: SERVER_OPERATION_BACKEND_LABELS,
},
InternodeOperationMetricDescriptor {
name: INTERNODE_OPERATION_LARGE_PAYLOADS_TOTAL,
labels: OPERATION_BACKEND_LABELS,
labels: SERVER_OPERATION_BACKEND_LABELS,
},
];
fn current_server_label() -> &'static str {
static STABLE_SERVER_LABEL: OnceLock<String> = OnceLock::new();
static FALLBACK_SERVER_LABEL: LazyLock<String> = LazyLock::new(rustfs_utils::get_local_ip_with_default);
if let Some(server) = STABLE_SERVER_LABEL.get() {
return server.as_str();
}
if let Some(server) = rustfs_common::try_get_global_local_node_name() {
let _ = STABLE_SERVER_LABEL.set(server);
if let Some(server) = STABLE_SERVER_LABEL.get() {
return server.as_str();
}
}
FALLBACK_SERVER_LABEL.as_str()
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct InternodeMetricsSnapshot {
pub sent_bytes_total: u64,
@@ -193,7 +213,7 @@ impl InternodeMetrics {
return;
}
self.sent_bytes_total.fetch_add(bytes, Ordering::Relaxed);
counter!("rustfs_system_network_internode_sent_bytes_total").increment(bytes);
counter!("rustfs_system_network_internode_sent_bytes_total", SERVER_LABEL => current_server_label()).increment(bytes);
}
pub fn record_sent_bytes_for_operation(&self, operation: &'static str, bytes: usize) {
@@ -207,7 +227,13 @@ impl InternodeMetrics {
if bytes == 0 {
return;
}
counter!(INTERNODE_OPERATION_SENT_BYTES_TOTAL, OPERATION_LABEL => operation, BACKEND_LABEL => backend).increment(bytes);
counter!(
INTERNODE_OPERATION_SENT_BYTES_TOTAL,
SERVER_LABEL => current_server_label(),
OPERATION_LABEL => operation,
BACKEND_LABEL => backend
)
.increment(bytes);
}
pub fn record_recv_bytes(&self, bytes: usize) {
@@ -216,7 +242,7 @@ impl InternodeMetrics {
return;
}
self.recv_bytes_total.fetch_add(bytes, Ordering::Relaxed);
counter!("rustfs_system_network_internode_recv_bytes_total").increment(bytes);
counter!("rustfs_system_network_internode_recv_bytes_total", SERVER_LABEL => current_server_label()).increment(bytes);
}
pub fn record_recv_bytes_for_operation(&self, operation: &'static str, bytes: usize) {
@@ -230,12 +256,18 @@ impl InternodeMetrics {
if bytes == 0 {
return;
}
counter!(INTERNODE_OPERATION_RECV_BYTES_TOTAL, OPERATION_LABEL => operation, BACKEND_LABEL => backend).increment(bytes);
counter!(
INTERNODE_OPERATION_RECV_BYTES_TOTAL,
SERVER_LABEL => current_server_label(),
OPERATION_LABEL => operation,
BACKEND_LABEL => backend
)
.increment(bytes);
}
pub fn record_outgoing_request(&self) {
self.outgoing_requests_total.fetch_add(1, Ordering::Relaxed);
counter!("rustfs_system_network_internode_requests_outgoing_total").increment(1);
counter!("rustfs_system_network_internode_requests_outgoing_total", SERVER_LABEL => current_server_label()).increment(1);
}
pub fn record_outgoing_request_for_operation(&self, operation: &'static str) {
@@ -244,13 +276,18 @@ impl InternodeMetrics {
pub fn record_outgoing_request_for_operation_and_backend(&self, operation: &'static str, backend: &'static str) {
self.record_outgoing_request();
counter!(INTERNODE_OPERATION_REQUESTS_OUTGOING_TOTAL, OPERATION_LABEL => operation, BACKEND_LABEL => backend)
.increment(1);
counter!(
INTERNODE_OPERATION_REQUESTS_OUTGOING_TOTAL,
SERVER_LABEL => current_server_label(),
OPERATION_LABEL => operation,
BACKEND_LABEL => backend
)
.increment(1);
}
pub fn record_incoming_request(&self) {
self.incoming_requests_total.fetch_add(1, Ordering::Relaxed);
counter!("rustfs_system_network_internode_requests_incoming_total").increment(1);
counter!("rustfs_system_network_internode_requests_incoming_total", SERVER_LABEL => current_server_label()).increment(1);
}
pub fn record_incoming_request_for_operation(&self, operation: &'static str) {
@@ -259,13 +296,18 @@ impl InternodeMetrics {
pub fn record_incoming_request_for_operation_and_backend(&self, operation: &'static str, backend: &'static str) {
self.record_incoming_request();
counter!(INTERNODE_OPERATION_REQUESTS_INCOMING_TOTAL, OPERATION_LABEL => operation, BACKEND_LABEL => backend)
.increment(1);
counter!(
INTERNODE_OPERATION_REQUESTS_INCOMING_TOTAL,
SERVER_LABEL => current_server_label(),
OPERATION_LABEL => operation,
BACKEND_LABEL => backend
)
.increment(1);
}
pub fn record_error(&self) {
self.errors_total.fetch_add(1, Ordering::Relaxed);
counter!("rustfs_system_network_internode_errors_total").increment(1);
counter!("rustfs_system_network_internode_errors_total", SERVER_LABEL => current_server_label()).increment(1);
}
pub fn record_error_for_operation(&self, operation: &'static str) {
@@ -274,13 +316,24 @@ impl InternodeMetrics {
pub fn record_error_for_operation_and_backend(&self, operation: &'static str, backend: &'static str) {
self.record_error();
counter!(INTERNODE_OPERATION_ERRORS_TOTAL, OPERATION_LABEL => operation, BACKEND_LABEL => backend).increment(1);
counter!(
INTERNODE_OPERATION_ERRORS_TOTAL,
SERVER_LABEL => current_server_label(),
OPERATION_LABEL => operation,
BACKEND_LABEL => backend
)
.increment(1);
}
pub fn record_duration_for_operation_and_backend(&self, operation: &'static str, backend: &'static str, duration: Duration) {
let duration_ms = duration.as_secs_f64() * 1000.0;
metrics::histogram!(INTERNODE_OPERATION_DURATION_MS, OPERATION_LABEL => operation, BACKEND_LABEL => backend)
.record(duration_ms);
metrics::histogram!(
INTERNODE_OPERATION_DURATION_MS,
SERVER_LABEL => current_server_label(),
OPERATION_LABEL => operation,
BACKEND_LABEL => backend
)
.record(duration_ms);
}
pub fn record_classified_error_for_operation_and_backend(
@@ -291,6 +344,7 @@ impl InternodeMetrics {
) {
counter!(
INTERNODE_OPERATION_CLASSIFIED_ERRORS_TOTAL,
SERVER_LABEL => current_server_label(),
OPERATION_LABEL => operation,
BACKEND_LABEL => backend,
CLASSIFICATION_LABEL => classification
@@ -306,6 +360,7 @@ impl InternodeMetrics {
) {
counter!(
INTERNODE_OPERATION_RETRIES_TOTAL,
SERVER_LABEL => current_server_label(),
OPERATION_LABEL => operation,
BACKEND_LABEL => backend,
CLASSIFICATION_LABEL => classification
@@ -321,6 +376,7 @@ impl InternodeMetrics {
) {
counter!(
INTERNODE_OPERATION_RETRY_SUCCESSES_TOTAL,
SERVER_LABEL => current_server_label(),
OPERATION_LABEL => operation,
BACKEND_LABEL => backend,
CLASSIFICATION_LABEL => classification
@@ -337,6 +393,7 @@ impl InternodeMetrics {
self.operation_http_versions_total.fetch_add(1, Ordering::Relaxed);
counter!(
INTERNODE_OPERATION_HTTP_VERSIONS_TOTAL,
SERVER_LABEL => current_server_label(),
OPERATION_LABEL => operation,
BACKEND_LABEL => backend,
HTTP_VERSION_LABEL => http_version
@@ -346,13 +403,24 @@ impl InternodeMetrics {
pub fn record_stall_timeout_for_operation_and_backend(&self, operation: &'static str, backend: &'static str) {
self.operation_stall_timeouts_total.fetch_add(1, Ordering::Relaxed);
counter!(INTERNODE_OPERATION_STALL_TIMEOUTS_TOTAL, OPERATION_LABEL => operation, BACKEND_LABEL => backend).increment(1);
counter!(
INTERNODE_OPERATION_STALL_TIMEOUTS_TOTAL,
SERVER_LABEL => current_server_label(),
OPERATION_LABEL => operation,
BACKEND_LABEL => backend
)
.increment(1);
}
pub fn record_write_shutdown_error_for_operation_and_backend(&self, operation: &'static str, backend: &'static str) {
self.operation_write_shutdown_errors_total.fetch_add(1, Ordering::Relaxed);
counter!(INTERNODE_OPERATION_WRITE_SHUTDOWN_ERRORS_TOTAL, OPERATION_LABEL => operation, BACKEND_LABEL => backend)
.increment(1);
counter!(
INTERNODE_OPERATION_WRITE_SHUTDOWN_ERRORS_TOTAL,
SERVER_LABEL => current_server_label(),
OPERATION_LABEL => operation,
BACKEND_LABEL => backend
)
.increment(1);
}
/// Record the payload size (bytes) of a completed internode operation into a histogram
@@ -360,15 +428,26 @@ impl InternodeMetrics {
/// (`ReadAll`/`ReadMultiple`/`WriteAll`) would benefit from being moved off the shared
/// control-plane channel (see docs/grpc-optimization P1).
pub fn record_operation_payload_bytes(&self, operation: &'static str, backend: &'static str, bytes: usize) {
metrics::histogram!(INTERNODE_OPERATION_PAYLOAD_BYTES, OPERATION_LABEL => operation, BACKEND_LABEL => backend)
.record(bytes as f64);
metrics::histogram!(
INTERNODE_OPERATION_PAYLOAD_BYTES,
SERVER_LABEL => current_server_label(),
OPERATION_LABEL => operation,
BACKEND_LABEL => backend
)
.record(bytes as f64);
}
/// Increment the large-payload counter for an operation+backend whose payload exceeded the
/// caller-configured warning threshold. Feeds alerting on large unary RPCs that contend with
/// latency-sensitive control-plane traffic on the shared connection.
pub fn record_large_operation_payload(&self, operation: &'static str, backend: &'static str) {
counter!(INTERNODE_OPERATION_LARGE_PAYLOADS_TOTAL, OPERATION_LABEL => operation, BACKEND_LABEL => backend).increment(1);
counter!(
INTERNODE_OPERATION_LARGE_PAYLOADS_TOTAL,
SERVER_LABEL => current_server_label(),
OPERATION_LABEL => operation,
BACKEND_LABEL => backend
)
.increment(1);
}
/// Count a decode that fell back to the JSON compatibility field because the msgpack `_bin`
@@ -377,13 +456,20 @@ impl InternodeMetrics {
/// dropped (grpc-optimization P2). `direction` is [`INTERNODE_MSGPACK_DIRECTION_REQUEST`] or
/// [`INTERNODE_MSGPACK_DIRECTION_RESPONSE`]; `message` is the low-cardinality value name.
pub fn record_msgpack_json_fallback(&self, direction: &'static str, message: &'static str) {
counter!(INTERNODE_MSGPACK_JSON_FALLBACK_TOTAL, DIRECTION_LABEL => direction, MESSAGE_LABEL => message).increment(1);
counter!(
INTERNODE_MSGPACK_JSON_FALLBACK_TOTAL,
SERVER_LABEL => current_server_label(),
DIRECTION_LABEL => direction,
MESSAGE_LABEL => message
)
.increment(1);
}
pub fn record_msgpack_json_decode(&self, direction: &'static str, message: &'static str, codec: &'static str) {
self.msgpack_json_decode_total.fetch_add(1, Ordering::Relaxed);
counter!(
INTERNODE_MSGPACK_JSON_DECODE_TOTAL,
SERVER_LABEL => current_server_label(),
DIRECTION_LABEL => direction,
MESSAGE_LABEL => message,
CODEC_LABEL => codec
@@ -395,6 +481,7 @@ impl InternodeMetrics {
self.msgpack_json_decode_error_total.fetch_add(1, Ordering::Relaxed);
counter!(
INTERNODE_MSGPACK_JSON_DECODE_ERROR_TOTAL,
SERVER_LABEL => current_server_label(),
DIRECTION_LABEL => direction,
MESSAGE_LABEL => message,
CODEC_LABEL => codec
@@ -420,7 +507,7 @@ impl InternodeMetrics {
/// enabled; after the strict flip the legacy fallback path is closed and the counter stays flat.
pub fn record_signature_v1_fallback(&self) {
self.signature_v1_fallback_total.fetch_add(1, Ordering::Relaxed);
counter!(INTERNODE_SIGNATURE_V1_FALLBACK_TOTAL).increment(1);
counter!(INTERNODE_SIGNATURE_V1_FALLBACK_TOTAL, SERVER_LABEL => current_server_label()).increment(1);
}
/// Count a mutating internode disk RPC that was accepted without a signature-bound canonical
@@ -431,14 +518,14 @@ impl InternodeMetrics {
/// mutations are rejected and the counter stays flat.
pub fn record_body_digest_fallback(&self) {
self.body_digest_fallback_total.fetch_add(1, Ordering::Relaxed);
counter!(INTERNODE_BODY_DIGEST_FALLBACK_TOTAL).increment(1);
counter!(INTERNODE_BODY_DIGEST_FALLBACK_TOTAL, SERVER_LABEL => current_server_label()).increment(1);
}
/// Count an accepted v1/v2 request that does not carry the replay-scoped signature. This is
/// the convergence signal for `RUSTFS_INTERNODE_RPC_REPLAY_SCOPE_STRICT`.
pub fn record_replay_scope_fallback(&self) {
self.replay_scope_fallback_total.fetch_add(1, Ordering::Relaxed);
counter!(INTERNODE_REPLAY_SCOPE_FALLBACK_TOTAL).increment(1);
counter!(INTERNODE_REPLAY_SCOPE_FALLBACK_TOTAL, SERVER_LABEL => current_server_label()).increment(1);
}
/// Count a body-bound internode RPC rejected because the replay-protection nonce cache was
@@ -447,12 +534,13 @@ impl InternodeMetrics {
/// mutation rate and writes are being refused — alert on this counter.
pub fn record_replay_cache_overflow(&self) {
self.replay_cache_overflow_total.fetch_add(1, Ordering::Relaxed);
counter!(INTERNODE_REPLAY_CACHE_OVERFLOW_TOTAL).increment(1);
counter!(INTERNODE_REPLAY_CACHE_OVERFLOW_TOTAL, SERVER_LABEL => current_server_label()).increment(1);
}
pub fn record_erasure_write_quorum_failure(&self, stage: &'static str, dominant_error: &'static str) {
counter!(
ERASURE_WRITE_QUORUM_FAILURES_TOTAL,
SERVER_LABEL => current_server_label(),
STAGE_LABEL => stage,
DOMINANT_ERROR_LABEL => dominant_error
)
@@ -464,11 +552,12 @@ impl InternodeMetrics {
self.dial_total_time_nanos.fetch_add(elapsed_nanos, Ordering::Relaxed);
let samples = self.dial_samples_total.fetch_add(1, Ordering::Relaxed) + 1;
let total = self.dial_total_time_nanos.load(Ordering::Relaxed);
gauge!("rustfs_system_network_internode_dial_avg_time_nanos").set(total as f64 / samples as f64);
gauge!("rustfs_system_network_internode_dial_avg_time_nanos", SERVER_LABEL => current_server_label())
.set(total as f64 / samples as f64);
if !success {
self.dial_errors_total.fetch_add(1, Ordering::Relaxed);
counter!("rustfs_system_network_internode_dial_errors_total").increment(1);
counter!("rustfs_system_network_internode_dial_errors_total", SERVER_LABEL => current_server_label()).increment(1);
}
let now_ms = SystemTime::now()
@@ -687,6 +776,9 @@ fn cluster_peer_health_keys() -> Vec<String> {
#[cfg(test)]
mod tests {
use super::*;
use metrics::with_local_recorder;
use metrics_util::debugging::DebuggingRecorder;
use std::collections::HashSet;
#[test]
fn snapshot_reports_recorded_values() {
@@ -750,22 +842,22 @@ mod tests {
fn operation_metric_descriptors_include_backend_and_operation_labels() {
assert_eq!(INTERNODE_OPERATION_METRICS.len(), 15);
for metric in &INTERNODE_OPERATION_METRICS[..6] {
assert_eq!(metric.labels, &[OPERATION_LABEL, BACKEND_LABEL]);
assert_eq!(metric.labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]);
}
for metric in &INTERNODE_OPERATION_METRICS[6..9] {
assert_eq!(metric.labels, &[OPERATION_LABEL, BACKEND_LABEL, CLASSIFICATION_LABEL]);
assert_eq!(metric.labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, CLASSIFICATION_LABEL]);
}
assert_eq!(
INTERNODE_OPERATION_METRICS[9].labels,
&[OPERATION_LABEL, BACKEND_LABEL, HTTP_VERSION_LABEL]
&[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, HTTP_VERSION_LABEL]
);
for metric in &INTERNODE_OPERATION_METRICS[10..12] {
assert_eq!(metric.labels, &[OPERATION_LABEL, BACKEND_LABEL]);
assert_eq!(metric.labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]);
}
assert_eq!(INTERNODE_OPERATION_METRICS[12].labels, &[STAGE_LABEL, DOMINANT_ERROR_LABEL]);
assert_eq!(INTERNODE_OPERATION_METRICS[12].labels, &[SERVER_LABEL, STAGE_LABEL, DOMINANT_ERROR_LABEL]);
// Payload histogram + large-payload counter carry operation+backend labels.
assert_eq!(INTERNODE_OPERATION_METRICS[13].labels, &[OPERATION_LABEL, BACKEND_LABEL]);
assert_eq!(INTERNODE_OPERATION_METRICS[14].labels, &[OPERATION_LABEL, BACKEND_LABEL]);
assert_eq!(INTERNODE_OPERATION_METRICS[13].labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]);
assert_eq!(INTERNODE_OPERATION_METRICS[14].labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]);
}
#[test]
@@ -843,6 +935,59 @@ mod tests {
);
}
#[test]
fn direct_internode_metrics_emit_stable_server_label() {
let recorder = DebuggingRecorder::new();
let snapshotter = recorder.snapshotter();
let metrics = InternodeMetrics::default();
with_local_recorder(&recorder, || {
metrics.record_sent_bytes_for_operation_and_backend(
INTERNODE_OPERATION_READ_FILE_STREAM,
INTERNODE_TRANSPORT_BACKEND_TCP_HTTP,
128,
);
metrics.record_recv_bytes_for_operation_and_backend(
INTERNODE_OPERATION_PUT_FILE_STREAM,
INTERNODE_TRANSPORT_BACKEND_TCP_HTTP,
256,
);
metrics.record_dial_result(Duration::from_millis(3), false);
});
let observed: Vec<(String, HashSet<String>, Option<String>)> = snapshotter
.snapshot()
.into_vec()
.into_iter()
.filter(|(composite, _, _, _)| {
matches!(
composite.key().name(),
"rustfs_system_network_internode_sent_bytes_total"
| "rustfs_system_network_internode_recv_bytes_total"
| INTERNODE_OPERATION_SENT_BYTES_TOTAL
| INTERNODE_OPERATION_RECV_BYTES_TOTAL
| "rustfs_system_network_internode_dial_avg_time_nanos"
| "rustfs_system_network_internode_dial_errors_total"
)
})
.map(|(composite, _, _, _)| {
let labels = composite.key().labels();
let keys = labels.clone().map(|label| label.key().to_string()).collect();
let server = labels
.filter(|label| label.key() == SERVER_LABEL)
.map(|label| label.value().to_string())
.next();
(composite.key().name().to_string(), keys, server)
})
.collect();
assert_eq!(observed.len(), 6);
for (name, keys, server) in observed {
assert!(keys.contains(SERVER_LABEL), "{name} must carry the server label");
assert!(server.is_some_and(|value| !value.is_empty()), "{name} server label must not be empty");
}
}
#[test]
fn msgpack_json_fallback_counter_records_without_panicking() {
// Smoke test: the counter accepts both directions and a static message label.
crates/obs/src/lib.rs
+1
View File
@@ -64,6 +64,7 @@ mod error;
mod global;
mod logging;
pub mod metrics;
mod node_identity;
mod telemetry;
pub use cleaner::*;
crates/obs/src/metrics/collectors/resource.rs
+20 -4
View File
@@ -22,6 +22,7 @@
use crate::metrics::report::PrometheusMetric;
use crate::metrics::schema::process_resource::*;
use crate::node_identity::SERVER_LABEL;
/// Resource statistics for metrics collection.
///
@@ -30,6 +31,8 @@ use crate::metrics::schema::process_resource::*;
/// this struct from their available data sources.
#[derive(Debug, Clone, Default)]
pub struct ResourceStats {
/// Stable local node identity for labeling node-local process resource metrics
pub server: String,
/// CPU usage as a percentage (can exceed 100% on multi-core systems)
pub cpu_percent: f64,
/// Resident memory usage in bytes
@@ -43,10 +46,14 @@ pub struct ResourceStats {
/// Uses the metric descriptors from `metrics_type::process_resource` module.
/// Returns a vector of Prometheus metrics for resource statistics.
pub fn collect_resource_metrics(stats: &ResourceStats) -> Vec<PrometheusMetric> {
let server_label = stats.server.as_str();
vec![
PrometheusMetric::from_descriptor(&PROCESS_CPU_PERCENT_MD, stats.cpu_percent),
PrometheusMetric::from_descriptor(&PROCESS_MEMORY_BYTES_MD, stats.memory_bytes as f64),
PrometheusMetric::from_descriptor(&PROCESS_UPTIME_SECONDS_MD, stats.uptime_seconds as f64),
PrometheusMetric::from_descriptor(&PROCESS_CPU_PERCENT_MD, stats.cpu_percent)
.with_label_owned(SERVER_LABEL, server_label.to_string()),
PrometheusMetric::from_descriptor(&PROCESS_MEMORY_BYTES_MD, stats.memory_bytes as f64)
.with_label_owned(SERVER_LABEL, server_label.to_string()),
PrometheusMetric::from_descriptor(&PROCESS_UPTIME_SECONDS_MD, stats.uptime_seconds as f64)
.with_label_owned(SERVER_LABEL, server_label.to_string()),
]
}
@@ -58,6 +65,7 @@ mod tests {
#[test]
fn test_collect_resource_metrics() {
let stats = ResourceStats {
server: "node1:9000".to_string(),
cpu_percent: 45.5,
memory_bytes: 1024 * 1024 * 256,
uptime_seconds: 7200,
@@ -67,6 +75,11 @@ mod tests {
report_metrics(&metrics);
assert_eq!(metrics.len(), 3);
assert!(
metrics
.iter()
.all(|m| m.labels.iter().any(|(k, v)| *k == SERVER_LABEL && v == "node1:9000"))
);
// Verify CPU metric
let cpu_metric_name = PROCESS_CPU_PERCENT_MD.get_full_metric_name();
@@ -97,13 +110,16 @@ mod tests {
for metric in &metrics {
assert_eq!(metric.value, 0.0);
assert!(metric.labels.is_empty());
assert_eq!(metric.labels.len(), 1);
assert_eq!(metric.labels[0].0, SERVER_LABEL);
assert!(metric.labels[0].1.is_empty());
}
}
#[test]
fn test_collect_resource_metrics_high_cpu() {
let stats = ResourceStats {
server: "node1:9000".to_string(),
cpu_percent: 150.0, // Can exceed 100% on multi-core systems
memory_bytes: 0,
uptime_seconds: 0,
crates/obs/src/metrics/collectors/system_cpu.rs
+28 -8
View File
@@ -25,11 +25,14 @@
use crate::metrics::report::PrometheusMetric;
use crate::metrics::schema::system_cpu::*;
use crate::metrics::schema::system_process::{PROCESS_CPU_USAGE_MD, PROCESS_CPU_UTILIZATION_MD};
use crate::node_identity::SERVER_LABEL;
use std::borrow::Cow;
/// System CPU statistics.
#[derive(Debug, Clone, Default)]
pub struct CpuStats {
/// Stable local node identity for labeling node-local CPU metrics
pub server: String,
/// Average CPU idle time (percentage, 0-100)
pub avg_idle: f64,
/// CPU load average over 1 minute
@@ -56,11 +59,16 @@ pub struct ProcessCpuStats {
/// Uses the metric descriptors from `metrics_type::system_cpu` module.
/// Returns a vector of Prometheus metrics for CPU statistics.
pub fn collect_cpu_metrics(stats: &CpuStats) -> Vec<PrometheusMetric> {
let server_label = stats.server.as_str();
vec![
PrometheusMetric::from_descriptor(&SYS_CPU_AVG_IDLE_MD, stats.avg_idle),
PrometheusMetric::from_descriptor(&SYS_CPU_LOAD_MD, stats.load_avg),
PrometheusMetric::from_descriptor(&SYS_CPU_LOAD_PERC_MD, stats.load_avg_perc),
PrometheusMetric::from_descriptor(&SYS_CPU_USAGE_PERC_MD, stats.usage_perc),
PrometheusMetric::from_descriptor(&SYS_CPU_AVG_IDLE_MD, stats.avg_idle)
.with_label_owned(SERVER_LABEL, server_label.to_string()),
PrometheusMetric::from_descriptor(&SYS_CPU_LOAD_MD, stats.load_avg)
.with_label_owned(SERVER_LABEL, server_label.to_string()),
PrometheusMetric::from_descriptor(&SYS_CPU_LOAD_PERC_MD, stats.load_avg_perc)
.with_label_owned(SERVER_LABEL, server_label.to_string()),
PrometheusMetric::from_descriptor(&SYS_CPU_USAGE_PERC_MD, stats.usage_perc)
.with_label_owned(SERVER_LABEL, server_label.to_string()),
]
}
@@ -91,10 +99,12 @@ pub fn collect_process_cpu_metrics(
mod tests {
use super::*;
use crate::metrics::report::report_metrics;
use crate::metrics::schema::system_process::{PROCESS_EXECUTABLE_NAME_LABEL, PROCESS_PID_LABEL};
#[test]
fn test_collect_cpu_metrics() {
let stats = CpuStats {
server: "node1:9000".to_string(),
avg_idle: 75.5,
load_avg: 1.5,
load_avg_perc: 37.5,
@@ -108,6 +118,11 @@ mod tests {
// Verify that metric names are properly generated from descriptors
assert!(metrics.iter().all(|m| m.name.starts_with("rustfs_system_cpu_")));
assert!(
metrics
.iter()
.all(|m| m.labels.iter().any(|(k, v)| *k == SERVER_LABEL && v == "node1:9000"))
);
}
#[test]
@@ -118,13 +133,16 @@ mod tests {
assert_eq!(metrics.len(), 4);
for metric in &metrics {
assert_eq!(metric.value, 0.0);
assert!(metric.labels.is_empty());
assert_eq!(metric.labels.len(), 1);
assert_eq!(metric.labels[0].0, SERVER_LABEL);
assert!(metric.labels[0].1.is_empty());
}
}
#[test]
fn system_cpu_metrics_export_total_usage_under_honest_name() {
let stats = CpuStats {
server: "node1:9000".to_string(),
avg_idle: 40.0,
load_avg: 1.0,
load_avg_perc: 25.0,
@@ -171,8 +189,9 @@ mod tests {
};
let labels = vec![
("process_pid", Cow::Borrowed("12345")),
("process_executable_name", Cow::Borrowed("rustfs")),
(SERVER_LABEL, Cow::Borrowed("node1:9000")),
(PROCESS_PID_LABEL, Cow::Borrowed("12345")),
(PROCESS_EXECUTABLE_NAME_LABEL, Cow::Borrowed("rustfs")),
];
let metrics = collect_process_cpu_metrics(&stats, Some(&labels));
@@ -180,7 +199,8 @@ mod tests {
// All metrics should have the labels
for metric in &metrics {
assert_eq!(metric.labels.len(), 2);
assert_eq!(metric.labels.len(), 3);
assert!(metric.labels.iter().any(|(k, v)| *k == SERVER_LABEL && v == "node1:9000"));
}
}
}
crates/obs/src/metrics/collectors/system_drive.rs
+32 -3
View File
@@ -24,7 +24,7 @@
use crate::metrics::report::PrometheusMetric;
use crate::metrics::schema::system_drive::*;
use crate::metrics::schema::system_process::PROCESS_DISK_IO_MD;
use crate::metrics::schema::system_process::{DIRECTION_LABEL, PROCESS_DISK_IO_MD};
use std::borrow::Cow;
/// Detailed drive statistics for a single drive.
@@ -223,8 +223,8 @@ pub fn collect_process_disk_metrics(
let mut read_metric = PrometheusMetric::from_descriptor(&PROCESS_DISK_IO_MD, stats.read_bytes as f64);
let mut write_metric = PrometheusMetric::from_descriptor(&PROCESS_DISK_IO_MD, stats.written_bytes as f64);
read_metric.labels.push(("direction", Cow::Borrowed("read")));
write_metric.labels.push(("direction", Cow::Borrowed("write")));
read_metric.labels.push((DIRECTION_LABEL, Cow::Borrowed("read")));
write_metric.labels.push((DIRECTION_LABEL, Cow::Borrowed("write")));
if let Some(l) = labels {
read_metric.labels.extend(l.iter().map(|(k, v)| (*k, v.clone())));
@@ -238,6 +238,7 @@ pub fn collect_process_disk_metrics(
mod tests {
use super::*;
use crate::metrics::report::report_metrics;
use crate::metrics::schema::system_process::{PROCESS_EXECUTABLE_NAME_LABEL, PROCESS_PID_LABEL};
use std::collections::BTreeSet;
fn assert_metric_label_keys(
@@ -371,4 +372,32 @@ mod tests {
assert!(offline.is_some());
assert_eq!(offline.map(|m| m.value), Some(2.0));
}
#[test]
fn test_collect_process_disk_metrics_with_node_and_process_labels() {
let stats = ProcessDiskStats {
read_bytes: 1024,
written_bytes: 2048,
};
let labels = vec![
(SERVER_LABEL, Cow::Borrowed("node1:9000")),
(PROCESS_PID_LABEL, Cow::Borrowed("12345")),
(PROCESS_EXECUTABLE_NAME_LABEL, Cow::Borrowed("rustfs")),
];
let metrics = collect_process_disk_metrics(&stats, Some(&labels));
assert_eq!(metrics.len(), 2);
assert_metric_label_keys(
&metrics,
&PROCESS_DISK_IO_MD,
1024.0,
&[
DIRECTION_LABEL,
SERVER_LABEL,
PROCESS_PID_LABEL,
PROCESS_EXECUTABLE_NAME_LABEL,
],
);
}
}
crates/obs/src/metrics/collectors/system_memory.rs
+37 -11
View File
@@ -25,11 +25,14 @@
use crate::metrics::report::PrometheusMetric;
use crate::metrics::schema::system_memory::*;
use crate::metrics::schema::system_process::{PROCESS_RESIDENT_MEMORY_BYTES_MD, PROCESS_VIRTUAL_MEMORY_BYTES_MD};
use crate::node_identity::SERVER_LABEL;
use std::borrow::Cow;
/// System memory statistics.
#[derive(Debug, Clone, Default)]
pub struct MemoryStats {
/// Stable local node identity for labeling node-local memory metrics
pub server: String,
/// Total memory in bytes
pub total: u64,
/// Used memory in bytes
@@ -64,15 +67,24 @@ pub struct ProcessMemoryStats {
/// Uses the metric descriptors from `metrics_type::system_memory` module.
/// Returns a vector of Prometheus metrics for memory statistics.
pub fn collect_memory_metrics(stats: &MemoryStats) -> Vec<PrometheusMetric> {
let server_label = stats.server.as_str();
vec![
PrometheusMetric::from_descriptor(&MEM_TOTAL_MD, stats.total as f64),
PrometheusMetric::from_descriptor(&MEM_USED_MD, stats.used as f64),
PrometheusMetric::from_descriptor(&MEM_USED_PERC_MD, stats.used_perc),
PrometheusMetric::from_descriptor(&MEM_FREE_MD, stats.free as f64),
PrometheusMetric::from_descriptor(&MEM_BUFFERS_MD, stats.buffers as f64),
PrometheusMetric::from_descriptor(&MEM_CACHE_MD, stats.cache as f64),
PrometheusMetric::from_descriptor(&MEM_SHARED_MD, stats.shared as f64),
PrometheusMetric::from_descriptor(&MEM_AVAILABLE_MD, stats.available as f64),
PrometheusMetric::from_descriptor(&MEM_TOTAL_MD, stats.total as f64)
.with_label_owned(SERVER_LABEL, server_label.to_string()),
PrometheusMetric::from_descriptor(&MEM_USED_MD, stats.used as f64)
.with_label_owned(SERVER_LABEL, server_label.to_string()),
PrometheusMetric::from_descriptor(&MEM_USED_PERC_MD, stats.used_perc)
.with_label_owned(SERVER_LABEL, server_label.to_string()),
PrometheusMetric::from_descriptor(&MEM_FREE_MD, stats.free as f64)
.with_label_owned(SERVER_LABEL, server_label.to_string()),
PrometheusMetric::from_descriptor(&MEM_BUFFERS_MD, stats.buffers as f64)
.with_label_owned(SERVER_LABEL, server_label.to_string()),
PrometheusMetric::from_descriptor(&MEM_CACHE_MD, stats.cache as f64)
.with_label_owned(SERVER_LABEL, server_label.to_string()),
PrometheusMetric::from_descriptor(&MEM_SHARED_MD, stats.shared as f64)
.with_label_owned(SERVER_LABEL, server_label.to_string()),
PrometheusMetric::from_descriptor(&MEM_AVAILABLE_MD, stats.available as f64)
.with_label_owned(SERVER_LABEL, server_label.to_string()),
]
}
@@ -104,10 +116,12 @@ pub fn collect_process_memory_metrics(
mod tests {
use super::*;
use crate::metrics::report::report_metrics;
use crate::metrics::schema::system_process::{PROCESS_EXECUTABLE_NAME_LABEL, PROCESS_PID_LABEL};
#[test]
fn test_collect_memory_metrics() {
let stats = MemoryStats {
server: "node1:9000".to_string(),
total: 16 * 1024 * 1024 * 1024, // 16 GB
used: 8 * 1024 * 1024 * 1024, // 8 GB
used_perc: 50.0,
@@ -123,6 +137,11 @@ mod tests {
assert_eq!(metrics.len(), 8);
assert!(metrics.iter().all(|m| m.name.starts_with("rustfs_system_memory_")));
assert!(
metrics
.iter()
.all(|m| m.labels.iter().any(|(k, v)| *k == SERVER_LABEL && v == "node1:9000"))
);
}
#[test]
@@ -133,7 +152,9 @@ mod tests {
assert_eq!(metrics.len(), 8);
for metric in &metrics {
assert_eq!(metric.value, 0.0);
assert!(metric.labels.is_empty());
assert_eq!(metric.labels.len(), 1);
assert_eq!(metric.labels[0].0, SERVER_LABEL);
assert!(metric.labels[0].1.is_empty());
}
}
@@ -157,13 +178,18 @@ mod tests {
virtual_mem: 1024 * 1024 * 1024,
};
let labels = vec![("process_pid", Cow::Borrowed("12345"))];
let labels = vec![
(SERVER_LABEL, Cow::Borrowed("node1:9000")),
(PROCESS_PID_LABEL, Cow::Borrowed("12345")),
(PROCESS_EXECUTABLE_NAME_LABEL, Cow::Borrowed("rustfs")),
];
let metrics = collect_process_memory_metrics(&stats, Some(&labels));
assert_eq!(metrics.len(), 2);
for metric in &metrics {
assert_eq!(metric.labels.len(), 1);
assert_eq!(metric.labels.len(), 3);
assert!(metric.labels.iter().any(|(k, v)| *k == SERVER_LABEL && v == "node1:9000"));
}
}
}
crates/obs/src/metrics/collectors/system_network.rs
+23 -6
View File
@@ -23,10 +23,13 @@
use crate::metrics::report::PrometheusMetric;
use crate::metrics::schema::system_network::*;
use crate::node_identity::SERVER_LABEL;
/// Network statistics for internode communication.
#[derive(Debug, Clone, Default)]
pub struct NetworkStats {
/// Stable local node identity for labeling node-local internode metrics
pub server: String,
/// Total number of failed internode calls
pub internode_errors_total: u64,
/// Total number of TCP dial timeouts and errors
@@ -44,12 +47,18 @@ pub struct NetworkStats {
/// Uses the metric descriptors from `metrics_type::system_network` module.
/// Returns a vector of Prometheus metrics for network statistics.
pub fn collect_network_metrics(stats: &NetworkStats) -> Vec<PrometheusMetric> {
let server_label = stats.server.as_str();
vec![
PrometheusMetric::from_descriptor(&INTERNODE_ERRORS_TOTAL_MD, stats.internode_errors_total as f64),
PrometheusMetric::from_descriptor(&INTERNODE_DIAL_ERRORS_TOTAL_MD, stats.internode_dial_errors_total as f64),
PrometheusMetric::from_descriptor(&INTERNODE_DIAL_AVG_TIME_NANOS_MD, stats.internode_dial_avg_time_nanos as f64),
PrometheusMetric::from_descriptor(&INTERNODE_SENT_BYTES_TOTAL_MD, stats.internode_sent_bytes_total as f64),
PrometheusMetric::from_descriptor(&INTERNODE_RECV_BYTES_TOTAL_MD, stats.internode_recv_bytes_total as f64),
PrometheusMetric::from_descriptor(&INTERNODE_ERRORS_TOTAL_MD, stats.internode_errors_total as f64)
.with_label_owned(SERVER_LABEL, server_label.to_string()),
PrometheusMetric::from_descriptor(&INTERNODE_DIAL_ERRORS_TOTAL_MD, stats.internode_dial_errors_total as f64)
.with_label_owned(SERVER_LABEL, server_label.to_string()),
PrometheusMetric::from_descriptor(&INTERNODE_DIAL_AVG_TIME_NANOS_MD, stats.internode_dial_avg_time_nanos as f64)
.with_label_owned(SERVER_LABEL, server_label.to_string()),
PrometheusMetric::from_descriptor(&INTERNODE_SENT_BYTES_TOTAL_MD, stats.internode_sent_bytes_total as f64)
.with_label_owned(SERVER_LABEL, server_label.to_string()),
PrometheusMetric::from_descriptor(&INTERNODE_RECV_BYTES_TOTAL_MD, stats.internode_recv_bytes_total as f64)
.with_label_owned(SERVER_LABEL, server_label.to_string()),
]
}
@@ -61,6 +70,7 @@ mod tests {
#[test]
fn test_collect_network_metrics() {
let stats = NetworkStats {
server: "node1:9000".to_string(),
internode_errors_total: 10,
internode_dial_errors_total: 5,
internode_dial_avg_time_nanos: 1_500_000, // 1.5ms
@@ -73,6 +83,11 @@ mod tests {
assert_eq!(metrics.len(), 5);
assert!(metrics.iter().all(|m| m.name.contains("internode")));
assert!(
metrics
.iter()
.all(|m| m.labels.iter().any(|(k, v)| *k == SERVER_LABEL && v == "node1:9000"))
);
}
#[test]
@@ -83,7 +98,9 @@ mod tests {
assert_eq!(metrics.len(), 5);
for metric in &metrics {
assert_eq!(metric.value, 0.0);
assert!(metric.labels.is_empty());
assert_eq!(metric.labels.len(), 1);
assert_eq!(metric.labels[0].0, SERVER_LABEL);
assert!(metric.labels[0].1.is_empty());
}
}
}
crates/obs/src/metrics/collectors/system_network_host.rs
+14 -2
View File
@@ -18,6 +18,7 @@ use crate::metrics::report::PrometheusMetric;
use crate::metrics::schema::system_network_host::{
DIRECTION_LABEL, HOST_NETWORK_IO_MD, HOST_NETWORK_IO_PER_INTERFACE_MD, INTERFACE_LABEL,
};
use crate::node_identity::SERVER_LABEL;
use std::borrow::Cow;
/// Network I/O statistics.
@@ -25,6 +26,8 @@ use std::borrow::Cow;
/// Contains host-wide network I/O totals and per-interface counters.
#[derive(Debug, Clone, Default)]
pub struct HostNetworkStats {
/// Stable local node identity for labeling host-wide network metrics.
pub server: String,
/// Total bytes received across observed host interfaces.
pub total_received: u64,
/// Total bytes transmitted across observed host interfaces.
@@ -47,7 +50,11 @@ pub fn collect_host_network_metrics(
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((SERVER_LABEL, Cow::Owned(stats.server.clone())));
received_metric.labels.push((DIRECTION_LABEL, Cow::Borrowed("received")));
transmitted_metric
.labels
.push((SERVER_LABEL, Cow::Owned(stats.server.clone())));
transmitted_metric
.labels
.push((DIRECTION_LABEL, Cow::Borrowed("transmitted")));
@@ -64,9 +71,13 @@ pub fn collect_host_network_metrics(
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((SERVER_LABEL, Cow::Owned(stats.server.clone())));
iface_received.labels.push((INTERFACE_LABEL, Cow::Owned(interface.clone())));
iface_received.labels.push((DIRECTION_LABEL, Cow::Borrowed("received")));
iface_transmitted
.labels
.push((SERVER_LABEL, Cow::Owned(stats.server.clone())));
iface_transmitted
.labels
.push((INTERFACE_LABEL, Cow::Owned(interface.clone())));
@@ -92,6 +103,7 @@ mod tests {
#[test]
fn host_network_metrics_use_dedicated_network_host_prefix() {
let stats = HostNetworkStats {
server: "node1:9000".to_string(),
total_received: 1024,
total_transmitted: 2048,
per_interface: vec![("eth0".to_string(), 512, 256)],
@@ -107,9 +119,9 @@ mod tests {
);
let total_keys: BTreeSet<&str> = metrics[0].labels.iter().map(|(key, _)| *key).collect();
assert_eq!(total_keys, BTreeSet::from([DIRECTION_LABEL]));
assert_eq!(total_keys, BTreeSet::from([SERVER_LABEL, DIRECTION_LABEL]));
let per_interface_keys: BTreeSet<&str> = metrics[2].labels.iter().map(|(key, _)| *key).collect();
assert_eq!(per_interface_keys, BTreeSet::from([DIRECTION_LABEL, INTERFACE_LABEL]));
assert_eq!(per_interface_keys, BTreeSet::from([SERVER_LABEL, DIRECTION_LABEL, INTERFACE_LABEL]));
}
}
crates/obs/src/metrics/collectors/system_process.rs
+20 -3
View File
@@ -24,6 +24,7 @@
use crate::metrics::report::PrometheusMetric;
use crate::metrics::schema::system_process::*;
use crate::node_identity::SERVER_LABEL;
use std::borrow::Cow;
use sysinfo::{Pid, ProcessStatus, System};
@@ -146,6 +147,8 @@ impl From<ProcessStatus> for ProcessStatusType {
/// Process statistics for the RustFS server process.
#[derive(Debug, Clone, Default)]
pub struct ProcessStats {
/// Stable local node identity for labeling node-local process metrics
pub server: String,
/// Total read locks held
pub locks_read_total: u64,
/// Total write locks held
@@ -190,6 +193,7 @@ pub struct ProcessStats {
///
/// Returns a vector of Prometheus metrics for process statistics.
pub fn collect_process_metrics(stats: &ProcessStats) -> Vec<PrometheusMetric> {
let server_label = stats.server.as_str();
let mut metrics = vec![
PrometheusMetric::from_descriptor(&PROCESS_LOCKS_READ_TOTAL_MD, stats.locks_read_total as f64),
PrometheusMetric::from_descriptor(&PROCESS_LOCKS_WRITE_TOTAL_MD, stats.locks_write_total as f64),
@@ -209,12 +213,18 @@ pub fn collect_process_metrics(stats: &ProcessStats) -> Vec<PrometheusMetric> {
PrometheusMetric::from_descriptor(&PROCESS_VIRTUAL_MEMORY_BYTES_MD, stats.virtual_memory_bytes as f64),
PrometheusMetric::from_descriptor(&PROCESS_VIRTUAL_MEMORY_MAX_BYTES_MD, stats.virtual_memory_max_bytes as f64),
];
for metric in &mut metrics {
metric.labels.push((SERVER_LABEL, Cow::Owned(server_label.to_string())));
}
// Add process status metric
let mut status_metric = PrometheusMetric::from_descriptor(&PROCESS_STATUS_MD, stats.status_value as f64);
status_metric
.labels
.push(("status", Cow::Owned(format!("{:?}", stats.status))));
.push((SERVER_LABEL, Cow::Owned(server_label.to_string())));
status_metric
.labels
.push((STATUS_LABEL, Cow::Owned(format!("{:?}", stats.status))));
metrics.push(status_metric);
metrics
@@ -235,6 +245,7 @@ mod tests {
#[test]
fn test_collect_process_metrics() {
let stats = ProcessStats {
server: "node1:9000".to_string(),
locks_read_total: 100,
locks_write_total: 50,
cpu_total_seconds: 1234.56,
@@ -261,6 +272,11 @@ mod tests {
// 17 original metrics + 1 status metric = 18
assert_eq!(metrics.len(), 18);
assert!(
metrics
.iter()
.all(|m| m.labels.iter().any(|(k, v)| *k == SERVER_LABEL && v == "node1:9000"))
);
// Verify uptime
let uptime_name = PROCESS_UPTIME_SECONDS_MD.get_full_metric_name();
@@ -288,6 +304,7 @@ mod tests {
// 17 original metrics + 1 status metric = 18
assert_eq!(metrics.len(), 18);
assert!(metrics.iter().all(|m| m.labels.iter().any(|(k, _)| *k == SERVER_LABEL)));
}
#[test]
@@ -296,7 +313,7 @@ mod tests {
let result = collect_process_attributes();
assert!(result.is_ok());
let attrs = result.unwrap();
let attrs = result.expect("current process attributes should be collectable");
assert!(attrs.pid > 0);
assert!(!attrs.executable_name.is_empty());
}
@@ -319,7 +336,7 @@ mod tests {
let labels = attrs.to_labels();
assert_eq!(labels.len(), 4);
assert_eq!(labels[0].0, "process_pid");
assert_eq!(labels[0].0, PROCESS_PID_LABEL);
assert_eq!(labels[0].1, "12345");
}
}
crates/obs/src/metrics/scheduler.rs
+20 -8
View File
@@ -92,6 +92,7 @@ use crate::metrics::schema::notification_target::{
NOTIFICATION_TARGET_FAILED_MESSAGES_MD, NOTIFICATION_TARGET_QUEUE_LENGTH_MD, NOTIFICATION_TARGET_TOTAL_MESSAGES_MD,
TARGET_ID as NOTIFICATION_TARGET_ID_LABEL, TARGET_TYPE as NOTIFICATION_TARGET_TYPE_LABEL,
};
use crate::metrics::schema::system_process::{PROCESS_EXECUTABLE_NAME_LABEL, PROCESS_PID_LABEL};
use crate::metrics::stats_collector::{
ProcessMetricBundle, collect_bucket_replication_bandwidth_stats, collect_bucket_replication_detail_stats,
collect_bucket_stats, collect_cluster_and_health_stats, collect_cluster_config_stats, collect_cluster_usage_metric_stats,
@@ -100,6 +101,7 @@ use crate::metrics::stats_collector::{
collect_process_metric_bundle_with, collect_replication_stats, collect_scanner_metric_stats,
collect_system_cpu_and_memory_stats_with,
};
use crate::node_identity::{SERVER_LABEL, current_local_node_identity};
use crate::telemetry::retire_metric_series;
use futures_util::FutureExt;
use rustfs_audit::audit_target_metrics;
@@ -1509,7 +1511,7 @@ pub fn init_metrics_runtime(token: CancellationToken) {
let token_clone = token.clone();
tokio::spawn(async move {
let labels = current_process_metric_labels();
let process_attribute_labels = current_process_attribute_labels();
let mut host_system = System::new_all();
let mut host_networks = Networks::new();
let mut process_sampler = ProcessSampler::new();
@@ -1560,6 +1562,7 @@ pub fn init_metrics_runtime(token: CancellationToken) {
}
if now >= next_system_run {
let labels = current_process_metric_labels(&process_attribute_labels);
#[cfg(feature = "gpu")]
let mut metrics =
collect_system_monitoring_metrics(&bundle, &labels, &mut host_system, &mut host_networks);
@@ -1686,24 +1689,33 @@ fn advance_deadline(deadline: &mut Instant, interval: Duration, now: Instant) {
}
}
fn current_process_metric_labels() -> Vec<(&'static str, Cow<'static, str>)> {
fn current_process_attribute_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)),
(PROCESS_PID_LABEL, Cow::Owned(attrs.pid.to_string())),
(PROCESS_EXECUTABLE_NAME_LABEL, Cow::Owned(attrs.executable_name)),
],
Err(err) => fallback_process_metric_labels(err),
Err(err) => fallback_process_attribute_labels(err),
}
}
fn fallback_process_metric_labels(err: ProcessAttributeError) -> Vec<(&'static str, Cow<'static, str>)> {
fn fallback_process_attribute_labels(err: ProcessAttributeError) -> Vec<(&'static str, Cow<'static, str>)> {
warn!(event = EVENT_METRICS_RUNTIME_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_METRICS_RUNTIME, collector = "process_metric_labels", result = "collect_failed", error = %err, "metrics runtime state changed");
vec![
("process_pid", Cow::Owned(std::process::id().to_string())),
("process_executable_name", Cow::Borrowed("unknown")),
(PROCESS_PID_LABEL, Cow::Owned(std::process::id().to_string())),
(PROCESS_EXECUTABLE_NAME_LABEL, Cow::Borrowed("unknown")),
]
}
fn current_process_metric_labels(
process_attribute_labels: &[(&'static str, Cow<'static, str>)],
) -> Vec<(&'static str, Cow<'static, str>)> {
let mut labels = Vec::with_capacity(process_attribute_labels.len() + 1);
labels.push((SERVER_LABEL, Cow::Owned(current_local_node_identity())));
labels.extend(process_attribute_labels.iter().map(|(key, value)| (*key, value.clone())));
labels
}
fn collect_system_monitoring_metrics(
bundle: &ProcessMetricBundle,
labels: &[(&'static str, Cow<'static, str>)],
crates/obs/src/metrics/schema/process_resource.rs
+4 -3
View File
@@ -14,6 +14,7 @@
#![allow(dead_code)]
use crate::node_identity::SERVER_LABEL;
use crate::{MetricDescriptor, MetricName, MetricSubsystem, new_gauge_md};
use std::sync::LazyLock;
@@ -22,7 +23,7 @@ pub static PROCESS_CPU_PERCENT_MD: LazyLock<MetricDescriptor> = LazyLock::new(||
new_gauge_md(
MetricName::Custom("cpu_percent".to_string()),
"CPU usage of the RustFS process as a percentage",
&[],
&[SERVER_LABEL],
MetricSubsystem::new("/process"),
)
});
@@ -32,7 +33,7 @@ pub static PROCESS_MEMORY_BYTES_MD: LazyLock<MetricDescriptor> = LazyLock::new(|
new_gauge_md(
MetricName::Custom("memory_bytes".to_string()),
"Resident memory usage of the RustFS process in bytes",
&[],
&[SERVER_LABEL],
MetricSubsystem::new("/process"),
)
});
@@ -42,7 +43,7 @@ pub static PROCESS_UPTIME_SECONDS_MD: LazyLock<MetricDescriptor> = LazyLock::new
new_gauge_md(
MetricName::Custom("uptime_seconds".to_string()),
"Uptime of the RustFS process in seconds",
&[],
&[SERVER_LABEL],
MetricSubsystem::new("/process"),
)
});
crates/obs/src/metrics/schema/system_cpu.rs
+24 -11
View File
@@ -14,24 +14,37 @@
#![allow(dead_code)]
use crate::node_identity::SERVER_LABEL;
use crate::{MetricDescriptor, MetricName, new_gauge_md, subsystems};
/// CPU system-related metric descriptors
use std::sync::LazyLock;
pub static SYS_CPU_AVG_IDLE_MD: LazyLock<MetricDescriptor> =
LazyLock::new(|| new_gauge_md(MetricName::SysCPUAvgIdle, "Average CPU idle time", &[], subsystems::SYSTEM_CPU));
pub static SYS_CPU_AVG_IDLE_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::SysCPUAvgIdle,
"Average CPU idle time",
&[SERVER_LABEL],
subsystems::SYSTEM_CPU,
)
});
pub static SYS_CPU_AVG_IOWAIT_MD: LazyLock<MetricDescriptor> =
LazyLock::new(|| new_gauge_md(MetricName::SysCPUAvgIOWait, "Average CPU IOWait time", &[], subsystems::SYSTEM_CPU));
pub static SYS_CPU_AVG_IOWAIT_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::SysCPUAvgIOWait,
"Average CPU IOWait time",
&[SERVER_LABEL],
subsystems::SYSTEM_CPU,
)
});
pub static SYS_CPU_LOAD_MD: LazyLock<MetricDescriptor> =
LazyLock::new(|| new_gauge_md(MetricName::SysCPULoad, "CPU load average 1min", &[], subsystems::SYSTEM_CPU));
LazyLock::new(|| new_gauge_md(MetricName::SysCPULoad, "CPU load average 1min", &[SERVER_LABEL], subsystems::SYSTEM_CPU));
pub static SYS_CPU_LOAD_PERC_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::SysCPULoadPerc,
"CPU load average 1min (percentage)",
&[],
&[SERVER_LABEL],
subsystems::SYSTEM_CPU,
)
});
@@ -40,19 +53,19 @@ pub static SYS_CPU_USAGE_PERC_MD: LazyLock<MetricDescriptor> = LazyLock::new(||
new_gauge_md(
MetricName::Custom("usage_perc".to_string()),
"Total CPU usage percentage across all measured CPU time categories",
&[],
&[SERVER_LABEL],
subsystems::SYSTEM_CPU,
)
});
pub static SYS_CPU_NICE_MD: LazyLock<MetricDescriptor> =
LazyLock::new(|| new_gauge_md(MetricName::SysCPUNice, "CPU nice time", &[], subsystems::SYSTEM_CPU));
LazyLock::new(|| new_gauge_md(MetricName::SysCPUNice, "CPU nice time", &[SERVER_LABEL], subsystems::SYSTEM_CPU));
pub static SYS_CPU_STEAL_MD: LazyLock<MetricDescriptor> =
LazyLock::new(|| new_gauge_md(MetricName::SysCPUSteal, "CPU steal time", &[], subsystems::SYSTEM_CPU));
LazyLock::new(|| new_gauge_md(MetricName::SysCPUSteal, "CPU steal time", &[SERVER_LABEL], subsystems::SYSTEM_CPU));
pub static SYS_CPU_SYSTEM_MD: LazyLock<MetricDescriptor> =
LazyLock::new(|| new_gauge_md(MetricName::SysCPUSystem, "CPU system time", &[], subsystems::SYSTEM_CPU));
LazyLock::new(|| new_gauge_md(MetricName::SysCPUSystem, "CPU system time", &[SERVER_LABEL], subsystems::SYSTEM_CPU));
pub static SYS_CPU_USER_MD: LazyLock<MetricDescriptor> =
LazyLock::new(|| new_gauge_md(MetricName::SysCPUUser, "CPU user time", &[], subsystems::SYSTEM_CPU));
LazyLock::new(|| new_gauge_md(MetricName::SysCPUUser, "CPU user time", &[SERVER_LABEL], subsystems::SYSTEM_CPU));
crates/obs/src/metrics/schema/system_memory.rs
+44 -13
View File
@@ -14,43 +14,74 @@
#![allow(dead_code)]
use crate::node_identity::SERVER_LABEL;
use crate::{MetricDescriptor, MetricName, new_gauge_md, subsystems};
use std::sync::LazyLock;
/// Total memory available on the node
pub static MEM_TOTAL_MD: LazyLock<MetricDescriptor> =
LazyLock::new(|| new_gauge_md(MetricName::MemTotal, "Total memory on the node", &[], subsystems::SYSTEM_MEMORY));
pub static MEM_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::MemTotal,
"Total memory on the node",
&[SERVER_LABEL],
subsystems::SYSTEM_MEMORY,
)
});
/// Memory currently in use on the node
pub static MEM_USED_MD: LazyLock<MetricDescriptor> =
LazyLock::new(|| new_gauge_md(MetricName::MemUsed, "Used memory on the node", &[], subsystems::SYSTEM_MEMORY));
LazyLock::new(|| new_gauge_md(MetricName::MemUsed, "Used memory on the node", &[SERVER_LABEL], subsystems::SYSTEM_MEMORY));
/// Percentage of total memory currently in use
pub static MEM_USED_PERC_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::MemUsedPerc,
"Used memory percentage on the node",
&[],
&[SERVER_LABEL],
subsystems::SYSTEM_MEMORY,
)
});
/// Memory not currently in use and available for allocation
pub static MEM_FREE_MD: LazyLock<MetricDescriptor> =
LazyLock::new(|| new_gauge_md(MetricName::MemFree, "Free memory on the node", &[], subsystems::SYSTEM_MEMORY));
LazyLock::new(|| new_gauge_md(MetricName::MemFree, "Free memory on the node", &[SERVER_LABEL], subsystems::SYSTEM_MEMORY));
/// Memory used for file buffers by the kernel
pub static MEM_BUFFERS_MD: LazyLock<MetricDescriptor> =
LazyLock::new(|| new_gauge_md(MetricName::MemBuffers, "Buffers memory on the node", &[], subsystems::SYSTEM_MEMORY));
pub static MEM_BUFFERS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::MemBuffers,
"Buffers memory on the node",
&[SERVER_LABEL],
subsystems::SYSTEM_MEMORY,
)
});
/// Memory used for caching file data by the kernel
pub static MEM_CACHE_MD: LazyLock<MetricDescriptor> =
LazyLock::new(|| new_gauge_md(MetricName::MemCache, "Cache memory on the node", &[], subsystems::SYSTEM_MEMORY));
pub static MEM_CACHE_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::MemCache,
"Cache memory on the node",
&[SERVER_LABEL],
subsystems::SYSTEM_MEMORY,
)
});
/// Memory shared between multiple processes
pub static MEM_SHARED_MD: LazyLock<MetricDescriptor> =
LazyLock::new(|| new_gauge_md(MetricName::MemShared, "Shared memory on the node", &[], subsystems::SYSTEM_MEMORY));
pub static MEM_SHARED_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::MemShared,
"Shared memory on the node",
&[SERVER_LABEL],
subsystems::SYSTEM_MEMORY,
)
});
/// Estimate of memory available for new applications without swapping
pub static MEM_AVAILABLE_MD: LazyLock<MetricDescriptor> =
LazyLock::new(|| new_gauge_md(MetricName::MemAvailable, "Available memory on the node", &[], subsystems::SYSTEM_MEMORY));
pub static MEM_AVAILABLE_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::MemAvailable,
"Available memory on the node",
&[SERVER_LABEL],
subsystems::SYSTEM_MEMORY,
)
});
crates/obs/src/metrics/schema/system_network.rs
+6 -5
View File
@@ -14,6 +14,7 @@
#![allow(dead_code)]
use crate::node_identity::SERVER_LABEL;
use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems};
use std::sync::LazyLock;
@@ -22,7 +23,7 @@ pub static INTERNODE_ERRORS_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new
new_counter_md(
MetricName::InternodeErrorsTotal,
"Total number of failed internode calls",
&[],
&[SERVER_LABEL],
subsystems::SYSTEM_NETWORK_INTERNODE,
)
});
@@ -32,7 +33,7 @@ pub static INTERNODE_DIAL_ERRORS_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock
new_counter_md(
MetricName::InternodeDialErrorsTotal,
"Total number of internode TCP dial timeouts and errors",
&[],
&[SERVER_LABEL],
subsystems::SYSTEM_NETWORK_INTERNODE,
)
});
@@ -42,7 +43,7 @@ pub static INTERNODE_DIAL_AVG_TIME_NANOS_MD: LazyLock<MetricDescriptor> = LazyLo
new_gauge_md(
MetricName::InternodeDialAvgTimeNanos,
"Average dial time of internode TCP calls in nanoseconds",
&[],
&[SERVER_LABEL],
subsystems::SYSTEM_NETWORK_INTERNODE,
)
});
@@ -52,7 +53,7 @@ pub static INTERNODE_SENT_BYTES_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock:
new_counter_md(
MetricName::InternodeSentBytesTotal,
"Total number of bytes sent to other peer nodes",
&[],
&[SERVER_LABEL],
subsystems::SYSTEM_NETWORK_INTERNODE,
)
});
@@ -62,7 +63,7 @@ pub static INTERNODE_RECV_BYTES_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock:
new_counter_md(
MetricName::InternodeRecvBytesTotal,
"Total number of bytes received from other peer nodes",
&[],
&[SERVER_LABEL],
subsystems::SYSTEM_NETWORK_INTERNODE,
)
});
crates/obs/src/metrics/schema/system_network_host.rs
+12 -4
View File
@@ -14,6 +14,7 @@
#![allow(dead_code)]
use crate::node_identity::SERVER_LABEL;
use crate::{MetricDescriptor, MetricName, new_counter_md, subsystems};
use std::sync::LazyLock;
@@ -25,7 +26,7 @@ pub static HOST_NETWORK_IO_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_counter_md(
MetricName::HostNetworkIO,
"Network bytes transferred across system network interfaces",
&[DIRECTION_LABEL],
&[SERVER_LABEL, DIRECTION_LABEL],
subsystems::SYSTEM_NETWORK_HOST,
)
});
@@ -35,7 +36,7 @@ pub static HOST_NETWORK_IO_PER_INTERFACE_MD: LazyLock<MetricDescriptor> = LazyLo
new_counter_md(
MetricName::HostNetworkIOPerInterface,
"Network bytes transferred across system network interfaces (per interface)",
&[INTERFACE_LABEL, DIRECTION_LABEL],
&[SERVER_LABEL, INTERFACE_LABEL, DIRECTION_LABEL],
subsystems::SYSTEM_NETWORK_HOST,
)
});
@@ -48,12 +49,19 @@ mod tests {
#[test]
fn host_network_descriptors_export_counter_labels() {
assert_eq!(HOST_NETWORK_IO_MD.metric_type, MetricType::Counter);
assert_eq!(HOST_NETWORK_IO_MD.variable_labels, vec![DIRECTION_LABEL.to_string()]);
assert_eq!(
HOST_NETWORK_IO_MD.variable_labels,
vec![SERVER_LABEL.to_string(), DIRECTION_LABEL.to_string()]
);
assert_eq!(HOST_NETWORK_IO_PER_INTERFACE_MD.metric_type, MetricType::Counter);
assert_eq!(
HOST_NETWORK_IO_PER_INTERFACE_MD.variable_labels,
vec![INTERFACE_LABEL.to_string(), DIRECTION_LABEL.to_string()]
vec![
SERVER_LABEL.to_string(),
INTERFACE_LABEL.to_string(),
DIRECTION_LABEL.to_string()
]
);
}
}
crates/obs/src/metrics/schema/system_process.rs
+37 -21
View File
@@ -14,15 +14,31 @@
#![allow(dead_code)]
use crate::node_identity::SERVER_LABEL;
use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems};
use std::sync::LazyLock;
pub const PROCESS_PID_LABEL: &str = "process_pid";
pub const PROCESS_EXECUTABLE_NAME_LABEL: &str = "process_executable_name";
pub const DIRECTION_LABEL: &str = "direction";
pub const STATUS_LABEL: &str = "status";
const PROCESS_LABELS: &[&str] = &[SERVER_LABEL];
const PROCESS_WITH_ATTRIBUTES_LABELS: &[&str] = &[SERVER_LABEL, PROCESS_PID_LABEL, PROCESS_EXECUTABLE_NAME_LABEL];
const PROCESS_DISK_IO_LABELS: &[&str] = &[
DIRECTION_LABEL,
SERVER_LABEL,
PROCESS_PID_LABEL,
PROCESS_EXECUTABLE_NAME_LABEL,
];
const PROCESS_STATUS_LABELS: &[&str] = &[SERVER_LABEL, STATUS_LABEL];
/// Number of current READ locks on this peer
pub static PROCESS_LOCKS_READ_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::ProcessLocksReadTotal,
"Number of current READ locks on this peer",
&[],
PROCESS_LABELS,
subsystems::SYSTEM_PROCESS,
)
});
@@ -32,7 +48,7 @@ pub static PROCESS_LOCKS_WRITE_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::
new_gauge_md(
MetricName::ProcessLocksWriteTotal,
"Number of current WRITE locks on this peer",
&[],
PROCESS_LABELS,
subsystems::SYSTEM_PROCESS,
)
});
@@ -42,7 +58,7 @@ pub static PROCESS_CPU_TOTAL_SECONDS_MD: LazyLock<MetricDescriptor> = LazyLock::
new_counter_md(
MetricName::ProcessCPUTotalSeconds,
"Total user and system CPU time spent in seconds",
&[],
PROCESS_LABELS,
subsystems::SYSTEM_PROCESS,
)
});
@@ -52,7 +68,7 @@ pub static PROCESS_GO_ROUTINE_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::n
new_gauge_md(
MetricName::ProcessGoRoutineTotal,
"Total number of go routines running",
&[],
PROCESS_LABELS,
subsystems::SYSTEM_PROCESS,
)
});
@@ -62,7 +78,7 @@ pub static PROCESS_IO_RCHAR_BYTES_MD: LazyLock<MetricDescriptor> = LazyLock::new
new_counter_md(
MetricName::ProcessIORCharBytes,
"Total bytes read by the process from the underlying storage system including cache, /proc/[pid]/io rchar",
&[],
PROCESS_LABELS,
subsystems::SYSTEM_PROCESS,
)
});
@@ -72,7 +88,7 @@ pub static PROCESS_IO_READ_BYTES_MD: LazyLock<MetricDescriptor> = LazyLock::new(
new_counter_md(
MetricName::ProcessIOReadBytes,
"Total bytes read by the process from the underlying storage system, /proc/[pid]/io read_bytes",
&[],
PROCESS_LABELS,
subsystems::SYSTEM_PROCESS,
)
});
@@ -82,7 +98,7 @@ pub static PROCESS_IO_WCHAR_BYTES_MD: LazyLock<MetricDescriptor> = LazyLock::new
new_counter_md(
MetricName::ProcessIOWCharBytes,
"Total bytes written by the process to the underlying storage system including page cache, /proc/[pid]/io wchar",
&[],
PROCESS_LABELS,
subsystems::SYSTEM_PROCESS,
)
});
@@ -92,7 +108,7 @@ pub static PROCESS_IO_WRITE_BYTES_MD: LazyLock<MetricDescriptor> = LazyLock::new
new_counter_md(
MetricName::ProcessIOWriteBytes,
"Total bytes written by the process to the underlying storage system, /proc/[pid]/io write_bytes",
&[],
PROCESS_LABELS,
subsystems::SYSTEM_PROCESS,
)
});
@@ -102,7 +118,7 @@ pub static PROCESS_START_TIME_SECONDS_MD: LazyLock<MetricDescriptor> = LazyLock:
new_gauge_md(
MetricName::ProcessStartTimeSeconds,
"Start time for RustFS process in seconds since Unix epoch",
&[],
PROCESS_LABELS,
subsystems::SYSTEM_PROCESS,
)
});
@@ -112,7 +128,7 @@ pub static PROCESS_UPTIME_SECONDS_MD: LazyLock<MetricDescriptor> = LazyLock::new
new_gauge_md(
MetricName::ProcessUptimeSeconds,
"Uptime for RustFS process in seconds",
&[],
PROCESS_LABELS,
subsystems::SYSTEM_PROCESS,
)
});
@@ -122,7 +138,7 @@ pub static PROCESS_FILE_DESCRIPTOR_LIMIT_TOTAL_MD: LazyLock<MetricDescriptor> =
new_gauge_md(
MetricName::ProcessFileDescriptorLimitTotal,
"Limit on total number of open file descriptors for the RustFS Server process",
&[],
PROCESS_LABELS,
subsystems::SYSTEM_PROCESS,
)
});
@@ -132,7 +148,7 @@ pub static PROCESS_FILE_DESCRIPTOR_OPEN_TOTAL_MD: LazyLock<MetricDescriptor> = L
new_gauge_md(
MetricName::ProcessFileDescriptorOpenTotal,
"Total number of open file descriptors by the RustFS Server process",
&[],
PROCESS_LABELS,
subsystems::SYSTEM_PROCESS,
)
});
@@ -142,7 +158,7 @@ pub static PROCESS_SYSCALL_READ_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock:
new_counter_md(
MetricName::ProcessSyscallReadTotal,
"Total read SysCalls to the kernel. /proc/[pid]/io syscr",
&[],
PROCESS_LABELS,
subsystems::SYSTEM_PROCESS,
)
});
@@ -152,7 +168,7 @@ pub static PROCESS_SYSCALL_WRITE_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock
new_counter_md(
MetricName::ProcessSyscallWriteTotal,
"Total write SysCalls to the kernel. /proc/[pid]/io syscw",
&[],
PROCESS_LABELS,
subsystems::SYSTEM_PROCESS,
)
});
@@ -162,7 +178,7 @@ pub static PROCESS_RESIDENT_MEMORY_BYTES_MD: LazyLock<MetricDescriptor> = LazyLo
new_gauge_md(
MetricName::ProcessResidentMemoryBytes,
"Resident memory size in bytes",
&[],
PROCESS_LABELS,
subsystems::SYSTEM_PROCESS,
)
});
@@ -172,7 +188,7 @@ pub static PROCESS_VIRTUAL_MEMORY_BYTES_MD: LazyLock<MetricDescriptor> = LazyLoc
new_gauge_md(
MetricName::ProcessVirtualMemoryBytes,
"Virtual memory size in bytes",
&[],
PROCESS_LABELS,
subsystems::SYSTEM_PROCESS,
)
});
@@ -182,7 +198,7 @@ pub static PROCESS_VIRTUAL_MEMORY_MAX_BYTES_MD: LazyLock<MetricDescriptor> = Laz
new_gauge_md(
MetricName::ProcessVirtualMemoryMaxBytes,
"Maximum virtual memory size in bytes",
&[],
PROCESS_LABELS,
subsystems::SYSTEM_PROCESS,
)
});
@@ -196,7 +212,7 @@ pub static PROCESS_CPU_USAGE_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::ProcessCPUUsage,
"The percentage of CPU in use by the process",
&[],
PROCESS_WITH_ATTRIBUTES_LABELS,
subsystems::SYSTEM_PROCESS,
)
});
@@ -206,7 +222,7 @@ pub static PROCESS_CPU_UTILIZATION_MD: LazyLock<MetricDescriptor> = LazyLock::ne
new_gauge_md(
MetricName::ProcessCPUUtilization,
"The amount of CPU in use by the process (considering multiple cores)",
&[],
PROCESS_WITH_ATTRIBUTES_LABELS,
subsystems::SYSTEM_PROCESS,
)
});
@@ -216,7 +232,7 @@ pub static PROCESS_DISK_IO_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::ProcessDiskIO,
"Disk bytes transferred by the process",
&[],
PROCESS_DISK_IO_LABELS,
subsystems::SYSTEM_PROCESS,
)
});
@@ -226,7 +242,7 @@ pub static PROCESS_STATUS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::ProcessStatus,
"Process status (0: Running, 1: Sleeping, 2: Zombie, 3: Other)",
&[],
PROCESS_STATUS_LABELS,
subsystems::SYSTEM_PROCESS,
)
});
crates/obs/src/metrics/stats_collector.rs
+33 -3
View File
@@ -33,6 +33,7 @@ use crate::metrics::{
obs_load_compression_total_from_memory, obs_load_data_usage_from_backend, obs_replication_site_stats_snapshot,
obs_resolve_object_store_handle,
};
use crate::node_identity::current_local_node_identity;
use chrono::Utc;
use rustfs_common::heal_channel::HealScanMode;
use rustfs_common::metrics::{ScannerMetricsReport, global_metrics};
@@ -539,12 +540,13 @@ pub async fn collect_disk_stats() -> Vec<DiskStats> {
disk_stats
}
fn build_system_cpu_stats(system: &System) -> CpuStats {
fn build_system_cpu_stats(system: &System, server: &str) -> CpuStats {
let cpu_usage = system.global_cpu_usage() as f64;
let cpu_count = system.cpus().len().max(1) as f64;
let load_avg = System::load_average().one;
CpuStats {
server: server.to_string(),
avg_idle: (100.0 - cpu_usage).max(0.0),
load_avg,
load_avg_perc: (load_avg / cpu_count) * 100.0,
@@ -552,11 +554,12 @@ fn build_system_cpu_stats(system: &System) -> CpuStats {
}
}
fn build_system_memory_stats(system: &System) -> MemoryStats {
fn build_system_memory_stats(system: &System, server: &str) -> MemoryStats {
let total = system.total_memory();
let used = system.used_memory();
MemoryStats {
server: server.to_string(),
total,
used,
used_perc: if total > 0 {
@@ -582,7 +585,8 @@ pub fn collect_system_cpu_and_memory_stats() -> (CpuStats, MemoryStats) {
pub fn collect_system_cpu_and_memory_stats_with(system: &mut System) -> (CpuStats, MemoryStats) {
system.refresh_cpu_all();
system.refresh_memory();
(build_system_cpu_stats(system), build_system_memory_stats(system))
let server = current_local_node_identity();
(build_system_cpu_stats(system, &server), build_system_memory_stats(system, &server))
}
/// Collect system CPU statistics from the current host.
@@ -692,6 +696,7 @@ fn process_metric_bundle_from_snapshots(
resource_snapshot: ProcessResourceSnapshot,
process_snapshot: ProcessSystemSnapshot,
) -> ProcessMetricBundle {
let server = current_local_node_identity();
let status = match process_snapshot.status {
ProcessStatusSnapshot::Running => ProcessStatusType::Running,
ProcessStatusSnapshot::Sleeping => ProcessStatusType::Sleeping,
@@ -700,11 +705,13 @@ fn process_metric_bundle_from_snapshots(
};
let resource_stats = ResourceStats {
server: server.clone(),
cpu_percent: resource_snapshot.cpu_percent,
memory_bytes: resource_snapshot.memory_bytes,
uptime_seconds: resource_snapshot.uptime_seconds,
};
let process_stats = ProcessStats {
server,
locks_read_total: process_snapshot.locks_read_total,
locks_write_total: process_snapshot.locks_write_total,
cpu_total_seconds: process_snapshot.cpu_total_seconds,
@@ -770,6 +777,7 @@ pub fn collect_host_network_stats_with(networks: &Networks) -> HostNetworkStats
}
HostNetworkStats {
server: current_local_node_identity(),
total_received,
total_transmitted,
per_interface,
@@ -794,6 +802,7 @@ pub fn collect_internode_network_stats() -> Option<NetworkStats> {
let snapshot = global_internode_metrics().snapshot();
Some(NetworkStats {
server: current_local_node_identity(),
internode_errors_total: snapshot.errors_total,
internode_dial_errors_total: snapshot.dial_errors_total,
internode_dial_avg_time_nanos: snapshot.dial_avg_time_nanos,
@@ -1307,6 +1316,27 @@ mod tests {
assert!(cluster_config_stats_from_backend_parities(Some(1), Some(overflow)).is_none());
}
#[tokio::test]
async fn node_local_resource_stats_use_stable_local_node_identity() {
let _guard = crate::node_identity::local_node_identity_test_guard().await;
let previous = rustfs_common::get_global_local_node_name().await;
rustfs_common::set_global_local_node_name("node1:9000").await;
let mut system = System::new_all();
let (cpu, memory) = collect_system_cpu_and_memory_stats_with(&mut system);
let host_network = collect_host_network_stats_with(&Networks::new());
let process_bundle =
process_metric_bundle_from_snapshots(ProcessResourceSnapshot::default(), ProcessSystemSnapshot::default());
assert_eq!(cpu.server, "node1:9000");
assert_eq!(memory.server, "node1:9000");
assert_eq!(host_network.server, "node1:9000");
assert_eq!(process_bundle.resource.server, "node1:9000");
assert_eq!(process_bundle.process.server, "node1:9000");
rustfs_common::set_global_local_node_name(&previous).await;
}
#[test]
fn erasure_set_stats_skip_unknown_backend_layout() {
let storage_info = storage_info_with_one_online_disk();
crates/obs/src/node_identity.rs
+34
View File
@@ -0,0 +1,34 @@
// 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.
pub(crate) const RUSTFS_NODE_ATTRIBUTE: &str = "rustfs.node";
pub(crate) const SERVER_LABEL: &str = "server";
#[cfg(test)]
static LOCAL_NODE_IDENTITY_TEST_LOCK: std::sync::LazyLock<tokio::sync::Mutex<()>> =
std::sync::LazyLock::new(|| tokio::sync::Mutex::new(()));
pub(crate) fn local_node_identity(local_ip: &str) -> String {
rustfs_common::try_get_global_local_node_name().unwrap_or_else(|| local_ip.to_string())
}
pub(crate) fn current_local_node_identity() -> String {
let local_ip = rustfs_utils::get_local_ip_with_default();
local_node_identity(&local_ip)
}
#[cfg(test)]
pub(crate) async fn local_node_identity_test_guard() -> tokio::sync::MutexGuard<'static, ()> {
LOCAL_NODE_IDENTITY_TEST_LOCK.lock().await
}
crates/obs/src/telemetry/resource.rs
+48 -5
View File
@@ -16,15 +16,20 @@
//!
//! A `Resource` describes the entity producing telemetry data. The resource
//! built here includes the service name, service version, deployment
//! environment, and the local machine IP address so that data can be
//! correlated across services in a distributed system.
//! environment, the stable RustFS node identity, and the local machine IP
//! address so that data can be correlated across services in a distributed
//! system.
use crate::config::OtelConfig;
use crate::node_identity::{RUSTFS_NODE_ATTRIBUTE, local_node_identity};
use opentelemetry::KeyValue;
use opentelemetry_sdk::Resource;
use opentelemetry_semantic_conventions::{
SCHEMA_URL,
attribute::{DEPLOYMENT_ENVIRONMENT_NAME, NETWORK_LOCAL_ADDRESS, SERVICE_VERSION as OTEL_SERVICE_VERSION},
attribute::{
DEPLOYMENT_ENVIRONMENT_NAME, NETWORK_LOCAL_ADDRESS, SERVICE_INSTANCE_ID as OTEL_SERVICE_INSTANCE_ID,
SERVICE_VERSION as OTEL_SERVICE_VERSION,
},
};
use rustfs_config::{APP_NAME, ENVIRONMENT, SERVICE_VERSION};
use rustfs_utils::get_local_ip_with_default;
@@ -38,12 +43,18 @@ use std::borrow::Cow;
/// [`SERVICE_VERSION`].
/// - `deployment.environment` — from `config.environment`, defaulting to
/// [`ENVIRONMENT`].
/// - `rustfs.node` / `service.instance.id` — the stable RustFS local node name
/// when available, falling back to the local IP during early startup.
/// - `network.local.address` — the primary local IP of the current host,
/// useful for identifying individual nodes in a cluster.
/// useful as an operational fallback when the stable node name is not yet
/// initialized.
///
/// All attributes are attached to the resource using the semantic conventions
/// schema URL to ensure compatibility with standard OTLP backends.
pub(super) fn build_resource(config: &OtelConfig) -> Resource {
let local_ip = get_local_ip_with_default();
let node_identity = local_node_identity(&local_ip);
Resource::builder()
.with_service_name(Cow::Borrowed(config.service_name.as_deref().unwrap_or(APP_NAME)).to_string())
.with_schema_url(
@@ -56,9 +67,41 @@ pub(super) fn build_resource(config: &OtelConfig) -> Resource {
DEPLOYMENT_ENVIRONMENT_NAME,
Cow::Borrowed(config.environment.as_deref().unwrap_or(ENVIRONMENT)).to_string(),
),
KeyValue::new(NETWORK_LOCAL_ADDRESS, get_local_ip_with_default()),
KeyValue::new(RUSTFS_NODE_ATTRIBUTE, node_identity.clone()),
KeyValue::new(OTEL_SERVICE_INSTANCE_ID, node_identity),
KeyValue::new(NETWORK_LOCAL_ADDRESS, local_ip),
],
SCHEMA_URL,
)
.build()
}
#[cfg(test)]
mod tests {
use super::*;
use opentelemetry::Key;
#[tokio::test]
async fn build_resource_uses_stable_local_node_identity() {
let _guard = crate::node_identity::local_node_identity_test_guard().await;
let previous = rustfs_common::get_global_local_node_name().await;
rustfs_common::set_global_local_node_name("node1:9000").await;
let resource = build_resource(&OtelConfig::default());
assert_eq!(
resource
.get(&Key::from_static_str(RUSTFS_NODE_ATTRIBUTE))
.map(|value| value.to_string()),
Some("node1:9000".to_string())
);
assert_eq!(
resource
.get(&Key::from_static_str(OTEL_SERVICE_INSTANCE_ID))
.map(|value| value.to_string()),
Some("node1:9000".to_string())
);
rustfs_common::set_global_local_node_name(&previous).await;
}
}