didbot_serve/

health.rs

//! The periodic line that says what the deployment currently holds, and the
//! state behind `GET /health`'s progress report.
//!
//! The development stack is operated by watching its log. Provisioning and
//! record writes each announce themselves as they happen, but a developer who
//! attaches to a quiet server sees nothing at all, and a developer scrolling
//! back through a busy one has no marker to count against. One line every ten
//! seconds fixes both: it is a heartbeat and a running total in the same
//! place.
//!
//! # Liveness versus progress
//!
//! A process that answers HTTP is not the same claim as one whose log is
//! advancing: a deadlocked lock, a registry stuck retrying a poisoned mutex,
//! or a runtime with every worker parked behind a hung syscall can all leave
//! axum answering `200` while nothing this tick would log is happening
//! underneath it. [`HealthState`] is what lets `GET /health` say more than
//! "a socket accepted" — it is the tick's own pulse, read back rather than
//! re-derived, so the two can never disagree about whether the tick that
//! *is* the deployment's sign of life is still ticking.
//!
//! `GET /health` itself stays the liveness check: it answers `200` whenever
//! this process can schedule a task at all, which is what a load balancer
//! should gate traffic on. `progress` in its body is the second claim,
//! carried in the same response rather than a separate route, so a caller
//! that only wants liveness reads one field and a caller diagnosing a stall
//! reads the rest of the same document.

use std::fmt;
use std::path::PathBuf;
use std::sync::atomic::{AtomicI64, AtomicU64, AtomicU8, Ordering};
use std::sync::{Arc, OnceLock};
use std::time::Duration;

use didbot_pds::{Registry, RegistryStats};

use didbot_tls::renew::AlertLevel;
use futures_core::future::BoxFuture;
use serde::Serialize;
use time::OffsetDateTime;
use tokio::sync::Notify;
use tokio::task::JoinHandle;
use tokio::time::MissedTickBehavior;
use tracing::{info, warn};

/// The commit this binary was built from, as `scripts/publish-image.sh`
/// passes it to the image build in `DIDBOT_REVISION`. `None` for a build
/// that was not given one, such as `cargo run`.
pub(crate) const REVISION: Option<&str> = option_env!("DIDBOT_REVISION");

/// How often the health line is written when a caller does not choose.
pub const DEFAULT_HEALTH_INTERVAL: Duration = Duration::from_secs(10);

/// The tick's pulse, shared with whoever answers `GET /health`.
///
/// Kept separate from [`HealthTick`] itself so a caller can hand the state to
/// the HTTP router and the join handle to nothing but the shutdown path —
/// the router has no business awaiting the task or aborting it, only reading
/// the counters.
#[derive(Default)]
pub struct HealthState {
    /// How many ticks have fired since the process started.
    ticks: AtomicU64,
    /// Unix seconds of the most recent tick, or `0` for "never yet" — the
    /// first tick fires one interval after startup, by design; see
    /// [`HealthTick::spawn`].
    last_tick_unix: AtomicI64,
    /// Unix seconds this state was created, which is startup for all
    /// practical purposes: it is built and handed to [`HealthTick::spawn`]
    /// in the same breath the tick task is.
    started_unix: i64,
    /// Whoever can answer for the certificates, once there is one — see
    /// [`HealthState::attach_certificates`] for why this is filled in after
    /// the router is already serving rather than passed to the constructor.
    certificates: OnceLock<Arc<dyn CertificateAlerts>>,
    /// The data directory whose volume [`HealthState::disk`] reads, and
    /// the watermarks its free space is measured against; see
    /// [`HealthState::attach_data_dir`].
    data_dir: OnceLock<(PathBuf, DiskWatermarks)>,
    /// The [`DiskAlert`] the last tick read, as its `u8`: `ok` before one
    /// has.
    disk_alert: AtomicU8,
}

impl fmt::Debug for HealthState {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("HealthState")
            .field("ticks", &self.ticks)
            .field("last_tick_unix", &self.last_tick_unix)
            .field("started_unix", &self.started_unix)
            .field("certificates", &self.certificates.get().is_some())
            .field("data_dir", &self.data_dir.get())
            .finish()
    }
}

/// Reads back each zone's certificate alert level.
///
/// A trait rather than the concrete `didbot_tls::fleet::CertificateFleet`
/// because that type is generic over its DNS provider, and `AppState` — one
/// type, shared by every route — would otherwise carry that parameter
/// through the whole router for the sake of one field on one response.
///
/// The future is boxed for the same reason the trait exists: an alert level
/// is read from behind a `tokio::sync::Mutex` the renewal loop also holds,
/// so answering is `async`, and an `async fn` in a trait is not object safe.
pub trait CertificateAlerts: Send + Sync {
    /// Each zone this deployment serves, and how loud its renewal state
    /// currently is, in configuration order.
    fn zone_alert_levels(&self) -> BoxFuture<'_, Vec<(String, AlertLevel)>>;
}

