opssignal_core/
routing.rs

1use crate::signal::{Severity, Signal};
2use serde::{Deserialize, Serialize};
3use std::time::Duration;
4
5// Unknown keys are rejected: a typo such as `enviroment: prod` would
6// otherwise be ignored and leave a match that every signal passes.
7#[derive(Debug, Clone, Deserialize, Serialize)]
8#[serde(deny_unknown_fields)]
9pub struct RouteMatch {
10    #[serde(default)]
11    pub severity: Vec<Severity>,
12    #[serde(default)]
13    pub environment: Option<String>,
14    #[serde(default)]
15    pub source: Option<String>,
16}
17
18/// How a route delivers to the providers it lists.
19#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Deserialize, Serialize)]
20#[serde(rename_all = "lowercase")]
21pub enum DeliveryMode {
22    /// Try the providers in order and stop at the first that succeeds,
23    /// so later providers are fallbacks.
24    #[default]
25    Failover,
26    /// Send to every provider at once. Each one is retried on its own,
27    /// so one that fails never causes a resend to one that succeeded.
28    Fanout,
29}
30
31// Unknown keys are rejected so a typo such as `mod: fanout` fails at
32// load time instead of quietly leaving the route on failover.
33#[derive(Debug, Clone, Deserialize, Serialize)]
34#[serde(deny_unknown_fields)]
35pub struct Route {
36    #[serde(rename = "match")]
37    pub match_rule: RouteMatch,
38    pub providers: Vec<String>,
39    #[serde(default)]
40    pub mode: DeliveryMode,
41    /// Sends a signal through this route at most once per window for
42    /// each dedup key; repeats inside the window are suppressed.
43    #[serde(default, skip_serializing_if = "Option::is_none")]
44    pub dedup: Option<Window>,
45    /// Caps how many signals this route sends, whatever their key.
46    #[serde(default, skip_serializing_if = "Option::is_none")]
47    pub rate_limit: Option<RateLimit>,
48}
49
50impl Route {
51    /// Whether the route keeps any dedup or rate-limit state.
52    pub fn is_throttled(&self) -> bool {
53        self.dedup.is_some() || self.rate_limit.is_some()
54    }
55
56    /// The route's providers, as `[slack, pagerduty]`, for messages.
57    pub fn describe(&self) -> String {
58        format!("[{}]", self.providers.join(", "))
59    }
60}
61
62/// At most `max` signals per `per`, counted over a sliding window.
63#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
64#[serde(deny_unknown_fields)]
65pub struct RateLimit {
66    pub max: u32,
67    pub per: Window,
68}
69
70/// A length of time written as a number and a unit: `30s`, `10m`,
71/// `2h` or `1d`. A bare number is seconds.
72#[derive(Debug, Clone, Copy, PartialEq, Eq)]
73pub struct Window(pub Duration);
74
75impl std::str::FromStr for Window {
76    type Err = String;
77
78    fn from_str(s: &str) -> Result<Self, Self::Err> {
79        let s = s.trim();
80        let split = s.find(|c: char| !c.is_ascii_digit()).unwrap_or(s.len());
81        let (num, unit) = s.split_at(split);
82        let n: u64 = num
83            .parse()
84            .map_err(|_| format!("invalid duration {s:?}; expected e.g. 30s, 10m, 2h or 1d"))?;
85        let secs = match unit.trim() {
86            "" | "s" => 1,
87            "m" => 60,
88            "h" => 3600,
89            "d" => 86400,
90            other => {
91                return Err(format!(
92                    "invalid duration unit {other:?} in {s:?}; expected s, m, h or d"
93                ))
94            }
95        };
96        Ok(Window(Duration::from_secs(n.saturating_mul(secs))))
97    }
98}
99
100impl std::fmt::Display for Window {
101    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
102        let secs = self.0.as_secs();
103        let (n, unit) = [(86400, "d"), (3600, "h"), (60, "m")]
104            .into_iter()
105            .find(|(size, _)| secs > 0 && secs.is_multiple_of(*size))
106            .map(|(size, unit)| (secs / size, unit))
107            .unwrap_or((secs, "s"));
108        write!(f, "{n}{unit}")
109    }
110}
111
112impl<'de> Deserialize<'de> for Window {
113    fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
114        #[derive(Deserialize)]
115        #[serde(untagged)]
116        enum Raw {
117            Secs(u64),
118            Text(String),
119        }
120        match Raw::deserialize(d)? {
121            Raw::Secs(n) => Ok(Window(Duration::from_secs(n))),
122            Raw::Text(s) => s.parse().map_err(serde::de::Error::custom),
123        }
124    }
125}
126
127impl Serialize for Window {
128    fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
129        s.collect_str(self)
130    }
131}
132
133#[derive(Debug, Clone, Deserialize, Serialize)]
134pub struct RoutingConfig {
135    pub routes: Vec<Route>,
136}
137
138/// Where one signal goes. Every target is delivered concurrently and
139/// independently; within a target the providers are tried in order
140/// until one succeeds.
141#[derive(Debug, Clone, PartialEq, Eq)]
142pub struct DeliveryPlan {
143    pub targets: Vec<Vec<String>>,
144    /// Why matching routes were held back by their `dedup` or
145    /// `rate_limit`, one line per route. Only [`RoutingConfig::plan_with`]
146    /// fills it.
147    pub suppressed: Vec<String>,
148}
149
150impl DeliveryPlan {
151    /// Every provider of the plan, in plan order.
152    pub fn providers(&self) -> impl Iterator<Item = &String> {
153        self.targets.iter().flatten()
154    }
155}
156
157impl RouteMatch {
158    fn matches(&self, signal: &Signal) -> bool {
159        if !self.severity.is_empty() && !self.severity.contains(&signal.severity) {
160            return false;
161        }
162        if let Some(env) = &self.environment {
163            if signal.environment.as_deref() != Some(env.as_str()) {
164                return false;
165            }
166        }
167        if let Some(src) = &self.source {
168            if signal.source != *src {
169                return false;
170            }
171        }
172        true
173    }
174}
175
176impl RoutingConfig {
177    /// Returns the provider names a signal should be delivered to, in
178    /// the order of [`RoutingConfig::plan`].
179    pub fn resolve(&self, signal: &Signal) -> Vec<String> {
180        self.plan(signal).providers().cloned().collect()
181    }
182
183    /// Groups the matching routes' providers into delivery targets.
184    ///
185    /// Each provider of a `fanout` route is a target of its own. The
186    /// providers of every matching `failover` route share one target,
187    /// placed where the first of those routes is, so a config without
188    /// `fanout` delivers exactly as it always has: one ordered list that
189    /// stops at the first success.
190    ///
191    /// CRITICAL INVARIANT: if no route matches, this MUST fall back to
192    /// "console" rather than returning an empty plan. A signal that
193    /// matches nothing must never silently disappear — that is the one
194    /// failure mode that destroys trust in an alerting tool. See the
195    /// architecture review: "unmatched signals always hit console at
196    /// minimum."
197    pub fn plan(&self, signal: &Signal) -> DeliveryPlan {
198        self.plan_with(signal, |_| Ok(()))
199    }
200
201    /// [`RoutingConfig::plan`] that first asks `admit` about each
202    /// matching route that sets `dedup` or `rate_limit`. A route it
203    /// turns down (with the reason) adds no providers.
204    ///
205    /// Unlike `plan`, this can return no targets: when routes matched
206    /// and every one of them was suppressed. That is a decision the
207    /// config asked for, not a signal that matched nothing, so it does
208    /// not fall back to console; `suppressed` says why.
209    pub fn plan_with<F>(&self, signal: &Signal, mut admit: F) -> DeliveryPlan
210    where
211        F: FnMut(&Route) -> Result<(), String>,
212    {
213        // A provider named by several matching routes is used once, at
214        // its first position, so it never gets the same signal twice and
215        // a failing provider is not retried again later in the same
216        // delivery.
217        let mut seen: Vec<&String> = Vec::new();
218        let mut targets: Vec<Vec<String>> = Vec::new();
219        let mut failover: Option<usize> = None;
220        let mut suppressed = Vec::new();
221        for route in self.routes.iter().filter(|r| r.match_rule.matches(signal)) {
222            if route.is_throttled() {
223                if let Err(reason) = admit(route) {
224                    suppressed.push(format!("route {}: {reason}", route.describe()));
225                    continue;
226                }
227            }
228            for name in &route.providers {
229                if seen.contains(&name) {
230                    continue;
231                }
232                seen.push(name);
233                match (route.mode, failover) {
234                    (DeliveryMode::Fanout, _) => targets.push(vec![name.clone()]),
235                    (DeliveryMode::Failover, Some(i)) => targets[i].push(name.clone()),
236                    (DeliveryMode::Failover, None) => {
237                        failover = Some(targets.len());
238                        targets.push(vec![name.clone()]);
239                    }
240                }
241            }
242        }
243
244        if targets.is_empty() && suppressed.is_empty() {
245            targets.push(vec!["console".to_string()]);
246        }
247        DeliveryPlan {
248            targets,
249            suppressed,
250        }
251    }
252}
253
254#[cfg(test)]
255mod tests {
256    use super::*;
257    use crate::signal::SignalInput;
258
259    fn sample_signal(severity: Severity, env: &str) -> Signal {
260        Signal::from_input(SignalInput {
261            source: "airflow".into(),
262            event_type: "task_failed".into(),
263            title: "test".into(),
264            severity,
265            environment: Some(env.to_string()),
266            ..Default::default()
267        })
268    }
269
270    #[test]
271    fn unmatched_signal_falls_back_to_console() {
272        let config = RoutingConfig {
273            routes: vec![Route {
274                match_rule: RouteMatch {
275                    severity: vec![Severity::Critical],
276                    environment: None,
277                    source: None,
278                },
279                providers: vec!["slack".into()],
280                mode: DeliveryMode::Failover,
281                dedup: None,
282                rate_limit: None,
283            }],
284        };
285        let signal = sample_signal(Severity::Info, "dev");
286        assert_eq!(config.resolve(&signal), vec!["console".to_string()]);
287    }
288
289    #[test]
290    fn matched_signal_routes_to_configured_provider() {
291        let config = RoutingConfig {
292            routes: vec![Route {
293                match_rule: RouteMatch {
294                    severity: vec![Severity::Critical, Severity::Error],
295                    environment: Some("prod".into()),
296                    source: None,
297                },
298                providers: vec!["slack".into()],
299                mode: DeliveryMode::Failover,
300                dedup: None,
301                rate_limit: None,
302            }],
303        };
304        let signal = sample_signal(Severity::Error, "prod");
305        assert_eq!(config.resolve(&signal), vec!["slack".to_string()]);
306    }
307
308    fn route(
309        severity: &[Severity],
310        environment: Option<&str>,
311        source: Option<&str>,
312        providers: &[&str],
313    ) -> Route {
314        Route {
315            match_rule: RouteMatch {
316                severity: severity.to_vec(),
317                environment: environment.map(str::to_string),
318                source: source.map(str::to_string),
319            },
320            providers: providers.iter().map(|p| p.to_string()).collect(),
321            mode: DeliveryMode::Failover,
322            dedup: None,
323            rate_limit: None,
324        }
325    }
326
327    fn fanout(route: Route) -> Route {
328        Route {
329            mode: DeliveryMode::Fanout,
330            ..route
331        }
332    }
333
334    fn targets(config: &RoutingConfig, signal: &Signal) -> Vec<Vec<String>> {
335        config.plan(signal).targets
336    }
337
338    fn names(targets: &[&[&str]]) -> Vec<Vec<String>> {
339        targets
340            .iter()
341            .map(|t| t.iter().map(|p| p.to_string()).collect())
342            .collect()
343    }
344
345    #[test]
346    fn empty_routing_config_falls_back_to_console() {
347        let config = RoutingConfig { routes: vec![] };
348        assert_eq!(
349            config.resolve(&sample_signal(Severity::Critical, "prod")),
350            vec!["console".to_string()]
351        );
352    }
353
354    #[test]
355    fn empty_match_matches_every_signal() {
356        let config = RoutingConfig {
357            routes: vec![route(&[], None, None, &["webhook"])],
358        };
359        assert_eq!(
360            config.resolve(&sample_signal(Severity::Info, "dev")),
361            vec!["webhook".to_string()]
362        );
363    }
364
365    #[test]
366    fn every_field_in_a_match_must_hold() {
367        let config = RoutingConfig {
368            routes: vec![route(
369                &[Severity::Error],
370                Some("prod"),
371                Some("airflow"),
372                &["slack"],
373            )],
374        };
375        assert_eq!(
376            config.resolve(&sample_signal(Severity::Error, "prod")),
377            vec!["slack".to_string()]
378        );
379        // Right severity and source, wrong environment.
380        assert_eq!(
381            config.resolve(&sample_signal(Severity::Error, "staging")),
382            vec!["console".to_string()]
383        );
384        // Right severity and environment, wrong source.
385        let mut other_source = sample_signal(Severity::Error, "prod");
386        other_source.source = "jenkins".into();
387        assert_eq!(config.resolve(&other_source), vec!["console".to_string()]);
388    }
389
390    #[test]
391    fn environment_rule_does_not_match_signal_without_environment() {
392        let config = RoutingConfig {
393            routes: vec![route(&[], Some("prod"), None, &["slack"])],
394        };
395        let mut signal = sample_signal(Severity::Error, "prod");
396        signal.environment = None;
397        assert_eq!(config.resolve(&signal), vec!["console".to_string()]);
398    }
399
400    #[test]
401    fn all_matching_routes_contribute_providers_in_config_order() {
402        let config = RoutingConfig {
403            routes: vec![
404                route(&[Severity::Error], None, None, &["slack"]),
405                route(&[Severity::Warning], None, None, &["console"]),
406                route(&[], Some("prod"), None, &["webhook"]),
407            ],
408        };
409        assert_eq!(
410            config.resolve(&sample_signal(Severity::Error, "prod")),
411            vec!["slack".to_string(), "webhook".to_string()]
412        );
413    }
414
415    #[test]
416    fn provider_named_by_several_routes_is_listed_once_at_its_first_position() {
417        let config = RoutingConfig {
418            routes: vec![
419                route(&[Severity::Error], None, None, &["slack", "webhook"]),
420                route(&[], Some("prod"), None, &["webhook", "slack"]),
421            ],
422        };
423        assert_eq!(
424            config.resolve(&sample_signal(Severity::Error, "prod")),
425            vec!["slack".to_string(), "webhook".to_string()]
426        );
427    }
428
429    #[test]
430    fn failover_routes_share_one_target() {
431        let config = RoutingConfig {
432            routes: vec![
433                route(&[Severity::Error], None, None, &["pagerduty", "teams"]),
434                route(&[], Some("prod"), None, &["slack"]),
435            ],
436        };
437        assert_eq!(
438            targets(&config, &sample_signal(Severity::Error, "prod")),
439            names(&[&["pagerduty", "teams", "slack"]])
440        );
441    }
442
443    #[test]
444    fn fanout_route_gives_each_provider_its_own_target() {
445        let config = RoutingConfig {
446            routes: vec![fanout(route(&[], None, None, &["slack", "pagerduty"]))],
447        };
448        assert_eq!(
449            targets(&config, &sample_signal(Severity::Error, "prod")),
450            names(&[&["slack"], &["pagerduty"]])
451        );
452    }
453
454    #[test]
455    fn mixed_modes_keep_the_failover_chain_where_its_first_route_is() {
456        let config = RoutingConfig {
457            routes: vec![
458                route(&[Severity::Error], None, None, &["teams"]),
459                fanout(route(&[], Some("prod"), None, &["slack", "pagerduty"])),
460                route(&[], None, None, &["email"]),
461            ],
462        };
463        assert_eq!(
464            targets(&config, &sample_signal(Severity::Error, "prod")),
465            names(&[&["teams", "email"], &["slack"], &["pagerduty"]])
466        );
467    }
468
469    #[test]
470    fn provider_in_both_modes_is_used_once_at_its_first_position() {
471        let config = RoutingConfig {
472            routes: vec![
473                fanout(route(&[], None, None, &["slack", "pagerduty"])),
474                route(&[], None, None, &["pagerduty", "teams"]),
475            ],
476        };
477        assert_eq!(
478            targets(&config, &sample_signal(Severity::Error, "prod")),
479            names(&[&["slack"], &["pagerduty"], &["teams"]])
480        );
481    }
482
483    #[test]
484    fn unmatched_fanout_route_still_falls_back_to_console() {
485        let config = RoutingConfig {
486            routes: vec![fanout(route(&[Severity::Critical], None, None, &["slack"]))],
487        };
488        assert_eq!(
489            targets(&config, &sample_signal(Severity::Info, "dev")),
490            names(&[&["console"]])
491        );
492    }
493
494    #[test]
495    fn mode_parses_and_defaults_to_failover() {
496        let routes: Vec<Route> = serde_yaml::from_str(
497            "- match: {}\n  providers: [slack]\n- match: {}\n  providers: [slack]\n  mode: fanout\n",
498        )
499        .expect("routes");
500        assert_eq!(routes[0].mode, DeliveryMode::Failover);
501        assert_eq!(routes[1].mode, DeliveryMode::Fanout);
502    }
503
504    #[test]
505    fn unknown_mode_or_route_key_is_rejected() {
506        let bad_mode: Result<Vec<Route>, _> =
507            serde_yaml::from_str("- match: {}\n  providers: [slack]\n  mode: broadcast\n");
508        assert!(bad_mode.is_err());
509        let typo: Result<Vec<Route>, _> =
510            serde_yaml::from_str("- match: {}\n  providers: [slack]\n  mod: fanout\n");
511        assert!(typo.is_err());
512    }
513
514    #[test]
515    fn plan_with_skips_routes_the_gate_turns_down() {
516        let mut limited = fanout(route(&[], None, None, &["pagerduty", "slack"]));
517        limited.dedup = Some("10m".parse().expect("window"));
518        let config = RoutingConfig {
519            routes: vec![limited, route(&[], None, None, &["teams", "slack"])],
520        };
521        let plan = config.plan_with(&sample_signal(Severity::Error, "prod"), |_| {
522            Err("duplicate".to_string())
523        });
524        // slack is still sent, now from the failover route.
525        assert_eq!(plan.targets, names(&[&["teams", "slack"]]));
526        assert_eq!(
527            plan.suppressed,
528            vec!["route [pagerduty, slack]: duplicate".to_string()]
529        );
530    }
531
532    #[test]
533    fn gate_is_asked_only_about_throttled_routes() {
534        let mut limited = route(&[], None, None, &["pagerduty"]);
535        limited.rate_limit = Some(RateLimit {
536            max: 1,
537            per: "1m".parse().expect("window"),
538        });
539        let config = RoutingConfig {
540            routes: vec![route(&[], None, None, &["slack"]), limited],
541        };
542        let mut asked = Vec::new();
543        let plan = config.plan_with(&sample_signal(Severity::Error, "prod"), |r| {
544            asked.push(r.providers.clone());
545            Ok(())
546        });
547        assert_eq!(asked, vec![vec!["pagerduty".to_string()]]);
548        assert_eq!(plan.targets, names(&[&["slack", "pagerduty"]]));
549    }
550
551    #[test]
552    fn every_matching_route_suppressed_gives_an_empty_plan_not_console() {
553        let mut limited = route(&[Severity::Error], None, None, &["pagerduty"]);
554        limited.dedup = Some("1m".parse().expect("window"));
555        let config = RoutingConfig {
556            routes: vec![limited],
557        };
558        let plan = config.plan_with(&sample_signal(Severity::Error, "prod"), |_| {
559            Err("rate limited".to_string())
560        });
561        assert!(plan.targets.is_empty());
562        assert_eq!(plan.suppressed.len(), 1);
563        // A signal no route matches still goes to console.
564        let unmatched = config.plan_with(&sample_signal(Severity::Info, "prod"), |_| {
565            Err("never asked".to_string())
566        });
567        assert_eq!(unmatched.targets, names(&[&["console"]]));
568        assert!(unmatched.suppressed.is_empty());
569    }
570
571    #[test]
572    fn windows_parse_and_print() {
573        for (text, secs, shown) in [
574            ("30s", 30, "30s"),
575            ("90", 90, "90s"),
576            ("10m", 600, "10m"),
577            ("120m", 7200, "2h"),
578            ("1d", 86400, "1d"),
579        ] {
580            let w: Window = text.parse().expect(text);
581            assert_eq!(w.0, Duration::from_secs(secs));
582            assert_eq!(w.to_string(), shown);
583        }
584        for bad in ["", "m", "10 minutes", "-5m", "1.5h"] {
585            assert!(bad.parse::<Window>().is_err(), "{bad}");
586        }
587    }
588
589    #[test]
590    fn unknown_match_key_is_rejected() {
591        let typo: Result<Vec<Route>, _> =
592            serde_yaml::from_str("- match: {enviroment: prod}\n  providers: [slack]\n");
593        assert!(typo.is_err());
594    }
595}
opssignal_core/routing.rs

opssignal_core/
routing.rs
