Skip to main content

max / makenotwork

10.0 KB · 220 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 while ticks.next().await {
53 let snapshot = http::check_health(&name, &health_config, expect).await;
54 info!(
55 "{}: {} ({}ms)",
56 name, snapshot.status, snapshot.response_time_ms
57 );
58 if let Err(e) = db::insert_health_check(&pool, &snapshot).await {
59 tracing::error!("{name}: failed to store health check: {e}");
60 }
61
62 // Alert AND open/close incidents only on a *confirmed*
63 // transition (debounced + restart-seeded above).
64 if let Some((from_status, to_status)) = gate.observe(snapshot.status) {
65 if let Some(ref alerter) = alerter {
66 let from = from_status.to_string();
67 let to = to_status.to_string();
68 if to_status == HealthStatus::Operational {
69 alerter.send_health_recovery(&name, &label, &from).await;
70 } else {
71 alerter
72 .send_health_alert(
73 &name,
74 &label,
75 &from,
76 &to,
77 snapshot.error.as_deref(),
78 )
79 .await;
80 }
81 }
82
83 match incident_action(from_status, to_status) {
84 IncidentAction::None => {}
85 IncidentAction::Open => {
86 if let Err(e) = db::insert_incident(
87 &pool,
88 &name,
89 &from_status.to_string(),
90 &to_status.to_string(),
91 )
92 .await
93 {
94 tracing::error!("{name}: failed to open incident: {e}");
95 }
96 }
97 IncidentAction::Close => {
98 if let Err(e) = db::close_open_incidents(&pool, &name).await {
99 tracing::error!("{name}: failed to close incidents: {e}");
100 }
101 }
102 IncidentAction::CloseAndOpen => {
103 if let Err(e) = db::close_and_open_incident(
104 &pool,
105 &name,
106 &from_status.to_string(),
107 &to_status.to_string(),
108 )
109 .await
110 {
111 tracing::error!("{name}: failed to close+open incident: {e}");
112 }
113 }
114 }
115 }
116
117 // Latency drift detection
118 if let Some(ref trending) = trending_config
119 && snapshot.status == HealthStatus::Operational
120 {
121 let baseline_cutoff = (chrono::Utc::now()
122 - chrono::Duration::hours(trending.baseline_window_hours as i64))
123 .to_rfc3339();
124 let baseline_data = db::get_response_times(&pool, &name, &baseline_cutoff)
125 .await
126 .unwrap_or_default();
127 let operational_times: Vec<i64> = baseline_data
128 .iter()
129 .filter(|(_, ms)| *ms > 0)
130 .map(|(_, ms)| *ms)
131 .collect();
132 let baseline = LatencyStats::from_times(&operational_times);
133 let recent = db::get_recent_response_times(&pool, &name, 3)
134 .await
135 .unwrap_or_default();
136
137 if let Some(ref bl) = baseline {
138 if let Some(msg) =
139 drift::detect_latency_drift(&recent, bl, trending.spike_threshold)
140 {
141 if !in_drift {
142 info!("{name}: {msg}");
143 if let Some(ref alerter) = alerter {
144 alerter.send_latency_drift_alert(&name, &label, &msg).await;
145 }
146 in_drift = true;
147 }
148 } else if in_drift {
149 info!("{name}: latency drift recovered");
150 if let Some(ref alerter) = alerter {
151 alerter.send_latency_recovery(&name, &label).await;
152 }
153 in_drift = false;
154 }
155 }
156 }
157 }
158 }));
159 }
160 }
161
162 // SSH banner checks, stored as health check entries with target name "{name}:ssh"
163 for (name, target_config) in configured_targets(config) {
164 if let Some(ssh_config) = target_config.ssh_banner {
165 let interval_secs = default_interval;
166 let pool = pool.clone();
167 let check_name = format!("{name}:ssh");
168 let label = target_config.label.clone();
169 let alerter = alerter.cloned();
170 let cancel = cancel.clone();
171
172 info!("{check_name}: SSH banner check every {interval_secs}s");
173
174 handles.push(tokio::spawn(async move {
175 let mut ticks = CheckInterval::new(interval_secs, cancel);
176 let seed = db::get_latest_health(&pool, &check_name)
177 .await
178 .ok()
179 .flatten()
180 .map(|s| s.status);
181 let mut gate = TransitionGate::seeded(seed, confirmations);
182 while ticks.next().await {
183 let snapshot = ssh_banner::check_ssh_banner(&check_name, &ssh_config).await;
184 info!(
185 "{}: {} ({}ms)",
186 check_name, snapshot.status, snapshot.response_time_ms
187 );
188 if let Err(e) = db::insert_health_check(&pool, &snapshot).await {
189 tracing::error!("{check_name}: failed to store SSH banner check: {e}");
190 }
191
192 if let Some((from_status, to_status)) = gate.observe(snapshot.status)
193 && let Some(ref alerter) = alerter
194 {
195 let from = from_status.to_string();
196 let to = to_status.to_string();
197 if to_status == HealthStatus::Operational {
198 alerter
199 .send_health_recovery(&check_name, &label, &from)
200 .await;
201 } else {
202 alerter
203 .send_health_alert(
204 &check_name,
205 &label,
206 &from,
207 &to,
208 snapshot.error.as_deref(),
209 )
210 .await;
211 }
212 }
213 }
214 }));
215 }
216 }
217
218 handles
219 }
220