From c4276b8d778db4a84fdb628145952dc2c9b8ae0c Mon Sep 17 00:00:00 2001 From: Jeff Larson Date: Sat, 7 Nov 2026 18:50:08 -0800 Subject: [PATCH] alerts: name the firing series in the webhook and email MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A rule folds every matching series into one scalar (agg across time AND series), so a firing alarm named only the rule — not which node/pod crossed. For a cluster-wide rule like node load average (agg=max over all nodes) that left the operator, and fixer, running `kubectl top nodes` by hand to find the hot box. When a rule fires, additionally resolve the ONE series that drove the breach — the series with the largest per-series aggregate for a `gt` rule, the smallest for `lt` — and attach its `attributes` JSONB (k8s.node.name, host.arch, …) to the webhook payload (new optional `attributes` field, omitted on resolve) and as a `Series:` line in the email. The attribution query mirrors eval_sql's window / predicate / rate shape exactly but GROUPs BY (service, attributes) instead of folding across them; it runs only when a rule actually fires, and is best-effort (no-data/error → no series, the page still goes out). No schema change — computed from the existing metrics.attributes column. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_013cYVqzH7Xfwea7fAozdQK7 --- server/src/alerts.rs | 239 ++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 226 insertions(+), 13 deletions(-) diff --git a/server/src/alerts.rs b/server/src/alerts.rs index b0c4603..30169b2 100644 --- a/server/src/alerts.rs +++ b/server/src/alerts.rs @@ -209,6 +209,11 @@ struct WebhookPayload<'a> { value: Option, threshold: f64, comparator: &'a str, + /// The attributes JSONB of the series that drove the breach (e.g. `k8s.node.name`, + /// `host.arch`), so a receiver can act on WHICH node/pod fired, not just the rule. Set + /// only on a firing transition; absent on resolve and when the series can't be resolved. + #[serde(skip_serializing_if = "Option::is_none")] + attributes: Option<&'a serde_json::Value>, } /// SMTP settings for emailing alert transitions. All fields come from the @@ -266,8 +271,44 @@ fn email_subject(rule_name: &str, state: &str) -> String { format!("[watcher] {}: {}", state.to_uppercase(), rule_name) } +/// Render a series' attributes JSONB as a compact, sorted `k=v, k=v` line for the email, or +/// None when it is absent or not an object. Bounded so a series with many resource +/// attributes can't make one page unreasonably large. +fn render_attrs(attrs: Option<&serde_json::Value>) -> Option { + let obj = attrs?.as_object()?; + if obj.is_empty() { + return None; + } + let mut pairs: Vec = obj + .iter() + .map(|(k, v)| match v { + serde_json::Value::String(s) => format!("{k}={s}"), + other => format!("{k}={other}"), + }) + .collect(); + pairs.sort(); + let mut line = pairs.join(", "); + const MAX: usize = 400; + if line.len() > MAX { + // Back off to a char boundary — `String::truncate` panics mid-codepoint, and an + // attribute value can be non-ASCII. + let mut end = MAX; + while !line.is_char_boundary(end) { + end -= 1; + } + line.truncate(end); + line.push_str(" …"); + } + Some(line) +} + /// Human-readable email body describing the rule and the observed value. -fn email_body(rule: &AlertRule, state: &str, value: Option) -> String { +fn email_body( + rule: &AlertRule, + state: &str, + value: Option, + attrs: Option<&serde_json::Value>, +) -> String { let scope = rule .service .as_deref() @@ -276,12 +317,18 @@ fn email_body(rule: &AlertRule, state: &str, value: Option) -> String { let observed = value .map(|v| v.to_string()) .unwrap_or_else(|| "no data".to_string()); + // Name the firing series (e.g. k8s.node.name=cluster-main) when known — the whole point + // of the attribution query, so the operator sees WHICH node/pod, not just the rule. + let series = render_attrs(attrs) + .map(|s| format!("Series: {s}\n")) + .unwrap_or_default(); format!( "Alert: {name}\n\ State: {state}\n\ Metric: {metric}{scope}\n\ Condition: {agg}({metric}) {cmp} {threshold} over {window}s\n\ - Observed: {observed}\n", + Observed: {observed}\n\ + {series}", name = rule.name, metric = rule.metric, agg = rule.agg, @@ -392,6 +439,86 @@ async fn resolve_rate(pool: &PgPool, rule: &AlertRule) -> anyhow::Result { Ok(monotonic.unwrap_or(false)) } +/// SQL returning the `attributes` JSONB of the ONE series that most drove the breach: +/// for a `gt` rule the series with the largest per-series aggregate, for `lt` the smallest. +/// Mirrors [`eval_sql`]'s window / predicate / rate shape exactly, but GROUPs BY the series +/// identity (`service, attributes`) instead of folding across series — so a firing alarm can +/// name the node/pod. Injection-safe on the same terms as `eval_sql`: only the enum- +/// whitelisted `agg_expr` and the comparator-derived `ASC`/`DESC` (both closed sets) are +/// interpolated; every operand is bound ($1 metric, $2 service, $3 window, $4 match, $5 exclude). +fn series_attrs_sql(agg: &str, rate: bool, comparator: &str) -> String { + let order = if comparator == "lt" { "ASC" } else { "DESC" }; + let filtered = "SELECT time, service, attributes, value + FROM metrics + WHERE name = $1 + AND ($2::text IS NULL OR service = $2) + AND ($4::jsonb IS NULL OR attributes @> $4) + AND ($5::jsonb IS NULL OR NOT (attributes @> $5)) + AND value IS NOT NULL + AND time >= now() - make_interval(secs => $3)"; + if agg == "increase" { + return format!( + "WITH pts AS ({filtered}), + steps AS ( + SELECT service, attributes, GREATEST(value - lag(value) OVER w, 0) AS step + FROM pts + WINDOW w AS (PARTITION BY service, attributes ORDER BY time) + ) + SELECT attributes FROM steps WHERE step IS NOT NULL + GROUP BY service, attributes ORDER BY sum(step) {order} LIMIT 1" + ); + } + if rate { + return format!( + "WITH pts AS ({filtered}), + rated AS ( + SELECT service, attributes, + CASE WHEN dt > 0 AND dv >= 0 THEN dv / dt ELSE 0 END AS value + FROM ( + SELECT service, attributes, + value - lag(value) OVER w AS dv, + extract(epoch FROM time - lag(time) OVER w) AS dt + FROM pts + WINDOW w AS (PARTITION BY service, attributes ORDER BY time) + ) d + WHERE dt IS NOT NULL + ) + SELECT attributes FROM rated + GROUP BY service, attributes ORDER BY {} {order} LIMIT 1", + agg_expr(agg) + ); + } + format!( + "WITH pts AS ({filtered}) + SELECT attributes FROM pts + GROUP BY service, attributes ORDER BY {} {order} LIMIT 1", + agg_expr(agg) + ) +} + +/// The firing series' attributes, best-effort: any error or no-data yields `None` and the +/// page still goes out (just without naming the series). Run only when a rule actually +/// fires, never on every breach tick. +async fn breaching_series_attrs( + pool: &PgPool, + rule: &AlertRule, + rate: bool, +) -> Option { + let sql = series_attrs_sql(&rule.agg, rate, &rule.comparator); + // AssertSqlSafe: same audit as eval_sql — only agg_expr() and the ASC/DESC literal are + // interpolated (both closed sets); metric/service/window/match/exclude are all bound. + sqlx::query_scalar::<_, serde_json::Value>(sqlx::AssertSqlSafe(sql)) + .bind(&rule.metric) + .bind(&rule.service) + .bind(rule.window_secs as f64) + .bind(&rule.match_attrs) + .bind(&rule.exclude_attrs) + .fetch_optional(pool) + .await + .ok() + .flatten() +} + /// Whether `value` breaches the rule. No data in the window (`None`) never fires. fn breached(value: Option, comparator: &str, threshold: f64) -> bool { match (value, comparator) { @@ -533,7 +660,8 @@ async fn evaluate( .execute(pool) .await?; if active { - fire(client, webhook, mailer, &rule, value).await; + let attrs = breaching_series_attrs(pool, &rule, rate).await; + fire(client, webhook, mailer, &rule, value, attrs.as_ref()).await; } } // Still breaching with an open event. A pending event that has now held @@ -548,7 +676,8 @@ async fn evaluate( .bind(value) .execute(pool) .await?; - fire(client, webhook, mailer, &rule, value).await; + let attrs = breaching_series_attrs(pool, &rule, rate).await; + fire(client, webhook, mailer, &rule, value, attrs.as_ref()).await; } } // Recovered. A firing event resolves and notifies as before; a pending @@ -561,7 +690,7 @@ async fn evaluate( .execute(pool) .await?; tracing::info!("alert '{}' resolved (value {:?})", rule.name, value); - notify(client, webhook, mailer, &rule, "resolved", value).await; + notify(client, webhook, mailer, &rule, "resolved", value, None).await; } else { sqlx::query("DELETE FROM alert_events WHERE id = $1") .bind(id) @@ -602,6 +731,7 @@ async fn fire( mailer: Option<&Mailer>, rule: &AlertRule, value: Option, + attrs: Option<&serde_json::Value>, ) { tracing::warn!( "alert '{}' firing: {} {} {} (value {:?})", @@ -611,7 +741,7 @@ async fn fire( rule.threshold, value ); - notify(client, webhook, mailer, rule, "firing", value).await; + notify(client, webhook, mailer, rule, "firing", value, attrs).await; } /// Ceiling on each sink's *total* delivery time inside `notify()`. This is @@ -644,10 +774,11 @@ async fn notify( rule: &AlertRule, state: &str, value: Option, + attrs: Option<&serde_json::Value>, ) { tokio::join!( - notify_webhook(client, webhook, rule, state, value), - notify_email(mailer, rule, state, value), + notify_webhook(client, webhook, rule, state, value, attrs), + notify_email(mailer, rule, state, value, attrs), ); } @@ -674,6 +805,7 @@ async fn notify_webhook( rule: &AlertRule, state: &str, value: Option, + attrs: Option<&serde_json::Value>, ) { let Some(url) = webhook else { return; @@ -686,6 +818,7 @@ async fn notify_webhook( value, threshold: rule.threshold, comparator: &rule.comparator, + attributes: attrs, }; // Propagate this span's context to the receiver. let mut headers = reqwest::header::HeaderMap::new(); @@ -703,12 +836,18 @@ async fn notify_webhook( /// Email the transition via `mailer`, if configured. Bounded end-to-end by /// [`NOTIFY_SINK_TIMEOUT`] — the whole SMTP conversation, not just each /// round trip within it; a failure or timeout is logged, never propagated. -async fn notify_email(mailer: Option<&Mailer>, rule: &AlertRule, state: &str, value: Option) { +async fn notify_email( + mailer: Option<&Mailer>, + rule: &AlertRule, + state: &str, + value: Option, + attrs: Option<&serde_json::Value>, +) { let Some(mailer) = mailer else { return; }; let subject = email_subject(&rule.name, state); - let body = email_body(rule, state, value); + let body = email_body(rule, state, value, attrs); deliver("email send", mailer.send(&subject, body)).await; } @@ -716,7 +855,8 @@ async fn notify_email(mailer: Option<&Mailer>, rule: &AlertRule, state: &str, va mod tests { use super::{ agg_expr, breached, build_webhook_client, email_body, email_subject, eval_sql, notify, - validate, AlertRule, Mailer, RuleConfig, MAX_FOR_SECS, NOTIFY_SINK_TIMEOUT, + series_attrs_sql, validate, AlertRule, Mailer, RuleConfig, MAX_FOR_SECS, + NOTIFY_SINK_TIMEOUT, }; use lettre::{AsyncSmtpTransport, Tokio1Executor}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; @@ -839,23 +979,94 @@ mod tests { #[test] fn email_body_includes_condition_and_value() { - let b = email_body(&rule(), "firing", Some(87.5)); + let b = email_body(&rule(), "firing", Some(87.5), None); assert!(b.contains("Alert: pod memory > 80%")); assert!(b.contains("State: firing")); assert!(b.contains("max(container.memory.usage) gt 80 over 300s")); assert!(b.contains("(service api)")); assert!(b.contains("Observed: 87.5")); + assert!(!b.contains("Series:")); // no attrs → no series line + } + + #[test] + fn email_body_names_the_firing_series() { + let attrs = serde_json::json!({"k8s.node.name": "cluster-main", "host.arch": "arm64"}); + let b = email_body(&rule(), "firing", Some(9.0), Some(&attrs)); + // Sorted k=v, so host.arch precedes k8s.node.name. + assert!(b.contains("Series: host.arch=arm64, k8s.node.name=cluster-main")); } #[test] fn email_body_handles_no_data_and_no_service() { let mut r = rule(); r.service = None; - let b = email_body(&r, "resolved", None); + let b = email_body(&r, "resolved", None, None); assert!(b.contains("Observed: no data")); assert!(!b.contains("(service")); } + #[test] + fn render_attrs_sorts_truncates_and_skips_empty() { + use super::render_attrs; + assert_eq!(render_attrs(None), None); + assert_eq!(render_attrs(Some(&serde_json::json!({}))), None); + assert_eq!(render_attrs(Some(&serde_json::json!("x"))), None); // not an object + let a = serde_json::json!({"b": "2", "a": "1"}); + assert_eq!(render_attrs(Some(&a)).unwrap(), "a=1, b=2"); + // A huge value is bounded: <= MAX(400) + " …"(4 bytes), and ends with the ellipsis. + let big = serde_json::json!({"k": "v".repeat(1000)}); + let r = render_attrs(Some(&big)).unwrap(); + assert!(r.len() <= 404 && r.ends_with('…')); + } + + #[test] + fn series_attrs_sql_groups_by_series_and_orders_by_breach_direction() { + // gt -> the largest per-series aggregate (DESC); lt -> the smallest (ASC); always one. + let gt = series_attrs_sql("max", false, "gt"); + assert!(gt.contains("GROUP BY service, attributes")); + assert!(gt.contains("ORDER BY max(value) DESC")); + assert!(gt.trim_end().ends_with("LIMIT 1")); + assert!(gt.contains("attributes @> $4")); // same predicates as eval_sql + let lt = series_attrs_sql("min", false, "lt"); + assert!(lt.contains("ORDER BY min(value) ASC")); + // rate + increase keep the per-series differencing, still grouped + limited to one. + let rate = series_attrs_sql("max", true, "gt"); + assert!(rate.contains("PARTITION BY service, attributes ORDER BY time")); + assert!(rate.contains("GROUP BY service, attributes ORDER BY max(value) DESC")); + let inc = series_attrs_sql("increase", false, "gt"); + assert!(inc.contains("ORDER BY sum(step) DESC")); + assert!(inc.trim_end().ends_with("LIMIT 1")); + } + + #[test] + fn webhook_payload_includes_attributes_only_when_present() { + let attrs = serde_json::json!({"k8s.node.name": "cluster-main"}); + let with = serde_json::to_string(&super::WebhookPayload { + rule: "r", + metric: "m", + service: None, + state: "firing", + value: Some(9.0), + threshold: 8.0, + comparator: "gt", + attributes: Some(&attrs), + }) + .unwrap(); + assert!(with.contains("\"attributes\":{\"k8s.node.name\":\"cluster-main\"}")); + let without = serde_json::to_string(&super::WebhookPayload { + rule: "r", + metric: "m", + service: None, + state: "resolved", + value: None, + threshold: 8.0, + comparator: "gt", + attributes: None, + }) + .unwrap(); + assert!(!without.contains("attributes")); // skip_serializing_if omits it + } + #[test] fn agg_expr_whitelist() { assert_eq!(agg_expr("avg"), "avg(value)"); @@ -968,6 +1179,7 @@ mod tests { &rule, "firing", Some(87.5), + None, ) .await; let elapsed = start.elapsed(); @@ -1089,6 +1301,7 @@ mod tests { &rule, "firing", Some(87.5), + None, ) .await; let elapsed = start.elapsed();