Skip to main content

max / makenotwork

Push PoM infra alerts to the MNW operator log Wire PoM's alerting pipeline into MNW's inbound ingestion endpoint so health, TLS, DNS, whois, backup, peer, route, latency, cors, scan, and monitoring alerts (plus their recoveries) land in the operator log, not just email/WAM. Server: expand AlertKind from the initial 3 to the 11-domain grouped taxonomy (health/tls/dns/whois/latency/cors/backup/peer/route/scan/ monitoring). Still a typed enum, rejected at deserialization. PoM: add mnw_url + alerts_ingest_token to AlertConfig (token redacted in Debug, overridable via POM_ALERTS_INGEST_TOKEN). Every alert funnels through fire_failure/fire_recovery, which now also push to MNW after the primary sink. push_mnw is durable: deliver once, and on failure enqueue a channel="mnw" row through the same pending_alerts queue WAM/email use, so it inherits the exponential backoff and dead-letter. As a secondary sink it does not re-record the ledger on retry (the primary channel already did). PoM's 24 categories fold onto the 11 domains; failure/recovery rides severity; dedup_key is "{kind}:{target}" so retries and repeats collapse server-side. Sink is a clean no-op unless both mnw_url and the token are set. pom-hetzner.toml documents the sink commented-out pending a reachability check on /api/internal from the PoM host. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-17 19:31 UTC
Commit: 453d9843ade77d122b3a9a83573d384b06265465
Parent: f49bd61
5 files changed, +279 insertions, -13 deletions
M pom/deploy/pom-hetzner.toml +9
@@ -144,3 +144,12 @@ on_missing = "log"
144 144 [alerts]
145 145 # postmark_token loaded from POM_POSTMARK_TOKEN env var
146 146 to = "pom-alerts@makenot.work"
147 +
148 + # MNW operator-log sink: also pushes alerts to POST {mnw_url}/api/internal/alerts.
149 + # Enable once ALERTS_INGEST_TOKEN is set in MNW prod .env AND the endpoint is
150 + # reachable from this host. Confirm reachability first: /api/internal may be
151 + # restricted at the public edge, in which case use the tailnet URL instead of
152 + # the public one below. The token loads from POM_ALERTS_INGEST_TOKEN env var
153 + # (never commit it here). Both mnw_url and the token must be present, else the
154 + # sink stays disabled.
155 + # mnw_url = "https://makenot.work"
M pom/src/alerts/mod.rs +219
@@ -57,6 +57,52 @@ fn backup_status_detail(status: &str, age_hours: Option<i64>) -> String {
57 57 }
58 58 }
59 59
60 + /// Map PoM's fine-grained [`AlertCategory`] onto MNW's domain-level alert kind
61 + /// (the `AlertKind` enum in the MNW server's `db/admin_alerts.rs`). Recoveries
62 + /// and sub-conditions fold onto their domain; the failure/recovery distinction
63 + /// rides on severity, not kind. A new PoM category needs a line here — and, if
64 + /// it introduces a new domain, a matching MNW `AlertKind` variant (else MNW
65 + /// rejects the push with 422).
66 + fn mnw_kind(category: AlertCategory) -> &'static str {
67 + use AlertCategory::*;
68 + match category {
69 + Health | Recovery => "health",
70 + TlsExpiry | TlsError | TlsRecovery => "tls",
71 + DnsMismatch | DnsRecovery => "dns",
72 + WhoisExpiry | WhoisError => "whois",
73 + LatencyDrift | LatencyRecovery | TestDurationDrift => "latency",
74 + CorsFailure | CorsRecovery => "cors",
75 + BackupStale | BackupRecovery => "backup",
76 + PeerMissing | PeerRecovery => "peer",
77 + RouteFailure | RouteRecovery => "route",
78 + ScanPipelineDegraded | ScanPipelineRecovery => "scan",
79 + MonitoringOffline | MonitoringRecovery => "monitoring",
80 + }
81 + }
82 +
83 + /// Map PoM's WAM priority string onto MNW's three-level severity. Recoveries do
84 + /// not carry a priority; the caller passes `"info"` for those directly.
85 + fn mnw_severity(priority: &str) -> &'static str {
86 + match priority {
87 + "critical" => "critical",
88 + "high" | "medium" => "warning",
89 + _ => "info",
90 + }
91 + }
92 +
93 + /// Truncate `s` to at most `max` bytes on a char boundary. MNW caps title at
94 + /// 200 and body at 5000; over-long input would be a 422, so clamp instead.
95 + fn truncate(s: &str, max: usize) -> String {
96 + if s.len() <= max {
97 + return s.to_string();
98 + }
99 + let mut end = max;
100 + while !s.is_char_boundary(end) {
101 + end -= 1;
102 + }
103 + s[..end].to_string()
104 + }
105 +
60 106 #[derive(Clone)]
61 107 pub struct Alerter {
62 108 config: AlertConfig,
@@ -101,9 +147,103 @@ impl Alerter {
101 147 instead of WAM tickets (set alerts.wam_url to route them to WAM)"
102 148 );
103 149 }
150 + match (&config.mnw_url, &config.alerts_ingest_token) {
151 + (Some(_), Some(_)) => info!("alerts: MNW sink enabled — alerts also push to the MNW operator log"),
152 + (Some(_), None) => warn!(
153 + "alerts: mnw_url set but alerts_ingest_token missing — MNW sink disabled \
154 + (set alerts.alerts_ingest_token or POM_ALERTS_INGEST_TOKEN)"
155 + ),
156 + _ => {}
157 + }
104 158 Self { config, client, pool, instance_name, wam_url }
105 159 }
106 160
161 + /// Push one alert to MNW's operator log (`POST /api/internal/alerts`).
162 + /// Returns `true` iff MNW accepted it. Returns `false` — never enqueues —
163 + /// when the sink is unconfigured; callers gate on config before treating a
164 + /// `false` as a delivery miss worth queuing. Bounded by a 10s timeout so MNW
165 + /// latency can't stall the probe loop. The server dedups on `dedup_key`
166 + /// (`{kind}:{target}`), so a retried or repeated push of one condition
167 + /// collapses rather than duplicating.
168 + async fn deliver_mnw(&self, kind: &str, key: &str, severity: &str, subject: &str, body: &str) -> bool {
169 + let (Some(base), Some(token)) =
170 + (self.config.mnw_url.as_deref(), self.config.alerts_ingest_token.as_deref())
171 + else {
172 + return false;
173 + };
174 + let payload = serde_json::json!({
175 + "source": "pom",
176 + "kind": kind,
177 + "severity": severity,
178 + "title": truncate(subject, 200),
179 + "body": truncate(body, 5000),
180 + "dedup_key": truncate(&format!("{kind}:{key}"), 200),
181 + "details": { "instance": self.instance_name, "target": key },
182 + });
183 + let url = format!("{}/api/internal/alerts", base.trim_end_matches('/'));
184 + let send_fut = self.client.post(&url).bearer_auth(token).json(&payload).send();
185 + match tokio::time::timeout(std::time::Duration::from_secs(10), send_fut).await {
186 + Ok(Ok(resp)) if resp.status().is_success() => {
187 + info!("mnw alert pushed: {kind}/{severity} {key}");
188 + true
189 + }
190 + Ok(Ok(resp)) => {
191 + let status = resp.status();
192 + let text = resp.text().await.unwrap_or_default();
193 + warn!("mnw alert push failed ({status}): {text}");
194 + false
195 + }
196 + Ok(Err(e)) => {
197 + warn!("mnw alert push error: {e}");
198 + false
199 + }
200 + Err(_) => {
201 + warn!("mnw alert push timed out: {kind}/{key}");
202 + false
203 + }
204 + }
205 + }
206 +
207 + /// Durable push to the MNW operator log: deliver now, and on failure enqueue
208 + /// for retry through the same `pending_alerts` queue WAM/email use. No-op
209 + /// unless both `mnw_url` and `alerts_ingest_token` are set.
210 + ///
211 + /// MNW is a *secondary* sink: the primary channel (WAM/email) already
212 + /// recorded this alert in the ledger, so a queued MNW row carries
213 + /// `channel = "mnw"` and, unlike WAM/email retries, does not re-record on
214 + /// delivery (see [`retry_pending`]) — it only needs to eventually land.
215 + /// `severity` derives from `priority` exactly as the retry path recomputes
216 + /// it (`None` → recovery → `"info"`), so live and retried pushes match.
217 + async fn push_mnw(&self, category: AlertCategory, key: &str, priority: Option<&str>, subject: &str, body: &str) {
218 + if self.config.mnw_url.is_none() || self.config.alerts_ingest_token.is_none() {
219 + return;
220 + }
221 + let kind = mnw_kind(category);
222 + let severity = priority.map(mnw_severity).unwrap_or("info");
223 + if self.deliver_mnw(kind, key, severity, subject, body).await {
224 + return;
225 + }
226 + let category = category.to_string();
227 + let pending = db::NewPendingAlert {
228 + alert_key: key,
229 + category: &category,
230 + channel: "mnw",
231 + subject,
232 + body,
233 + priority,
234 + source: Some("pom"),
235 + source_ref: None,
236 + from_status: None,
237 + to_status: None,
238 + error: None,
239 + };
240 + if let Err(e) = db::enqueue_pending_alert(&self.pool, &pending).await {
241 + warn!("failed to enqueue undelivered mnw alert for retry: {e}");
242 + } else {
243 + warn!("mnw alert to {key} undelivered; queued for retry");
244 + }
245 + }
246 +
107 247 /// Whether a FAILURE alert for `target` is within its cooldown.
108 248 ///
109 249 /// A recorded recovery *newer* than the last failure clears the cooldown: the
@@ -233,7 +373,11 @@ impl Alerter {
233 373 info!("alert cooldown active for {}, skipping", meta.key);
234 374 return;
235 375 }
376 + // Primary sink (WAM) first so a slow MNW push can't delay the ticket;
377 + // capture what push_mnw needs before `meta` is moved into dispatch.
378 + let (category, key) = (meta.category, meta.key);
236 379 self.dispatch_wam(subject, body, priority, source, source_ref, meta).await;
380 + self.push_mnw(category, key, Some(priority), subject, body).await;
237 381 }
238 382
239 383 /// Fire a recovery: recovery-cooldown-gate, then dispatch by email. The single
@@ -243,7 +387,9 @@ impl Alerter {
243 387 info!("recovery cooldown active for {}, skipping", meta.key);
244 388 return;
245 389 }
390 + let (category, key) = (meta.category, meta.key);
246 391 self.dispatch_email(subject, body, meta).await;
392 + self.push_mnw(category, key, None, subject, body).await;
247 393 }
248 394
249 395 /// Dispatch a WAM failure alert: attempt delivery (WAM, falling back to
@@ -314,6 +460,15 @@ impl Alerter {
314 460 /// signal the caller to delete the row; otherwise leave it for a later tick.
315 461 /// Called by the retry task in `serve`.
316 462 pub async fn retry_pending(&self, p: &db::PendingAlertRow) -> bool {
463 + // MNW is a secondary sink: the primary channel already wrote this alert
464 + // to the ledger, so a successful retry only clears the queued row — it
465 + // must not re-record (that would double-count and skew the cooldown).
466 + // Reconstruct kind/severity the same way the live push derives them.
467 + if p.channel == "mnw" {
468 + let kind = p.category.parse::<AlertCategory>().map(mnw_kind).unwrap_or("health");
469 + let severity = p.priority.as_deref().map(mnw_severity).unwrap_or("info");
470 + return self.deliver_mnw(kind, &p.alert_key, severity, &p.subject, &p.body).await;
471 + }
317 472 let delivered = match p.channel.as_str() {
318 473 "wam" => {
319 474 self.wam_ticket(
@@ -395,6 +550,68 @@ impl Alerter {
395 550 mod tests {
396 551 use super::*;
397 552
553 + #[tokio::test]
554 + async fn mnw_retry_routes_to_sink_and_never_records() {
555 + // A queued mnw row must retry via the MNW sink, not the WAM/email path,
556 + // and must NOT write the ledger (the primary channel already did). With
557 + // the sink unconfigured (test_alerter has no mnw_url), delivery fails and
558 + // the row stays queued — and the cooldown ledger is untouched.
559 + let pool = db::connect_in_memory().await.unwrap();
560 + let alerter = test_alerter(pool.clone());
561 + let row = db::PendingAlertRow {
562 + id: 1,
563 + alert_key: "makenot.work".to_string(),
564 + category: "tls_expiry".to_string(),
565 + channel: "mnw".to_string(),
566 + subject: "TLS expiring".to_string(),
567 + body: "cert expires soon".to_string(),
568 + priority: Some("high".to_string()),
569 + source: Some("pom".to_string()),
570 + source_ref: None,
571 + from_status: None,
572 + to_status: None,
573 + error: None,
574 + attempts: 0,
575 + };
576 + assert!(!alerter.retry_pending(&row).await, "unconfigured sink cannot deliver");
577 + assert!(
578 + !alerter.is_within_cooldown("makenot.work").await,
579 + "mnw retry must not record to the ledger"
580 + );
581 + }
582 +
583 + #[test]
584 + fn mnw_kind_folds_failure_and_recovery_to_same_domain() {
585 + // A condition and its recovery must map to one kind so they share a
586 + // dedup thread in the MNW log; the failure/recovery split rides severity.
587 + assert_eq!(mnw_kind(AlertCategory::TlsExpiry), "tls");
588 + assert_eq!(mnw_kind(AlertCategory::TlsError), "tls");
589 + assert_eq!(mnw_kind(AlertCategory::TlsRecovery), "tls");
590 + assert_eq!(mnw_kind(AlertCategory::DnsMismatch), "dns");
591 + assert_eq!(mnw_kind(AlertCategory::DnsRecovery), "dns");
592 + assert_eq!(mnw_kind(AlertCategory::TestDurationDrift), "latency");
593 + assert_eq!(mnw_kind(AlertCategory::MonitoringOffline), "monitoring");
594 + }
595 +
596 + #[test]
597 + fn mnw_severity_collapses_priorities_to_three_levels() {
598 + assert_eq!(mnw_severity("critical"), "critical");
599 + assert_eq!(mnw_severity("high"), "warning");
600 + assert_eq!(mnw_severity("medium"), "warning");
601 + assert_eq!(mnw_severity("low"), "info");
602 + assert_eq!(mnw_severity("anything-else"), "info");
603 + }
604 +
605 + #[test]
606 + fn truncate_respects_cap_and_char_boundaries() {
607 + assert_eq!(truncate("short", 200), "short");
608 + assert_eq!(truncate("abcdef", 3), "abc");
609 + // Cutting mid multi-byte char steps back to the previous boundary rather
610 + // than panicking on a non-boundary slice.
611 + let s = "aé"; // 'é' is two bytes; byte index 2 is not a char boundary
612 + assert_eq!(truncate(s, 2), "a");
613 + }
614 +
398 615 fn test_alerter(pool: SqlitePool) -> Alerter {
399 616 let config = AlertConfig {
400 617 postmark_token: None, // dev mode
@@ -402,6 +619,8 @@ mod tests {
402 619 from: "PoM Alerts <pom-alerts@makenot.work>".to_string(),
403 620 cooldown_secs: 300,
404 621 wam_url: None,
622 + mnw_url: None,
623 + alerts_ingest_token: None,
405 624 };
406 625 Alerter::new(config, pool, "test-instance".to_string())
407 626 }
M pom/src/config.rs +17
@@ -42,6 +42,13 @@ pub struct AlertConfig {
42 42 pub cooldown_secs: u64,
43 43 /// WAM ticket manager URL (tailnet). When set, alerts also create WAM tickets.
44 44 pub wam_url: Option<String>,
45 + /// MNW base URL (e.g. "https://makenot.work"). When set together with
46 + /// `alerts_ingest_token`, alerts are also pushed to MNW's operator log via
47 + /// `POST /api/internal/alerts`. Either unset disables the MNW sink.
48 + pub mnw_url: Option<String>,
49 + /// Bearer token for MNW's alert-ingestion endpoint. Can also be set via the
50 + /// `POM_ALERTS_INGEST_TOKEN` env var. Distinct from any CLI service token.
51 + pub alerts_ingest_token: Option<String>,
45 52 }
46 53
47 54 impl std::fmt::Debug for AlertConfig {
@@ -52,6 +59,8 @@ impl std::fmt::Debug for AlertConfig {
52 59 .field("from", &self.from)
53 60 .field("cooldown_secs", &self.cooldown_secs)
54 61 .field("wam_url", &self.wam_url)
62 + .field("mnw_url", &self.mnw_url)
63 + .field("alerts_ingest_token", &self.alerts_ingest_token.as_ref().map(|_| "***"))
55 64 .finish()
56 65 }
57 66 }
@@ -482,6 +491,11 @@ impl Config {
482 491 {
483 492 alerts.postmark_token = Some(token);
484 493 }
494 + if let Ok(token) = std::env::var("POM_ALERTS_INGEST_TOKEN")
495 + && let Some(ref mut alerts) = config.alerts
496 + {
497 + alerts.alerts_ingest_token = Some(token);
498 + }
485 499 if let Ok(token) = std::env::var("POM_API_TOKEN") {
486 500 config.serve.api_token = Some(token);
487 501 }
@@ -1230,9 +1244,12 @@ expected_routes = ["/status"]
1230 1244 from: "PoM".to_string(),
1231 1245 cooldown_secs: 300,
1232 1246 wam_url: None,
1247 + mnw_url: None,
1248 + alerts_ingest_token: Some("super-secret-ingest-token".to_string()),
1233 1249 };
1234 1250 let rendered = format!("{alerts:?}");
1235 1251 assert!(!rendered.contains("super-secret-token"), "token must be redacted in Debug");
1252 + assert!(!rendered.contains("super-secret-ingest-token"), "ingest token must be redacted in Debug");
1236 1253 assert!(rendered.contains("***"), "redaction marker expected");
1237 1254
1238 1255 let serve = ServeConfig {
M server/migrations/169_admin_alerts.sql +4 -4
@@ -9,10 +9,10 @@
9 9 -- ops signals for the operator, never creator-facing.
10 10 --
11 11 -- `source` names the reporting agent ("pom", "mt", ...). `kind` is the alert
12 - -- class the agent assigns ("health", "tls_expiry", "dns_change") and `severity`
13 - -- the urgency ("critical"/"warning"/"info"); both are closed sets enforced at
14 - -- the application layer. Stored as TEXT (not an enum type) so extending either
15 - -- set is a code change, not a migration.
12 + -- domain the agent assigns ("health"/"tls"/"dns"/"whois"/"backup"/... — see the
13 + -- AlertKind enum) and `severity` the urgency ("critical"/"warning"/"info");
14 + -- both are closed sets enforced at the application layer. Stored as TEXT (not an
15 + -- enum type) so extending either set is a code change, not a migration.
16 16 -- `dedup_key` lets an agent collapse repeated firings of the same condition
17 17 -- (e.g. one row per (source, dedup_key) flap) without the server modeling agent
18 18 -- state; it is nullable for one-shot alerts.
M server/src/db/admin_alerts.rs +30 -9
@@ -7,17 +7,30 @@
7 7 use serde::Deserialize;
8 8 use sqlx::PgPool;
9 9
10 - /// Alert class an agent may report. Closed set, deserialized straight from the
11 - /// wire (`#[serde(rename_all = "snake_case")]` maps `TlsExpiry` → `"tls_expiry"`
12 - /// etc.), so an unknown value is rejected when the request body is parsed —
13 - /// there is no runtime membership check. Extending the taxonomy is a variant
14 - /// here plus its `as_str`, not a migration (the column is TEXT).
10 + /// Alert domain an agent may report. Closed set, deserialized straight from the
11 + /// wire (`#[serde(rename_all = "snake_case")]`), so an unknown value is rejected
12 + /// when the request body is parsed — there is no runtime membership check.
13 + /// Extending the taxonomy is a variant here plus its `as_str`, not a migration
14 + /// (the column is TEXT).
15 + ///
16 + /// These are domain-level, not per-event: PoM's finer categories (tls expiry vs
17 + /// tls error vs tls recovery) all fold onto `Tls`, with the failure/recovery
18 + /// distinction carried by [`AlertSeverity`] instead. Kept small so the operator
19 + /// log filters by system, not by every sub-condition.
15 20 #[derive(Debug, Clone, Copy, Deserialize)]
16 21 #[serde(rename_all = "snake_case")]
17 22 pub enum AlertKind {
18 23 Health,
19 - TlsExpiry,
20 - DnsChange,
24 + Tls,
25 + Dns,
26 + Whois,
27 + Latency,
28 + Cors,
29 + Backup,
30 + Peer,
31 + Route,
32 + Scan,
33 + Monitoring,
21 34 }
22 35
23 36 impl AlertKind {
@@ -25,8 +38,16 @@ impl AlertKind {
25 38 pub fn as_str(self) -> &'static str {
26 39 match self {
27 40 Self::Health => "health",
28 - Self::TlsExpiry => "tls_expiry",
29 - Self::DnsChange => "dns_change",
41 + Self::Tls => "tls",
42 + Self::Dns => "dns",
43 + Self::Whois => "whois",
44 + Self::Latency => "latency",
45 + Self::Cors => "cors",
46 + Self::Backup => "backup",
47 + Self::Peer => "peer",
48 + Self::Route => "route",
49 + Self::Scan => "scan",
50 + Self::Monitoring => "monitoring",
30 51 }
31 52 }
32 53 }