feat: dependency-aware monitoring.

Author: uplg

CHANGELOG.md

@@ -9,6 +9,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **Dependency-aware alerting** (`depends_on`) and **display groups** (`group`)
+  on monitors. When a monitor goes down, the alert is annotated with topology
+  context: `"caused by X"` if an upstream it depends on is also down (symptom),
+  or `"impacts: A, B, C"` if all its upstreams are up (root cause with blast
+  radius). Every monitor still alerts independently — annotations are additive,
+  nothing is suppressed. The dependency graph is validated as a DAG at load
+  (Kahn's algorithm); cycles and unknown references are rejected. On the status
+  page, monitors are grouped by their `group` field under section headers. The
+  webhook payload carries structural `cause` and `impacted` fields.
 - **Mutual surveillance / dead-man's switch** via a `[health]` section and
   `[[peers]]`. A node emits an outbound heartbeat to each peer's `ping_url` only
   while it is locally healthy (scheduler ticking *and* database writable), so a

Makefile

@@ -0,0 +1,23 @@
+# Local quality gate — mirrors .github/workflows/ci.yml so you can run the exact
+# same checks before pushing. `make gate` must be green for CI to pass.
+.PHONY: gate fmt clippy deny test fix
+
+# The full gate: formatting, lints, license/advisory/ban checks, and tests.
+gate: fmt clippy deny test
+
+fmt:
+	cargo fmt --all -- --check
+
+clippy:
+	cargo clippy --workspace --all-targets --locked -- -D warnings
+
+deny:
+	cargo deny check
+
+test:
+	cargo test --workspace --locked
+
+# Auto-fix what can be fixed (formatting + machine-applicable clippy suggestions).
+fix:
+	cargo fmt --all
+	cargo clippy --workspace --all-targets --fix --allow-dirty --allow-staged

config.example.toml

@@ -112,6 +112,14 @@ default_retention_days = 90
 # For tcp:  target is "host:port"; up = TCP connect succeeds.
 # For push: no target; the job calls POST /api/push/{id} (see below).
 # TLS certificate expiry is checked automatically for https:// monitors.
+#
+# Topology (optional):
+#   group = "infra"              # display group on the status page
+#   depends_on = ["db", "cache"] # upstream monitors this one depends on
+# When a monitor goes down, the alert is annotated:
+#   - "caused by X" if an upstream it depends on is also down (symptom)
+#   - "impacts: A, B, C" if all its upstreams are up (root cause, blast radius)
+# The dependency graph must be acyclic (validated at load).
 
 [[monitors]]
 id = "website"
@@ -120,6 +128,8 @@ kind = "http"
 target = "https://example.com"
 interval_secs = 60
 timeout_secs = 10
+group = "app"
+depends_on = ["api"]
 # expected_status = 200   # optional; default: any 2xx counts as up
 # degraded_over_ms = 800  # optional; up but slower than this = degraded (yellow)
 # slo_latency_ms = 500    # optional; the 24h p95 is flagged met/breached against this
@@ -136,6 +146,8 @@ kind = "http"
 target = "https://api.example.com/health"
 interval_secs = 30
 timeout_secs = 10
+group = "app"
+depends_on = ["database"]
 # Body assertions (http only):
 # keyword = "operational"   # body must contain this (keyword_invert = true → must NOT)
 # json_query = "$.status"   # JSONPath (RFC 9535) evaluated against a JSON body
@@ -149,6 +161,7 @@ kind = "tcp"
 target = "db.example.com:5432"
 interval_secs = 60
 timeout_secs = 5
+group = "infra"
 
 # Heartbeat / push monitor: down if no ping arrives within `interval_secs`.
 # The job calls: curl -fsS -X POST \

crates/hora-core/src/config.rs

@@ -557,6 +557,16 @@ pub struct Monitor {
     /// Override how long this monitor's checks are kept before pruning.
     #[serde(default)]
     pub retention_days: Option<u16>,
+    /// Display group for the status page (monitors sharing a group are rendered
+    /// together under a section header). Ungrouped monitors appear last.
+    #[serde(default)]
+    pub group: Option<String>,
+    /// Upstream monitor ids this one depends on. When set, a down alert is
+    /// annotated with the probable upstream cause (if any upstream is also
+    /// down) or with the dependents it impacts (if this is a root cause).
+    /// Validated as a DAG at load time.
+    #[serde(default)]
+    pub depends_on: Option<Vec<String>>,
 }
 
 // Manual `Debug` (rather than derived) so credentials never leak: `target` and
@@ -586,6 +596,8 @@ impl std::fmt::Debug for Monitor {
             .field("push_token", &self.push_token)
             .field("check_cert", &self.check_cert)
             .field("retention_days", &self.retention_days)
+            .field("group", &self.group)
+            .field("depends_on", &self.depends_on)
             .finish()
     }
 }
