1use std::path::Path;
62use std::sync::{Arc, Mutex};
63use std::time::Duration;
64
65use bytes::Bytes;
66use http_body_util::{BodyExt, Full};
67use mira_core::json::Json;
68use mira_core::query::{self, Op, Search, Signal, Target, Value};
69use yaml_rust2::Yaml;
70
71use crate::api::{self, Api};
72use crate::config;
73
74pub struct Rules {
76 pub every: Duration,
78 pub link_base: String,
82 pub rules: Vec<Rule>,
83 pub targets: Vec<Target_>,
84}
85
86pub struct Rule {
87 pub name: String,
88 query: Search,
90 of: Option<Search>,
92 over: Duration,
93 metric: Metric,
94 cmp: Cmp,
95 threshold: f64,
96 hold: Duration,
99 severity: String,
100 notify: Vec<usize>,
103}
104
105pub struct Target_ {
112 pub name: String,
113 url: String,
114 format: Format,
115 key: String,
117}
118
119#[derive(Clone, Copy, PartialEq, Eq)]
120enum Format {
121 Slack,
122 Discord,
123 Pagerduty,
124 Json,
126}
127
128#[derive(Clone, Copy, PartialEq, Eq)]
129enum Metric {
130 Count,
131 Ratio,
132}
133
134#[derive(Clone, Copy)]
135enum Cmp {
136 Gt,
137 Gte,
138 Lt,
139 Lte,
140}
141
142impl Cmp {
143 fn holds(self, v: f64, t: f64) -> bool {
144 match self {
145 Cmp::Gt => v > t,
146 Cmp::Gte => v >= t,
147 Cmp::Lt => v < t,
148 Cmp::Lte => v <= t,
149 }
150 }
151
152 fn as_str(self) -> &'static str {
153 match self {
154 Cmp::Gt => ">",
155 Cmp::Gte => ">=",
156 Cmp::Lt => "<",
157 Cmp::Lte => "<=",
158 }
159 }
160}
161
162#[derive(Default, Clone)]
164pub struct State {
165 since: Option<i64>,
169 firing: Option<i64>,
170 value: f64,
171 matched: usize,
172 total: Option<usize>,
173 error: Option<String>,
176 at: i64,
177}
178
179impl State {
180 fn advance(&mut self, breaching: bool, now: i64, hold: i64) -> Option<bool> {
190 match (breaching, self.since, self.firing) {
191 (false, _, Some(_)) => {
192 (self.since, self.firing) = (None, None);
193 Some(false)
194 }
195 (false, _, None) => {
196 self.since = None;
197 None
198 }
199 (true, None, _) => {
200 self.since = Some(now);
201 (hold == 0).then(|| {
204 self.firing = Some(now);
205 true
206 })
207 }
208 (true, Some(began), None) if now - began >= hold => {
209 self.firing = Some(now);
210 Some(true)
211 }
212 (true, Some(_), _) => None,
213 }
214 }
215
216 fn phase(&self) -> &'static str {
217 match (self.firing.is_some(), self.since.is_some()) {
218 (true, _) => "firing",
219 (_, true) => "pending",
220 _ => "ok",
221 }
222 }
223}
224
225impl Rules {
228 pub fn off() -> Rules {
231 Rules {
232 every: Duration::from_secs(15),
233 link_base: String::new(),
234 rules: Vec::new(),
235 targets: Vec::new(),
236 }
237 }
238
239 pub fn load(path: &Path) -> Result<Rules, String> {
240 let text = std::fs::read_to_string(path).map_err(|e| format!("{}: {e}", path.display()))?;
241 Rules::parse(&text).map_err(|e| format!("{}: {e}", path.display()))
242 }
243
244 pub fn parse(text: &str) -> Result<Rules, String> {
245 let doc = api::parse(text)?;
246 api::known(&doc, &["every", "link_base", "notify", "rules"])?;
247 let every = match doc["every"].as_str() {
248 Some(s) => config::duration(s).map_err(|e| format!("every: {e}"))?,
249 None => Duration::from_secs(15),
250 };
251 let link_base = match &doc["link_base"] {
252 Yaml::BadValue | Yaml::Null => String::new(),
253 y => y
254 .as_str()
255 .ok_or("link_base must be a quoted string")?
256 .trim_end_matches('/')
257 .to_owned(),
258 };
259 let targets = match &doc["notify"] {
260 Yaml::BadValue | Yaml::Null => Vec::new(),
261 Yaml::Array(a) => a.iter().map(target).collect::<Result<Vec<_>, _>>()?,
262 _ => return Err("`notify` must be a list of webhook targets".into()),
263 };
264 let rules = match &doc["rules"] {
265 Yaml::Array(a) => a
266 .iter()
267 .map(|y| rule(y, &targets))
268 .collect::<Result<Vec<_>, _>>()?,
269 _ => return Err("`rules` must be a list of rules".into()),
270 };
271 for (i, r) in rules.iter().enumerate() {
276 if rules[..i].iter().any(|o| o.name == r.name) {
277 return Err(format!("two rules named {:?}", r.name));
278 }
279 }
280 Ok(Rules {
281 every,
282 link_base,
283 rules,
284 targets,
285 })
286 }
287}
288
289fn target(y: &Yaml) -> Result<Target_, String> {
290 api::known(y, &["name", "url", "format", "key"])?;
291 let name = y["name"]
292 .as_str()
293 .ok_or("a notify target needs a quoted `name`")?
294 .to_owned();
295 let url = y["url"]
296 .as_str()
297 .ok_or_else(|| format!("notify {name:?}: needs a quoted `url`"))?
298 .to_owned();
299 let format = match y["format"].as_str().unwrap_or("json") {
300 "slack" => Format::Slack,
301 "discord" => Format::Discord,
302 "pagerduty" => Format::Pagerduty,
303 "json" => Format::Json,
304 other => {
305 return Err(format!(
306 "notify {name:?}: unknown format {other:?}; expected slack discord pagerduty json"
307 ));
308 }
309 };
310 if url.starts_with("https://") && !cfg!(feature = "webhook-tls") {
311 return Err(format!(
314 "notify {name:?}: this build posts over HTTP only. Rebuild with \
315 `--features webhook-tls` for a direct https:// target, or point it \
316 at a local egress proxy."
317 ));
318 }
319 if !url.starts_with("http://") && !url.starts_with("https://") {
320 return Err(format!("notify {name:?}: url must be http:// or https://"));
321 }
322 if format == Format::Pagerduty && y["key"].as_str().unwrap_or_default().is_empty() {
323 return Err(format!(
324 "notify {name:?}: pagerduty needs `key`, the Events v2 routing key"
325 ));
326 }
327 Ok(Target_ {
328 name,
329 url,
330 format,
331 key: y["key"].as_str().unwrap_or_default().to_owned(),
332 })
333}
334
335fn rule(y: &Yaml, targets: &[Target_]) -> Result<Rule, String> {
336 api::known(
337 y,
338 &[
339 "name", "query", "of", "over", "when", "for", "severity", "notify",
340 ],
341 )?;
342 let name = y["name"]
343 .as_str()
344 .ok_or("a rule needs a quoted `name`")?
345 .to_owned();
346 let at = |e: String| format!("rule {name:?}: {e}");
347 let query = windowless(&y["query"], "query").map_err(at)?;
348 let of = match &y["of"] {
349 Yaml::BadValue | Yaml::Null => None,
350 d => Some(windowless(d, "of").map_err(at)?),
351 };
352 let over = config::duration(y["over"].as_str().unwrap_or("1m")).map_err(&at)?;
353 let hold = config::duration(y["for"].as_str().unwrap_or("0s")).map_err(&at)?;
354 let (metric, cmp, threshold) = when(y["when"].as_str().unwrap_or_default()).map_err(at)?;
355 if metric == Metric::Ratio && of.is_none() {
356 return Err(at("`ratio` needs `of`, the denominator query".into()));
357 }
358 if metric == Metric::Count && of.is_some() {
359 return Err(at(
360 "`of` is the denominator of a `ratio`; `count` has none".into()
361 ));
362 }
363 let notify = match &y["notify"] {
364 Yaml::BadValue | Yaml::Null => Vec::new(),
368 Yaml::Array(a) => a
369 .iter()
370 .map(|n| {
371 let n = n
372 .as_str()
373 .ok_or_else(|| at("notify names are strings".into()))?;
374 targets
375 .iter()
376 .position(|t| t.name == n)
377 .ok_or_else(|| at(format!("notify {n:?} is not a target in `notify`")))
378 })
379 .collect::<Result<Vec<_>, _>>()?,
380 _ => return Err(at("`notify` must be a list of target names".into())),
381 };
382 Ok(Rule {
383 name,
384 query,
385 of,
386 over,
387 metric,
388 cmp,
389 threshold,
390 hold,
391 severity: y["severity"].as_str().unwrap_or("warning").to_owned(),
392 notify,
393 })
394}
395
396fn windowless(doc: &Yaml, field: &str) -> Result<Search, String> {
403 if doc.is_badvalue() || doc.is_null() {
404 return Err(format!("`{field}` is required and is a query document"));
405 }
406 for k in ["from", "to", "limit", "after"] {
407 if !doc[k].is_badvalue() {
408 return Err(format!(
409 "{field}: `{k}` is the engine's; the window is `over` and the \
410 limit is always zero because this counts rather than reads"
411 ));
412 }
413 }
414 let mut s = api::search_doc(doc, 0)?;
415 s.limit = 0;
416 Ok(s)
417}
418
419fn when(s: &str) -> Result<(Metric, Cmp, f64), String> {
424 let bad = || {
425 format!(
426 "when: expected `count <op> <number>` or `ratio <op> <number>`, \
427 op one of > >= < <=, got {s:?}"
428 )
429 };
430 let (cmp, at) = [
432 (Cmp::Gte, ">="),
433 (Cmp::Lte, "<="),
434 (Cmp::Gt, ">"),
435 (Cmp::Lt, "<"),
436 ]
437 .into_iter()
438 .find_map(|(c, sym)| s.find(sym).map(|i| (c, (i, sym.len()))))
439 .ok_or_else(bad)?;
440 let metric = match s[..at.0].trim() {
441 "count" => Metric::Count,
442 "ratio" => Metric::Ratio,
443 _ => return Err(bad()),
444 };
445 let rhs = s[at.0 + at.1..].trim();
446 let (num, scale) = match rhs.strip_suffix('%') {
449 Some(n) => (n.trim(), 0.01),
450 None => (rhs, 1.0),
451 };
452 let v: f64 = num.parse().map_err(|_| bad())?;
453 Ok((metric, cmp, v * scale))
454}
455
456pub struct Engine {
459 pub rules: Rules,
460 state: Mutex<Vec<State>>,
461}
462
463impl Default for Engine {
464 fn default() -> Engine {
465 Engine::new(Rules::off())
466 }
467}
468
469impl Engine {
470 pub fn new(rules: Rules) -> Engine {
471 let state = Mutex::new(vec![State::default(); rules.rules.len()]);
472 Engine { rules, state }
473 }
474
475 pub async fn tick(&self, api: &Api) {
477 let now = api::now_nanos();
478 for (i, r) in self.rules.rules.iter().enumerate() {
479 let (value, matched, total, error) = match count_pair(api, r, now).await {
480 Ok(v) => v,
481 Err(e) => {
482 let mut st = self.state.lock().expect("alert state");
483 st[i].error = Some(e.clone());
484 st[i].at = now;
485 tracing::warn!(rule = %r.name, error = %e, "alert rule failed");
486 continue;
487 }
488 };
489 let breaching = r.cmp.holds(value, r.threshold);
490
491 let event = {
495 let mut st = self.state.lock().expect("alert state");
496 let s = &mut st[i];
497 (s.value, s.matched, s.total, s.error, s.at) = (value, matched, total, error, now);
498 s.advance(breaching, now, r.hold.as_nanos() as i64)
499 .map(|firing| (firing, s.clone()))
500 };
501 if let Some((firing, snapshot)) = event {
502 self.dispatch(r, &snapshot, firing).await;
503 }
504 }
505 }
506
507 async fn dispatch(&self, r: &Rule, s: &State, firing: bool) {
508 tracing::info!(
509 rule = %r.name, severity = %r.severity, value = s.value,
510 state = if firing { "firing" } else { "resolved" },
511 "alert"
512 );
513 let link = link(&self.rules.link_base, r);
514 for &t in &r.notify {
515 let t = &self.rules.targets[t];
516 let body = payload(t, r, s, firing, &link);
517 if let Err(e) = post(&t.url, body).await {
518 tracing::warn!(rule = %r.name, target = %t.name, error = %e, "webhook failed");
519 }
520 }
521 }
522
523 pub fn json(&self) -> String {
525 let st = self.state.lock().expect("alert state");
526 let mut j = Json::new();
527 j.obj(|j| {
528 j.key("alerts");
529 j.arr(|j| {
530 for (r, s) in self.rules.rules.iter().zip(st.iter()) {
531 j.obj(|j| {
532 j.key("name");
533 j.str(&r.name);
534 j.key("state");
535 j.str(s.phase());
536 j.key("severity");
537 j.str(&r.severity);
538 j.key("metric");
539 j.str(match r.metric {
540 Metric::Count => "count",
541 Metric::Ratio => "ratio",
542 });
543 j.key("op");
544 j.str(r.cmp.as_str());
545 j.key("threshold");
546 j.f64(r.threshold);
547 j.key("value");
548 j.f64(s.value);
549 j.key("matched");
550 j.u64(s.matched as u64);
551 j.key("total");
552 match s.total {
553 Some(t) => j.u64(t as u64),
554 None => j.null(),
555 }
556 j.key("over_nano");
557 j.u64_str(r.over.as_nanos() as u64);
558 j.key("for_nano");
559 j.u64_str(r.hold.as_nanos() as u64);
560 j.key("since");
561 match s.since {
562 Some(t) => j.i64_str(t),
563 None => j.null(),
564 }
565 j.key("firing_since");
566 match s.firing {
567 Some(t) => j.i64_str(t),
568 None => j.null(),
569 }
570 j.key("evaluated_at");
571 j.i64_str(s.at);
572 j.key("signal");
573 j.str(signal_name(r));
574 j.key("filter");
577 j.str(&filter_of(r));
578 j.key("link");
579 j.str(&link(&self.rules.link_base, r));
580 j.key("error");
581 match &s.error {
582 Some(e) => j.str(e),
583 None => j.null(),
584 }
585 });
586 }
587 });
588 j.key("every_nano");
589 j.u64_str(self.rules.every.as_nanos() as u64);
590 });
591 j.into_string()
592 }
593}
594
595async fn count_pair(
601 api: &Api,
602 r: &Rule,
603 now: i64,
604) -> Result<(f64, usize, Option<usize>, Option<String>), String> {
605 let from = now - r.over.as_nanos() as i64;
606 let matched = count(api, &r.query, from, now).await?;
607 let total = match &r.of {
608 Some(q) => Some(count(api, q, from, now).await?),
609 None => None,
610 };
611 let value = match (r.metric, total) {
612 (Metric::Count, _) => matched as f64,
613 (Metric::Ratio, Some(0)) | (Metric::Ratio, None) => 0.0,
617 (Metric::Ratio, Some(t)) => matched as f64 / t as f64,
618 };
619 Ok((value, matched, total, None))
620}
621
622async fn count(api: &Api, q: &Search, from: i64, to: i64) -> Result<usize, String> {
623 let mut q = q.clone();
624 (q.from, q.to, q.limit) = (from, to, 0);
625 let dir = api.data_dir.clone();
626 let open = api.open(q.signal.dir()).await;
627 tokio::task::spawn_blocking(move || query::search_open(&dir, &q, &open))
631 .await
632 .map_err(|e| e.to_string())?
633 .map(|r| r.stats.rows_matched)
634 .map_err(|e| e.to_string())
635}
636
637fn link(base: &str, r: &Rule) -> String {
648 if base.is_empty() {
649 return String::new();
650 }
651 let range = format!("-{}s", r.over.as_secs().max(1));
652 format!(
653 "{base}/#/{}?q={}&range={range}",
654 signal_name(r),
655 urlencode(&filter_of(r))
656 )
657}
658
659fn signal_name(r: &Rule) -> &'static str {
660 match r.query.signal {
661 Signal::Logs => "logs",
662 Signal::Traces => "traces",
663 }
664}
665
666fn filter_of(r: &Rule) -> String {
674 let q: Vec<String> = r
675 .query
676 .terms
677 .iter()
678 .map(|t| {
679 let (kind, key) = match &t.target {
680 Target::Field(f) => ("field", f.as_str()),
681 Target::Attr(a) => ("attr", a.as_str()),
682 };
683 let v = match &t.value {
686 Value::Str(s) => s.clone(),
687 Value::Int(i) => i.to_string(),
688 Value::Double(d) => d.to_string(),
689 Value::Bool(b) => b.to_string(),
690 };
691 let v = if v.contains(' ') || v.is_empty() {
692 format!("\"{}\"", v.replace('"', ""))
693 } else {
694 v
695 };
696 format!("{kind}:{key}{}{v}", op_symbol(t.op))
697 })
698 .collect();
699 q.join(" ")
700}
701
702fn op_symbol(op: Op) -> &'static str {
703 match op {
704 Op::Eq => "=",
705 Op::Ne => "!=",
706 Op::Lt => "<",
707 Op::Lte => "<=",
708 Op::Gt => ">",
709 Op::Gte => ">=",
710 Op::Contains => "~",
711 }
712}
713
714fn urlencode(s: &str) -> String {
719 let mut out = String::with_capacity(s.len());
720 for b in s.bytes() {
721 match b {
722 b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
723 out.push(b as char);
724 }
725 _ => out.push_str(&format!("%{b:02X}")),
726 }
727 }
728 out
729}
730
731fn summary(r: &Rule, s: &State, firing: bool) -> String {
735 let value = match r.metric {
736 Metric::Ratio => format!("{:.2}%", s.value * 100.0),
737 Metric::Count => format!("{:.0}", s.value),
738 };
739 let threshold = match r.metric {
740 Metric::Ratio => format!("{:.2}%", r.threshold * 100.0),
741 Metric::Count => format!("{:.0}", r.threshold),
742 };
743 let head = if firing { "FIRING" } else { "RESOLVED" };
744 let over = human(r.over);
745 match (r.metric, s.total) {
746 (Metric::Ratio, Some(t)) => format!(
747 "[{head}] {} — {} {} {} over {over} ({} of {t} records)",
748 r.name,
749 value,
750 r.cmp.as_str(),
751 threshold,
752 s.matched
753 ),
754 _ => format!(
755 "[{head}] {} — {} records {} {} over {over}",
756 r.name,
757 value,
758 r.cmp.as_str(),
759 threshold
760 ),
761 }
762}
763
764fn human(d: Duration) -> String {
765 let s = d.as_secs();
766 match s {
767 0 => "0s".into(),
768 s if s % 86_400 == 0 => format!("{}d", s / 86_400),
769 s if s % 3_600 == 0 => format!("{}h", s / 3_600),
770 s if s % 60 == 0 => format!("{}m", s / 60),
771 s => format!("{s}s"),
772 }
773}
774
775fn payload(t: &Target_, r: &Rule, s: &State, firing: bool, link: &str) -> String {
776 let text = summary(r, s, firing);
777 let mut j = Json::new();
778 match t.format {
779 Format::Slack => j.obj(|j| {
780 j.key("text");
781 j.str(&match link.is_empty() {
784 true => text.clone(),
785 false => format!("{text}\n<{link}|open in Mira>"),
786 });
787 }),
788 Format::Discord => j.obj(|j| {
789 j.key("content");
790 j.str(&match link.is_empty() {
791 true => text.clone(),
792 false => format!("{text}\n{link}"),
793 });
794 }),
795 Format::Pagerduty => j.obj(|j| {
796 j.key("routing_key");
797 j.str(&t.key);
798 j.key("event_action");
799 j.str(if firing { "trigger" } else { "resolve" });
800 j.key("dedup_key");
803 j.str(&r.name);
804 j.key("payload");
805 j.obj(|j| {
806 j.key("summary");
807 j.str(&text);
808 j.key("severity");
809 j.str(match r.severity.as_str() {
813 s @ ("critical" | "error" | "warning" | "info") => s,
814 _ => "warning",
815 });
816 j.key("source");
817 j.str("mira");
818 });
819 if !link.is_empty() {
820 j.key("links");
821 j.arr(|j| {
822 j.obj(|j| {
823 j.key("href");
824 j.str(link);
825 j.key("text");
826 j.str("open in Mira");
827 });
828 });
829 }
830 }),
831 Format::Json => j.raw(&alert_json(r, s, firing, link)),
832 }
833 j.into_string()
834}
835
836fn alert_json(r: &Rule, s: &State, firing: bool, link: &str) -> String {
837 let mut j = Json::new();
838 j.obj(|j| {
839 j.key("rule");
840 j.str(&r.name);
841 j.key("state");
842 j.str(if firing { "firing" } else { "resolved" });
843 j.key("severity");
844 j.str(&r.severity);
845 j.key("summary");
846 j.str(&summary(r, s, firing));
847 j.key("value");
848 j.f64(s.value);
849 j.key("threshold");
850 j.f64(r.threshold);
851 j.key("matched");
852 j.u64(s.matched as u64);
853 j.key("total");
854 match s.total {
855 Some(t) => j.u64(t as u64),
856 None => j.null(),
857 }
858 j.key("over_nano");
859 j.u64_str(r.over.as_nanos() as u64);
860 j.key("at");
861 j.i64_str(s.at);
862 j.key("link");
863 j.str(link);
864 });
865 j.into_string()
866}
867
868async fn post(url: &str, body: String) -> Result<(), String> {
875 let req = hyper::Request::builder()
876 .method(hyper::Method::POST)
877 .uri(url)
878 .header(hyper::header::CONTENT_TYPE, "application/json")
879 .body(Full::new(Bytes::from(body)))
880 .map_err(|e| e.to_string())?;
881 let fut = client().request(req);
882 let resp = tokio::time::timeout(WEBHOOK_TIMEOUT, fut)
883 .await
884 .map_err(|_| format!("no response in {}", human(WEBHOOK_TIMEOUT)))?
885 .map_err(|e| e.to_string())?;
886 let status = resp.status();
887 let _ = resp.into_body().collect().await;
890 match status.is_success() {
891 true => Ok(()),
892 false => Err(format!("HTTP {}", status.as_u16())),
893 }
894}
895
896const WEBHOOK_TIMEOUT: Duration = Duration::from_secs(10);
897
898type Client = hyper_util::client::legacy::Client<Connector, Full<Bytes>>;
899
900#[cfg(not(feature = "webhook-tls"))]
901type Connector = hyper_util::client::legacy::connect::HttpConnector;
902
903#[cfg(feature = "webhook-tls")]
904type Connector = hyper_rustls::HttpsConnector<hyper_util::client::legacy::connect::HttpConnector>;
905
906fn client() -> &'static Client {
909 static C: std::sync::OnceLock<Client> = std::sync::OnceLock::new();
910 C.get_or_init(|| {
911 let b = hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new());
912 #[cfg(not(feature = "webhook-tls"))]
913 {
914 b.build_http()
915 }
916 #[cfg(feature = "webhook-tls")]
917 {
918 b.build(
919 hyper_rustls::HttpsConnectorBuilder::new()
920 .with_webpki_roots()
921 .https_or_http()
922 .enable_http1()
923 .build(),
924 )
925 }
926 })
927}
928
929pub fn router(api: Api) -> axum::Router {
930 axum::Router::new()
931 .route("/api/v1/alerts", axum::routing::get(handler))
932 .with_state(api)
933}
934
935async fn handler(axum::extract::State(api): axum::extract::State<Api>) -> axum::response::Response {
940 use axum::response::IntoResponse;
941 (
942 [(axum::http::header::CONTENT_TYPE, "application/json")],
943 api.alerts.json(),
944 )
945 .into_response()
946}
947
948pub fn spawn(api: Api) {
950 let engine = Arc::clone(&api.alerts);
951 if engine.rules.rules.is_empty() {
952 return;
953 }
954 tracing::info!(
955 rules = engine.rules.rules.len(),
956 targets = engine.rules.targets.len(),
957 every = %human(engine.rules.every),
958 "alerting"
959 );
960 tokio::spawn(async move {
961 let mut tick = tokio::time::interval(engine.rules.every);
962 tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
966 loop {
967 tick.tick().await;
968 engine.tick(&api).await;
969 }
970 });
971}
972
973#[cfg(test)]
974mod tests {
975 use super::*;
976
977 const DOC: &str = r#"{
978 "every": "5s",
979 "link_base": "https://mira.example.com/",
980 "notify": [ { "name": "oncall", "url": "http://127.0.0.1:9/hook", "format": "slack" } ],
981 "rules": [
982 { "name": "checkout-errors",
983 "over": "1m", "for": "2m", "severity": "critical", "notify": ["oncall"],
984 "query": { "signal": "traces", "where": [
985 { "attr": "service.name", "eq": "checkout" },
986 { "field": "status_code", "eq": 2 } ] },
987 "of": { "signal": "traces", "where": [
988 { "attr": "service.name", "eq": "checkout" } ] },
989 "when": "ratio > 5%" },
990 { "name": "any-log", "query": { "signal": "logs" }, "when": "count>=1" }
991 ]
992 }"#;
993
994 #[test]
1001 fn the_shipped_example_rules_file_parses() {
1002 let r = Rules::parse(include_str!("../../../docs/e2e/alerts.kyaml")).expect("alerts.kyaml");
1003 assert_eq!(r.every, Duration::from_secs(15));
1004 assert_eq!(r.link_base, "http://localhost:4318");
1005 assert!(r.targets.is_empty());
1008 let names: Vec<&str> = r.rules.iter().map(|x| x.name.as_str()).collect();
1009 assert_eq!(
1010 names,
1011 [
1012 "shop-error-rate",
1013 "checkout-p95-latency",
1014 "card-declines",
1015 "inventory-outage"
1016 ]
1017 );
1018 let p95 = &r.rules[1];
1021 assert!(matches!(p95.metric, Metric::Ratio));
1022 assert!(p95.of.is_some());
1023 assert!((p95.threshold - 0.05).abs() < 1e-12);
1024
1025 let filters: Vec<String> = r.rules.iter().map(filter_of).collect();
1030 assert_eq!(
1031 filters,
1032 [
1033 "field:status_code=2",
1034 "attr:service.name=checkout field:duration_nano>250000000",
1035 "attr:exception.type=payments.CardDeclined",
1036 "attr:service.name=inventory field:severity_number>=21",
1037 ]
1038 );
1039 }
1040
1041 #[test]
1042 fn a_rules_file_parses_to_what_it_says() {
1043 let r = Rules::parse(DOC).unwrap();
1044 assert_eq!(r.every, Duration::from_secs(5));
1045 assert_eq!(r.link_base, "https://mira.example.com");
1047 assert_eq!(r.rules.len(), 2);
1048 let a = &r.rules[0];
1049 assert_eq!(a.threshold, 0.05);
1050 assert_eq!(a.hold, Duration::from_secs(120));
1051 assert_eq!(a.notify, vec![0]);
1052 assert!(a.of.is_some());
1053 assert_eq!(a.query.limit, 0);
1055 assert_eq!(r.rules[1].severity, "warning");
1056 }
1057
1058 #[test]
1059 fn a_percentile_threshold_is_a_ratio_threshold() {
1060 let r = Rules::parse(
1064 r#"{ "rules": [ { "name": "p95", "over": "5m", "when": "ratio > 5%",
1065 "query": { "signal": "traces", "where": [
1066 { "field": "duration_nano", "gt": 500000000 } ] },
1067 "of": { "signal": "traces" } } ] }"#,
1068 )
1069 .unwrap();
1070 let rule = &r.rules[0];
1071 assert!(matches!(rule.metric, Metric::Ratio));
1072 assert!(rule.cmp.holds(0.06, rule.threshold));
1073 assert!(!rule.cmp.holds(0.04, rule.threshold));
1074 }
1075
1076 #[test]
1083 fn every_operator_and_scalar_survives_the_trip_through_a_filter_box() {
1084 let r = Rules::parse(
1085 r#"{ "rules": [ { "name": "all-ops", "over": "1m", "when": "count > 0",
1086 "query": { "signal": "logs", "where": [
1087 { "attr": "service.name", "ne": "checkout" },
1088 { "field": "severity_number", "lt": 17 },
1089 { "field": "severity_number", "lte": 16 },
1090 { "attr": "http.route", "contains": "/api" },
1091 { "attr": "sampling.ratio", "eq": 0.25 },
1092 { "attr": "deployment.canary", "eq": true },
1093 { "attr": "http.target", "eq": "GET /a b" },
1094 { "attr": "empty", "eq": "" } ] } } ] }"#,
1095 )
1096 .unwrap();
1097 assert_eq!(
1098 filter_of(&r.rules[0]),
1099 "attr:service.name!=checkout field:severity_number<17 \
1100 field:severity_number<=16 attr:http.route~/api attr:sampling.ratio=0.25 \
1101 attr:deployment.canary=true attr:http.target=\"GET /a b\" attr:empty=\"\""
1102 );
1103 }
1104
1105 #[test]
1106 fn every_way_to_write_a_rule_wrong_is_refused_by_name() {
1107 let bad = |doc: &str, want: &str| {
1110 let e = Rules::parse(doc).err().expect("should not have parsed");
1111 assert!(e.contains(want), "{e:?} should mention {want:?}");
1112 };
1113 bad(
1114 r#"{ "rules": [ { "name": "a", "query": {}, "when": "count ~ 1" } ] }"#,
1115 "when",
1116 );
1117 bad(
1118 r#"{ "rules": [ { "name": "a", "query": {}, "when": "p95 > 1" } ] }"#,
1119 "when",
1120 );
1121 bad(
1122 r#"{ "rules": [ { "name": "a", "query": {}, "when": "ratio > 1" } ] }"#,
1123 "of",
1124 );
1125 bad(
1126 r#"{ "rules": [ { "name": "a", "query": {}, "of": {}, "when": "count > 1" } ] }"#,
1127 "denominator",
1128 );
1129 bad(
1132 r#"{ "rules": [ { "name": "a", "query": { "from": "-1h" }, "when": "count > 1" } ] }"#,
1133 "over",
1134 );
1135 bad(
1136 r#"{ "rules": [ { "name": "a", "when": "count > 1" } ] }"#,
1137 "required",
1138 );
1139 bad(
1140 r#"{ "rules": [ { "name": "a", "query": {}, "when": "count > 1", "nope": "x" } ] }"#,
1141 "nope",
1142 );
1143 bad(
1144 r#"{ "rules": [ { "name": "a", "query": {}, "when": "count>1", "notify": ["ghost"] } ] }"#,
1145 "ghost",
1146 );
1147 bad(
1150 r#"{ "notify": [ { "name": "n", "url": "http://x/" } ],
1151 "rules": [ { "name": "a", "query": {}, "when": "count>1", "notify": "n" } ] }"#,
1152 "list of target names",
1153 );
1154 bad(
1155 r#"{ "notify": [ { "name": "n", "url": "http://x/" } ],
1156 "rules": [ { "name": "a", "query": {}, "when": "count>1", "notify": [7] } ] }"#,
1157 "notify names are strings",
1158 );
1159 bad(
1160 r#"{ "rules": [ { "name": "a", "query": {}, "when": "count>1" },
1161 { "name": "a", "query": {}, "when": "count>1" } ] }"#,
1162 "two rules named",
1163 );
1164 bad(
1165 r#"{ "notify": [ { "name": "pd", "url": "http://x/", "format": "pagerduty" } ], "rules": [] }"#,
1166 "routing key",
1167 );
1168 bad(
1169 r#"{ "notify": [ { "name": "n", "url": "ftp://x/" } ], "rules": [] }"#,
1170 "http://",
1171 );
1172 bad(r#"{ "every": "soon", "rules": [] }"#, "every");
1175 bad(r#"{ "link_base": 4318, "rules": [] }"#, "link_base");
1176 bad(r#"{ "notify": { "name": "n" }, "rules": [] }"#, "notify");
1177 bad(r#"{ "rules": { "name": "a" } }"#, "rules");
1178 bad(r#"{ "rules": [], "alerts": [] }"#, "alerts");
1179 bad(
1180 r#"{ "notify": [ { "url": "http://x/" } ], "rules": [] }"#,
1181 "name",
1182 );
1183 bad(r#"{ "notify": [ { "name": "n" } ], "rules": [] }"#, "url");
1184 bad(
1185 r#"{ "notify": [ { "name": "n", "url": "http://x/", "format": "email" } ], "rules": [] }"#,
1186 "email",
1187 );
1188 bad(
1189 r#"{ "notify": [ { "name": "n", "url": "http://x/", "to": "me" } ], "rules": [] }"#,
1190 "to",
1191 );
1192 }
1193
1194 #[test]
1200 fn a_rules_file_is_loaded_by_path_and_names_the_path_when_it_cannot_be() {
1201 let dir = std::env::temp_dir().join(format!("mira-rules-{}", std::process::id()));
1202 std::fs::create_dir_all(&dir).expect("mkdir");
1203 let path = dir.join("alerts.kyaml");
1204 std::fs::write(&path, DOC).expect("write");
1205 let r = Rules::load(&path).expect("load");
1206 assert_eq!(r.rules.len(), 2);
1207
1208 std::fs::write(&path, "{ rules: nope }").expect("write");
1209 let e = Rules::load(&path).err().expect("should not have parsed");
1210 assert!(e.contains("alerts.kyaml") && e.contains("rules"), "{e}");
1211
1212 let missing = dir.join("gone.kyaml");
1213 let e = Rules::load(&missing).err().expect("should not have opened");
1214 assert!(e.contains("gone.kyaml"), "{e}");
1215 std::fs::remove_dir_all(&dir).ok();
1216 }
1217
1218 #[test]
1224 fn a_rule_can_fire_on_too_little_rather_than_too_much() {
1225 let r = Rules::parse(
1226 r#"{ "rules": [
1227 { "name": "traffic-gone", "over": "5m", "when": "count < 100",
1228 "query": { "signal": "traces" } },
1229 { "name": "success-rate", "over": "5m", "when": "ratio <= 99%",
1230 "query": { "signal": "traces", "where": [ { "field": "status_code", "eq": 1 } ] },
1231 "of": { "signal": "traces" } } ] }"#,
1232 )
1233 .expect("rules");
1234
1235 let quiet = &r.rules[0];
1236 assert_eq!(quiet.cmp.as_str(), "<");
1237 assert!(quiet.cmp.holds(3.0, 100.0));
1238 assert!(!quiet.cmp.holds(100.0, 100.0));
1239
1240 let rate = &r.rules[1];
1241 assert_eq!(rate.cmp.as_str(), "<=");
1242 assert!((rate.threshold - 0.99).abs() < 1e-12);
1243 assert!(rate.cmp.holds(0.99, 0.99));
1244 assert!(!rate.cmp.holds(0.999, 0.99));
1245
1246 let s = State {
1248 value: 3.0,
1249 matched: 3,
1250 total: None,
1251 at: 0,
1252 ..State::default()
1253 };
1254 assert_eq!(
1255 summary(quiet, &s, true),
1256 "[FIRING] traffic-gone — 3 records < 100 over 5m"
1257 );
1258 }
1259
1260 #[cfg(not(feature = "webhook-tls"))]
1261 #[test]
1262 fn an_https_target_is_refused_at_load_by_a_build_that_cannot_dial_it() {
1263 let e =
1264 Rules::parse(r#"{ "notify": [ { "name": "s", "url": "https://x/" } ], "rules": [] }"#)
1265 .err()
1266 .expect("an https target should not load in this build");
1267 assert!(e.contains("webhook-tls"), "{e:?}");
1268 }
1269
1270 #[test]
1274 fn for_needs_a_sustained_breach_not_a_repeated_one() {
1275 let e = Engine::new(Rules::parse(DOC).unwrap());
1276 let hold = e.rules.rules[0].hold.as_nanos() as i64;
1277 let step = |breaching: bool, now: i64| -> (Option<bool>, &'static str) {
1278 let mut st = e.state.lock().unwrap();
1279 let s = &mut st[0];
1280 (s.advance(breaching, now, hold), s.phase())
1281 };
1282 const MIN: i64 = 60_000_000_000;
1283 assert_eq!(step(true, 0), (None, "pending"));
1284 assert_eq!(step(false, MIN), (None, "ok"));
1286 assert_eq!(step(true, 2 * MIN), (None, "pending"));
1287 assert_eq!(step(true, 3 * MIN), (None, "pending"));
1288 assert_eq!(step(true, 4 * MIN), (Some(true), "firing"));
1289 assert_eq!(step(true, 5 * MIN), (None, "firing"));
1291 assert_eq!(step(false, 6 * MIN), (Some(false), "ok"));
1292 }
1293
1294 #[test]
1295 fn a_link_lands_on_the_rows_that_fired() {
1296 let r = Rules::parse(DOC).unwrap();
1297 let l = link(&r.link_base, &r.rules[0]);
1298 assert!(l.starts_with("https://mira.example.com/#/traces?q="), "{l}");
1299 assert!(l.contains("attr%3Aservice.name%3Dcheckout"), "{l}");
1302 assert!(l.contains("field%3Astatus_code%3D2"), "{l}");
1303 assert!(l.ends_with("&range=-60s"), "{l}");
1304 assert_eq!(link("", &r.rules[0]), "");
1306 }
1307
1308 #[test]
1309 fn each_format_says_the_same_thing_in_its_own_words() {
1310 let r = Rules::parse(DOC).unwrap();
1311 let rule = &r.rules[0];
1312 let s = State {
1313 value: 0.12,
1314 matched: 24,
1315 total: Some(200),
1316 ..State::default()
1317 };
1318 let link = link(&r.link_base, rule);
1319 let text = summary(rule, &s, true);
1320 assert!(text.contains("FIRING"), "{text}");
1321 assert!(text.contains("12.00%"), "{text}");
1322 assert!(text.contains("24 of 200"), "{text}");
1323 assert!(text.contains("over 1m"), "{text}");
1324
1325 let slack = payload(&r.targets[0], rule, &s, true, &link);
1326 assert!(slack.starts_with(r#"{"text":"[FIRING]"#), "{slack}");
1327 assert!(slack.contains("|open in Mira>"), "{slack}");
1328
1329 let pd = Target_ {
1330 name: "pd".into(),
1331 url: "http://x/".into(),
1332 format: Format::Pagerduty,
1333 key: "rk".into(),
1334 };
1335 let fire = payload(&pd, rule, &s, true, &link);
1336 assert!(fire.contains(r#""event_action":"trigger""#), "{fire}");
1337 assert!(fire.contains(r#""dedup_key":"checkout-errors""#), "{fire}");
1338 assert!(fire.contains(r#""severity":"critical""#), "{fire}");
1339 let clear = payload(&pd, rule, &s, false, &link);
1340 assert!(clear.contains(r#""event_action":"resolve""#), "{clear}");
1341 assert!(
1343 clear.contains(r#""dedup_key":"checkout-errors""#),
1344 "{clear}"
1345 );
1346
1347 let raw = Target_ {
1348 format: Format::Json,
1349 ..Target_ {
1350 name: "j".into(),
1351 url: "http://x/".into(),
1352 format: Format::Json,
1353 key: String::new(),
1354 }
1355 };
1356 let j = payload(&raw, rule, &s, true, &link);
1357 assert!(j.contains(r#""rule":"checkout-errors""#), "{j}");
1358 assert!(j.contains(r#""state":"firing""#), "{j}");
1359 assert!(j.contains(r#""total":200"#), "{j}");
1360 let counted = payload(&raw, &r.rules[1], &State::default(), true, "");
1364 assert!(counted.contains(r#""total":null"#), "{counted}");
1365 assert!(counted.contains(r#""matched":0"#), "{counted}");
1366
1367 let dis = Target_ {
1371 name: "d".into(),
1372 url: "http://x/".into(),
1373 format: Format::Discord,
1374 key: String::new(),
1375 };
1376 let d = payload(&dis, rule, &s, true, &link);
1377 assert!(d.starts_with(r#"{"content":"[FIRING]"#), "{d}");
1378 assert!(d.contains(&link), "{d}");
1379
1380 for t in [&r.targets[0], &dis] {
1383 let p = payload(t, rule, &s, true, "");
1384 assert!(p.ends_with(r#"over 1m (24 of 200 records)"}"#), "{p}");
1385 }
1386 }
1387
1388 #[test]
1389 fn a_severity_pagerduty_does_not_know_degrades_rather_than_400s() {
1390 let mut r = Rules::parse(DOC).unwrap();
1391 r.rules[0].severity = "sev1".into();
1392 let pd = Target_ {
1393 name: "pd".into(),
1394 url: "http://x/".into(),
1395 format: Format::Pagerduty,
1396 key: "rk".into(),
1397 };
1398 let out = payload(&pd, &r.rules[0], &State::default(), true, "");
1399 assert!(out.contains(r#""severity":"warning""#), "{out}");
1400 }
1401
1402 #[test]
1403 fn an_idle_service_is_not_a_hundred_percent_error_rate() {
1404 let r = Rules::parse(DOC).unwrap();
1408 let rule = &r.rules[0];
1409 assert!(!rule.cmp.holds(0.0, rule.threshold));
1410 }
1411
1412 #[test]
1413 fn durations_round_trip_the_way_they_were_written() {
1414 assert_eq!(human(Duration::from_secs(60)), "1m");
1415 assert_eq!(human(Duration::from_secs(90)), "90s");
1416 assert_eq!(human(Duration::from_secs(7200)), "2h");
1417 assert_eq!(human(Duration::from_secs(86_400)), "1d");
1418 assert_eq!(human(Duration::ZERO), "0s");
1419 }
1420
1421 #[tokio::test]
1436 async fn a_webhook_that_refuses_the_page_is_a_failure_that_names_the_status() {
1437 use std::io::{Read, Write};
1438 let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
1439 let url = format!("http://{}/hook", listener.local_addr().expect("addr"));
1440 let seen = std::thread::spawn(move || {
1441 let (mut sock, _) = listener.accept().expect("accept");
1442 sock.set_read_timeout(Some(Duration::from_secs(10)))
1446 .expect("timeout");
1447 sock.set_write_timeout(Some(Duration::from_secs(10)))
1448 .expect("timeout");
1449 let mut bodies = Vec::new();
1450 for i in 0..2 {
1451 let mut head = Vec::new();
1452 let mut byte = [0u8; 1];
1453 while !head.ends_with(b"\r\n\r\n") && sock.read(&mut byte).unwrap_or(0) == 1 {
1454 head.push(byte[0]);
1455 }
1456 let text = String::from_utf8_lossy(&head).to_lowercase();
1457 let len: usize = text
1458 .split("content-length:")
1459 .nth(1)
1460 .and_then(|t| t.split("\r\n").next())
1461 .and_then(|t| t.trim().parse().ok())
1462 .unwrap_or(0);
1463 let mut body = vec![0u8; len];
1464 let read = sock.read_exact(&mut body).is_ok();
1465 bodies.push(match read {
1466 true => String::from_utf8_lossy(&body).into_owned(),
1467 false => "<nothing arrived on this connection>".into(),
1468 });
1469 const PAGE: usize = 1 << 20;
1476 let reply: Vec<u8> = match i {
1477 0 => {
1478 let mut r = format!(
1479 "HTTP/1.1 500 Internal Server Error\r\ncontent-length: {PAGE}\r\n\r\n"
1480 )
1481 .into_bytes();
1482 r.extend(std::iter::repeat_n(b'x', PAGE));
1483 r
1484 }
1485 _ => b"HTTP/1.1 204 No Content\r\ncontent-length: 0\r\n\r\n".to_vec(),
1486 };
1487 let _ = sock.write_all(&reply);
1488 }
1489 listener.set_nonblocking(true).expect("nonblocking");
1490 (bodies, listener.accept().is_ok())
1491 });
1492
1493 assert_eq!(
1494 post(&url, r#"{"text":"first"}"#.into()).await,
1495 Err("HTTP 500".to_owned())
1496 );
1497 assert_eq!(post(&url, r#"{"text":"second"}"#.into()).await, Ok(()));
1500 let (bodies, reconnected) = seen.join().expect("receiver");
1501 assert_eq!(
1502 bodies,
1503 [r#"{"text":"first"}"#, r#"{"text":"second"}"#],
1504 "both bodies arrived intact, down one socket"
1505 );
1506 assert!(
1507 !reconnected,
1508 "the refusal's body was drained, so the pooled connection survived it"
1509 );
1510
1511 assert_eq!(
1515 post("http://[bad", "{}".into()).await,
1516 Err("invalid authority".to_owned())
1517 );
1518 }
1519
1520 #[test]
1521 fn the_alerts_document_reports_every_rule_including_the_quiet_ones() {
1522 let e = Engine::new(Rules::parse(DOC).unwrap());
1523 let j = e.json();
1524 assert!(j.contains(r#""name":"checkout-errors""#), "{j}");
1525 assert!(j.contains(r#""state":"ok""#), "{j}");
1526 assert!(j.contains(r#""name":"any-log""#), "{j}");
1527 assert!(j.contains(r#""over_nano":"60000000000""#), "{j}");
1530 assert!(j.contains(r#""error":null"#), "{j}");
1531 }
1532}