/// What `GET /health` reports about certificate renewal.
///
/// `worst` is one field an alarm expression can match on without parsing a
/// list, and `zones` is what an operator reading the same response needs in
/// order to know which zone to go and look at — the pair
/// `didbot_tls::fleet::CertificateFleet` already draws the distinction
/// between.
///
/// A deployment serving no configured certificate answers `ok` over an empty
/// `zones`, which is `CertificateFleet::alert_level`'s own answer for a
/// fleet of no zones: the worst level of nothing is `ok`.
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CertificateHealth {
    /// The loudest level any zone is at.
    pub worst: &'static str,
    /// Each zone and its own level.
    pub zones: Vec<ZoneCertificateHealth>,
}

/// What `GET /health` reports about the data volume.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct DiskHealth {
    /// Bytes this server may still write there. See
    /// [`didbot_pds::volume`] for which blocks that leaves out.
    pub free_bytes: u64,
    /// The volume's size.
    pub total_bytes: u64,
    /// The lowest watermark the free space is below. A string an alarm can
    /// match on, as `certificates.worst` is.
    pub alert: DiskAlert,
}

/// Which free-space watermark the data volume is below.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize)]
#[serde(rename_all = "lowercase")]
#[repr(u8)]
pub enum DiskAlert {
    /// Above both.
    Ok,
    /// Below [`DiskWatermarks::warning_free_percent`].
    Warning,
    /// Below [`DiskWatermarks::critical_free_percent`].
    Critical,
}

impl DiskAlert {
    /// The name `GET /health` and the journal give it.
    #[must_use]
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Ok => "ok",
            Self::Warning => "warning",
            Self::Critical => "critical",
        }
    }

    fn from_u8(value: u8) -> Option<Self> {
        [Self::Ok, Self::Warning, Self::Critical]
            .into_iter()
            .find(|alert| *alert as u8 == value)
    }
}

/// The free space, as a percentage of the data volume, below which
/// `GET /health`'s `disk.alert` and the journal raise an alert. The
/// `[capacity] disk_*_free_percent` keys set them.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct DiskWatermarks {
    /// Below this, `warning`.
    pub warning_free_percent: u8,
    /// Below this, `critical`.
    pub critical_free_percent: u8,
}

impl Default for DiskWatermarks {
    fn default() -> Self {
        Self {
            warning_free_percent: 20,
            critical_free_percent: 10,
        }
    }
}

impl DiskWatermarks {
    /// The alert `free` bytes of `total` is at.
    #[must_use]
    pub fn alert(&self, free: u64, total: u64) -> DiskAlert {
        let below = |percent: u8| u128::from(free) * 100 < u128::from(total) * u128::from(percent);
        if below(self.critical_free_percent) {
            DiskAlert::Critical
        } else if below(self.warning_free_percent) {
            DiskAlert::Warning
        } else {
            DiskAlert::Ok
        }
    }

    /// The watermark `alert` is below, in percent. `None` for `ok`.
    fn percent(&self, alert: DiskAlert) -> Option<u8> {
        match alert {
            DiskAlert::Ok => None,
            DiskAlert::Warning => Some(self.warning_free_percent),
            DiskAlert::Critical => Some(self.critical_free_percent),
        }
    }
}

/// One zone's entry in [`CertificateHealth::zones`].
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ZoneCertificateHealth {
    /// The zone hostname, as configured.
    pub zone: String,
    /// That zone's current alert level.
    pub level: &'static str,
}

impl HealthState {
    /// A fresh state, stamped with the current time as its start.
    pub fn new() -> Arc<Self> {
        Arc::new(Self {
            started_unix: OffsetDateTime::now_utc().unix_timestamp(),
            ..Self::default()
        })
    }

    /// Records that a tick just fired.
    fn record_tick(&self) {
        self.ticks.fetch_add(1, Ordering::Relaxed);
        self.last_tick_unix.store(
            OffsetDateTime::now_utc().unix_timestamp(),
            Ordering::Relaxed,
        );
    }

    /// Names who can answer for the certificates.
    ///
    /// Separate from [`HealthState::new`] because of the order the binary
    /// builds things in: the router — and this state with it — exists before
    /// the ACME fleet does, since the fleet is bootstrapped by the code that
    /// is handed the finished router to serve. The first call wins and later
    /// ones are ignored, so a second fleet cannot quietly replace the one the
    /// listener is actually serving certificates from.
    pub fn attach_certificates(&self, alerts: Arc<dyn CertificateAlerts>) {
        let _ = self.certificates.set(alerts);
    }

