mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 08:18:18 +00:00
feat(ilm): expose lifecycle expiry scanner counters (#5262)
Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
@@ -786,6 +786,9 @@ pub struct Metrics {
|
||||
scanner_expiry_queue_missed: AtomicU64,
|
||||
scanner_expiry_queued_total: AtomicU64,
|
||||
scanner_expiry_missed_total: AtomicU64,
|
||||
scanner_expiry_blocked_total: AtomicU64,
|
||||
scanner_expiry_not_enqueued_total: AtomicU64,
|
||||
scanner_expiry_delete_failed_total: AtomicU64,
|
||||
scanner_transition_queue_capacity: AtomicU64,
|
||||
scanner_transition_queued: AtomicU64,
|
||||
scanner_transition_active: AtomicU64,
|
||||
@@ -1051,6 +1054,12 @@ pub struct ScannerLifecycleExpirySnapshot {
|
||||
pub queue_missed: u64,
|
||||
pub scanner_queued: u64,
|
||||
pub scanner_missed: u64,
|
||||
#[serde(default)]
|
||||
pub scanner_blocked: u64,
|
||||
#[serde(default)]
|
||||
pub scanner_not_enqueued: u64,
|
||||
#[serde(default)]
|
||||
pub delete_failed: u64,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
|
||||
@@ -1755,6 +1764,9 @@ impl Metrics {
|
||||
scanner_expiry_queue_missed: AtomicU64::new(0),
|
||||
scanner_expiry_queued_total: AtomicU64::new(0),
|
||||
scanner_expiry_missed_total: AtomicU64::new(0),
|
||||
scanner_expiry_blocked_total: AtomicU64::new(0),
|
||||
scanner_expiry_not_enqueued_total: AtomicU64::new(0),
|
||||
scanner_expiry_delete_failed_total: AtomicU64::new(0),
|
||||
scanner_transition_queue_capacity: AtomicU64::new(0),
|
||||
scanner_transition_queued: AtomicU64::new(0),
|
||||
scanner_transition_active: AtomicU64::new(0),
|
||||
@@ -1985,9 +1997,18 @@ impl Metrics {
|
||||
self.scanner_expiry_queued_total.fetch_add(count, Ordering::Relaxed);
|
||||
} else {
|
||||
self.scanner_expiry_missed_total.fetch_add(count, Ordering::Relaxed);
|
||||
self.scanner_expiry_not_enqueued_total.fetch_add(count, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
|
||||
pub fn record_scanner_expiry_blocked(&self, count: u64) {
|
||||
self.scanner_expiry_blocked_total.fetch_add(count, Ordering::Relaxed);
|
||||
}
|
||||
|
||||
pub fn record_scanner_expiry_delete_failed(&self, count: u64) {
|
||||
self.scanner_expiry_delete_failed_total.fetch_add(count, Ordering::Relaxed);
|
||||
}
|
||||
|
||||
pub fn record_scanner_transition_enqueue_result(&self, count: u64, queued: bool) {
|
||||
self.record_scanner_ilm_enqueue_result(count, queued);
|
||||
if queued {
|
||||
@@ -2850,6 +2871,9 @@ impl Metrics {
|
||||
queue_missed: self.scanner_expiry_queue_missed.load(Ordering::Relaxed),
|
||||
scanner_queued: self.scanner_expiry_queued_total.load(Ordering::Relaxed),
|
||||
scanner_missed: self.scanner_expiry_missed_total.load(Ordering::Relaxed),
|
||||
scanner_blocked: self.scanner_expiry_blocked_total.load(Ordering::Relaxed),
|
||||
scanner_not_enqueued: self.scanner_expiry_not_enqueued_total.load(Ordering::Relaxed),
|
||||
delete_failed: self.scanner_expiry_delete_failed_total.load(Ordering::Relaxed),
|
||||
};
|
||||
m.lifecycle_transition = ScannerLifecycleTransitionSnapshot {
|
||||
current_queue_capacity: self.scanner_transition_queue_capacity.load(Ordering::Relaxed),
|
||||
@@ -3385,6 +3409,8 @@ mod tests {
|
||||
});
|
||||
metrics.record_scanner_expiry_enqueue_result(6, true);
|
||||
metrics.record_scanner_expiry_enqueue_result(2, false);
|
||||
metrics.record_scanner_expiry_blocked(4);
|
||||
metrics.record_scanner_expiry_delete_failed(1);
|
||||
|
||||
let report = metrics.report().await;
|
||||
|
||||
@@ -3395,6 +3421,9 @@ mod tests {
|
||||
assert_eq!(report.lifecycle_expiry.queue_missed, 3);
|
||||
assert_eq!(report.lifecycle_expiry.scanner_queued, 6);
|
||||
assert_eq!(report.lifecycle_expiry.scanner_missed, 2);
|
||||
assert_eq!(report.lifecycle_expiry.scanner_blocked, 4);
|
||||
assert_eq!(report.lifecycle_expiry.scanner_not_enqueued, 2);
|
||||
assert_eq!(report.lifecycle_expiry.delete_failed, 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -212,6 +212,18 @@ fn record_scanner_lifecycle_enqueue_result(src: &LcEventSrc, count: u64, queued:
|
||||
}
|
||||
}
|
||||
|
||||
fn record_scanner_lifecycle_expiry_blocked(src: &LcEventSrc, count: u64) {
|
||||
if matches!(src, LcEventSrc::Scanner) {
|
||||
global_metrics().record_scanner_expiry_blocked(count);
|
||||
}
|
||||
}
|
||||
|
||||
fn record_scanner_lifecycle_expiry_delete_failed(src: &LcEventSrc, count: u64) {
|
||||
if matches!(src, LcEventSrc::Scanner) {
|
||||
global_metrics().record_scanner_expiry_delete_failed(count);
|
||||
}
|
||||
}
|
||||
|
||||
fn record_scanner_transition_enqueue_result(src: &LcEventSrc, count: u64, queued: bool) {
|
||||
if matches!(src, LcEventSrc::Scanner) {
|
||||
global_metrics().record_scanner_transition_enqueue_result(count, queued);
|
||||
@@ -845,6 +857,7 @@ impl ExpiryState {
|
||||
if deleted {
|
||||
trace.emit(EVENT_LIFECYCLE_DELETE_COMPLETED, "delete_completed", None);
|
||||
} else {
|
||||
record_scanner_lifecycle_expiry_delete_failed(&v.src, 1);
|
||||
trace.emit(
|
||||
EVENT_LIFECYCLE_DELETE_FAILED,
|
||||
"delete_failed",
|
||||
@@ -3495,6 +3508,7 @@ async fn enqueue_expiry_for_existing_object_group(
|
||||
}
|
||||
};
|
||||
if blocked_by_replication {
|
||||
record_scanner_lifecycle_expiry_blocked(context.src, 1);
|
||||
continue;
|
||||
}
|
||||
apply_existing_object_expiry(context.api.clone(), object, event, context.src).await;
|
||||
@@ -5504,6 +5518,7 @@ mod tests {
|
||||
assert_eq!(state.stats.missed_tasks(), 1);
|
||||
let after = global_metrics().report().await.lifecycle_expiry;
|
||||
assert!(after.scanner_missed >= before.scanner_missed.saturating_add(1));
|
||||
assert!(after.scanner_not_enqueued >= before.scanner_not_enqueued.saturating_add(1));
|
||||
let observed = observed.lock().expect("observability test events should not poison");
|
||||
assert!(observed.contains(&(EVENT_LIFECYCLE_EXPIRED_DETECTED, "detected", None)));
|
||||
assert!(observed.contains(&(EVENT_LIFECYCLE_NOT_ENQUEUED, "not_enqueued", Some("worker_unavailable"))));
|
||||
@@ -5607,6 +5622,8 @@ mod tests {
|
||||
assert!(!second);
|
||||
assert_eq!(state.stats.pending_tasks(), 1);
|
||||
assert_eq!(state.stats.missed_tasks(), 1);
|
||||
let after = global_metrics().report().await.lifecycle_expiry;
|
||||
assert!(after.scanner_not_enqueued >= 1);
|
||||
let observed = observed.lock().expect("observability test events should not poison");
|
||||
assert_eq!(
|
||||
observed
|
||||
|
||||
@@ -192,6 +192,9 @@ fn to_madmin_scanner_metrics(metrics: rustfs_common::metrics::ScannerMetricsRepo
|
||||
queue_missed: metrics.lifecycle_expiry.queue_missed,
|
||||
scanner_queued: metrics.lifecycle_expiry.scanner_queued,
|
||||
scanner_missed: metrics.lifecycle_expiry.scanner_missed,
|
||||
scanner_blocked: metrics.lifecycle_expiry.scanner_blocked,
|
||||
scanner_not_enqueued: metrics.lifecycle_expiry.scanner_not_enqueued,
|
||||
delete_failed: metrics.lifecycle_expiry.delete_failed,
|
||||
},
|
||||
usage_freshness: MadminScannerUsageFreshnessSnapshot {
|
||||
dirty_pending_buckets: metrics.usage_freshness.dirty_pending_buckets,
|
||||
@@ -646,6 +649,9 @@ mod test {
|
||||
queue_missed: 3,
|
||||
scanner_queued: 6,
|
||||
scanner_missed: 2,
|
||||
scanner_blocked: 4,
|
||||
scanner_not_enqueued: 2,
|
||||
delete_failed: 1,
|
||||
},
|
||||
lifecycle_transition: rustfs_common::metrics::ScannerLifecycleTransitionSnapshot {
|
||||
current_queue_capacity: 16,
|
||||
@@ -672,6 +678,9 @@ mod test {
|
||||
assert_eq!(scanner.lifecycle_expiry.queue_missed, 3);
|
||||
assert_eq!(scanner.lifecycle_expiry.scanner_queued, 6);
|
||||
assert_eq!(scanner.lifecycle_expiry.scanner_missed, 2);
|
||||
assert_eq!(scanner.lifecycle_expiry.scanner_blocked, 4);
|
||||
assert_eq!(scanner.lifecycle_expiry.scanner_not_enqueued, 2);
|
||||
assert_eq!(scanner.lifecycle_expiry.delete_failed, 1);
|
||||
assert_eq!(scanner.lifecycle_transition.current_queue_capacity, 16);
|
||||
assert_eq!(scanner.lifecycle_transition.current_queued, 5);
|
||||
assert_eq!(scanner.lifecycle_transition.current_active, 2);
|
||||
|
||||
@@ -468,6 +468,12 @@ pub struct ScannerLifecycleExpirySnapshot {
|
||||
pub scanner_queued: u64,
|
||||
#[serde(rename = "scanner_missed", default)]
|
||||
pub scanner_missed: u64,
|
||||
#[serde(rename = "scanner_blocked", default)]
|
||||
pub scanner_blocked: u64,
|
||||
#[serde(rename = "scanner_not_enqueued", default)]
|
||||
pub scanner_not_enqueued: u64,
|
||||
#[serde(rename = "delete_failed", default)]
|
||||
pub delete_failed: u64,
|
||||
}
|
||||
|
||||
impl ScannerLifecycleExpirySnapshot {
|
||||
@@ -479,6 +485,9 @@ impl ScannerLifecycleExpirySnapshot {
|
||||
self.queue_missed = self.queue_missed.saturating_add(other.queue_missed);
|
||||
self.scanner_queued = self.scanner_queued.saturating_add(other.scanner_queued);
|
||||
self.scanner_missed = self.scanner_missed.saturating_add(other.scanner_missed);
|
||||
self.scanner_blocked = self.scanner_blocked.saturating_add(other.scanner_blocked);
|
||||
self.scanner_not_enqueued = self.scanner_not_enqueued.saturating_add(other.scanner_not_enqueued);
|
||||
self.delete_failed = self.delete_failed.saturating_add(other.delete_failed);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1508,6 +1517,9 @@ mod tests {
|
||||
queue_missed: 3,
|
||||
scanner_queued: 5,
|
||||
scanner_missed: 2,
|
||||
scanner_blocked: 4,
|
||||
scanner_not_enqueued: 1,
|
||||
delete_failed: 1,
|
||||
},
|
||||
lifecycle_transition: ScannerLifecycleTransitionSnapshot {
|
||||
current_queue_capacity: 8,
|
||||
@@ -1541,6 +1553,9 @@ mod tests {
|
||||
queue_missed: 2,
|
||||
scanner_queued: 6,
|
||||
scanner_missed: 4,
|
||||
scanner_blocked: 7,
|
||||
scanner_not_enqueued: 5,
|
||||
delete_failed: 2,
|
||||
},
|
||||
lifecycle_transition: ScannerLifecycleTransitionSnapshot {
|
||||
current_queue_capacity: 4,
|
||||
@@ -1567,6 +1582,9 @@ mod tests {
|
||||
assert_eq!(scanner.lifecycle_expiry.queue_missed, 5);
|
||||
assert_eq!(scanner.lifecycle_expiry.scanner_queued, 11);
|
||||
assert_eq!(scanner.lifecycle_expiry.scanner_missed, 6);
|
||||
assert_eq!(scanner.lifecycle_expiry.scanner_blocked, 11);
|
||||
assert_eq!(scanner.lifecycle_expiry.scanner_not_enqueued, 6);
|
||||
assert_eq!(scanner.lifecycle_expiry.delete_failed, 3);
|
||||
assert_eq!(scanner.lifecycle_transition.current_queue_capacity, 12);
|
||||
assert_eq!(scanner.lifecycle_transition.current_queued, 5);
|
||||
assert_eq!(scanner.lifecycle_transition.current_active, 3);
|
||||
|
||||
Reference in New Issue
Block a user