gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-21 11:56:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(obs): complete metric dimension coverage (#6314)

Browse Source 
Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-08-20 22:43:40 +08:00
committed by GitHub
parent d65fba9142
commit 830e553a3c
15 changed files with 821 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/common/src/metrics.rs
+109 -4
View File
@@ -781,6 +781,19 @@ struct ScannerBucketDriveResultValue {
last_seen: u64,
}
#[derive(Clone, Debug, Eq, Hash, PartialEq)]
struct ScannerActiveBucketDriveKey {
source: String,
bucket: String,
drive: String,
}
#[derive(Clone, Copy, Debug)]
struct ScannerActiveBucketDriveValue {
count: u64,
started_at: Timestamp,
}
// ---------------------------------------------------------------------------
// Metrics
// ---------------------------------------------------------------------------
@@ -813,6 +826,7 @@ pub struct Metrics {
scanner_set_scans_active: AtomicU64,
scanner_disk_bucket_scan_states: Mutex<HashMap<ScannerDiskBucketScanKey, ScannerDiskBucketScanState>>,
scanner_bucket_drive_results: Mutex<ScannerBucketDriveResults>,
scanner_active_bucket_drive_scans: Mutex<HashMap<ScannerActiveBucketDriveKey, ScannerActiveBucketDriveValue>>,
scanner_bucket_drive_result_clock: AtomicU64,
current_scan_cycle_bucket_drive_results_start: Mutex<HashMap<ScannerBucketDriveResultKey, u64>>,
last_scan_cycle_bucket_drive_results: Mutex<Vec<ScannerBucketDriveResultSnapshot>>,
@@ -1045,6 +1059,15 @@ pub struct ScannerBucketDriveResultSnapshot {
pub count: u64,
}
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct ScannerActiveBucketDriveSnapshot {
pub source: String,
pub bucket: String,
pub drive: String,
pub count: u64,
pub age_seconds: u64,
}
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct ScannerReplicationRepairSnapshot {
pub source: String,
@@ -1387,6 +1410,8 @@ pub struct ScannerRuntimeDetailsReport {
pub current_cycle_bucket_drive_results: Vec<ScannerBucketDriveResultSnapshot>,
#[serde(default)]
pub last_cycle_bucket_drive_results: Vec<ScannerBucketDriveResultSnapshot>,
#[serde(default)]
pub active_bucket_drive_scans: Vec<ScannerActiveBucketDriveSnapshot>,
}
impl CurrentCycle {
@@ -1746,7 +1771,7 @@ pub fn emit_scan_cycle_deferred(duration: Duration) {
metrics::counter!(OTEL_SCANNER_CYCLES, "result" => SCAN_CYCLE_RESULT_DEFERRED_LABEL).increment(1);
}
pub fn emit_scan_bucket_drive_complete(success: bool, bucket: &str, disk: &str, duration: Duration) {
pub fn emit_scan_bucket_drive_complete(_source: ScannerWorkSource, success: bool, bucket: &str, disk: &str, duration: Duration) {
let result = if success { "success" } else { "error" };
global_metrics().record_scanner_bucket_drive_result(bucket, disk, result);
metrics::counter!(
@@ -1764,7 +1789,7 @@ pub fn emit_scan_bucket_drive_complete(success: bool, bucket: &str, disk: &str,
.record(duration.as_secs_f64());
}
pub fn emit_scan_bucket_drive_partial(bucket: &str, disk: &str, duration: Duration) {
pub fn emit_scan_bucket_drive_partial(_source: ScannerWorkSource, bucket: &str, disk: &str, duration: Duration) {
global_metrics().record_scanner_bucket_drive_result(bucket, disk, SCAN_CYCLE_RESULT_PARTIAL_LABEL);
metrics::counter!(
OTEL_SCANNER_BUCKETS_SCANNED,
@@ -1817,6 +1842,7 @@ impl Metrics {
scanner_set_scans_active: AtomicU64::new(0),
scanner_disk_bucket_scan_states: Mutex::new(HashMap::new()),
scanner_bucket_drive_results: Mutex::new(ScannerBucketDriveResults::default()),
scanner_active_bucket_drive_scans: Mutex::new(HashMap::new()),
scanner_bucket_drive_result_clock: AtomicU64::new(0),
current_scan_cycle_bucket_drive_results_start: Mutex::new(HashMap::new()),
last_scan_cycle_bucket_drive_results: Mutex::new(Vec::new()),
@@ -2308,8 +2334,45 @@ impl Metrics {
}
}
pub fn record_scan_bucket_drive_start(&self) {
pub fn record_scan_bucket_drive_start(&self, source: ScannerWorkSource, bucket: &str, drive: &str) {
self.operations[Metric::ScanBucketDriveStart as usize].fetch_add(1, Ordering::Relaxed);
if bucket.is_empty() || drive.is_empty() {
return;
}
let key = ScannerActiveBucketDriveKey {
source: source.as_str().to_string(),
bucket: bucket.to_string(),
drive: drive.to_string(),
};
let mut active = self
.scanner_active_bucket_drive_scans
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
active
.entry(key)
.and_modify(|value| value.count = value.count.saturating_add(1))
.or_insert(ScannerActiveBucketDriveValue {
count: 1,
started_at: Timestamp::now(),
});
}
pub fn record_scan_bucket_drive_end(&self, source: ScannerWorkSource, bucket: &str, drive: &str) {
let key = ScannerActiveBucketDriveKey {
source: source.as_str().to_string(),
bucket: bucket.to_string(),
drive: drive.to_string(),
};
let mut active = self
.scanner_active_bucket_drive_scans
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
if let Some(value) = active.get_mut(&key) {
value.count = value.count.saturating_sub(1);
if value.count == 0 {
active.remove(&key);
}
}
}
pub fn record_scan_bucket_drive_failure(&self) {
@@ -2782,6 +2845,26 @@ impl Metrics {
} else {
Vec::new()
};
let now = Timestamp::now();
let mut active_bucket_drive_scans = self
.scanner_active_bucket_drive_scans
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.iter()
.map(|(key, value)| ScannerActiveBucketDriveSnapshot {
source: key.source.clone(),
bucket: key.bucket.clone(),
drive: key.drive.clone(),
count: value.count,
age_seconds: timestamp_elapsed_seconds_since(now, value.started_at),
})
.collect::<Vec<_>>();
active_bucket_drive_scans.sort_by(|left, right| {
left.source
.cmp(&right.source)
.then_with(|| left.bucket.cmp(&right.bucket))
.then_with(|| left.drive.cmp(&right.drive))
});
ScannerRuntimeDetailsReport {
disk_bucket_scan_states: self.scanner_disk_bucket_scan_state_snapshots(),
bucket_drive_results: self.scanner_bucket_drive_result_counter_snapshots(),
@@ -2791,6 +2874,7 @@ impl Metrics {
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.clone(),
active_bucket_drive_scans,
}
}
@@ -4371,7 +4455,7 @@ mod tests {
#[tokio::test]
async fn report_includes_bucket_drive_scan_starts() {
let metrics = Metrics::new();
metrics.record_scan_bucket_drive_start();
metrics.record_scan_bucket_drive_start(ScannerWorkSource::Usage, "bucket-a", "/mnt/data/1");
metrics.record_scan_bucket_drive_failure();
let report = metrics.report().await;
@@ -4380,6 +4464,27 @@ mod tests {
assert_eq!(report.life_time_ops.get("scan_bucket_drive_failure"), Some(&1));
}
#[tokio::test]
async fn active_bucket_drive_snapshot_is_structured_and_retired_on_end() {
let metrics = Metrics::new();
metrics.record_scan_bucket_drive_start(ScannerWorkSource::Usage, "bucket-a", "/mnt/data/1");
metrics.record_scan_bucket_drive_start(ScannerWorkSource::Usage, "bucket-a", "/mnt/data/1");
let active = metrics.scanner_runtime_details_report().active_bucket_drive_scans;
assert_eq!(active.len(), 1);
assert_eq!(active[0].source, ScannerWorkSource::Usage.as_str());
assert_eq!(active[0].bucket, "bucket-a");
assert_eq!(active[0].drive, "/mnt/data/1");
assert_eq!(active[0].count, 2);
metrics.record_scan_bucket_drive_end(ScannerWorkSource::Usage, "bucket-a", "/mnt/data/1");
assert_eq!(metrics.scanner_runtime_details_report().active_bucket_drive_scans[0].count, 1);
metrics.record_scan_bucket_drive_end(ScannerWorkSource::Usage, "bucket-a", "/mnt/data/1");
assert!(metrics.scanner_runtime_details_report().active_bucket_drive_scans.is_empty());
metrics.record_scan_bucket_drive_start(ScannerWorkSource::Usage, "", "/mnt/data/1");
assert!(metrics.scanner_runtime_details_report().active_bucket_drive_scans.is_empty());
}
#[tokio::test]
async fn report_includes_structured_bucket_drive_results() {
let metrics = Metrics::new();
crates/obs/src/metrics/collectors/ilm.rs
+104 -1
View File
@@ -54,12 +54,37 @@ pub(crate) struct IlmActionTaskStats {
pub(crate) value: u64,
}
#[derive(Debug, Clone, Default)]
pub(crate) struct IlmQueueTaskStats {
pub(crate) action: String,
pub(crate) state: String,
pub(crate) value: u64,
}
#[derive(Debug, Clone, Default)]
pub(crate) struct IlmTaskEventStats {
pub(crate) action: String,
pub(crate) result: String,
pub(crate) value: u64,
}
#[derive(Debug, Clone, Default)]
pub(crate) struct IlmBackpressureStats {
pub(crate) action: String,
pub(crate) reason: String,
pub(crate) value: u64,
}
/// ILM statistics with runtime-local node identity and bounded action/state details.
#[derive(Debug, Clone, Default)]
pub(crate) struct IlmRuntimeStats {
pub(crate) server: String,
pub(crate) stats: IlmStats,
pub(crate) action_tasks: Vec<IlmActionTaskStats>,
pub(crate) queue_tasks: Vec<IlmQueueTaskStats>,
pub(crate) task_events: Vec<IlmTaskEventStats>,
pub(crate) backpressure: Vec<IlmBackpressureStats>,
pub(crate) versions_scanned: u64,
}
fn is_live_action_task_state(state: &str) -> bool {
@@ -112,6 +137,30 @@ pub(crate) fn collect_ilm_runtime_metrics(stats: &IlmRuntimeStats) -> Vec<Promet
}),
);
metrics.extend(stats.queue_tasks.iter().map(|task| {
PrometheusMetric::from_descriptor(&ILM_TASKS_MD, task.value as f64)
.with_label_owned(SERVER_LABEL, stats.server.clone())
.with_label_owned(ACTION_LABEL, task.action.clone())
.with_label_owned(QUEUE_STATE_LABEL, task.state.clone())
}));
metrics.extend(stats.task_events.iter().map(|event| {
PrometheusMetric::from_descriptor(&ILM_TASK_EVENTS_MD, event.value as f64)
.with_label_owned(SERVER_LABEL, stats.server.clone())
.with_label_owned(ACTION_LABEL, event.action.clone())
.with_label_owned(RESULT_LABEL, event.result.clone())
}));
metrics.extend(stats.backpressure.iter().map(|event| {
PrometheusMetric::from_descriptor(&ILM_QUEUE_BACKPRESSURE_MD, event.value as f64)
.with_label_owned(SERVER_LABEL, stats.server.clone())
.with_label_owned(ACTION_LABEL, event.action.clone())
.with_label_owned(REASON_LABEL, event.reason.clone())
}));
metrics.push(
PrometheusMetric::from_descriptor(&ILM_VERSIONS_SCANNED_BY_SERVER_MD, stats.versions_scanned as f64)
.with_label_owned(SERVER_LABEL, stats.server.clone())
.with_label_owned(SOURCE_LABEL, "lifecycle".to_string()),
);
metrics
}
@@ -135,6 +184,22 @@ mod tests {
let runtime_stats = IlmRuntimeStats {
server: "node1:9000".to_string(),
stats,
queue_tasks: vec![IlmQueueTaskStats {
action: "transition".to_string(),
state: "pending".to_string(),
value: 8,
}],
task_events: vec![IlmTaskEventStats {
action: "transition".to_string(),
result: "completed".to_string(),
value: 7,
}],
backpressure: vec![IlmBackpressureStats {
action: "transition".to_string(),
reason: "queue_full".to_string(),
value: 2,
}],
versions_scanned: 1000000,
action_tasks: vec![
IlmActionTaskStats {
action: "expiry".to_string(),
@@ -156,7 +221,7 @@ mod tests {
let metrics = collect_ilm_runtime_metrics(&runtime_stats);
assert_eq!(metrics.len(), 11);
assert_eq!(metrics.len(), 15);
let pending = metrics.iter().find(|m| m.value == 100.0);
assert!(pending.is_some());
@@ -178,6 +243,44 @@ mod tests {
});
assert!(transition_timeout.is_none());
let transition_queue = metrics.iter().find(|m| {
m.name == ILM_TASKS_MD.get_full_metric_name()
&& m.labels
.iter()
.any(|(name, value)| *name == ACTION_LABEL && value.as_ref() == "transition")
&& m.labels
.iter()
.any(|(name, value)| *name == QUEUE_STATE_LABEL && value.as_ref() == "pending")
});
assert_eq!(transition_queue.map(|metric| metric.value), Some(8.0));
let completed = metrics.iter().find(|m| {
m.name == ILM_TASK_EVENTS_MD.get_full_metric_name()
&& m.labels
.iter()
.any(|(name, value)| *name == RESULT_LABEL && value.as_ref() == "completed")
});
assert_eq!(completed.map(|metric| metric.value), Some(7.0));
let backpressure = metrics.iter().find(|m| {
m.name == ILM_QUEUE_BACKPRESSURE_MD.get_full_metric_name()
&& m.labels
.iter()
.any(|(name, value)| *name == REASON_LABEL && value.as_ref() == "queue_full")
});
assert_eq!(backpressure.map(|metric| metric.value), Some(2.0));
let version_detail = metrics.iter().find(|m| {
m.name == ILM_VERSIONS_SCANNED_BY_SERVER_MD.get_full_metric_name()
&& m.labels
.iter()
.any(|(name, value)| *name == SERVER_LABEL && value.as_ref() == "node1:9000")
&& m.labels
.iter()
.any(|(name, value)| *name == SOURCE_LABEL && value.as_ref() == "lifecycle")
});
assert_eq!(version_detail.map(|metric| metric.value), Some(1000000.0));
let transition_active = metrics.iter().find(|m| {
m.name == ILM_ACTION_TASKS_MD.get_full_metric_name()
&& m.labels
crates/obs/src/metrics/collectors/mod.rs
+4 -1
View File
@@ -59,9 +59,12 @@ pub use cluster_iam::{IamStats, collect_iam_metrics};
pub use cluster_usage::{BucketUsageStats, ClusterUsageStats, collect_bucket_usage_metrics, collect_cluster_usage_metrics};
pub use compression::{CompressionClusterStats, collect_compression_cluster_metrics};
pub use dial9::{Dial9Stats, collect_current_dial9_metrics, collect_dial9_metrics, is_dial9_enabled};
pub(crate) use ilm::{IlmActionTaskStats, IlmRuntimeStats, collect_ilm_runtime_metrics};
pub(crate) use ilm::{
IlmActionTaskStats, IlmBackpressureStats, IlmQueueTaskStats, IlmRuntimeStats, IlmTaskEventStats, collect_ilm_runtime_metrics,
};
pub use ilm::{IlmStats, collect_ilm_metrics};
pub use node::{DiskStats, collect_node_metrics};
pub(crate) use notification::collect_notification_runtime_metrics;
pub use notification::{NotificationStats, collect_notification_metrics};
pub(crate) use notification_target::{NotificationTargetRuntimeStats, collect_notification_target_runtime_metrics};
pub use notification_target::{NotificationTargetStats, collect_notification_target_metrics};
crates/obs/src/metrics/collectors/notification.rs
+57 -2
View File
@@ -19,9 +19,12 @@
use crate::metrics::report::PrometheusMetric;
use crate::metrics::schema::cluster_notification::{
NOTIFICATION_CURRENT_SEND_IN_PROGRESS_MD, NOTIFICATION_EVENTS_ERRORS_TOTAL_MD, NOTIFICATION_EVENTS_SENT_TOTAL_MD,
NOTIFICATION_EVENTS_SKIPPED_TOTAL_MD,
NOTIFICATION_CURRENT_SEND_IN_PROGRESS_BY_SERVER_MD, NOTIFICATION_CURRENT_SEND_IN_PROGRESS_MD,
NOTIFICATION_EVENTS_ERRORS_TOTAL_BY_SERVER_MD, NOTIFICATION_EVENTS_ERRORS_TOTAL_MD,
NOTIFICATION_EVENTS_SENT_TOTAL_BY_SERVER_MD, NOTIFICATION_EVENTS_SENT_TOTAL_MD,
NOTIFICATION_EVENTS_SKIPPED_TOTAL_BY_SERVER_MD, NOTIFICATION_EVENTS_SKIPPED_TOTAL_MD, SERVER,
};
use std::borrow::Cow;
/// Notification statistics.
#[derive(Debug, Clone, Default)]
@@ -49,6 +52,30 @@ pub fn collect_notification_metrics(stats: &NotificationStats) -> Vec<Prometheus
]
}
/// Collects the legacy aggregate metrics and node-local runtime siblings.
pub(crate) fn collect_notification_runtime_metrics(stats: &NotificationStats, server: &str) -> Vec<PrometheusMetric> {
let mut metrics = collect_notification_metrics(stats);
if server.is_empty() {
return metrics;
}
let server_label: Cow<'static, str> = Cow::Owned(server.to_string());
metrics.extend([
PrometheusMetric::from_descriptor(
&NOTIFICATION_CURRENT_SEND_IN_PROGRESS_BY_SERVER_MD,
stats.current_send_in_progress as f64,
)
.with_label(SERVER, server_label.clone()),
PrometheusMetric::from_descriptor(&NOTIFICATION_EVENTS_ERRORS_TOTAL_BY_SERVER_MD, stats.events_errors_total as f64)
.with_label(SERVER, server_label.clone()),
PrometheusMetric::from_descriptor(&NOTIFICATION_EVENTS_SENT_TOTAL_BY_SERVER_MD, stats.events_sent_total as f64)
.with_label(SERVER, server_label.clone()),
PrometheusMetric::from_descriptor(&NOTIFICATION_EVENTS_SKIPPED_TOTAL_BY_SERVER_MD, stats.events_skipped_total as f64)
.with_label(SERVER, server_label),
]);
metrics
}
#[cfg(test)]
mod tests {
use super::*;
@@ -86,4 +113,32 @@ mod tests {
assert!(metric.labels.is_empty());
}
}
#[test]
fn runtime_metrics_keep_aggregate_and_add_server_siblings() {
let stats = NotificationStats {
current_send_in_progress: 5,
events_errors_total: 10,
events_sent_total: 100,
events_skipped_total: 2,
};
let metrics = collect_notification_runtime_metrics(&stats, "node1:9000");
assert_eq!(metrics.len(), 8);
assert_eq!(metrics.iter().filter(|metric| metric.labels.is_empty()).count(), 4);
assert_eq!(metrics.iter().filter(|metric| metric.labels.len() == 1).count(), 4);
assert!(metrics.iter().filter(|metric| metric.labels.len() == 1).all(|metric| {
metric
.labels
.iter()
.any(|(name, value)| *name == SERVER && value == "node1:9000")
}));
}
#[test]
fn runtime_metrics_do_not_publish_empty_server_series() {
let metrics = collect_notification_runtime_metrics(&NotificationStats::default(), "");
assert_eq!(metrics.len(), 4);
assert!(metrics.iter().all(|metric| metric.labels.is_empty()));
}
}
crates/obs/src/metrics/collectors/scanner.rs
+64 -1
View File
@@ -184,6 +184,15 @@ pub struct ScannerBucketDriveResultStats {
pub count: u64,
}
#[derive(Debug, Clone, Default)]
pub struct ScannerActiveBucketDriveStats {
pub source: String,
pub bucket: String,
pub drive: String,
pub count: u64,
pub age_seconds: u64,
}
/// Scanner statistics with runtime-local node identity and bounded source/result details.
#[derive(Debug, Clone, Default)]
pub(crate) struct ScannerRuntimeStats {
@@ -195,6 +204,7 @@ pub(crate) struct ScannerRuntimeStats {
pub(crate) bucket_drive_results: Vec<ScannerBucketDriveResultStats>,
pub(crate) current_cycle_bucket_drive_results: Vec<ScannerBucketDriveResultStats>,
pub(crate) last_cycle_bucket_drive_results: Vec<ScannerBucketDriveResultStats>,
pub(crate) active_bucket_drive_scans: Vec<ScannerActiveBucketDriveStats>,
}
/// Collects scanner metrics from the given stats.
@@ -452,6 +462,23 @@ fn collect_scanner_metrics_with_runtime(stats: &ScannerStats, runtime: Option<&S
&runtime.last_cycle_bucket_drive_results,
Some("last"),
);
for active in &runtime.active_bucket_drive_scans {
let labels = |metric: PrometheusMetric| {
metric
.with_label_owned(SERVER_LABEL, runtime.server.clone())
.with_label_owned(SOURCE_LABEL, active.source.clone())
.with_label_owned(BUCKET_LABEL, active.bucket.clone())
.with_label_owned(DRIVE_LABEL, active.drive.clone())
};
metrics.push(labels(PrometheusMetric::from_descriptor(
&SCANNER_ACTIVE_BUCKET_DRIVE_SCANS_MD,
active.count as f64,
)));
metrics.push(labels(PrometheusMetric::from_descriptor(
&SCANNER_ACTIVE_BUCKET_DRIVE_SCAN_AGE_SECONDS_MD,
active.age_seconds as f64,
)));
}
}
metrics
@@ -566,6 +593,13 @@ mod tests {
result: "error".to_string(),
count: 2,
}],
active_bucket_drive_scans: vec![ScannerActiveBucketDriveStats {
source: "usage".to_string(),
bucket: "photos".to_string(),
drive: "/data1".to_string(),
count: 2,
age_seconds: 7,
}],
stats: ScannerStats {
bucket_scans_finished: 100,
bucket_scans_started: 100,
@@ -642,7 +676,7 @@ mod tests {
let metrics = collect_scanner_runtime_metrics(&stats);
report_metrics(&metrics);
assert_eq!(metrics.len(), 90);
assert_eq!(metrics.len(), 92);
let objects = metrics.iter().find(|m| m.value == 1000000.0);
assert!(objects.is_some());
@@ -656,6 +690,35 @@ mod tests {
assert_eq!(active_paths.map(|m| m.value), Some(4.0));
assert_eq!(active_paths.map(|m| m.labels.len()), Some(0));
let active_bucket_drive = metrics
.iter()
.find(|m| m.name == SCANNER_ACTIVE_BUCKET_DRIVE_SCANS_MD.get_full_metric_name())
.expect("active bucket-drive metric");
assert_eq!(active_bucket_drive.value, 2.0);
assert!(
active_bucket_drive
.labels
.iter()
.any(|(name, value)| *name == SOURCE_LABEL && value == "usage")
);
assert!(
active_bucket_drive
.labels
.iter()
.any(|(name, value)| *name == BUCKET_LABEL && value == "photos")
);
assert!(
active_bucket_drive
.labels
.iter()
.any(|(name, value)| *name == DRIVE_LABEL && value == "/data1")
);
let active_age = metrics
.iter()
.find(|m| m.name == SCANNER_ACTIVE_BUCKET_DRIVE_SCAN_AGE_SECONDS_MD.get_full_metric_name())
.expect("active bucket-drive age metric");
assert_eq!(active_age.value, 7.0);
let bucket_drive_result = metrics
.iter()
.find(|m| m.name == SCANNER_BUCKET_DRIVE_RESULT_TOTAL_MD.get_full_metric_name());
crates/obs/src/metrics/collectors/system_drive.rs
+17 -1
View File
@@ -60,6 +60,10 @@ pub struct DriveDetailedStats {
pub api_latency_micros: Option<u64>,
/// Health status (1=healthy, 0=unhealthy)
pub health: u8,
/// Total successful write operations when backed by a real disk metric.
pub writes_total: Option<u64>,
/// Total successful delete operations when backed by a real disk metric.
pub deletes_total: Option<u64>,
/// Reads per second when backed by a real iostat sample
pub reads_per_sec: Option<f64>,
/// Kilobytes read per second when backed by a real iostat sample
@@ -282,6 +286,12 @@ pub(crate) fn collect_drive_runtime_detailed_metrics(stats: &[DriveRuntimeDetail
if let Some(value) = stat.stats.perc_util {
push_drive_metric(&mut metrics, &DRIVE_PERC_UTIL_MD, value, server_label, drive_label);
}
if let Some(value) = stat.stats.writes_total {
push_drive_metric(&mut metrics, &DRIVE_WRITES_TOTAL_MD, value as f64, server_label, drive_label);
}
if let Some(value) = stat.stats.deletes_total {
push_drive_metric(&mut metrics, &DRIVE_DELETES_TOTAL_MD, value as f64, server_label, drive_label);
}
if let Some(labels) = &topology_labels {
if let Some(disk_id) = stat.disk_id.as_ref().filter(|disk_id| !disk_id.is_empty()) {
metrics.push(
@@ -449,6 +459,8 @@ mod tests {
waiting_io: Some(3),
api_latency_micros: Some(1500),
health: 1,
writes_total: Some(11),
deletes_total: Some(4),
reads_per_sec: Some(100.0),
reads_kb_per_sec: Some(1024.0),
reads_await: Some(5.5),
@@ -462,7 +474,7 @@ mod tests {
let metrics = collect_drive_runtime_detailed_metrics(&stats);
report_metrics(&metrics);
assert_eq!(metrics.len(), 34);
assert_eq!(metrics.len(), 36);
// Verify total bytes metric
let total_bytes_name = DRIVE_TOTAL_BYTES_MD.get_full_metric_name();
@@ -503,6 +515,8 @@ mod tests {
API_LABEL,
],
);
assert_metric_label_keys(&metrics, &DRIVE_WRITES_TOTAL_MD, 11.0, &[SERVER_LABEL, DRIVE_LABEL]);
assert_metric_label_keys(&metrics, &DRIVE_DELETES_TOTAL_MD, 4.0, &[SERVER_LABEL, DRIVE_LABEL]);
}
#[test]
@@ -524,6 +538,8 @@ mod tests {
waiting_io: None,
api_latency_micros: None,
health: 1,
writes_total: None,
deletes_total: None,
reads_per_sec: None,
reads_kb_per_sec: None,
reads_await: None,
crates/obs/src/metrics/scheduler.rs
+93 -8
View File
@@ -62,7 +62,7 @@ use crate::metrics::collectors::{
collect_memory_metrics,
collect_network_metrics,
collect_node_metrics,
collect_notification_metrics,
collect_notification_runtime_metrics,
collect_notification_target_runtime_metrics,
collect_process_attributes,
collect_process_cpu_metrics,
@@ -120,12 +120,13 @@ use crate::metrics::schema::notification_target::{
};
use crate::metrics::schema::scanner::{
BUCKET_LABEL as SCANNER_BUCKET_LABEL, CYCLE_SCOPE_LABEL as SCANNER_CYCLE_SCOPE_LABEL, DRIVE_LABEL as SCANNER_DRIVE_LABEL,
RESULT_LABEL as SCANNER_RESULT_LABEL, SCANNER_BUCKET_DRIVE_RESULT_TOTAL_MD, SCANNER_CYCLE_BUCKET_DRIVE_RESULT_MD,
RESULT_LABEL as SCANNER_RESULT_LABEL, SCANNER_ACTIVE_BUCKET_DRIVE_SCAN_AGE_SECONDS_MD, SCANNER_ACTIVE_BUCKET_DRIVE_SCANS_MD,
SCANNER_BUCKET_DRIVE_RESULT_TOTAL_MD, SCANNER_CYCLE_BUCKET_DRIVE_RESULT_MD, SOURCE_LABEL as SCANNER_SOURCE_LABEL,
};
use crate::metrics::schema::system_drive::{
API_LABEL as DRIVE_API_LABEL, DISK_ID_LABEL, DRIVE_API_CALLS_MD, DRIVE_API_LATENCY_BY_API_MD, DRIVE_HEALING_MD,
DRIVE_INDEX_LABEL, DRIVE_INFO_MD, DRIVE_LABEL, DRIVE_OFFLINE_DURATION_SECONDS_MD, DRIVE_RUNTIME_STATE_MD, DRIVE_SCANNING_MD,
POOL_INDEX_LABEL, SET_INDEX_LABEL, STATE_LABEL as DRIVE_STATE_LABEL,
API_LABEL as DRIVE_API_LABEL, DISK_ID_LABEL, DRIVE_API_CALLS_MD, DRIVE_API_LATENCY_BY_API_MD, DRIVE_DELETES_TOTAL_MD,
DRIVE_HEALING_MD, DRIVE_INDEX_LABEL, DRIVE_INFO_MD, DRIVE_LABEL, DRIVE_OFFLINE_DURATION_SECONDS_MD, DRIVE_RUNTIME_STATE_MD,
DRIVE_SCANNING_MD, DRIVE_WRITES_TOTAL_MD, POOL_INDEX_LABEL, SET_INDEX_LABEL, STATE_LABEL as DRIVE_STATE_LABEL,
};
use crate::metrics::schema::system_process::{PROCESS_EXECUTABLE_NAME_LABEL, PROCESS_PID_LABEL};
use crate::metrics::stats_collector::{
@@ -303,15 +304,33 @@ type AuditTargetKey = (String, String); // (server, target_id)
type NotificationLegacyTargetKey = (String, String); // (target_id, target_type)
type NotificationTargetKey = (String, String, String); // (server, target_id, target_type)
type DriveTopologyKey = (String, String, String, String, String); // (server, drive, pool, set, drive_index)
type DriveBasicKey = (String, String); // (server, drive)
type DriveTopologyApiKey = (String, String, String, String, String, String); // (server, drive, pool, set, drive_index, api)
type DriveInfoKey = (String, String, String, String, String, String); // (server, drive, pool, set, drive_index, disk_id)
type ScannerCycleBucketDriveResultKey = (String, String, String, String, String); // (server, cycle_scope, bucket, drive, result)
type ScannerBucketDriveResultKey = (String, String, String, String); // (server, bucket, drive, result)
type ScannerActiveBucketDriveKey = (String, String, String, String); // (server, source, bucket, drive)
fn drive_info_live_keys(stats: &[DriveRuntimeDetailedStats]) -> HashSet<DriveInfoKey> {
stats.iter().filter_map(drive_info_key).collect()
}
fn drive_basic_live_keys(stats: &[DriveRuntimeDetailedStats]) -> HashSet<DriveBasicKey> {
stats
.iter()
.map(|stat| (stat.stats.server.clone(), stat.stats.drive.clone()))
.collect()
}
fn retire_drive_basic_metric_series(key: &DriveBasicKey) -> usize {
let labels = [
(SERVER_LABEL, Cow::Owned(key.0.clone())),
(DRIVE_LABEL, Cow::Owned(key.1.clone())),
];
retire_metric_series(&DRIVE_WRITES_TOTAL_MD.get_full_metric_name(), &labels)
+ retire_metric_series(&DRIVE_DELETES_TOTAL_MD.get_full_metric_name(), &labels)
}
fn drive_topology_live_keys(stats: &[DriveRuntimeDetailedStats]) -> HashSet<DriveTopologyKey> {
stats.iter().filter_map(drive_topology_key).collect()
}
@@ -469,6 +488,25 @@ fn retire_scanner_bucket_drive_result_metric_series(key: &ScannerBucketDriveResu
retire_metric_series(&SCANNER_BUCKET_DRIVE_RESULT_TOTAL_MD.get_full_metric_name(), &labels)
}
fn scanner_active_bucket_drive_live_keys(stats: &ScannerRuntimeStats) -> HashSet<ScannerActiveBucketDriveKey> {
stats
.active_bucket_drive_scans
.iter()
.map(|active| (stats.server.clone(), active.source.clone(), active.bucket.clone(), active.drive.clone()))
.collect()
}
fn retire_scanner_active_bucket_drive_metric_series(key: &ScannerActiveBucketDriveKey) -> usize {
let labels = [
(SERVER_LABEL, Cow::Owned(key.0.clone())),
(SCANNER_SOURCE_LABEL, Cow::Owned(key.1.clone())),
(SCANNER_BUCKET_LABEL, Cow::Owned(key.2.clone())),
(SCANNER_DRIVE_LABEL, Cow::Owned(key.3.clone())),
];
retire_metric_series(&SCANNER_ACTIVE_BUCKET_DRIVE_SCANS_MD.get_full_metric_name(), &labels)
+ retire_metric_series(&SCANNER_ACTIVE_BUCKET_DRIVE_SCAN_AGE_SECONDS_MD.get_full_metric_name(), &labels)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Default)]
pub struct MetricsRuntimeCollectorHealthSnapshot {
pub healthy_collectors: u8,
@@ -1841,6 +1879,7 @@ pub fn init_metrics_runtime(token: CancellationToken) {
let token_clone = token.clone();
tokio::spawn(async move {
let mut interval = metrics_interval(node_interval, Duration::ZERO);
let mut prev_drive_basic_keys: HashSet<DriveBasicKey> = HashSet::new();
let mut prev_drive_info_keys: HashSet<DriveInfoKey> = HashSet::new();
let mut prev_drive_topology_keys: HashSet<DriveTopologyKey> = HashSet::new();
let mut prev_drive_topology_api_keys: HashSet<DriveTopologyApiKey> = HashSet::new();
@@ -1851,6 +1890,7 @@ pub fn init_metrics_runtime(token: CancellationToken) {
run_metrics_collector_tick(health, MetricsCollectorTaskId::NodeDiskStats, "node_disk_stats", async {
let (disk_stats, drive_stats, drive_counts) = collect_disk_and_system_drive_runtime_stats().await;
let current_drive_info_keys = drive_info_live_keys(&drive_stats);
let current_drive_basic_keys = drive_basic_live_keys(&drive_stats);
let current_drive_topology_keys = drive_topology_live_keys(&drive_stats);
let current_drive_topology_api_keys = drive_topology_api_live_keys(&drive_stats);
let retire_drive_info_keys = if has_seen_drive_info_snapshot {
@@ -1858,6 +1898,11 @@ pub fn init_metrics_runtime(token: CancellationToken) {
} else {
Vec::new()
};
let retire_drive_basic_keys = if has_seen_drive_info_snapshot {
prev_drive_basic_keys.difference(&current_drive_basic_keys).cloned().collect::<Vec<_>>()
} else {
Vec::new()
};
let retire_drive_topology_keys = if has_seen_drive_info_snapshot {
prev_drive_topology_keys.difference(&current_drive_topology_keys).cloned().collect::<Vec<_>>()
} else {
@@ -1872,6 +1917,7 @@ pub fn init_metrics_runtime(token: CancellationToken) {
Vec::new()
};
prev_drive_info_keys = current_drive_info_keys;
prev_drive_basic_keys = current_drive_basic_keys;
prev_drive_topology_keys = current_drive_topology_keys;
prev_drive_topology_api_keys = current_drive_topology_api_keys;
has_seen_drive_info_snapshot = true;
@@ -1882,6 +1928,9 @@ pub fn init_metrics_runtime(token: CancellationToken) {
for key in retire_drive_info_keys {
let _ = retire_drive_info_metric_series(&key);
}
for key in retire_drive_basic_keys {
let _ = retire_drive_basic_metric_series(&key);
}
for key in retire_drive_topology_keys {
let _ = retire_drive_topology_metric_series(&key);
}
@@ -2106,14 +2155,14 @@ pub fn init_metrics_runtime(token: CancellationToken) {
_ = interval.tick() => {
run_metrics_collector_tick(health, MetricsCollectorTaskId::NotificationStats, "notification_stats", async {
let snapshot = notification_metrics_snapshot();
let mut metrics = collect_notification_metrics(&NotificationStats {
let server = current_local_node_identity();
let mut metrics = collect_notification_runtime_metrics(&NotificationStats {
current_send_in_progress: snapshot.current_send_in_progress,
events_errors_total: snapshot.events_errors_total,
events_sent_total: snapshot.events_sent_total,
events_skipped_total: snapshot.events_skipped_total,
});
}, &server);
let server = current_local_node_identity();
let target_stats = notification_target_metrics().await
.into_iter()
.map(|snapshot| NotificationTargetRuntimeStats {
@@ -2173,6 +2222,7 @@ pub fn init_metrics_runtime(token: CancellationToken) {
let mut has_seen_scanner_snapshot = false;
let mut prev_scanner_cycle_bucket_drive_result_keys: HashSet<ScannerCycleBucketDriveResultKey> = HashSet::new();
let mut prev_scanner_bucket_drive_result_keys: HashSet<ScannerBucketDriveResultKey> = HashSet::new();
let mut prev_scanner_active_bucket_drive_keys: HashSet<ScannerActiveBucketDriveKey> = HashSet::new();
loop {
tokio::select! {
_ = interval.tick() => {
@@ -2189,9 +2239,11 @@ pub fn init_metrics_runtime(token: CancellationToken) {
let mut retire_scanner_cycle_bucket_drive_result_keys = Vec::new();
let mut retire_scanner_bucket_drive_result_keys = Vec::new();
let mut retire_scanner_active_bucket_drive_keys = Vec::new();
if let Some(stats) = collect_scanner_runtime_metric_stats().await {
let current_cycle_keys = scanner_cycle_bucket_drive_result_live_keys(&stats);
let current_keys = scanner_bucket_drive_result_live_keys(&stats);
let current_active_keys = scanner_active_bucket_drive_live_keys(&stats);
if has_seen_scanner_snapshot {
retire_scanner_cycle_bucket_drive_result_keys = prev_scanner_cycle_bucket_drive_result_keys
.difference(&current_cycle_keys)
@@ -2201,9 +2253,14 @@ pub fn init_metrics_runtime(token: CancellationToken) {
.difference(&current_keys)
.cloned()
.collect();
retire_scanner_active_bucket_drive_keys = prev_scanner_active_bucket_drive_keys
.difference(&current_active_keys)
.cloned()
.collect();
}
prev_scanner_cycle_bucket_drive_result_keys = current_cycle_keys;
prev_scanner_bucket_drive_result_keys = current_keys;
prev_scanner_active_bucket_drive_keys = current_active_keys;
has_seen_scanner_snapshot = true;
metrics.extend(collect_scanner_runtime_metrics(&stats));
}
@@ -2217,6 +2274,9 @@ pub fn init_metrics_runtime(token: CancellationToken) {
for key in retire_scanner_bucket_drive_result_keys {
let _ = retire_scanner_bucket_drive_result_metric_series(&key);
}
for key in retire_scanner_active_bucket_drive_keys {
let _ = retire_scanner_active_bucket_drive_metric_series(&key);
}
},
).await;
}
@@ -2495,6 +2555,7 @@ fn collect_system_monitoring_metrics(
#[cfg(test)]
mod tests {
use super::*;
use crate::metrics::collectors::scanner::ScannerActiveBucketDriveStats;
use std::collections::{HashMap, HashSet};
use std::time::Duration;
use tokio::time::Instant;
@@ -2723,6 +2784,30 @@ mod tests {
assert!(current.contains(&("server-a".to_string(), "logs".to_string(), "/data1".to_string(), "success".to_string(),)));
}
#[test]
fn scanner_active_bucket_drive_keys_detect_completed_scans() {
let previous = scanner_active_bucket_drive_live_keys(&ScannerRuntimeStats {
server: "server-a".to_string(),
active_bucket_drive_scans: vec![ScannerActiveBucketDriveStats {
source: "usage".to_string(),
bucket: "photos".to_string(),
drive: "/data1".to_string(),
count: 1,
age_seconds: 3,
}],
..Default::default()
});
let current = scanner_active_bucket_drive_live_keys(&ScannerRuntimeStats {
server: "server-a".to_string(),
..Default::default()
});
assert!(
previous
.difference(&current)
.any(|key| key == &("server-a".to_string(), "usage".to_string(), "photos".to_string(), "/data1".to_string()))
);
}
#[test]
fn replication_proxy_bucket_keys_detect_removed_buckets() {
let previous = repl_proxy_bucket_live_keys(&[BucketReplicationRuntimeStats {
crates/obs/src/metrics/schema/cluster_notification.rs
+40
View File
@@ -15,6 +15,10 @@
use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems};
use std::sync::LazyLock;
pub const SERVER: &str = "server";
const SERVER_LABELS: [&str; 1] = [SERVER];
pub static NOTIFICATION_CURRENT_SEND_IN_PROGRESS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::NotificationCurrentSendInProgress,
@@ -24,6 +28,15 @@ pub static NOTIFICATION_CURRENT_SEND_IN_PROGRESS_MD: LazyLock<MetricDescriptor>
)
});
pub static NOTIFICATION_CURRENT_SEND_IN_PROGRESS_BY_SERVER_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::Custom("current_send_in_progress_by_server".to_string()),
"Number of concurrent async Send calls active to all targets by server",
&SERVER_LABELS,
subsystems::NOTIFICATION,
)
});
pub static NOTIFICATION_EVENTS_ERRORS_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_counter_md(
MetricName::NotificationEventsErrorsTotal,
@@ -33,6 +46,15 @@ pub static NOTIFICATION_EVENTS_ERRORS_TOTAL_MD: LazyLock<MetricDescriptor> = Laz
)
});
pub static NOTIFICATION_EVENTS_ERRORS_TOTAL_BY_SERVER_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_counter_md(
MetricName::Custom("events_errors_total_by_server".to_string()),
"Events that failed to be sent to the targets by server",
&SERVER_LABELS,
subsystems::NOTIFICATION,
)
});
pub static NOTIFICATION_EVENTS_SENT_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_counter_md(
MetricName::NotificationEventsSentTotal,
@@ -42,6 +64,15 @@ pub static NOTIFICATION_EVENTS_SENT_TOTAL_MD: LazyLock<MetricDescriptor> = LazyL
)
});
pub static NOTIFICATION_EVENTS_SENT_TOTAL_BY_SERVER_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_counter_md(
MetricName::Custom("events_sent_total_by_server".to_string()),
"Total number of events sent to the targets by server",
&SERVER_LABELS,
subsystems::NOTIFICATION,
)
});
pub static NOTIFICATION_EVENTS_SKIPPED_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_counter_md(
MetricName::NotificationEventsSkippedTotal,
@@ -50,3 +81,12 @@ pub static NOTIFICATION_EVENTS_SKIPPED_TOTAL_MD: LazyLock<MetricDescriptor> = La
subsystems::NOTIFICATION,
)
});
pub static NOTIFICATION_EVENTS_SKIPPED_TOTAL_BY_SERVER_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_counter_md(
MetricName::Custom("events_skipped_total_by_server".to_string()),
"Notification dispatch attempts skipped before delivery by server",
&SERVER_LABELS,
subsystems::NOTIFICATION,
)
});
crates/obs/src/metrics/schema/entry/metric_name.rs
+4
View File
@@ -371,6 +371,8 @@ pub enum MetricName {
DriveWaitingIO,
DriveAPILatencyMicros,
DriveHealth,
DriveWritesTotal,
DriveDeletesTotal,
DriveOfflineCount,
DriveOnlineCount,
@@ -780,6 +782,8 @@ impl MetricName {
Self::DriveWaitingIO => "waiting_io".to_string(),
Self::DriveAPILatencyMicros => "api_latency_micros".to_string(),
Self::DriveHealth => "health".to_string(),
Self::DriveWritesTotal => "writes_total".to_string(),
Self::DriveDeletesTotal => "deletes_total".to_string(),
Self::DriveOfflineCount => "offline_count".to_string(),
Self::DriveOnlineCount => "online_count".to_string(),
crates/obs/src/metrics/schema/ilm.rs
+40
View File
@@ -18,6 +18,10 @@ use std::sync::LazyLock;
pub const SERVER_LABEL: &str = "server";
pub const ACTION_LABEL: &str = "action";
pub const STATE_LABEL: &str = "state";
pub const QUEUE_STATE_LABEL: &str = "queue_state";
pub const RESULT_LABEL: &str = "result";
pub const REASON_LABEL: &str = "reason";
pub const SOURCE_LABEL: &str = "source";
pub static ILM_ACTION_TASKS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
@@ -28,6 +32,33 @@ pub static ILM_ACTION_TASKS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
)
});
pub static ILM_TASKS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::Custom("tasks".to_string()),
"Current ILM task counts by server, action, and queue state",
&[SERVER_LABEL, ACTION_LABEL, QUEUE_STATE_LABEL],
subsystems::ILM,
)
});
pub static ILM_TASK_EVENTS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_counter_md(
MetricName::Custom("task_events_total".to_string()),
"ILM task events by server, action, and result",
&[SERVER_LABEL, ACTION_LABEL, RESULT_LABEL],
subsystems::ILM,
)
});
pub static ILM_QUEUE_BACKPRESSURE_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_counter_md(
MetricName::Custom("queue_backpressure_total".to_string()),
"ILM queue backpressure events by server, action, and reason",
&[SERVER_LABEL, ACTION_LABEL, REASON_LABEL],
subsystems::ILM,
)
});
pub static ILM_EXPIRY_PENDING_TASKS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::IlmExpiryPendingTasks,
@@ -108,3 +139,12 @@ pub static ILM_VERSIONS_SCANNED_MD: LazyLock<MetricDescriptor> = LazyLock::new(|
subsystems::ILM,
)
});
pub static ILM_VERSIONS_SCANNED_BY_SERVER_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_counter_md(
MetricName::Custom("versions_scanned_by_server".to_string()),
"ILM lifecycle-checked object versions by server and source",
&[SERVER_LABEL, SOURCE_LABEL],
subsystems::ILM,
)
});
crates/obs/src/metrics/schema/scanner.rs
+18
View File
@@ -59,6 +59,24 @@ pub static SCANNER_CYCLE_BUCKET_DRIVE_RESULT_MD: LazyLock<MetricDescriptor> = La
)
});
pub static SCANNER_ACTIVE_BUCKET_DRIVE_SCANS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::Custom("active_bucket_drive_scans".to_string()),
"Current active scanner bucket-drive scans by server, source, bucket, and drive",
&[SERVER_LABEL, SOURCE_LABEL, BUCKET_LABEL, DRIVE_LABEL],
subsystems::SCANNER,
)
});
pub static SCANNER_ACTIVE_BUCKET_DRIVE_SCAN_AGE_SECONDS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::Custom("active_bucket_drive_scan_age_seconds".to_string()),
"Age of the oldest active scanner bucket-drive scan by server, source, bucket, and drive",
&[SERVER_LABEL, SOURCE_LABEL, BUCKET_LABEL, DRIVE_LABEL],
subsystems::SCANNER,
)
});
pub static SCANNER_BUCKET_SCANS_FINISHED_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_counter_md(
MetricName::ScannerBucketScansFinished,
crates/obs/src/metrics/schema/system_drive.rs
+18
View File
@@ -259,6 +259,24 @@ pub static DRIVE_HEALTH_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
)
});
pub static DRIVE_WRITES_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_counter_md(
MetricName::DriveWritesTotal,
"Total successful write operations on a drive",
&ALL_DRIVE_LABELS[..],
subsystems::SYSTEM_DRIVE,
)
});
pub static DRIVE_DELETES_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_counter_md(
MetricName::DriveDeletesTotal,
"Total successful delete operations on a drive",
&ALL_DRIVE_LABELS[..],
subsystems::SYSTEM_DRIVE,
)
});
pub static DRIVE_OFFLINE_COUNT_MD: LazyLock<MetricDescriptor> =
LazyLock::new(|| new_gauge_md(MetricName::DriveOfflineCount, "Count of offline drives", &[], subsystems::SYSTEM_DRIVE));
crates/obs/src/metrics/stats_collector.rs
+206 -5
View File
@@ -18,15 +18,15 @@
//! RustFS internal sources (storage layer, bucket monitor, system info)
//! and convert them to the Stats structs used by collectors.
use crate::metrics::collectors::scanner::{ScannerBucketDriveResultStats, ScannerSourceWorkStats};
use crate::metrics::collectors::scanner::{ScannerActiveBucketDriveStats, ScannerBucketDriveResultStats, ScannerSourceWorkStats};
use crate::metrics::collectors::{
ApiRequestMetricSupport, ApiRequestStats, BucketReplicationBacklogStats, BucketReplicationBandwidthStats,
BucketReplicationRuntimeStats, BucketReplicationStats, BucketReplicationTargetBacklogStats, BucketReplicationTargetFlowStats,
BucketReplicationTargetStats, BucketStats, BucketUsageStats, ClusterConfigStats, ClusterHealthStats, ClusterStats,
ClusterUsageStats, CompressionClusterStats, CpuStats, DiskStats, DriveCountStats, DriveDetailedStats,
DriveRuntimeDetailedStats, ErasureSetStats, HostNetworkStats, IamStats, IlmActionTaskStats, IlmRuntimeStats, IlmStats,
MemoryStats, NetworkStats, ProcessStats, ProcessStatusType, ReplicationStats, ResourceStats, ScannerRuntimeStats,
ScannerStats,
DriveRuntimeDetailedStats, ErasureSetStats, HostNetworkStats, IamStats, IlmActionTaskStats, IlmBackpressureStats,
IlmQueueTaskStats, IlmRuntimeStats, IlmStats, IlmTaskEventStats, MemoryStats, NetworkStats, ProcessStats, ProcessStatusType,
ReplicationStats, ResourceStats, ScannerRuntimeStats, ScannerStats,
};
use crate::metrics::runtime_sources::{ObsIlmRuntimeSnapshot, bucket_monitor_handle, iam_metrics_snapshot, ilm_runtime_snapshot};
use crate::metrics::{
@@ -38,7 +38,10 @@ use crate::metrics::{
use crate::node_identity::current_local_node_identity;
use jiff::Timestamp;
use rustfs_common::heal_channel::HealScanMode;
use rustfs_common::metrics::{ScannerBucketDriveResultSnapshot, ScannerMetricsReport, ScannerSourceWorkSnapshot, global_metrics};
use rustfs_common::metrics::{
ScannerActiveBucketDriveSnapshot, ScannerBucketDriveResultSnapshot, ScannerMetricsReport, ScannerSourceWorkSnapshot,
global_metrics,
};
use rustfs_io_metrics::internode_metrics::global_internode_metrics;
use rustfs_io_metrics::{
ProcessResourceSnapshot, ProcessSampler, ProcessStatusSnapshot, ProcessSystemSnapshot, s3_op_metrics_snapshot,
@@ -835,6 +838,8 @@ pub(crate) async fn collect_disk_and_system_drive_runtime_stats()
drive_api_latency_micros(metrics.last_minute.values().map(|action| (action.count, action.acc_time)))
}),
health: if is_online { 1 } else { 0 },
writes_total: disk.metrics.as_ref().map(|metrics| metrics.total_writes),
deletes_total: disk.metrics.as_ref().map(|metrics| metrics.total_deletes),
reads_per_sec: None,
reads_kb_per_sec: None,
reads_await: None,
@@ -1275,6 +1280,110 @@ fn ilm_action_task_stats(ilm: &ObsIlmRuntimeSnapshot) -> Vec<IlmActionTaskStats>
]
}
fn ilm_queue_task_stats(metrics: &ScannerMetricsReport) -> Vec<IlmQueueTaskStats> {
let expiry = &metrics.lifecycle_expiry;
let transition = &metrics.lifecycle_transition;
vec![
IlmQueueTaskStats {
action: "expiry".to_string(),
state: "pending".to_string(),
value: expiry.current_queued,
},
IlmQueueTaskStats {
action: "expiry".to_string(),
state: "active".to_string(),
value: expiry.current_active,
},
IlmQueueTaskStats {
action: "transition".to_string(),
state: "pending".to_string(),
value: transition.current_queued,
},
IlmQueueTaskStats {
action: "transition".to_string(),
state: "active".to_string(),
value: transition.current_active,
},
IlmQueueTaskStats {
action: "transition".to_string(),
state: "compensation_running".to_string(),
value: transition.compensation_running,
},
]
}
fn ilm_task_event_stats(metrics: &ScannerMetricsReport) -> Vec<IlmTaskEventStats> {
let expiry = &metrics.lifecycle_expiry;
let transition = &metrics.lifecycle_transition;
vec![
IlmTaskEventStats {
action: "expiry".to_string(),
result: "queued".to_string(),
value: expiry.scanner_queued,
},
IlmTaskEventStats {
action: "expiry".to_string(),
result: "missed".to_string(),
value: expiry.scanner_missed,
},
IlmTaskEventStats {
action: "expiry".to_string(),
result: "blocked".to_string(),
value: expiry.scanner_blocked,
},
IlmTaskEventStats {
action: "expiry".to_string(),
result: "not_enqueued".to_string(),
value: expiry.scanner_not_enqueued,
},
IlmTaskEventStats {
action: "expiry".to_string(),
result: "failed".to_string(),
value: expiry.delete_failed,
},
IlmTaskEventStats {
action: "transition".to_string(),
result: "queued".to_string(),
value: transition.scanner_queued,
},
IlmTaskEventStats {
action: "transition".to_string(),
result: "missed".to_string(),
value: transition.scanner_missed,
},
IlmTaskEventStats {
action: "transition".to_string(),
result: "completed".to_string(),
value: transition.completed,
},
IlmTaskEventStats {
action: "transition".to_string(),
result: "failed".to_string(),
value: transition.failed,
},
]
}
fn ilm_backpressure_stats(metrics: &ScannerMetricsReport) -> Vec<IlmBackpressureStats> {
vec![
IlmBackpressureStats {
action: "expiry".to_string(),
reason: "queue_missed".to_string(),
value: metrics.lifecycle_expiry.queue_missed,
},
IlmBackpressureStats {
action: "transition".to_string(),
reason: "queue_full".to_string(),
value: metrics.lifecycle_transition.queue_full,
},
IlmBackpressureStats {
action: "transition".to_string(),
reason: "send_timeout".to_string(),
value: metrics.lifecycle_transition.queue_send_timeout,
},
]
}
/// Collect ILM metrics from the current lifecycle runtime state.
pub async fn collect_ilm_metric_stats() -> Option<IlmStats> {
collect_ilm_runtime_metric_stats().await.map(|stats| stats.stats)
@@ -1288,6 +1397,10 @@ pub(crate) async fn collect_ilm_runtime_metric_stats() -> Option<IlmRuntimeStats
Some(IlmRuntimeStats {
server: current_local_node_identity(),
action_tasks: ilm_action_task_stats(&ilm),
queue_tasks: ilm_queue_task_stats(&metrics),
task_events: ilm_task_event_stats(&metrics),
backpressure: ilm_backpressure_stats(&metrics),
versions_scanned,
stats: IlmStats {
expiry_pending_tasks: ilm.expiry_pending_tasks,
transition_active_tasks: ilm.transition_active_tasks,
@@ -1377,6 +1490,27 @@ fn scanner_bucket_drive_result_stats(results: &[ScannerBucketDriveResultSnapshot
stats
}
fn scanner_active_bucket_drive_stats(results: &[ScannerActiveBucketDriveSnapshot]) -> Vec<ScannerActiveBucketDriveStats> {
let mut stats = results
.iter()
.filter(|result| !result.source.is_empty() && !result.bucket.is_empty() && !result.drive.is_empty() && result.count > 0)
.map(|result| ScannerActiveBucketDriveStats {
source: result.source.clone(),
bucket: result.bucket.clone(),
drive: result.drive.clone(),
count: result.count,
age_seconds: result.age_seconds,
})
.collect::<Vec<_>>();
stats.sort_by(|left, right| {
left.source
.cmp(&right.source)
.then_with(|| left.bucket.cmp(&right.bucket))
.then_with(|| left.drive.cmp(&right.drive))
});
stats
}
pub async fn collect_scanner_metric_stats() -> Option<ScannerStats> {
collect_scanner_runtime_metric_stats().await.map(|stats| stats.stats)
}
@@ -1418,6 +1552,7 @@ pub(crate) async fn collect_scanner_runtime_metric_stats() -> Option<ScannerRunt
&runtime_details.current_cycle_bucket_drive_results,
),
last_cycle_bucket_drive_results: scanner_bucket_drive_result_stats(&runtime_details.last_cycle_bucket_drive_results),
active_bucket_drive_scans: scanner_active_bucket_drive_stats(&runtime_details.active_bucket_drive_scans),
stats: ScannerStats {
bucket_scans_finished,
bucket_scans_started,
@@ -1984,6 +2119,72 @@ mod tests {
assert_eq!(scanner_lifecycle_checked_versions(&report), 37);
}
#[test]
fn ilm_detail_stats_keep_expiry_and_transition_results_separate() {
let report = ScannerMetricsReport {
lifecycle_expiry: rustfs_common::metrics::ScannerLifecycleExpirySnapshot {
current_queued: 2,
current_active: 1,
scanner_queued: 10,
scanner_missed: 3,
delete_failed: 4,
..Default::default()
},
lifecycle_transition: rustfs_common::metrics::ScannerLifecycleTransitionSnapshot {
current_queued: 5,
current_active: 6,
queue_full: 7,
queue_send_timeout: 8,
scanner_queued: 11,
completed: 12,
failed: 13,
..Default::default()
},
..Default::default()
};
let queues = ilm_queue_task_stats(&report);
assert!(
queues
.iter()
.any(|task| task.action == "expiry" && task.state == "pending" && task.value == 2)
);
assert!(
queues
.iter()
.any(|task| task.action == "transition" && task.state == "active" && task.value == 6)
);
let events = ilm_task_event_stats(&report);
assert!(
events
.iter()
.any(|event| event.action == "expiry" && event.result == "failed" && event.value == 4)
);
assert!(
events
.iter()
.any(|event| event.action == "transition" && event.result == "completed" && event.value == 12)
);
assert!(
events
.iter()
.any(|event| event.action == "transition" && event.result == "failed" && event.value == 13)
);
let backpressure = ilm_backpressure_stats(&report);
assert!(
backpressure
.iter()
.any(|event| event.action == "transition" && event.reason == "queue_full" && event.value == 7)
);
assert!(
backpressure
.iter()
.any(|event| event.action == "transition" && event.reason == "send_timeout" && event.value == 8)
);
}
#[test]
fn scanner_source_work_stats_sorts_and_skips_empty_source() {
let stats = scanner_source_work_stats(&[
crates/scanner/src/scanner_io/guards.rs
+36 -2
View File
@@ -147,11 +147,19 @@ impl Drop for DiskBucketScanActiveGuard {
pub(super) struct BucketDriveFailureGuard {
failed: bool,
source: rustfs_common::metrics::ScannerWorkSource,
bucket: String,
drive: String,
}
impl BucketDriveFailureGuard {
pub(super) fn new() -> Self {
Self { failed: true }
pub(super) fn new(source: rustfs_common::metrics::ScannerWorkSource, bucket: &str, drive: &str) -> Self {
Self {
failed: true,
source,
bucket: bucket.to_string(),
drive: drive.to_string(),
}
}
pub(super) fn mark_not_failed(&mut self) {
@@ -161,6 +169,7 @@ impl BucketDriveFailureGuard {
impl Drop for BucketDriveFailureGuard {
fn drop(&mut self) {
global_metrics().record_scan_bucket_drive_end(self.source, &self.bucket, &self.drive);
if self.failed {
global_metrics().record_scan_bucket_drive_failure();
}
@@ -272,3 +281,28 @@ pub(super) fn record_set_scan_failure(first_err: &mut Option<Error>, err: Error)
pub(super) fn scanner_task_join_error(stage: &str, err: tokio::task::JoinError) -> Error {
Error::other(format!("{stage} task join failed: {err}"))
}
#[cfg(test)]
mod tests {
use super::*;
use rustfs_common::metrics::{ScannerWorkSource, global_metrics};
#[test]
fn bucket_drive_failure_guard_retires_active_scan_on_drop() {
let source = ScannerWorkSource::Usage;
let bucket = "__guard_active_lifecycle_test__";
let drive = "/__guard_active_lifecycle_test__";
global_metrics().record_scan_bucket_drive_start(source, bucket, drive);
{
let mut guard = BucketDriveFailureGuard::new(source, bucket, drive);
guard.mark_not_failed();
}
assert!(
!global_metrics()
.scanner_runtime_details_report()
.active_bucket_drive_scans
.iter()
.any(|active| active.source == source.as_str() && active.bucket == bucket && active.drive == drive)
);
}
}
crates/scanner/src/scanner_io/io_disk.rs
+11 -7
View File
@@ -147,8 +147,12 @@ impl ScannerIODisk for Disk {
let drive_start = std::time::Instant::now();
let bucket = cache.info.name.clone();
let disk_path = self.path().to_string_lossy().to_string();
global_metrics().record_scan_bucket_drive_start();
let mut failure_guard = BucketDriveFailureGuard::new();
let source = match scan_mode {
HealScanMode::Deep => rustfs_common::metrics::ScannerWorkSource::Bitrot,
HealScanMode::Normal | HealScanMode::Unknown => rustfs_common::metrics::ScannerWorkSource::Usage,
};
global_metrics().record_scan_bucket_drive_start(source, &bucket, &disk_path);
let mut failure_guard = BucketDriveFailureGuard::new(source, &bucket, &disk_path);
let _guard = self.start_scan();
let mut cache = cache;
@@ -196,32 +200,32 @@ impl ScannerIODisk for Disk {
match result {
Ok(mut data_usage_info) => {
done_drive();
emit_scan_bucket_drive_complete(true, &bucket, &disk_path, drive_start.elapsed());
emit_scan_bucket_drive_complete(source, true, &bucket, &disk_path, drive_start.elapsed());
data_usage_info.info.last_update = Some(SystemTime::now());
failure_guard.mark_not_failed();
Ok(ScannerDiskScanOutcome::Complete(data_usage_info))
}
Err(ScannerError::PartialCache(mut partial_cache)) => {
done_drive();
emit_scan_bucket_drive_partial(&bucket, &disk_path, drive_start.elapsed());
emit_scan_bucket_drive_partial(source, &bucket, &disk_path, drive_start.elapsed());
partial_cache.info.last_update.get_or_insert_with(SystemTime::now);
failure_guard.mark_not_failed();
Ok(ScannerDiskScanOutcome::Partial(*partial_cache))
}
Err(ScannerError::NamespaceNotFoundCache(mut partial_cache)) => {
done_drive();
emit_scan_bucket_drive_partial(&bucket, &disk_path, drive_start.elapsed());
emit_scan_bucket_drive_partial(source, &bucket, &disk_path, drive_start.elapsed());
partial_cache.info.last_update.get_or_insert_with(SystemTime::now);
failure_guard.mark_not_failed();
Ok(ScannerDiskScanOutcome::NamespaceNotFound(*partial_cache))
}
Err(e) => {
if ctx.is_cancelled() {
emit_scan_bucket_drive_partial(&bucket, &disk_path, drive_start.elapsed());
emit_scan_bucket_drive_partial(source, &bucket, &disk_path, drive_start.elapsed());
failure_guard.mark_not_failed();
} else {
done_drive();
emit_scan_bucket_drive_complete(false, &bucket, &disk_path, drive_start.elapsed());
emit_scan_bucket_drive_complete(source, false, &bucket, &disk_path, drive_start.elapsed());
}
Err(StorageError::other(format!("Failed to scan data folder: {e}")))
}