From 30dc04c94bfd17dc081994784daff507191311a8 Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 30 Jul 2026 12:57:22 +0800 Subject: [PATCH] fix(obs): label node-local metrics by server (#5465) 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 --- Cargo.lock | 2 + crates/io-metrics/Cargo.toml | 2 + crates/io-metrics/src/internode_metrics.rs | 253 ++++++++++++++---- crates/obs/src/lib.rs | 1 + crates/obs/src/metrics/collectors/resource.rs | 24 +- .../obs/src/metrics/collectors/system_cpu.rs | 36 ++- .../src/metrics/collectors/system_drive.rs | 35 ++- .../src/metrics/collectors/system_memory.rs | 48 +++- .../src/metrics/collectors/system_network.rs | 29 +- .../metrics/collectors/system_network_host.rs | 16 +- .../src/metrics/collectors/system_process.rs | 23 +- crates/obs/src/metrics/scheduler.rs | 28 +- .../src/metrics/schema/process_resource.rs | 7 +- crates/obs/src/metrics/schema/system_cpu.rs | 35 ++- .../obs/src/metrics/schema/system_memory.rs | 57 +++- .../obs/src/metrics/schema/system_network.rs | 11 +- .../src/metrics/schema/system_network_host.rs | 16 +- .../obs/src/metrics/schema/system_process.rs | 58 ++-- crates/obs/src/metrics/stats_collector.rs | 36 ++- crates/obs/src/node_identity.rs | 34 +++ crates/obs/src/telemetry/resource.rs | 53 +++- 21 files changed, 640 insertions(+), 164 deletions(-) create mode 100644 crates/obs/src/node_identity.rs diff --git a/Cargo.lock b/Cargo.lock index 3cde4a2ad..a2c01b043 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9353,7 +9353,9 @@ dependencies = [ "metrics", "metrics-util", "num_cpus", + "rustfs-common", "rustfs-s3-ops", + "rustfs-utils", "sysinfo", "thiserror 2.0.19", "tokio", diff --git a/crates/io-metrics/Cargo.toml b/crates/io-metrics/Cargo.toml index eb8d6fa22..d77725ae5 100644 --- a/crates/io-metrics/Cargo.toml +++ b/crates/io-metrics/Cargo.toml @@ -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"] } diff --git a/crates/io-metrics/src/internode_metrics.rs b/crates/io-metrics/src/internode_metrics.rs index 0aa69e6fb..790536b4a 100644 --- a/crates/io-metrics/src/internode_metrics.rs +++ b/crates/io-metrics/src/internode_metrics.rs @@ -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 = OnceLock::new(); + static FALLBACK_SERVER_LABEL: LazyLock = 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 { #[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, Option)> = 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. diff --git a/crates/obs/src/lib.rs b/crates/obs/src/lib.rs index b2919d9b2..55b079e6b 100644 --- a/crates/obs/src/lib.rs +++ b/crates/obs/src/lib.rs @@ -64,6 +64,7 @@ mod error; mod global; mod logging; pub mod metrics; +mod node_identity; mod telemetry; pub use cleaner::*; diff --git a/crates/obs/src/metrics/collectors/resource.rs b/crates/obs/src/metrics/collectors/resource.rs index 7d7e0ee68..011859ce6 100644 --- a/crates/obs/src/metrics/collectors/resource.rs +++ b/crates/obs/src/metrics/collectors/resource.rs @@ -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 { + 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, diff --git a/crates/obs/src/metrics/collectors/system_cpu.rs b/crates/obs/src/metrics/collectors/system_cpu.rs index 302e9fbe0..5e829a8b4 100644 --- a/crates/obs/src/metrics/collectors/system_cpu.rs +++ b/crates/obs/src/metrics/collectors/system_cpu.rs @@ -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 { + 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")); } } } diff --git a/crates/obs/src/metrics/collectors/system_drive.rs b/crates/obs/src/metrics/collectors/system_drive.rs index 521ee3f59..7cda25a07 100644 --- a/crates/obs/src/metrics/collectors/system_drive.rs +++ b/crates/obs/src/metrics/collectors/system_drive.rs @@ -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, + ], + ); + } } diff --git a/crates/obs/src/metrics/collectors/system_memory.rs b/crates/obs/src/metrics/collectors/system_memory.rs index 78a4d625e..c9e319365 100644 --- a/crates/obs/src/metrics/collectors/system_memory.rs +++ b/crates/obs/src/metrics/collectors/system_memory.rs @@ -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 { + 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")); } } } diff --git a/crates/obs/src/metrics/collectors/system_network.rs b/crates/obs/src/metrics/collectors/system_network.rs index 4392e398c..8bb660de0 100644 --- a/crates/obs/src/metrics/collectors/system_network.rs +++ b/crates/obs/src/metrics/collectors/system_network.rs @@ -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 { + 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()); } } } diff --git a/crates/obs/src/metrics/collectors/system_network_host.rs b/crates/obs/src/metrics/collectors/system_network_host.rs index 9dccd9d20..083ad8d69 100644 --- a/crates/obs/src/metrics/collectors/system_network_host.rs +++ b/crates/obs/src/metrics/collectors/system_network_host.rs @@ -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])); } } diff --git a/crates/obs/src/metrics/collectors/system_process.rs b/crates/obs/src/metrics/collectors/system_process.rs index ebdcb2516..708c1e975 100644 --- a/crates/obs/src/metrics/collectors/system_process.rs +++ b/crates/obs/src/metrics/collectors/system_process.rs @@ -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 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 { + 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::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"); } } diff --git a/crates/obs/src/metrics/scheduler.rs b/crates/obs/src/metrics/scheduler.rs index e5a002aa0..189c0c04d 100644 --- a/crates/obs/src/metrics/scheduler.rs +++ b/crates/obs/src/metrics/scheduler.rs @@ -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>)], diff --git a/crates/obs/src/metrics/schema/process_resource.rs b/crates/obs/src/metrics/schema/process_resource.rs index 9c0c03e34..5bac84f5a 100644 --- a/crates/obs/src/metrics/schema/process_resource.rs +++ b/crates/obs/src/metrics/schema/process_resource.rs @@ -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 = 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 = 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 = LazyLock::new new_gauge_md( MetricName::Custom("uptime_seconds".to_string()), "Uptime of the RustFS process in seconds", - &[], + &[SERVER_LABEL], MetricSubsystem::new("/process"), ) }); diff --git a/crates/obs/src/metrics/schema/system_cpu.rs b/crates/obs/src/metrics/schema/system_cpu.rs index 350b8310d..f0a3ea784 100644 --- a/crates/obs/src/metrics/schema/system_cpu.rs +++ b/crates/obs/src/metrics/schema/system_cpu.rs @@ -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 = - LazyLock::new(|| new_gauge_md(MetricName::SysCPUAvgIdle, "Average CPU idle time", &[], subsystems::SYSTEM_CPU)); +pub static SYS_CPU_AVG_IDLE_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::SysCPUAvgIdle, + "Average CPU idle time", + &[SERVER_LABEL], + subsystems::SYSTEM_CPU, + ) +}); -pub static SYS_CPU_AVG_IOWAIT_MD: LazyLock = - LazyLock::new(|| new_gauge_md(MetricName::SysCPUAvgIOWait, "Average CPU IOWait time", &[], subsystems::SYSTEM_CPU)); +pub static SYS_CPU_AVG_IOWAIT_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::SysCPUAvgIOWait, + "Average CPU IOWait time", + &[SERVER_LABEL], + subsystems::SYSTEM_CPU, + ) +}); pub static SYS_CPU_LOAD_MD: LazyLock = - 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 = 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 = 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 = - 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 = - 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 = - 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 = - 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)); diff --git a/crates/obs/src/metrics/schema/system_memory.rs b/crates/obs/src/metrics/schema/system_memory.rs index de45de6fd..0ffa9f193 100644 --- a/crates/obs/src/metrics/schema/system_memory.rs +++ b/crates/obs/src/metrics/schema/system_memory.rs @@ -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 = - LazyLock::new(|| new_gauge_md(MetricName::MemTotal, "Total memory on the node", &[], subsystems::SYSTEM_MEMORY)); +pub static MEM_TOTAL_MD: LazyLock = 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 = - 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 = 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 = - 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 = - LazyLock::new(|| new_gauge_md(MetricName::MemBuffers, "Buffers memory on the node", &[], subsystems::SYSTEM_MEMORY)); +pub static MEM_BUFFERS_MD: LazyLock = 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 = - LazyLock::new(|| new_gauge_md(MetricName::MemCache, "Cache memory on the node", &[], subsystems::SYSTEM_MEMORY)); +pub static MEM_CACHE_MD: LazyLock = 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 = - LazyLock::new(|| new_gauge_md(MetricName::MemShared, "Shared memory on the node", &[], subsystems::SYSTEM_MEMORY)); +pub static MEM_SHARED_MD: LazyLock = 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 = - LazyLock::new(|| new_gauge_md(MetricName::MemAvailable, "Available memory on the node", &[], subsystems::SYSTEM_MEMORY)); +pub static MEM_AVAILABLE_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::MemAvailable, + "Available memory on the node", + &[SERVER_LABEL], + subsystems::SYSTEM_MEMORY, + ) +}); diff --git a/crates/obs/src/metrics/schema/system_network.rs b/crates/obs/src/metrics/schema/system_network.rs index 6ac2bc75d..5571951fd 100644 --- a/crates/obs/src/metrics/schema/system_network.rs +++ b/crates/obs/src/metrics/schema/system_network.rs @@ -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 = 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 = 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 = 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 = 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 = LazyLock: new_counter_md( MetricName::InternodeRecvBytesTotal, "Total number of bytes received from other peer nodes", - &[], + &[SERVER_LABEL], subsystems::SYSTEM_NETWORK_INTERNODE, ) }); diff --git a/crates/obs/src/metrics/schema/system_network_host.rs b/crates/obs/src/metrics/schema/system_network_host.rs index b7d26572c..45cc89ffe 100644 --- a/crates/obs/src/metrics/schema/system_network_host.rs +++ b/crates/obs/src/metrics/schema/system_network_host.rs @@ -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 = 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 = 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() + ] ); } } diff --git a/crates/obs/src/metrics/schema/system_process.rs b/crates/obs/src/metrics/schema/system_process.rs index 34b46157c..ecddbc9c3 100644 --- a/crates/obs/src/metrics/schema/system_process.rs +++ b/crates/obs/src/metrics/schema/system_process.rs @@ -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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = LazyLock::new(|| { new_gauge_md( MetricName::ProcessStatus, "Process status (0: Running, 1: Sleeping, 2: Zombie, 3: Other)", - &[], + PROCESS_STATUS_LABELS, subsystems::SYSTEM_PROCESS, ) }); diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index 743722921..7f14aa599 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -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 { 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 { 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(); diff --git a/crates/obs/src/node_identity.rs b/crates/obs/src/node_identity.rs new file mode 100644 index 000000000..2128db816 --- /dev/null +++ b/crates/obs/src/node_identity.rs @@ -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> = + 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 +} diff --git a/crates/obs/src/telemetry/resource.rs b/crates/obs/src/telemetry/resource.rs index a323047f9..37bda8a5f 100644 --- a/crates/obs/src/telemetry/resource.rs +++ b/crates/obs/src/telemetry/resource.rs @@ -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; + } +}