use core::time::Duration;
use crate::model::{Confidence, MeasuredValue, Measurement, MetricState, Severity, SystemSnapshot};
use crate::units::{Percent, format_age, format_duration};
use super::super::{DiagnosticRule, Evidence, Finding, HistoryWindow, Thresholds};
use super::{as_count, as_percent, ratio};
pub const COLLECTOR_BEHIND: &str = "collector.falling_behind";
pub const SNAPSHOT_STALE: &str = "collector.snapshot_stale";
pub const SELF_OVERHEAD: &str = "self.overhead_above_budget";
const CRITICAL_BUDGET_MULTIPLE: f64 = 2.0;
#[derive(Clone, Copy, Debug)]
pub struct CollectorBehindRule {
thresholds: Thresholds,
}
impl CollectorBehindRule {
#[must_use]
pub const fn new(thresholds: Thresholds) -> Self {
Self { thresholds }
}
}
impl DiagnosticRule for CollectorBehindRule {
fn id(&self) -> &'static str {
COLLECTOR_BEHIND
}
fn evaluate(&self, current: &SystemSnapshot, history: &HistoryWindow<'_>) -> Option<Finding> {
let interval = history.expected_interval();
if interval.is_zero() {
return None;
}
let lag = current.health.lag.as_secs_f64();
let watch = Thresholds::intervals_as_seconds(
interval,
self.thresholds.collector_lag_watch_intervals,
);
let critical = Thresholds::intervals_as_seconds(
interval,
self.thresholds.collector_lag_critical_intervals,
);
let severity = super::escalate(lag >= watch, lag >= critical)?;
let evidence = vec![
Evidence::current(Measurement::new(
"lag",
MeasuredValue::Duration(current.health.lag),
)),
Evidence::current(Measurement::new(
"sample interval",
MeasuredValue::Duration(interval),
)),
Evidence::current(Measurement::new(
"fast collection p95",
MeasuredValue::Duration(current.health.fast.p95_duration),
)),
Evidence::current(Measurement::new(
"dropped samples",
MeasuredValue::Count(current.health.dropped_samples),
)),
Evidence::current(Measurement::new(
"coalesced samples",
MeasuredValue::Count(current.health.coalesced_samples),
)),
];
let intervals = ratio(lag, interval.as_secs_f64()).map_or_else(String::new, |value| {
format!(" ({value:.1} sample intervals)")
});
let summary = format!(
"The newest snapshot is {} behind live{intervals}. Displayed values are real \
measurements taken later than intended, not predictions; expensive enrichment is \
reduced before samples are dropped.",
format_age(current.health.lag),
);
Some(
Finding::new(
COLLECTOR_BEHIND,
severity,
"Collector falling behind",
summary,
Confidence::High,
)
.with_evidence(evidence),
)
}
}
#[derive(Clone, Copy, Debug)]
pub struct SnapshotStaleRule {
thresholds: Thresholds,
}
impl SnapshotStaleRule {
#[must_use]
pub const fn new(thresholds: Thresholds) -> Self {
Self { thresholds }
}
}
fn stale_headline_metrics(snapshot: &SystemSnapshot) -> (usize, Duration) {
let mut count = 0usize;
let mut oldest = Duration::ZERO;
let mut note = |age: Option<Duration>| {
if let Some(age) = age {
count = count.saturating_add(1);
oldest = oldest.max(age);
}
};
note(stale_age(&snapshot.cpu.total));
note(stale_age(&snapshot.memory.available));
note(stale_age(&snapshot.memory.used));
note(stale_age(&snapshot.memory.swap.used));
note(stale_age(&snapshot.load));
for disk in &snapshot.disks {
note(stale_age(&disk.read));
note(stale_age(&disk.write));
}
for interface in &snapshot.networks {
note(stale_age(&interface.rx));
note(stale_age(&interface.tx));
}
(count, oldest)
}
fn stale_age<T>(state: &MetricState<T>) -> Option<Duration> {
match state {
MetricState::Stale { age, .. } => Some(*age),
_ => None,
}
}
impl DiagnosticRule for SnapshotStaleRule {
fn id(&self) -> &'static str {
SNAPSHOT_STALE
}
fn evaluate(&self, current: &SystemSnapshot, history: &HistoryWindow<'_>) -> Option<Finding> {
let interval = history.expected_interval();
if interval.is_zero() {
return None;
}
let watch =
Thresholds::intervals_as_seconds(interval, self.thresholds.stale_watch_intervals);
let critical =
Thresholds::intervals_as_seconds(interval, self.thresholds.stale_critical_intervals);
let (stale_count, oldest) = stale_headline_metrics(current);
let gap = if current.has_valid_interval() {
current.elapsed
} else {
Duration::ZERO
};
let worst = oldest.max(gap).as_secs_f64();
let triggered = stale_count > 0 || gap.as_secs_f64() >= watch;
if !triggered || worst < watch {
return None;
}
let severity = if worst >= critical {
Severity::Critical
} else {
Severity::Watch
};
let mut evidence = vec![
Evidence::current(Measurement::new(
"sample interval",
MeasuredValue::Duration(interval),
)),
Evidence::current(Measurement::new(
"interval since previous sample",
MeasuredValue::Duration(gap),
)),
Evidence::current(Measurement::new(
"stale headline metrics",
MeasuredValue::Count(as_count(stale_count)),
)),
];
if !oldest.is_zero() {
evidence.push(Evidence::current(Measurement::new(
"oldest retained value",
MeasuredValue::Duration(oldest),
)));
}
let mut summary = String::new();
if stale_count > 0 {
summary.push_str(&format!(
"{stale_count} headline metric(s) are showing retained values, the oldest {} old. ",
format_age(oldest)
));
}
if gap.as_secs_f64() >= watch {
summary.push_str(&format!(
"The interval since the previous sample was {}, against a configured {}. ",
format_duration(gap),
format_duration(interval)
));
}
summary.push_str(
"Readings either side of a gap are not comparable, and rates across it were not \
computed from an assumed interval.",
);
Some(
Finding::new(
SNAPSHOT_STALE,
severity,
"Snapshot data stale",
summary,
Confidence::High,
)
.with_evidence(evidence),
)
}
}
#[derive(Clone, Copy, Debug)]
pub struct SelfOverheadRule {
thresholds: Thresholds,
}
impl SelfOverheadRule {
#[must_use]
pub const fn new(thresholds: Thresholds) -> Self {
Self { thresholds }
}
}
impl DiagnosticRule for SelfOverheadRule {
fn id(&self) -> &'static str {
SELF_OVERHEAD
}
fn evaluate(&self, current: &SystemSnapshot, _history: &HistoryWindow<'_>) -> Option<Finding> {
let overhead = current.health.self_overhead.as_ref()?;
let thresholds = &self.thresholds;
let cpu_budget = f64::from(thresholds.self_cpu_budget_percent);
let rss_budget = thresholds.self_rss_budget_bytes as f64;
let sample_budget = thresholds.self_sample_budget();
let cpu_over =
ratio(f64::from(overhead.cpu.value()), cpu_budget).filter(|over| *over >= 1.0);
let rss_over = ratio(overhead.rss_bytes as f64, rss_budget).filter(|over| *over >= 1.0);
let sample_over = ratio(
current.health.fast.p95_duration.as_secs_f64(),
sample_budget.as_secs_f64(),
)
.filter(|over| *over >= 1.0);
let exceeded: Vec<(&str, f64)> = [
("cpu", cpu_over),
("resident memory", rss_over),
("sample duration", sample_over),
]
.into_iter()
.filter_map(|(label, over)| over.map(|over| (label, over)))
.collect();
let worst = exceeded
.iter()
.map(|(_, over)| *over)
.fold(0.0f64, f64::max);
if exceeded.is_empty() {
return None;
}
let severity = if worst >= CRITICAL_BUDGET_MULTIPLE {
Severity::Critical
} else {
Severity::Watch
};
let mut evidence = vec![
Evidence::current(Measurement::new(
"self cpu",
MeasuredValue::Percent(overhead.cpu),
)),
Evidence::current(Measurement::new(
"self cpu budget",
MeasuredValue::Percent(as_percent(thresholds.self_cpu_budget_percent)),
)),
Evidence::current(Measurement::new(
"self resident memory",
MeasuredValue::Bytes(overhead.rss_bytes),
)),
Evidence::current(Measurement::new(
"self resident memory budget",
MeasuredValue::Bytes(thresholds.self_rss_budget_bytes),
)),
Evidence::current(Measurement::new(
"history bytes",
MeasuredValue::Bytes(overhead.history_bytes),
)),
Evidence::current(Measurement::new(
"fast collection p95",
MeasuredValue::Duration(current.health.fast.p95_duration),
)),
Evidence::current(Measurement::new(
"fast collection budget",
MeasuredValue::Duration(sample_budget),
)),
];
if let Some(&open_files) = overhead.open_files.fresh() {
evidence.push(Evidence::current(Measurement::new(
"open files",
MeasuredValue::Count(u64::from(open_files)),
)));
}
let named: Vec<String> = exceeded
.iter()
.map(|(label, over)| format!("{label} {over:.1}x budget"))
.collect();
let summary = format!(
"monitrs is over its own budget: {}. A monitor that costs more than what it measures \
is a bug in the monitor, not a property of the system.",
named.join(", ")
);
Some(
Finding::new(
SELF_OVERHEAD,
severity,
"monitrs self-overhead above budget",
summary,
Confidence::High,
)
.with_evidence(evidence),
)
}
}
#[must_use]
pub fn budget_share(measured: f64, budget: f64) -> Option<Percent> {
let share = ratio(measured, budget)?;
#[allow(clippy::cast_possible_truncation)]
let percent = (share * 100.0) as f32;
Percent::new(percent)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::diagnostics::fixtures::{
Timeline, set_cpu, set_health, set_self_overhead, snapshot,
};
const MIB: u64 = 1024 * 1024;
fn behind_rule() -> CollectorBehindRule {
CollectorBehindRule::new(Thresholds::default().sanitized())
}
fn stale_rule() -> SnapshotStaleRule {
SnapshotStaleRule::new(Thresholds::default().sanitized())
}
fn overhead_rule() -> SelfOverheadRule {
SelfOverheadRule::new(Thresholds::default().sanitized())
}
fn timeline() -> Timeline {
Timeline::new(Duration::from_secs(1))
}
#[test]
fn a_collector_keeping_up_produces_no_finding() {
let mut timeline = timeline();
let current = timeline.push_many(5, |snapshot| {
set_cpu(snapshot, 10.0);
set_health(
snapshot,
Duration::from_millis(120),
Duration::from_millis(40),
);
});
assert!(
behind_rule()
.evaluate(¤t, &timeline.window())
.is_none()
);
}
#[test]
fn lag_beyond_two_intervals_is_a_watch() {
let mut timeline = timeline();
let current = timeline.push_many(5, |snapshot| {
set_health(
snapshot,
Duration::from_millis(2_500),
Duration::from_millis(400),
);
});
let finding = behind_rule()
.evaluate(¤t, &timeline.window())
.expect("2.5s of lag on a 1s interval");
assert_eq!(finding.severity, Severity::Watch);
assert_eq!(finding.confidence, Confidence::High);
assert!(
finding.summary.contains("2.5 sample intervals"),
"{}",
finding.summary
);
}
#[test]
fn lag_beyond_five_intervals_is_critical() {
let mut timeline = timeline();
let current = timeline.push_many(5, |snapshot| {
set_health(snapshot, Duration::from_secs(9), Duration::from_millis(900));
});
let finding = behind_rule()
.evaluate(¤t, &timeline.window())
.expect("9s of lag on a 1s interval");
assert_eq!(finding.severity, Severity::Critical);
let labels: Vec<&str> = finding
.evidence
.iter()
.map(|item| item.measurement.label)
.collect();
assert!(labels.contains(&"lag"), "{labels:?}");
assert!(labels.contains(&"dropped samples"), "{labels:?}");
assert!(labels.contains(&"coalesced samples"), "{labels:?}");
}
#[test]
fn lag_is_judged_against_the_configured_interval_not_one_second() {
let mut slow = Timeline::new(Duration::from_secs(5));
let current = slow.push_many(3, |snapshot| {
set_health(
snapshot,
Duration::from_millis(2_500),
Duration::from_millis(400),
);
});
assert!(
behind_rule().evaluate(¤t, &slow.window()).is_none(),
"§8.1 forbids assuming a one-second interval"
);
}
#[test]
fn a_fresh_snapshot_is_not_stale() {
let mut timeline = timeline();
let current = timeline.push_many(5, |snapshot| set_cpu(snapshot, 10.0));
assert!(
stale_rule()
.evaluate(¤t, &timeline.window())
.is_none()
);
}
#[test]
fn retained_headline_values_are_reported_with_their_age() {
let mut timeline = timeline();
let mut current = timeline.push_many(5, |snapshot| set_cpu(snapshot, 10.0));
current.cpu.total = current.cpu.total.into_stale(Duration::from_secs(4));
let finding = stale_rule()
.evaluate(¤t, &timeline.window())
.expect("a retained value four seconds old");
assert_eq!(finding.severity, Severity::Watch);
assert!(
finding.summary.contains("retained values"),
"{}",
finding.summary
);
let count = finding
.evidence
.iter()
.find(|item| item.measurement.label == "stale headline metrics")
.expect("the count is evidence");
assert_eq!(count.measurement.value, MeasuredValue::Count(1));
}
#[test]
fn a_briefly_retained_value_is_not_worth_a_warning() {
let mut timeline = timeline();
let mut current = timeline.push_many(5, |snapshot| set_cpu(snapshot, 10.0));
current.cpu.total = current.cpu.total.into_stale(Duration::from_millis(1_500));
assert!(
stale_rule()
.evaluate(¤t, &timeline.window())
.is_none(),
"one and a half intervals is within tolerance"
);
}
#[test]
fn a_long_gap_between_samples_is_reported_as_stale_data() {
let mut timeline = timeline();
timeline.push_many(3, |snapshot| set_cpu(snapshot, 10.0));
let mut current = timeline.build(|snapshot| set_cpu(snapshot, 10.0));
current.elapsed = Duration::from_secs(30);
let finding = stale_rule()
.evaluate(¤t, &timeline.window())
.expect("a thirty second gap on a one second interval");
assert_eq!(finding.severity, Severity::Critical);
assert!(finding.summary.contains("30s"), "{}", finding.summary);
assert!(
finding.summary.contains("not comparable"),
"{}",
finding.summary
);
}
#[test]
fn the_first_snapshot_is_warming_up_rather_than_stale() {
let timeline = timeline();
let current = snapshot();
assert!(!current.has_valid_interval());
assert!(
stale_rule()
.evaluate(¤t, &timeline.window())
.is_none()
);
}
#[test]
fn no_measured_overhead_produces_no_finding() {
let mut timeline = timeline();
let current = timeline.push_many(3, |snapshot| set_cpu(snapshot, 10.0));
assert!(current.health.self_overhead.is_none());
assert!(
overhead_rule()
.evaluate(¤t, &timeline.window())
.is_none()
);
}
#[test]
fn overhead_inside_budget_produces_no_finding() {
let mut timeline = timeline();
let current = timeline.push_many(3, |snapshot| {
set_health(snapshot, Duration::ZERO, Duration::from_millis(80));
set_self_overhead(snapshot, 0.8, 30 * MIB);
});
assert!(
overhead_rule()
.evaluate(¤t, &timeline.window())
.is_none()
);
}
#[test]
fn our_own_cpu_above_budget_is_reported_against_the_budget() {
let mut timeline = timeline();
let current = timeline.push_many(3, |snapshot| {
set_health(snapshot, Duration::ZERO, Duration::from_millis(80));
set_self_overhead(snapshot, 3.0, 30 * MIB);
});
let finding = overhead_rule()
.evaluate(¤t, &timeline.window())
.expect("3% against a 2% budget");
assert_eq!(finding.severity, Severity::Watch);
assert!(
finding.summary.contains("cpu 1.5x budget"),
"{}",
finding.summary
);
let labels: Vec<&str> = finding
.evidence
.iter()
.map(|item| item.measurement.label)
.collect();
assert!(labels.contains(&"self cpu budget"), "{labels:?}");
assert!(labels.contains(&"history bytes"), "{labels:?}");
assert!(labels.contains(&"open files"), "{labels:?}");
}
#[test]
fn double_the_budget_is_critical_and_names_every_breach() {
let mut timeline = timeline();
let current = timeline.push_many(3, |snapshot| {
set_health(snapshot, Duration::ZERO, Duration::from_millis(600));
set_self_overhead(snapshot, 9.0, 120 * MIB);
});
let finding = overhead_rule()
.evaluate(¤t, &timeline.window())
.expect("every budget is exceeded");
assert_eq!(finding.severity, Severity::Critical);
for expected in ["cpu", "resident memory", "sample duration"] {
assert!(finding.summary.contains(expected), "{}", finding.summary);
}
}
#[test]
fn a_slow_collection_alone_is_enough_to_report_overhead() {
let mut timeline = timeline();
let current = timeline.push_many(3, |snapshot| {
set_health(snapshot, Duration::ZERO, Duration::from_millis(250));
set_self_overhead(snapshot, 0.5, 20 * MIB);
});
let finding = overhead_rule()
.evaluate(¤t, &timeline.window())
.expect("250ms against a 200ms budget");
assert!(
finding.summary.contains("sample duration"),
"{}",
finding.summary
);
}
#[test]
fn a_budget_share_of_a_zero_budget_is_undefined_rather_than_infinite() {
assert!(budget_share(1.0, 0.0).is_none());
let share = budget_share(1.0, 4.0).expect("a quarter of the budget");
assert!((share.value() - 25.0).abs() < f32::EPSILON);
}
}