    /// Names the data directory whose volume [`HealthState::disk`] reads,
    /// and the watermarks its free space is measured against.
    ///
    /// The first call wins, as it does for
    /// [`HealthState::attach_certificates`]. A state nothing names has no
    /// volume to read, which is every in-memory run.
    pub fn attach_data_dir(&self, dir: PathBuf, watermarks: DiskWatermarks) {
        let _ = self.data_dir.set((dir, watermarks));
    }

    /// The data volume's free and total bytes, read as this is called.
    ///
    /// Read on the blocking pool, because `statvfs` asks the filesystem, and
    /// a volume that has stopped answering must not hold an async worker.
    /// `None` when no data directory was named, and when the read fails,
    /// which is logged.
    pub async fn disk(&self) -> Option<DiskHealth> {
        let (dir, watermarks) = self.data_dir.get()?.clone();
        match tokio::task::spawn_blocking(move || didbot_pds::volume::Space::of(&dir)).await {
            Ok(Ok(space)) => Some(DiskHealth {
                free_bytes: space.free,
                total_bytes: space.total,
                alert: watermarks.alert(space.free, space.total),
            }),
            Ok(Err(error)) => {
                warn!(%error, "could not read the data volume's free space");
                None
            }
            Err(error) => {
                warn!(%error, "the read of the data volume's free space did not finish");
                None
            }
        }
    }

    /// Writes a line when `disk` has crossed a watermark since the last
    /// reading: `warn` going down, `info` coming back up. A first reading
    /// below a watermark is a crossing. Answers the alert it wrote about.
    pub(crate) fn watch_disk(&self, disk: &DiskHealth) -> Option<DiskAlert> {
        let watermarks = self.data_dir.get()?.1;
        let previous =
            DiskAlert::from_u8(self.disk_alert.swap(disk.alert as u8, Ordering::Relaxed))
                .unwrap_or(DiskAlert::Ok);
        let free_percent = disk.free_bytes.saturating_mul(100) / disk.total_bytes.max(1);
        if disk.alert > previous {
            warn!(
                free_bytes = disk.free_bytes,
                total_bytes = disk.total_bytes,
                free_percent,
                alert = disk.alert.as_str(),
                watermark_percent = watermarks.percent(disk.alert),
                "the data volume's free space fell below a watermark"
            );
        } else if disk.alert < previous {
            info!(
                free_bytes = disk.free_bytes,
                total_bytes = disk.total_bytes,
                free_percent,
                alert = disk.alert.as_str(),
                watermark_percent = watermarks.percent(previous),
                "the data volume's free space rose back above a watermark"
            );
        } else {
            return None;
        }
        Some(disk.alert)
    }

    /// The certificate half of the health response.
    ///
    /// `async`, and so kept out of [`HealthState::snapshot`]: the levels are
    /// read from the renewal trackers themselves rather than from a copy
    /// something has to remember to refresh, which is what makes a `warning`
    /// in this response and a `warning` in the log the same fact.
    pub async fn certificates(&self) -> CertificateHealth {
        let levels = match self.certificates.get() {
            Some(alerts) => alerts.zone_alert_levels().await,
            None => Vec::new(),
        };
        let worst = levels
            .iter()
            .map(|(_, level)| *level)
            .max()
            .unwrap_or(AlertLevel::Ok);
        CertificateHealth {
            worst: worst.as_str(),
            zones: levels
                .into_iter()
                .map(|(zone, level)| ZoneCertificateHealth {
                    zone,
                    level: level.as_str(),
                })
                .collect(),
        }
    }

    /// A point-in-time read, safe to serialize into an HTTP response.
    pub fn snapshot(&self) -> HealthSnapshot {
        let last = self.last_tick_unix.load(Ordering::Relaxed);
        let now = OffsetDateTime::now_utc().unix_timestamp();
        HealthSnapshot {
            ticks: self.ticks.load(Ordering::Relaxed),
            uptime_secs: (now - self.started_unix).max(0),
            last_tick_secs_ago: (last != 0).then(|| (now - last).max(0)),
        }
    }
}

/// What `GET /health` reports beyond liveness: whether the tick that logs
/// this deployment's own pulse is still advancing, and how recently.
///
/// `last_tick_secs_ago` absent means no tick has landed since startup yet,
/// which is expected for the first [`DEFAULT_HEALTH_INTERVAL`] of a fresh
/// process and worth alerting on past it — see `plan/deploy.md`'s "alert on
/// the things that fail quietly".
#[derive(Debug, Clone, Copy, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct HealthSnapshot {
    /// How many ticks have fired.
    pub ticks: u64,
    /// Seconds since this process started.
    pub uptime_secs: i64,
    /// Seconds since the last tick, or absent if none has fired yet.
    pub last_tick_secs_ago: Option<i64>,
}