@@ -887,6 +899,8 @@ fn validate(config: &Config) -> anyhow::Result<()> {
         }
     }
 
+    crate::topology::validate_dag(&config.monitors)?;
+
     validate_peers(config, &seen, &channel_names)?;
     Ok(())
 }
@@ -1780,4 +1794,76 @@ mod tests {
         assert!(!dump.contains("sup3rsecret"), "url secret leaked: {dump}");
         assert!(!dump.contains("tok3n"), "ping token leaked: {dump}");
     }
+
+    #[test]
+    fn parses_group_and_depends_on() {
+        let config = parse(
+            r#"
+            [page]
+            [server]
+            [[monitors]]
+            id = "db"
+            name = "DB"
+            kind = "tcp"
+            target = "db:5432"
+            interval_secs = 30
+            group = "infra"
+            [[monitors]]
+            id = "api"
+            name = "API"
+            target = "https://api.example"
+            interval_secs = 30
+            group = "app"
+            depends_on = ["db"]
+        "#,
+        );
+        assert_eq!(config.monitors[0].group.as_deref(), Some("infra"));
+        assert_eq!(
+            config.monitors[1].depends_on.as_deref(),
+            Some(&vec!["db".to_owned()][..])
+        );
+        validate(&config).expect("valid topology");
+    }
+
+    #[test]
+    fn rejects_depends_on_unknown_monitor() {
+        let config = parse(
+            r#"
+            [page]
+            [server]
+            [[monitors]]
+            id = "api"
+            name = "API"
+            target = "https://api.example"
+            interval_secs = 30
+            depends_on = ["ghost"]
+        "#,
+        );
+        let error = validate(&config).unwrap_err().to_string();
+        assert!(error.contains("unknown monitor"), "got: {error}");
+    }
+
+    #[test]
+    fn rejects_dependency_cycle() {
+        let config = parse(
+            r#"
+            [page]
+            [server]
+            [[monitors]]
+            id = "a"
+            name = "A"
+            target = "https://a.example"
+            interval_secs = 30
+            depends_on = ["b"]
+            [[monitors]]
+            id = "b"
+            name = "B"
+            target = "https://b.example"
+            interval_secs = 30
+            depends_on = ["a"]
+        "#,
+        );
+        let error = validate(&config).unwrap_err().to_string();
+        assert!(error.contains("cycle"), "got: {error}");
+    }
 }

crates/hora-core/src/db.rs

@@ -451,6 +451,27 @@ pub async fn cert_all(pool: &SqlitePool) -> sqlx::Result<HashMap<String, i64>> {
     Ok(rows.into_iter().collect())
 }
 
+/// Current status from the recent checks (newest first): a single failure only
+/// counts as `degraded` until `threshold` consecutive failures confirm `down`.
+#[must_use]
+pub fn derive_status(recent: &[Latest], threshold: i64) -> &'static str {
+    let Some(latest) = recent.first() else {
+        return "unknown";
+    };
+    match latest.status {
+        1 => "up",
+        2 => "degraded",
+        _ => {
+            let needed = usize::try_from(threshold).unwrap_or(usize::MAX);
+            if recent.len() >= needed && recent.iter().all(|check| check.status == 0) {
+                "down"
+            } else {
+                "degraded"
+            }
+        }
+    }
+}
+
 /// Background task: periodically prune each monitor's history to its retention,
 /// and drop any data left behind by monitors removed from the config. A shutdown
 /// signal lets it stop between ticks instead of being aborted.

