1use std::collections::{BTreeMap, VecDeque};
12use std::path::{Path, PathBuf};
13use std::sync::atomic::{AtomicBool, Ordering};
14use std::sync::mpsc::{Receiver, SyncSender, TrySendError};
15use std::sync::{Arc, Mutex};
16use std::time::Duration;
17
18use serde::{Deserialize, Serialize};
19use serde_json::{Value, json};
20
21use super::state::{NEVER_UP_MS, Notify, State, Status, Thresholds};
22use super::store::{Check, Db};
23use super::target::{self, Ctx, Outcome};
24use super::{Kind, MAX_PER_ORG, Monitor, Settings, auto, dir, human, read_json, write_json};
25use crate::app::Apps;
26use crate::error::{Error, Result};
27use crate::org::OrgId;
28use crate::secrets::Secrets;
29use crate::stack::controller::now_ms;
30
31pub const WORKERS: usize = 8;
33pub const QUEUE: usize = 256;
35const AUTO_SYNC_MS: u64 = 60_000;
37const ROLLUP_MS: u64 = 300_000;
39const DETAILS_KEPT: usize = 256;
41
42pub const WAITING_FOR_APP: &str = "waiting for the app's first live deployment";
44pub const WAITING_FOR_SERVICE: &str = "waiting for the service's first replica in rotation";
46
47pub type AllowPrivate = Arc<dyn Fn() -> bool + Send + Sync>;
49
50#[derive(Debug, Clone, Default, Serialize, Deserialize)]
52#[serde(default)]
53pub struct Stored {
54 pub state: State,
55 pub last: Option<Outcome>,
56}
57
58#[derive(Debug, Clone, Copy, Default)]
59struct Slot {
60 next: u64,
61 running: bool,
62}
63
64struct Job {
65 org: OrgId,
66 monitor: Monitor,
67}
68
69type DetailKey = (OrgId, String, String);
70
71struct Inner {
72 state: PathBuf,
73 apps: Apps,
74 secrets: Arc<Secrets>,
75 allow_private: AllowPrivate,
76 public_url: Option<String>,
77 edit: Mutex<()>,
79 dbs: Mutex<BTreeMap<OrgId, Arc<Mutex<Db>>>>,
80 slots: Mutex<BTreeMap<(OrgId, String), Slot>>,
81 details: Mutex<VecDeque<(DetailKey, Value)>>,
82 stop: Arc<AtomicBool>,
83 tls: Arc<rustls::ClientConfig>,
84}
85
86#[derive(Clone)]
88pub struct Monitors {
89 inner: Arc<Inner>,
90}
91
92pub(super) fn fnv(s: &str) -> u64 {
93 s.bytes().fold(0xcbf2_9ce4_8422_2325, |h, b| {
94 (h ^ u64::from(b)).wrapping_mul(0x0100_0000_01b3)
95 })
96}
97
98fn jitter(max_ms: u64) -> i64 {
100 if max_ms == 0 {
101 return 0;
102 }
103 let n = std::time::SystemTime::now()
104 .duration_since(std::time::UNIX_EPOCH)
105 .map(|d| d.subsec_nanos())
106 .unwrap_or(0);
107 let r = fnv(&n.to_string()) % (2 * max_ms + 1);
108 r as i64 - max_ms as i64
109}
110
111impl Monitors {
112 pub fn new(
113 state: &Path,
114 apps: Apps,
115 secrets: Arc<Secrets>,
116 allow_private: AllowPrivate,
117 public_url: Option<String>,
118 ) -> Monitors {
119 Monitors {
120 inner: Arc::new(Inner {
121 state: state.to_path_buf(),
122 apps,
123 secrets,
124 allow_private,
125 public_url: public_url.map(|u| u.trim_end_matches('/').to_string()),
126 edit: Mutex::new(()),
127 dbs: Mutex::new(BTreeMap::new()),
128 slots: Mutex::new(BTreeMap::new()),
129 details: Mutex::new(VecDeque::new()),
130 stop: Arc::new(AtomicBool::new(false)),
131 tls: crate::net::default_tls(),
132 }),
133 }
134 }
135
136 pub fn with_tls(mut self, tls: Arc<rustls::ClientConfig>) -> Monitors {
138 if let Some(i) = Arc::get_mut(&mut self.inner) {
139 i.tls = tls;
140 }
141 self
142 }
143
144 fn path(&self, org: &OrgId, file: &str) -> PathBuf {
145 dir(&self.inner.state, org).join(file)
146 }
147
148 pub(crate) fn db(&self, org: &OrgId) -> Result<Arc<Mutex<Db>>> {
149 let mut dbs = self.inner.dbs.lock().unwrap();
150 if let Some(d) = dbs.get(org) {
151 return Ok(d.clone());
152 }
153 let d = Arc::new(Mutex::new(Db::open(&self.path(org, "monitors.db"))?));
154 dbs.insert(org.clone(), d.clone());
155 Ok(d)
156 }
157
158 pub fn orgs(&self) -> Vec<OrgId> {
160 let mut out = vec![OrgId::default_org()];
161 if let Ok(rd) = std::fs::read_dir(self.inner.state.join("orgs")) {
162 for e in rd.flatten() {
163 if let Some(o) = e.file_name().to_str().and_then(|n| OrgId::new(n).ok()) {
164 if !out.contains(&o) {
165 out.push(o);
166 }
167 }
168 }
169 }
170 out
171 }
172
173 pub fn list(&self, org: &OrgId) -> Result<Vec<Monitor>> {
176 read_json(&self.path(org, "monitors.json"))
177 }
178
179 fn save(&self, org: &OrgId, all: &[Monitor]) -> Result<()> {
180 write_json(&self.path(org, "monitors.json"), &all)
181 }
182
183 pub fn get(&self, org: &OrgId, name: &str) -> Result<Monitor> {
184 self.list(org)?
185 .into_iter()
186 .find(|m| m.name == name)
187 .ok_or_else(|| Error::NotFound(format!("monitor {name}")))
188 }
189
190 pub fn settings(&self, org: &OrgId) -> Result<Settings> {
191 read_json(&self.path(org, "settings.json"))
192 }
193
194 pub fn set_settings(&self, org: &OrgId, s: &Settings) -> Result<()> {
195 for a in &s.exclude_apps {
196 crate::app::validate_app_name(a)?;
197 }
198 for x in &s.exclude_services {
199 let ok = x.split_once('/').is_some_and(|(st, sv)| {
200 crate::stack::validate_stack_name(st).is_ok()
201 && super::validate_service_name(sv).is_ok()
202 });
203 if !ok {
204 return Err(Error::invalid(format!(
205 "exclude_services: {x:?} is not <stack>/<service>"
206 )));
207 }
208 }
209 let _g = self.inner.edit.lock().unwrap();
210 write_json(&self.path(org, "settings.json"), s)
211 }
212
213 fn check_refs(&self, org: &OrgId, m: &Monitor) -> Result<()> {
216 if m.kind == Kind::App {
217 let a = m.app.as_deref().unwrap_or_default();
218 self.inner.apps.get(org, a)?;
219 }
220 if m.kind == Kind::Service {
221 let (st, sv) = (
222 m.stack.as_deref().unwrap_or_default(),
223 m.service.as_deref().unwrap_or_default(),
224 );
225 let d = self
226 .inner
227 .apps
228 .controller()
229 .definition(&crate::stack::qualified(org, st))
230 .map_err(|_| Error::NotFound(format!("stack {st} in org {org}")))?;
231 if !d.file.services.contains_key(sv) {
232 return Err(Error::NotFound(format!("service {sv} in stack {st}")));
233 }
234 }
235 for s in m.secrets() {
236 self.inner.secrets.inspect(org, s).map_err(|e| {
237 if e.is_not_found() {
238 Error::invalid(format!("no secret {s} in org {org}: create it first"))
239 } else {
240 e
241 }
242 })?;
243 }
244 Ok(())
245 }
246
247 pub fn create(&self, org: &OrgId, mut m: Monitor) -> Result<Monitor> {
248 m.validate()?;
249 self.check_refs(org, &m)?;
250 let _g = self.inner.edit.lock().unwrap();
251 let mut all = self.list(org)?;
252 if all.iter().any(|x| x.name == m.name) {
253 return Err(Error::invalid(format!("monitor {} exists", m.name)));
254 }
255 if all.len() >= MAX_PER_ORG {
256 return Err(Error::invalid(format!(
257 "at most {MAX_PER_ORG} monitors per org"
258 )));
259 }
260 let now = now_ms() / 1000;
261 (m.created_at, m.updated_at) = (now, now);
262 all.push(m.clone());
263 self.save(org, &all)?;
264 self.due_soon(org, &m.name);
265 Ok(m)
266 }
267
268 pub fn update(
271 &self,
272 org: &OrgId,
273 name: &str,
274 patch: serde_json::Map<String, Value>,
275 ) -> Result<Monitor> {
276 if patch.get("name").is_some_and(|n| n.as_str() != Some(name)) {
277 return Err(Error::invalid(
278 "a monitor's name cannot change; create another",
279 ));
280 }
281 let _g = self.inner.edit.lock().unwrap();
282 let mut all = self.list(org)?;
283 let cur = all
284 .iter_mut()
285 .find(|m| m.name == name)
286 .ok_or_else(|| Error::NotFound(format!("monitor {name}")))?;
287 let mut v = serde_json::to_value(&*cur)?;
288 let o = v.as_object_mut().expect("a monitor is an object");
289 for (k, x) in patch {
290 if matches!(k.as_str(), "created_at" | "updated_at" | "auto") {
291 continue;
292 }
293 if x.is_null() {
294 o.remove(&k);
295 } else {
296 o.insert(k, x);
297 }
298 }
299 let mut n: Monitor =
300 serde_json::from_value(v).map_err(|e| Error::invalid(format!("bad monitor: {e}")))?;
301 n.validate()?;
302 self.check_refs(org, &n)?;
303 n.updated_at = now_ms() / 1000;
304 *cur = n.clone();
305 self.save(org, &all)?;
306 drop(_g);
307 self.edit_state(org, name, State::reset_counts)?;
308 self.due_soon(org, name);
309 Ok(n)
310 }
311
312 pub fn delete(&self, org: &OrgId, name: &str) -> Result<Monitor> {
315 let _g = self.inner.edit.lock().unwrap();
316 let mut all = self.list(org)?;
317 let i = all
318 .iter()
319 .position(|m| m.name == name)
320 .ok_or_else(|| Error::NotFound(format!("monitor {name}")))?;
321 let m = all.remove(i);
322 self.save(org, &all)?;
323 if m.auto {
324 let mut s = self.settings(org)?;
325 let (list, x) = match m.kind {
326 Kind::Service => (
327 &mut s.exclude_services,
328 auto::exclusion(
329 m.stack.as_deref().unwrap_or_default(),
330 m.service.as_deref().unwrap_or_default(),
331 ),
332 ),
333 _ => (&mut s.exclude_apps, m.app.clone().unwrap_or_default()),
334 };
335 if !list.contains(&x) {
336 list.push(x);
337 write_json(&self.path(org, "settings.json"), &s)?;
338 }
339 }
340 drop(_g);
341 self.db(org)?.lock().unwrap().forget(name)?;
342 self.inner
343 .slots
344 .lock()
345 .unwrap()
346 .remove(&(org.clone(), name.to_string()));
347 Ok(m)
348 }
349
350 pub fn following(&self, org: &OrgId, targets: &[auto::Target]) -> Result<Vec<String>> {
353 Ok(self
354 .list(org)?
355 .into_iter()
356 .filter(|m| targets.iter().any(|t| t.owns(m)))
357 .map(|m| m.name)
358 .collect())
359 }
360
361 pub fn remove_following(&self, org: &OrgId, targets: &[auto::Target]) -> Result<Vec<String>> {
367 let _g = self.inner.edit.lock().unwrap();
368 let mut all = self.list(org)?;
369 let mut gone = Vec::new();
370 all.retain(|m| {
371 let keep = !targets.iter().any(|t| t.owns(m));
372 if !keep {
373 gone.push(m.name.clone());
374 }
375 keep
376 });
377 if !gone.is_empty() {
378 self.save(org, &all)?;
379 }
380 let mut s = self.settings(org)?;
381 if targets.iter().any(|t| t.excluded(&s)) {
382 for t in targets {
383 match t {
384 auto::Target::App(a) => s.exclude_apps.retain(|x| x != a),
385 auto::Target::Service { stack, service } => {
386 let x = auto::exclusion(stack, service);
387 s.exclude_services.retain(|e| *e != x);
388 }
389 }
390 }
391 write_json(&self.path(org, "settings.json"), &s)?;
392 }
393 drop(_g);
394 for n in &gone {
395 self.db(org)?.lock().unwrap().forget(n)?;
396 self.inner
397 .slots
398 .lock()
399 .unwrap()
400 .remove(&(org.clone(), n.clone()));
401 }
402 Ok(gone)
403 }
404
405 pub fn set_paused(&self, org: &OrgId, name: &str, paused: bool) -> Result<Monitor> {
406 let mut p = serde_json::Map::new();
407 p.insert("paused".into(), json!(paused));
408 self.update(org, name, p)
409 }
410
411 fn edit_state(&self, org: &OrgId, name: &str, f: impl FnOnce(&mut State)) -> Result<()> {
412 let db = self.db(org)?;
413 let db = db.lock().unwrap();
414 let mut s: Stored = db.load_state(name)?.unwrap_or_default();
415 f(&mut s.state);
416 db.save_state(name, &s)
417 }
418
419 pub(crate) fn stored(&self, org: &OrgId, name: &str) -> Result<Stored> {
420 Ok(self
421 .db(org)?
422 .lock()
423 .unwrap()
424 .load_state(name)?
425 .unwrap_or_default())
426 }
427
428 fn due_soon(&self, org: &OrgId, name: &str) {
429 let mut slots = self.inner.slots.lock().unwrap();
430 let s = slots.entry((org.clone(), name.to_string())).or_default();
431 s.next = now_ms() + 1000;
432 }
433
434 pub fn sync_auto(&self, org: &OrgId) -> Result<()> {
440 let settings = self.settings(org)?;
441 let mut cands = auto::apps(&self.inner.apps, org)?;
442 cands.extend(auto::stack_services(&self.inner.apps, org));
443 let _g = self.inner.edit.lock().unwrap();
444 let mut all = self.list(org)?;
445 let (gone, added) = auto::reconcile(&mut all, &settings, &cands, now_ms() / 1000);
446 if !gone.is_empty() || !added.is_empty() {
447 self.save(org, &all)?;
448 }
449 drop(_g);
450 for n in &gone {
451 self.db(org)?.lock().unwrap().forget(n)?;
452 }
453 for n in &added {
454 eprintln!("isb serve: monitor: {org}/{n}: watching its domain");
455 self.due_soon(org, n);
456 }
457 Ok(())
458 }
459
460 pub fn start(&self) {
464 let (tx, rx) = std::sync::mpsc::sync_channel::<Job>(QUEUE);
465 let rx = Arc::new(Mutex::new(rx));
466 for i in 0..WORKERS {
467 let (me, rx) = (self.clone(), rx.clone());
468 let _ = std::thread::Builder::new()
469 .name(format!("isb-monitor-{i}"))
470 .spawn(move || me.worker(&rx));
471 }
472 let me = self.clone();
473 let _ = std::thread::Builder::new()
474 .name("isb-monitor".into())
475 .spawn(move || me.scheduler(&tx));
476 }
477
478 pub fn shutdown(&self) {
479 self.inner.stop.store(true, Ordering::SeqCst);
480 }
481
482 pub fn stopper(&self) -> Arc<AtomicBool> {
484 self.inner.stop.clone()
485 }
486
487 fn scheduler(&self, tx: &SyncSender<Job>) {
488 let (mut synced, mut rolled) = (0u64, now_ms());
489 while !self.inner.stop.load(Ordering::SeqCst) {
490 let now = now_ms();
491 if now.saturating_sub(synced) >= AUTO_SYNC_MS {
492 for o in self.orgs() {
493 if let Err(e) = self.sync_auto(&o) {
494 eprintln!("isb serve: monitor: {o}: own monitors: {e}");
495 }
496 }
497 synced = now;
498 }
499 if now.saturating_sub(rolled) >= ROLLUP_MS {
500 self.rollup_all(now);
501 rolled = now;
502 }
503 self.tick(now, tx);
504 std::thread::sleep(Duration::from_secs(1));
505 }
506 }
507
508 fn rollup_all(&self, now: u64) {
509 for o in self.orgs() {
510 if !self.path(&o, "monitors.db").exists() {
511 continue;
512 }
513 if let Err(e) = self.db(&o).and_then(|d| d.lock().unwrap().rollup(now)) {
514 eprintln!("isb serve: monitor: {o}: history rollup: {e}");
515 }
516 }
517 }
518
519 fn tick(&self, now: u64, tx: &SyncSender<Job>) {
521 let mut seen = Vec::new();
522 for org in self.orgs() {
523 for m in self.list(&org).unwrap_or_default() {
524 let key = (org.clone(), m.name.clone());
525 seen.push(key.clone());
526 if m.paused {
527 continue;
528 }
529 let mut slots = self.inner.slots.lock().unwrap();
530 let s = slots.entry(key.clone()).or_insert_with(|| Slot {
531 next: now + fnv(&format!("{}/{}", org, m.name)) % (m.interval.min(60) * 1000),
533 running: false,
534 });
535 if s.running || now < s.next {
536 continue;
537 }
538 match tx.try_send(Job {
539 org: org.clone(),
540 monitor: m,
541 }) {
542 Ok(()) => s.running = true,
543 Err(TrySendError::Full(_)) => return,
544 Err(TrySendError::Disconnected(_)) => return,
545 }
546 }
547 }
548 self.inner
549 .slots
550 .lock()
551 .unwrap()
552 .retain(|k, _| seen.contains(k));
553 }
554
555 fn worker(&self, rx: &Mutex<Receiver<Job>>) {
556 loop {
557 let job = match rx.lock().unwrap().recv_timeout(Duration::from_secs(1)) {
558 Ok(j) => j,
559 Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
560 if self.inner.stop.load(Ordering::SeqCst) {
561 return;
562 }
563 continue;
564 }
565 Err(_) => return,
566 };
567 let started = now_ms();
568 let done = if target::is_stopped(&self.inner.apps, &job.org, &job.monitor) {
569 self.rest(&job.org, &job.monitor)
570 } else {
571 let o = self.check(&job.org, &job.monitor);
572 self.record(&job.org, &job.monitor, o)
573 };
574 if let Err(e) = done {
575 eprintln!("isb serve: monitor: {}/{}: {e}", job.org, job.monitor.name);
576 }
577 let iv = job.monitor.interval * 1000;
578 let mut slots = self.inner.slots.lock().unwrap();
579 if let Some(s) = slots.get_mut(&(job.org, job.monitor.name)) {
580 s.running = false;
581 s.next = (started + iv).saturating_add_signed(jitter((iv / 20).min(2000)));
582 }
583 }
584 }
585
586 pub fn check(&self, org: &OrgId, m: &Monitor) -> Outcome {
590 if m.follows()
591 && self
592 .stored(org, &m.name)
593 .map(|s| s.state.status == Status::Pending)
594 .unwrap_or(true)
595 && !target::is_live(&self.inner.apps, org, m)
596 {
597 let why = if m.kind == Kind::App {
598 WAITING_FOR_APP
599 } else {
600 WAITING_FOR_SERVICE
601 };
602 return Outcome {
603 at: now_ms(),
604 ok: false,
605 error: Some(why.into()),
606 ..Default::default()
607 };
608 }
609 let ctx = Ctx {
610 apps: &self.inner.apps,
611 ctl: self.inner.apps.controller(),
612 secrets: &self.inner.secrets,
613 allow_private: (self.inner.allow_private)(),
614 tls: self.inner.tls.clone(),
615 };
616 target::run(&ctx, org, m, now_ms())
617 }
618
619 pub fn rest(&self, org: &OrgId, m: &Monitor) -> Result<()> {
622 match self.get(org, &m.name) {
623 Ok(cur) if !cur.paused => {}
624 _ => return Ok(()),
625 }
626 let db = self.db(org)?;
627 let db = db.lock().unwrap();
628 let mut s: Stored = db.load_state(&m.name)?.unwrap_or_default();
629 let now = now_ms();
630 if s.state.stop(now) {
631 db.close_incident(&m.name, now)?;
632 db.save_state(&m.name, &s)?;
633 }
634 Ok(())
635 }
636
637 pub fn record(&self, org: &OrgId, m: &Monitor, o: Outcome) -> Result<()> {
639 match self.get(org, &m.name) {
641 Ok(cur) if !cur.paused => {}
642 _ => return Ok(()),
643 }
644 let db = self.db(org)?;
645 let db = db.lock().unwrap();
646 let mut s: Stored = db.load_state(&m.name)?.unwrap_or_default();
647 let t = Thresholds {
648 failures: m.failure_threshold,
649 recoveries: m.recovery_threshold,
650 };
651 let step = s.state.observe(o.ok, o.at, t);
652 db.insert(
653 &m.name,
654 &Check {
655 at: o.at,
656 ok: o.ok,
657 latency_ms: o.latency_ms,
658 status: o.status,
659 error: o.error.clone(),
660 pending: step.pending,
661 },
662 )?;
663 if let Notify::NeverUp { since } = step.notify {
664 let why = format!("never came up: {}", o.error.as_deref().unwrap_or("failed"));
665 db.open_incident(&m.name, since, Some(&why))?;
666 }
667 match step.changed {
668 Some(Status::Down) => {
669 let since = s.state.failing_since.unwrap_or(o.at);
670 db.open_incident(&m.name, since, o.error.as_deref())?;
671 }
672 Some(Status::Up) => db.close_incident(&m.name, o.at)?,
673 _ => {}
674 }
675 let cert = self.cert_due(m, &o, &mut s.state);
676 s.last = Some(o.clone());
677 db.save_state(&m.name, &s)?;
678 drop(db);
679 match step.notify {
680 Notify::Down { since, flapping } => {
681 self.emit_down(org, m, &o, since, flapping, s.state.fails)
682 }
683 Notify::NeverUp { since } => self.emit_never_up(org, m, &o, since, s.state.fails),
684 Notify::Up {
685 down_since,
686 downtime_ms,
687 } => self.emit_up(org, m, &o, down_since, downtime_ms),
688 Notify::None => {}
689 }
690 if let Some(days) = cert {
691 self.emit_cert(org, m, &o, days);
692 }
693 Ok(())
694 }
695
696 fn cert_due(&self, m: &Monitor, o: &Outcome, s: &mut State) -> Option<i64> {
699 let exp = o.cert_expires?;
700 if m.cert_expiry_days == 0 || s.cert_warned == Some(exp) {
701 return None;
702 }
703 let days = (exp as i64 - (o.at / 1000) as i64).div_euclid(86_400);
704 (days <= i64::from(m.cert_expiry_days)).then(|| {
705 s.cert_warned = Some(exp);
706 days
707 })
708 }
709
710 fn subject(&self, org: &OrgId, m: &Monitor) -> (String, String) {
716 if let (Kind::Service, Some(st), Some(sv)) = (m.kind, &m.stack, &m.service) {
717 return (crate::stack::qualified(org, st), sv.clone());
718 }
719 if let Some(a) = m
720 .app
721 .as_deref()
722 .and_then(|a| self.inner.apps.get(org, a).ok())
723 {
724 if let Ok(st) = a.spec.stack() {
725 return (crate::stack::qualified(org, &st), a.spec.name);
726 }
727 }
728 (crate::stack::qualified(org, "@monitors"), m.name.clone())
729 }
730
731 pub fn link(&self, org: &OrgId, name: &str) -> Option<String> {
733 let base = self.inner.public_url.as_deref()?;
734 Some(format!("{base}/orgs/{org}/uptime/{name}"))
735 }
736
737 fn base_details(&self, org: &OrgId, m: &Monitor, o: &Outcome) -> Value {
738 json!({
739 "monitor": m.name,
740 "type": m.kind,
741 "app": m.app,
742 "stack": m.stack,
743 "service": m.service,
744 "url": o.url.clone().unwrap_or_else(|| m.target()),
745 "status": o.status,
746 "latency_ms": o.latency_ms,
747 "error": o.error,
748 "via": o.via,
749 "note": o.note,
750 "checked_at": o.at,
751 "link": self.link(org, &m.name),
752 })
753 }
754
755 fn emit(
756 &self,
757 org: &OrgId,
758 m: &Monitor,
759 kind: &str,
760 level: &str,
761 message: String,
762 details: Value,
763 ) {
764 let (stack, service) = self.subject(org, m);
765 {
766 let mut d = self.inner.details.lock().unwrap();
767 if d.len() >= DETAILS_KEPT {
768 d.pop_front();
769 }
770 d.push_back(((org.clone(), kind.to_string(), message.clone()), details));
771 }
772 self.inner
773 .apps
774 .controller()
775 .event(kind, level, &stack, &service, message);
776 }
777
778 fn emit_down(
779 &self,
780 org: &OrgId,
781 m: &Monitor,
782 o: &Outcome,
783 since: u64,
784 flapping: bool,
785 fails: u32,
786 ) {
787 let mut d = self.base_details(org, m, o);
788 d["down_since"] = json!(since);
789 d["failures"] = json!(fails);
790 d["flapping"] = json!(flapping);
791 self.emit(
792 org,
793 m,
794 "monitor.down",
795 "error",
796 down_message(m, o, fails, flapping),
797 d,
798 );
799 }
800
801 fn emit_never_up(&self, org: &OrgId, m: &Monitor, o: &Outcome, since: u64, fails: u32) {
804 let mut d = self.base_details(org, m, o);
805 d["down_since"] = json!(since);
806 d["failures"] = json!(fails);
807 d["flapping"] = json!(false);
808 d["never_up"] = json!(true);
809 self.emit(
810 org,
811 m,
812 "monitor.down",
813 "error",
814 never_up_message(m, o, fails),
815 d,
816 );
817 }
818
819 fn emit_up(&self, org: &OrgId, m: &Monitor, o: &Outcome, down_since: u64, downtime_ms: u64) {
820 let mut d = self.base_details(org, m, o);
821 d["down_since"] = json!(down_since);
822 d["downtime_ms"] = json!(downtime_ms);
823 d["downtime"] = json!(human(downtime_ms));
824 self.emit(
825 org,
826 m,
827 "monitor.up",
828 "info",
829 up_message(m, o, downtime_ms),
830 d,
831 );
832 }
833
834 fn emit_cert(&self, org: &OrgId, m: &Monitor, o: &Outcome, days: i64) {
835 let mut d = self.base_details(org, m, o);
836 d["cert_expires_at"] = json!(o.cert_expires);
837 d["cert_days_left"] = json!(days);
838 self.emit(
839 org,
840 m,
841 "monitor.cert_expiring",
842 "warn",
843 cert_message(m, o, days),
844 d,
845 );
846 }
847
848 pub fn details(&self, org: &OrgId, kind: &str, message: &str) -> Option<Value> {
850 let d = self.inner.details.lock().unwrap();
851 d.iter()
852 .rev()
853 .find(|((o, k, msg), _)| o == org && k == kind && msg == message)
854 .map(|(_, v)| v.clone())
855 }
856}
857
858fn what(m: &Monitor, o: &Outcome) -> String {
859 o.url.clone().unwrap_or_else(|| m.target())
860}
861
862pub fn down_message(m: &Monitor, o: &Outcome, fails: u32, flapping: bool) -> String {
863 let why = o.error.as_deref().unwrap_or("failed");
864 let mut s = format!(
865 "Monitor {} is DOWN: {}: {why} ({fails} failed check{} in a row)",
866 m.name,
867 what(m, o),
868 if fails == 1 { "" } else { "s" }
869 );
870 if flapping {
871 s.push_str("; it is flapping, so further changes are held until it is stable for 30 min");
872 }
873 s
874}
875
876pub fn never_up_message(m: &Monitor, o: &Outcome, fails: u32) -> String {
877 let why = o.error.as_deref().unwrap_or("failed");
878 format!(
879 "Monitor {} never came up: {}: {why} (no successful check in {} min, {fails} failed check{})",
880 m.name,
881 what(m, o),
882 NEVER_UP_MS / 60_000,
883 if fails == 1 { "" } else { "s" }
884 )
885}
886
887pub fn up_message(m: &Monitor, o: &Outcome, downtime_ms: u64) -> String {
888 let answer = match (o.status, o.latency_ms) {
889 (Some(st), Some(l)) => format!("answered HTTP {st} in {l} ms"),
890 (None, Some(l)) => format!("answered in {l} ms"),
891 _ => "answered".into(),
892 };
893 format!(
894 "Monitor {} is UP again after {}: {} {answer}",
895 m.name,
896 human(downtime_ms),
897 what(m, o)
898 )
899}
900
901pub fn cert_message(m: &Monitor, o: &Outcome, days: i64) -> String {
902 let on = o
903 .cert_expires
904 .map(|e| {
905 let d = (e / 86_400) as i64;
906 let (y, mo, da) = civil(d);
907 format!(" ({y:04}-{mo:02}-{da:02})")
908 })
909 .unwrap_or_default();
910 let when = if days < 0 {
911 "has expired".to_string()
912 } else {
913 format!("expires in {days} day{}", if days == 1 { "" } else { "s" })
914 };
915 format!(
916 "Monitor {}: the TLS certificate of {} {when}{on}",
917 m.name,
918 what(m, o)
919 )
920}
921
922fn civil(z: i64) -> (i64, i64, i64) {
924 let z = z + 719_468;
925 let era = z.div_euclid(146_097);
926 let doe = z - era * 146_097;
927 let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
928 let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
929 let mp = (5 * doy + 2) / 153;
930 let d = doy - (153 * mp + 2) / 5 + 1;
931 let m = if mp < 10 { mp + 3 } else { mp - 9 };
932 (yoe + era * 400 + i64::from(m <= 2), m, d)
933}
934
935#[cfg(test)]
936#[path = "service_tests.rs"]
937mod tests;