Skip to main content

max / makenotwork

12.0 KB · 254 lines History Blame Raw
1 //! Background health and SSH-banner check tasks, including the transition gate,
2 //! incident bookkeeping, and latency-drift alerting.
3
4 use tokio::task::JoinHandle;
5 use tracing::info;
6
7 use pom::alerts::Alerter;
8 use pom::checks::{drift, http, ssh_banner};
9 use pom::config::Config;
10 use pom::db;
11 use pom::types::{HealthStatus, LatencyStats};
12
13 use super::super::incident::{IncidentAction, incident_action};
14 use super::super::transition::TransitionGate;
15 use super::{CheckInterval, configured_targets};
16
17 pub(crate) fn spawn_health_tasks(
18 config: &Config,
19 pool: &sqlx::SqlitePool,
20 cancel: &tokio_util::sync::CancellationToken,
21 alerter: Option<&Alerter>,
22 ) -> Vec<JoinHandle<()>> {
23 let default_interval = config.serve.interval_secs;
24 let confirmations = config.serve.confirmations;
25 let mut handles = Vec::new();
26
27 for (name, target_config) in configured_targets(config) {
28 if let Some(health_config) = target_config.health {
29 let interval_secs = health_config.interval_secs.unwrap_or(default_interval);
30 let pool = pool.clone();
31 let name = name.clone();
32 let label = target_config.label.clone();
33 let alerter = alerter.cloned();
34 let cancel = cancel.clone();
35 let trending_config = health_config.trending.clone();
36
37 info!("{name}: health check every {interval_secs}s");
38
39 handles.push(tokio::spawn(async move {
40 let mut ticks = CheckInterval::new(interval_secs, cancel);
41 let expect = health_config.expect.as_ref();
42 let mut in_drift = false;
43 // Seed the N-of-M transition gate from the last stored status so a
44 // restart of a persistently-degraded target doesn't re-fire, and a
45 // transient blip is debounced before it pages/opens an incident.
46 let seed = db::get_latest_health(&pool, &name)
47 .await
48 .ok()
49 .flatten()
50 .map(|s| s.status);
51 let mut gate = TransitionGate::seeded(seed, confirmations);
52
53 // RECONCILE THE INCIDENT LEDGER AGAINST THE SEED, before the
54 // first tick.
55 //
56 // Seeding is what stops a restart re-firing a transition, and it
57 // is also how a restart can LOSE one. An incident closes on a
58 // confirmed transition back to Operational; if the process
59 // restarts across that recovery, the gate seeds Operational,
60 // observes Operational, reports no transition, and the incident
61 // stays open forever with nothing to reopen the question.
62 //
63 // Measured on astra 2026-08-23: incident 307, `mt`, opened
64 // 2026-08-16 16:57 with `ended_at` still NULL a week later. Two
65 // sibling incidents opened in the same minute by the same blip
66 // (`mnw` 308, `htpy` 309) both closed an hour later, and mt's
67 // health had been Operational at 100% uptime throughout. The
68 // instance had read `failed` ever since, which is a monitor
69 // reporting its own bookkeeping as an outage.
70 //
71 // Idempotent and once per target per start, so it costs one
72 // query at boot and nothing afterwards. Deliberately does NOT
73 // alert: this closes a stale record, and a recovery notice for
74 // an outage that ended a week ago is noise.
75 if seed == Some(HealthStatus::Operational) {
76 match db::close_open_incidents(&pool, &name).await {
77 Ok(0) => {}
78 Ok(n) => info!(
79 "{name}: closed {n} incident(s) left open across a restart; health reads operational"
80 ),
81 Err(e) => {
82 tracing::error!("{name}: failed to reconcile open incidents: {e}");
83 }
84 }
85 }
86 while ticks.next().await {
87 let snapshot = http::check_health(&name, &health_config, expect).await;
88 info!(
89 "{}: {} ({}ms)",
90 name, snapshot.status, snapshot.response_time_ms
91 );
92 if let Err(e) = db::insert_health_check(&pool, &snapshot).await {
93 tracing::error!("{name}: failed to store health check: {e}");
94 }
95
96 // Alert AND open/close incidents only on a *confirmed*
97 // transition (debounced + restart-seeded above).
98 if let Some((from_status, to_status)) = gate.observe(snapshot.status) {
99 if let Some(ref alerter) = alerter {
100 let from = from_status.to_string();
101 let to = to_status.to_string();
102 if to_status == HealthStatus::Operational {
103 alerter.send_health_recovery(&name, &label, &from).await;
104 } else {
105 alerter
106 .send_health_alert(
107 &name,
108 &label,
109 &from,
110 &to,
111 snapshot.error.as_deref(),
112 )
113 .await;
114 }
115 }
116
117 match incident_action(from_status, to_status) {
118 IncidentAction::None => {}
119 IncidentAction::Open => {
120 if let Err(e) = db::insert_incident(
121 &pool,
122 &name,
123 &from_status.to_string(),
124 &to_status.to_string(),
125 )
126 .await
127 {
128 tracing::error!("{name}: failed to open incident: {e}");
129 }
130 }
131 IncidentAction::Close => {
132 if let Err(e) = db::close_open_incidents(&pool, &name).await {
133 tracing::error!("{name}: failed to close incidents: {e}");
134 }
135 }
136 IncidentAction::CloseAndOpen => {
137 if let Err(e) = db::close_and_open_incident(
138 &pool,
139 &name,
140 &from_status.to_string(),
141 &to_status.to_string(),
142 )
143 .await
144 {
145 tracing::error!("{name}: failed to close+open incident: {e}");
146 }
147 }
148 }
149 }
150
151 // Latency drift detection
152 if let Some(ref trending) = trending_config
153 && snapshot.status == HealthStatus::Operational
154 {
155 let baseline_cutoff = (chrono::Utc::now()
156 - chrono::Duration::hours(trending.baseline_window_hours as i64))
157 .to_rfc3339();
158 let baseline_data = db::get_response_times(&pool, &name, &baseline_cutoff)
159 .await
160 .unwrap_or_default();
161 let operational_times: Vec<i64> = baseline_data
162 .iter()
163 .filter(|(_, ms)| *ms > 0)
164 .map(|(_, ms)| *ms)
165 .collect();
166 let baseline = LatencyStats::from_times(&operational_times);
167 let recent = db::get_recent_response_times(&pool, &name, 3)
168 .await
169 .unwrap_or_default();
170
171 if let Some(ref bl) = baseline {
172 if let Some(msg) =
173 drift::detect_latency_drift(&recent, bl, trending.spike_threshold)
174 {
175 if !in_drift {
176 info!("{name}: {msg}");
177 if let Some(ref alerter) = alerter {
178 alerter.send_latency_drift_alert(&name, &label, &msg).await;
179 }
180 in_drift = true;
181 }
182 } else if in_drift {
183 info!("{name}: latency drift recovered");
184 if let Some(ref alerter) = alerter {
185 alerter.send_latency_recovery(&name, &label).await;
186 }
187 in_drift = false;
188 }
189 }
190 }
191 }
192 }));
193 }
194 }
195
196 // SSH banner checks, stored as health check entries with target name "{name}:ssh"
197 for (name, target_config) in configured_targets(config) {
198 if let Some(ssh_config) = target_config.ssh_banner {
199 let interval_secs = default_interval;
200 let pool = pool.clone();
201 let check_name = format!("{name}:ssh");
202 let label = target_config.label.clone();
203 let alerter = alerter.cloned();
204 let cancel = cancel.clone();
205
206 info!("{check_name}: SSH banner check every {interval_secs}s");
207
208 handles.push(tokio::spawn(async move {
209 let mut ticks = CheckInterval::new(interval_secs, cancel);
210 let seed = db::get_latest_health(&pool, &check_name)
211 .await
212 .ok()
213 .flatten()
214 .map(|s| s.status);
215 let mut gate = TransitionGate::seeded(seed, confirmations);
216 while ticks.next().await {
217 let snapshot = ssh_banner::check_ssh_banner(&check_name, &ssh_config).await;
218 info!(
219 "{}: {} ({}ms)",
220 check_name, snapshot.status, snapshot.response_time_ms
221 );
222 if let Err(e) = db::insert_health_check(&pool, &snapshot).await {
223 tracing::error!("{check_name}: failed to store SSH banner check: {e}");
224 }
225
226 if let Some((from_status, to_status)) = gate.observe(snapshot.status)
227 && let Some(ref alerter) = alerter
228 {
229 let from = from_status.to_string();
230 let to = to_status.to_string();
231 if to_status == HealthStatus::Operational {
232 alerter
233 .send_health_recovery(&check_name, &label, &from)
234 .await;
235 } else {
236 alerter
237 .send_health_alert(
238 &check_name,
239 &label,
240 &from,
241 &to,
242 snapshot.error.as_deref(),
243 )
244 .await;
245 }
246 }
247 }
248 }));
249 }
250 }
251
252 handles
253 }
254