crates/hora-core/src/lib.rs

@@ -9,6 +9,7 @@ pub mod peer;
 pub mod probe;
 pub mod scheduler;
 pub mod supervisor;
+pub mod topology;
 
 /// Seconds in a day (UTC), shared across the time-bucketing logic.
 pub const SECONDS_PER_DAY: i64 = 86_400;

crates/hora-core/src/peer.rs

@@ -351,6 +351,8 @@ pub(crate) fn spawn_watch(
                         Event::Down {
                             monitor: &peer.name,
                             error: outcome.error.as_deref(),
+                            cause: None,
+                            impacted: &[],
                         },
                     )
                     .await;

crates/hora-core/src/probe.rs

@@ -296,6 +296,8 @@ mod tests {
             push_token: None,
             check_cert: None,
             retention_days: None,
+            group: None,
+            depends_on: None,
         }
     }
 

crates/hora-core/src/scheduler.rs

@@ -13,6 +13,7 @@ use tracing::{error, info, warn};
 use crate::config::{Config, Kind, Monitor};
 use crate::notifications::Notifiers;
 use crate::probe::Outcome;
+use crate::topology;
 use crate::{db, probe};
 
 /// The level a monitor was most recently alerted at, so we never re-alert the
@@ -111,12 +112,18 @@ async fn run(
             consecutive_degraded = 0;
             if consecutive_down >= threshold && alerted != AlertLevel::Down {
                 error!(monitor = %monitor.id, failures = consecutive_down, "confirmed down");
+                let snapshot = config.borrow().clone();
+                let (cause, impacted_names) =
+                    down_context(&snapshot, &pool, &monitor, threshold).await;
+                let impacted_refs: Vec<&str> = impacted_names.iter().map(String::as_str).collect();
                 notifier
                     .load_full()
                     .dispatch(
                         Event::Down {
                             monitor: &monitor.name,
                             error: outcome.error.as_deref(),
+                            cause: cause.as_deref(),
+                            impacted: &impacted_refs,
                         },
                         monitor.notify.as_deref(),
                     )
@@ -169,6 +176,39 @@ async fn heartbeat_outcome(pool: &SqlitePool, monitor: &Monitor) -> Option<Outco
     heartbeat_outcome_for(pool, &monitor.id, monitor.interval_secs).await
 }
 
+/// Compute topology annotation for a down alert: the nearest down upstream
+/// (`cause`) if any, or the list of impacted dependents (`impacted`) if this
+/// monitor is a root cause. Returns `(None, vec![])` when the monitor has no
+/// topology configured.
+async fn down_context(
+    config: &Config,
+    pool: &SqlitePool,
+    monitor: &Monitor,
+    threshold: u32,
+) -> (Option<String>, Vec<String>) {
+    let threshold_i64 = i64::from(threshold.max(1));
+
+    let upstreams = topology::transitive_upstreams(&config.monitors, &monitor.id);
+    for up_id in &upstreams {
+        let Ok(recent) = db::recent_checks(pool, up_id, threshold_i64).await else {
+            continue;
+        };
+        if db::derive_status(&recent, threshold_i64) == "down"
+            && let Some(name) = topology::monitor_name(&config.monitors, up_id)
+        {
+            return (Some(name.to_owned()), Vec::new());
+        }
+    }
+
+    let dependents = topology::transitive_dependents(&config.monitors, &monitor.id);
+    let impacted: Vec<String> = dependents
+        .iter()
+        .filter_map(|dep_id| topology::monitor_name(&config.monitors, dep_id).map(String::from))
+        .collect();
+
+    (None, impacted)
+}
+
 /// Evaluate a heartbeat from the stored pings for `id` against `interval_secs`:
 /// down (and record it) when none arrived within the interval, up when one did.
 /// `None` means no heartbeat yet (or a read error), leaving the status unknown -