Skip to main content

max / makenotwork

9.2 KB · 306 lines History Blame Raw
1 //! Scan-pipeline health check, polls `<host>/admin/uploads/health.json` on
2 //! a target makenotwork instance and applies the audit-doc thresholds.
3 //!
4 //! Thresholds (audit doc § 6):
5 //! - Per-layer error rate > 10% over 1h → degraded
6 //! - Per-layer success count == 0 over 24h → down (layer fully unavailable)
7 //! (we use the cheaper proxy "no clean response in 24h" via
8 //! `last_clean_secs_ago` from the upstream payload).
9 //! - Queue depth > 50 pending → degraded
10 //! - Stuck-scan count > 5 → degraded (workers falling behind)
11 //! - Held-for-review total > 100 → degraded (review backlog growing)
12 //!
13 //! Anything that fires "down" elevates the overall status to Unreachable;
14 //! anything "degraded" elevates to Degraded. Multiple issues all reported
15 //! in the `issues` vec.
16
17 use std::time::Duration;
18
19 use tracing::instrument;
20
21 use crate::types::{ScanLayerSnapshot, ScanPipelineCheckResult};
22
23 const QUEUE_PENDING_THRESHOLD: i64 = 50;
24 const STUCK_SCAN_THRESHOLD: i64 = 5;
25 const HELD_BACKLOG_THRESHOLD: i64 = 100;
26 const ERROR_RATE_PCT_THRESHOLD: i32 = 10;
27 const LAYER_DOWN_AGE_SECS: i64 = 24 * 3600;
28
29 #[derive(Debug, serde::Deserialize)]
30 struct UpstreamLayer {
31 layer: String,
32 total_1h: i64,
33 success_1h: i64,
34 error_1h: i64,
35 last_clean_secs_ago: Option<i64>,
36 }
37
38 #[derive(Debug, serde::Deserialize)]
39 struct UpstreamHealth {
40 queue_pending: i64,
41 queue_running: i64,
42 queue_stuck: i64,
43 held_total: i64,
44 layers: Vec<UpstreamLayer>,
45 }
46
47 #[instrument(skip_all)]
48 pub async fn check_scan_pipeline(
49 target_name: &str,
50 base_url: &str,
51 timeout_secs: u64,
52 ) -> ScanPipelineCheckResult {
53 let checked_at = chrono::Utc::now().to_rfc3339();
54 let url = format!(
55 "{}/admin/uploads/health.json",
56 base_url.trim_end_matches('/')
57 );
58
59 let client = match crate::tls::https_client_builder()
60 .timeout(Duration::from_secs(timeout_secs))
61 .build()
62 {
63 Ok(c) => c,
64 Err(e) => return unreachable(target_name, &checked_at, format!("client build: {e}")),
65 };
66
67 let response = match client.get(&url).send().await {
68 Ok(r) => r,
69 Err(e) => return unreachable(target_name, &checked_at, format!("request: {e}")),
70 };
71
72 let status = response.status();
73 if !status.is_success() {
74 return unreachable(
75 target_name,
76 &checked_at,
77 format!("HTTP {}", status.as_u16()),
78 );
79 }
80
81 let body: UpstreamHealth = match response.json().await {
82 Ok(b) => b,
83 Err(e) => return unreachable(target_name, &checked_at, format!("parse: {e}")),
84 };
85
86 classify(target_name, &checked_at, body)
87 }
88
89 fn unreachable(target: &str, checked_at: &str, msg: String) -> ScanPipelineCheckResult {
90 ScanPipelineCheckResult {
91 target: target.to_string(),
92 status: "unreachable".to_string(),
93 queue_pending: 0,
94 queue_running: 0,
95 queue_stuck: 0,
96 held_total: 0,
97 layers: Vec::new(),
98 issues: vec![format!("scan-pipeline health endpoint: {msg}")],
99 checked_at: checked_at.to_string(),
100 error: Some(msg),
101 }
102 }
103
104 fn classify(target: &str, checked_at: &str, body: UpstreamHealth) -> ScanPipelineCheckResult {
105 let mut issues: Vec<String> = Vec::new();
106 let mut any_down = false;
107 let mut any_degraded = false;
108
109 if body.queue_pending > QUEUE_PENDING_THRESHOLD {
110 issues.push(format!(
111 "queue depth {} > {}",
112 body.queue_pending, QUEUE_PENDING_THRESHOLD
113 ));
114 any_degraded = true;
115 }
116 if body.queue_stuck > STUCK_SCAN_THRESHOLD {
117 issues.push(format!(
118 "stuck scans {} > {}",
119 body.queue_stuck, STUCK_SCAN_THRESHOLD
120 ));
121 any_degraded = true;
122 }
123 if body.held_total > HELD_BACKLOG_THRESHOLD {
124 issues.push(format!(
125 "held backlog {} > {}",
126 body.held_total, HELD_BACKLOG_THRESHOLD
127 ));
128 any_degraded = true;
129 }
130
131 let layers: Vec<ScanLayerSnapshot> = body
132 .layers
133 .into_iter()
134 .map(|l| {
135 let mut layer_status = "operational";
136 let error_rate_pct = if l.total_1h > 0 {
137 (100 * l.error_1h / l.total_1h) as i32
138 } else {
139 0
140 };
141
142 let truly_idle = l.total_1h == 0
143 && l.last_clean_secs_ago
144 .is_none_or(|s| s > LAYER_DOWN_AGE_SECS);
145 if truly_idle {
146 layer_status = "down";
147 any_down = true;
148 issues.push(format!("layer '{}': no clean response in >24h", l.layer));
149 } else if error_rate_pct > ERROR_RATE_PCT_THRESHOLD {
150 layer_status = "degraded";
151 any_degraded = true;
152 issues.push(format!(
153 "layer '{}': error rate {}% > {}%",
154 l.layer, error_rate_pct, ERROR_RATE_PCT_THRESHOLD
155 ));
156 }
157
158 ScanLayerSnapshot {
159 layer: l.layer,
160 total_1h: l.total_1h,
161 success_1h: l.success_1h,
162 error_1h: l.error_1h,
163 last_clean_secs_ago: l.last_clean_secs_ago,
164 status: layer_status.to_string(),
165 }
166 })
167 .collect();
168
169 let overall = if any_down {
170 "unreachable"
171 } else if any_degraded {
172 "degraded"
173 } else {
174 "operational"
175 };
176
177 ScanPipelineCheckResult {
178 target: target.to_string(),
179 status: overall.to_string(),
180 queue_pending: body.queue_pending,
181 queue_running: body.queue_running,
182 queue_stuck: body.queue_stuck,
183 held_total: body.held_total,
184 layers,
185 issues,
186 checked_at: checked_at.to_string(),
187 error: None,
188 }
189 }
190
191 #[cfg(test)]
192 mod tests {
193 use super::*;
194
195 fn body(layers: Vec<(&str, i64, i64, i64, Option<i64>)>) -> UpstreamHealth {
196 UpstreamHealth {
197 queue_pending: 0,
198 queue_running: 0,
199 queue_stuck: 0,
200 held_total: 0,
201 layers: layers
202 .into_iter()
203 .map(|(name, total, success, errors, last)| UpstreamLayer {
204 layer: name.to_string(),
205 total_1h: total,
206 success_1h: success,
207 error_1h: errors,
208 last_clean_secs_ago: last,
209 })
210 .collect(),
211 }
212 }
213
214 #[test]
215 fn empty_pipeline_is_operational() {
216 let b = UpstreamHealth {
217 queue_pending: 0,
218 queue_running: 0,
219 queue_stuck: 0,
220 held_total: 0,
221 layers: vec![],
222 };
223 let r = classify("t", "now", b);
224 assert_eq!(r.status, "operational");
225 assert!(r.issues.is_empty());
226 }
227
228 #[test]
229 fn queue_depth_threshold_degrades() {
230 let mut b = body(vec![]);
231 b.queue_pending = 99;
232 let r = classify("t", "now", b);
233 assert_eq!(r.status, "degraded");
234 assert!(r.issues.iter().any(|i| i.contains("queue depth")));
235 }
236
237 #[test]
238 fn stuck_scans_threshold_degrades() {
239 let mut b = body(vec![]);
240 b.queue_stuck = 6;
241 let r = classify("t", "now", b);
242 assert_eq!(r.status, "degraded");
243 }
244
245 #[test]
246 fn held_backlog_threshold_degrades() {
247 let mut b = body(vec![]);
248 b.held_total = 150;
249 let r = classify("t", "now", b);
250 assert_eq!(r.status, "degraded");
251 }
252
253 #[test]
254 fn high_layer_error_rate_degrades() {
255 // 20 of 100 are errors = 20% > 10% threshold.
256 let b = body(vec![("malwarebazaar", 100, 80, 20, Some(60))]);
257 let r = classify("t", "now", b);
258 assert_eq!(r.status, "degraded");
259 assert!(r.issues.iter().any(|i| i.contains("error rate")));
260 }
261
262 #[test]
263 fn layer_silent_for_24h_is_down() {
264 let b = body(vec![("clamav", 0, 0, 0, None)]);
265 let r = classify("t", "now", b);
266 assert_eq!(r.status, "unreachable");
267 assert!(
268 r.issues
269 .iter()
270 .any(|i| i.contains("no clean response in >24h"))
271 );
272 }
273
274 #[test]
275 fn layer_silent_with_stale_clean_is_down() {
276 // 2 days ago = 172_800 secs ago, beyond 24h threshold.
277 let b = body(vec![("urlhaus", 0, 0, 0, Some(172_800))]);
278 let r = classify("t", "now", b);
279 assert_eq!(r.status, "unreachable");
280 }
281
282 #[test]
283 fn layer_recently_clean_is_operational_even_with_no_recent_volume() {
284 // 0 scans in last hour but a clean response 30 min ago → operational.
285 let b = body(vec![("yara", 0, 0, 0, Some(1800))]);
286 let r = classify("t", "now", b);
287 assert_eq!(r.status, "operational");
288 }
289
290 #[test]
291 fn down_layer_beats_degraded_signals() {
292 let mut b = body(vec![("malwarebazaar", 0, 0, 0, None)]);
293 b.queue_pending = 99;
294 let r = classify("t", "now", b);
295 assert_eq!(r.status, "unreachable", "down beats degraded");
296 }
297
298 #[test]
299 fn boundary_error_rate_does_not_trigger() {
300 // exactly 10%, at the boundary, not over.
301 let b = body(vec![("malwarebazaar", 100, 90, 10, Some(60))]);
302 let r = classify("t", "now", b);
303 assert_eq!(r.status, "operational");
304 }
305 }
306