/// A running health tick, stopped by [`HealthTick::stop`] or by dropping it.
///
/// The task holds only an `Arc<dyn Registry>` and never keeps the process
/// alive on its own: a tokio task dies with its runtime, and this handle
/// aborts it on drop so a server that has finished shutting down does not
/// leave a timer writing lines about a registry nobody is serving.
#[derive(Debug)]
pub struct HealthTick {
    handle: JoinHandle<()>,
    stop: Arc<Notify>,
    state: Arc<HealthState>,
}

impl HealthTick {
    /// Starts writing a health line every `interval`, with its own
    /// [`HealthState`] a caller can read back through [`HealthTick::state`].
    ///
    /// The first line comes after one interval rather than immediately: at
    /// startup the counts are all zero and the banner has just said as much.
    ///
    /// A tick that arrives late is delayed rather than replayed. Catching up
    /// on missed ticks would print several identical lines in a row after the
    /// machine came back from a suspend, which says nothing a single current
    /// line does not.
    pub fn spawn(registry: Arc<dyn Registry>, interval: Duration) -> Self {
        let stop = Arc::new(Notify::new());
        let notified = stop.clone();
        let state = HealthState::new();
        let ticked = state.clone();
        let handle = tokio::spawn(async move {
            let mut ticker = tokio::time::interval(interval);
            ticker.set_missed_tick_behavior(MissedTickBehavior::Delay);
            // `interval` yields its first tick immediately.
            ticker.tick().await;
            loop {
                tokio::select! {
                    // Biased so that a stop that arrives with a due tick wins.
                    // A line written after the shutdown log line would read as
                    // a server that had not actually stopped.
                    biased;
                    () = notified.notified() => break,
                    _ = ticker.tick() => {
                        let disk = ticked.disk().await;
                        log_stats(&registry.stats(), disk.as_ref());
                        if let Some(disk) = &disk {
                            ticked.watch_disk(disk);
                        }
                        crate::refusal::flush();
                        ticked.record_tick();
                    }
                }
            }
        });
        Self {
            handle,
            stop,
            state,
        }
    }

    /// The pulse this tick is keeping, for wiring into the HTTP router.
    pub fn state(&self) -> Arc<HealthState> {
        self.state.clone()
    }

    /// Stops the tick and waits for the task to finish.
    ///
    /// Awaiting the join is what makes "stopped" true rather than "asked to
    /// stop": a caller that returns immediately can still race one last line
    /// into the log after its own shutdown message.
    pub async fn stop(mut self) {
        // `notify_one` rather than `notify_waiters`: it stores a permit, so a
        // stop that arrives before the task reaches its `select!` still stops
        // it instead of being dropped on the floor.
        self.stop.notify_one();
        // A task that has already finished, or whose runtime is going away,
        // is not an error here; the abort in `Drop` covers what is left.
        let _ = (&mut self.handle).await;
    }
}

impl Drop for HealthTick {
    fn drop(&mut self) {
        self.handle.abort();
    }
}

/// Writes one health line.
///
/// Counts only, and a fixed number of them. Nothing here can name an account
/// or quote a record, so the line is safe to leave running for the life of a
/// process — and nothing here grows with what the deployment holds, so it
/// stays one line. A real account carries dozens of collections and their
/// names are long; rendered inline, every ten seconds, they are most of the
/// log. `bot.did.stats` answers which collections, on request.
///
/// `disk_free` and `disk_total` are the data volume's, in bytes, so the
/// journal holds a history of how fast it filled. A run with no data
/// directory writes neither.
fn log_stats(stats: &RegistryStats, disk: Option<&DiskHealth>) {
    info!(
        accounts = stats.accounts,
        records = stats.records.records,
        bytes = stats.records.bytes,
        collections = stats.records.collections.len(),
        unreadable = stats.records.unreadable,
        disk_free = disk.map(|disk| disk.free_bytes),
        disk_total = disk.map(|disk| disk.total_bytes),
        "health"
    );
}

/// The fleet answers for its own zones.
///
/// Written here rather than in `didbot-tls` because [`CertificateAlerts`] is
/// this crate's trait: the HTTP surface decides what `GET /health` says, and
/// `didbot-tls` stays a library that knows nothing about a health route.
impl<D> CertificateAlerts for didbot_tls::fleet::CertificateFleet<D>
where
    D: didbot_tls::dns::AcmeDnsChallenge + 'static,
{
    fn zone_alert_levels(&self) -> BoxFuture<'_, Vec<(String, AlertLevel)>> {
        Box::pin(self.zone_alert_levels())
    }
}