max / makenotwork

3.9 KB · 108 lines History Blame Raw

1	//! Scan-pipeline health-check task. Polls the makenotwork health endpoint
2	//! on a per-target interval, applies thresholds (audit doc § 6), and fires
3	//! alerts on operational → degraded / unreachable transitions.
4
5	use tokio::task::JoinHandle;
6	use tracing::{info, warn};
7
8	use pom::alerts::Alerter;
9	use pom::checks::scan_pipeline;
10	use pom::config::Config;
11	use pom::db;
12
13	use super::{CheckInterval, configured_targets};
14
15	pub(crate) fn spawn_scan_pipeline_tasks(
16	config: &Config,
17	pool: &sqlx::SqlitePool,
18	cancel: &tokio_util::sync::CancellationToken,
19	alerter: Option<&Alerter>,
20	) -> Vec<JoinHandle<()>> {
21	let mut handles = Vec::new();
22
23	for (name, target_config) in configured_targets(config) {
24	let Some(sp_config) = target_config.scan_pipeline else {
25	continue;
26	};
27
28	let name = name.clone();
29	let label = target_config.label.clone();
30	let alerter = alerter.cloned();
31	let pool = pool.clone();
32	let cancel = cancel.clone();
33
34	info!(
35	"{name}: scan-pipeline check every {}s against {}",
36	sp_config.interval_secs, sp_config.base_url,
37	);
38
39	handles.push(tokio::spawn(async move {
40	let mut ticks = CheckInterval::new(sp_config.interval_secs, cancel);
41
42	// Track previous status so we only fire alerts on transitions. Seed
43	// from the ledger so a restart while the pipeline is already-degraded
44	// doesn't re-fire: if the latest scan-pipeline alert is the degraded
45	// (not the recovery) one, start from that non-operational status.
46	let mut previous_status: Option<String> = db::get_latest_alert_matching(
47	&pool,
48	&format!("scan_pipeline:{name}"),
49	"scan_pipeline_%",
50	)
51	.await
52	.ok()
53	.flatten()
54	.filter(\|a\| a.alert_type == "scan_pipeline_degraded")
55	.and_then(\|a\| a.to_status);
56
57	while ticks.next().await {
58	let result = scan_pipeline::check_scan_pipeline(
59	&name,
60	&sp_config.base_url,
61	sp_config.timeout_secs,
62	)
63	.await;
64
65	if result.issues.is_empty() && result.error.is_none() {
66	info!(
67	"{name}: scan pipeline operational (queue p={}/r={}, held={})",
68	result.queue_pending, result.queue_running, result.held_total
69	);
70	} else {
71	warn!(
72	target = %name,
73	status = %result.status,
74	issues = ?result.issues,
75	"scan pipeline non-operational"
76	);
77	}
78
79	// Persist so /status.json can render the latest result from the
80	// ledger rather than re-probing on every viewer poll.
81	if let Err(e) = db::insert_scan_pipeline_check(&pool, &result).await {
82	tracing::error!("{name}: failed to store scan pipeline check: {e}");
83	}
84
85	// Fire alerts on status transitions only.
86	if let Some(ref alerter) = alerter {
87	let prev_ok = previous_status
88	.as_deref()
89	.is_none_or(\|s\| s == "operational");
90	let now_ok = result.status == "operational";
91
92	if prev_ok && !now_ok {
93	alerter
94	.send_scan_pipeline_alert(&name, &label, &result.status, &result.issues)
95	.await;
96	} else if !prev_ok && now_ok {
97	alerter.send_scan_pipeline_recovery(&name, &label).await;
98	}
99	}
100
101	previous_status = Some(result.status);
102	}
103	}));
104	}
105
106	handles
107	}
108