Skip to main content

isb_apps/monitor/
service.rs

1//! The monitoring service of a daemon: definitions, the scheduler, the
2//! workers that run checks, and the events they raise.
3//!
4//! One scheduler thread wakes every second and hands due checks to a fixed
5//! pool of workers through a bounded queue ([`WORKERS`], [`QUEUE`]), so a
6//! hundred slow targets never start a hundred threads, and a check never
7//! runs twice at once. A monitor's first check comes at a phase spread by
8//! its name, later ones every interval plus a little jitter, so monitors
9//! created together do not fire together.
10
11use 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
31/// Checks running at once, at most.
32pub const WORKERS: usize = 8;
33/// Due checks waiting for a worker, at most; more wait for the next tick.
34pub const QUEUE: usize = 256;
35/// How often apps and stacks are looked at for their own monitors.
36const AUTO_SYNC_MS: u64 = 60_000;
37/// How often the history is rolled up and pruned.
38const ROLLUP_MS: u64 = 300_000;
39/// Event details kept for the notifier to pick up.
40const DETAILS_KEPT: usize = 256;
41
42/// Why an app's monitor has not checked yet.
43pub const WAITING_FOR_APP: &str = "waiting for the app's first live deployment";
44/// Why a stack service's monitor has not checked yet.
45pub const WAITING_FOR_SERVICE: &str = "waiting for the service's first replica in rotation";
46
47/// Is the address policy relaxed (the platform's private-targets setting)?
48pub type AllowPrivate = Arc<dyn Fn() -> bool + Send + Sync>;
49
50/// A monitor's state as kept: the state machine and its last check.
51#[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    /// Held across read-modify-write of an org's files.
78    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/// The monitoring service.
87#[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
98/// A little randomness for the schedule, without a crate.
99fn 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    /// Trust these TLS roots instead of the public set (tests).
137    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    /// The orgs with state on this daemon.
159    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    // ---- definitions ----
174
175    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    /// What a monitor needs to exist: its app or stack service, and its
214    /// secrets.
215    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    /// Change fields of a monitor: `patch` holds the fields to set (null
269    /// puts a field back to its default). The name stays.
270    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    /// Remove a monitor and its history. An app's or a stack service's own
313    /// monitor stays away: it joins the org's exclusions.
314    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    /// The monitors that follow any of `targets`: their own, and the app
351    /// and service monitors made for them.
352    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    /// Remove the monitors that follow `targets`, with their history, as
362    /// the targets themselves go. Unlike [`Monitors::delete`] the targets
363    /// leave the org's exclusions rather than join them, so an app made
364    /// again under the same name gets its own monitor again. Returns the
365    /// names removed.
366    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    // ---- apps' and stack services' own monitors ----
435
436    /// Give every app and compose stack service with a served domain its
437    /// own monitor, and remove those whose target is gone, has no domains,
438    /// or opted out ([`auto`]).
439    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    // ---- running ----
461
462    /// Start the scheduler and the workers.
463    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    /// Set by [`Monitors::shutdown`]: what the heartbeat stops on too.
483    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    /// Queue every check that is due.
520    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                    // The first check: a phase spread by name, within a minute.
532                    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    /// Run one check now. An app's or stack service's monitor that has not
587    /// been up yet does not look at it until it has a live deployment: the
588    /// check fails as "waiting", which the state machine keeps as pending.
589    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    /// Its app or stack service is stopped: no check. The first time, the
620    /// monitor turns stopped and an open incident closes, telling no one.
621    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    /// Keep a check's result, step the state machine, raise events.
638    pub fn record(&self, org: &OrgId, m: &Monitor, o: Outcome) -> Result<()> {
639        // A monitor removed or paused while its check ran: drop the result.
640        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    /// Days left on a certificate that is due a warning, once per
697    /// certificate.
698    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    // ---- events ----
711
712    /// Where a monitor's events go: an app's service (so channel rules on
713    /// apps and projects match), a stack service's own, else
714    /// `<org>/@monitors`, service = the monitor.
715    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    /// The web UI's page of a monitor, when isb knows its public URL.
732    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    /// A new monitor that has not had one successful check in
802    /// [`NEVER_UP_MS`]: a `monitor.down` saying so, once.
803    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    /// The details of an event this service raised, for the notifier.
849    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
922/// The date of a day number since 1970-01-01.
923fn 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;