knut-bifrost 0.3.0

//! A tiny live `/metrics` HTTP endpoint (plan.md §7), feature `metrics-http`.
//!
//! The poll loop already folds a cumulative [`Metrics`] after every round and
//! [`render_prometheus`] turns it into Prometheus text — but `--metrics-file`
//! only writes that text once, on shutdown. A Prometheus scraper wants to
//! *pull* it live. This module serves exactly that `render_prometheus` output
//! over HTTP at `/metrics`, so each scrape sees the current counters.
//!
//! Design, kept deliberately small:
//!
//! - **Pure-Rust, blocking, no async runtime.** [`tiny_http`] is a tiny
//!   blocking HTTP server; it runs on its own background thread, so it never
//!   touches the tokio poll loop. The dependency is behind the `metrics-http`
//!   feature, so the default / core build stays dependency-light.
//! - **Shared snapshot via `Arc<Mutex<Metrics>>`.** The loop owns the metrics
//!   and *publishes* a clone into the shared cell after each round (see
//!   [`MetricsHandle`]); a scrape briefly locks the cell, clones out the
//!   current [`Metrics`], releases the lock, and renders. So a scrape always
//!   returns a *consistent* snapshot and never blocks the loop for longer than
//!   a `Metrics` clone (a handful of `u64`s).
//! - **`/metrics` only.** `GET /metrics` → `200` + `text/plain; version=0.0.4`
//!   (the Prometheus exposition content type); every other path → `404`.
//!
//! ```no_run
//! # #[cfg(feature = "metrics-http")]
//! # fn demo() -> std::io::Result<()> {
//! use knut_bifrost::metrics_http::MetricsServer;
//!
//! // The handle is the shared snapshot cell; publish into it after each round.
//! let server = MetricsServer::start("127.0.0.1:9090")?;
//! let handle = server.handle();
//! // … in the poll loop, after folding a round, publish the current metrics:
//! // handle.publish(listener.metrics());
//! # let _ = handle;
//! # Ok(()) }
//! ```

use std::io;
use std::net::ToSocketAddrs;
use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;

use crate::metrics::{Metrics, render_prometheus};

/// The Prometheus text-exposition content type (format version 0.0.4). Scrapers
/// key off this to parse the body as the exposition format.
const PROMETHEUS_CONTENT_TYPE: &[u8] = b"text/plain; version=0.0.4";

/// A cheap, cloneable handle to the snapshot the [`MetricsServer`] serves. The
/// poll loop holds one and calls [`publish`](Self::publish) after every round;
/// the server thread reads it on each scrape. Cloning shares the same cell.
#[derive(Clone)]
pub struct MetricsHandle {
    inner: Arc<Mutex<Metrics>>,
}

impl MetricsHandle {
    /// Publish the latest [`Metrics`] for subsequent scrapes to render. Clones
    /// the counters into the shared cell under a very short-lived lock, so the
    /// loop never blocks on a scrape (and a scrape never sees a half-updated
    /// snapshot). Poison from a panicked scrape thread is recovered — a metrics
    /// snapshot can never be logically corrupt, so we just take the inner value.
    pub fn publish(&self, metrics: &Metrics) {
        let mut guard = self.inner.lock().unwrap_or_else(|e| e.into_inner());
        *guard = metrics.clone();
    }

    /// Render the current snapshot as Prometheus text. The server calls this on
    /// each `GET /metrics`; exposed so the snapshot path can be unit-tested
    /// without a socket.
    pub fn render(&self) -> String {
        let guard = self.inner.lock().unwrap_or_else(|e| e.into_inner());
        render_prometheus(&guard)
    }
}

/// A running `/metrics` HTTP server on a background thread. Dropping it stops
/// accepting and joins the thread (the server unblocks via `tiny_http`'s
/// `unblock`), so the endpoint lives exactly as long as this value.
pub struct MetricsServer {
    handle: MetricsHandle,
    server: Arc<tiny_http::Server>,
    thread: Option<JoinHandle<()>>,
}

impl MetricsServer {
    /// Bind `addr` (e.g. `127.0.0.1:9090`) and start serving `/metrics` on a
    /// background thread. The snapshot starts empty; drive it by publishing
    /// through [`handle`](Self::handle) after each round. Returns an
    /// [`io::Error`] if the address can't be parsed or bound.
    pub fn start(addr: impl ToSocketAddrs) -> io::Result<Self> {
        Self::start_with(addr, Metrics::default())
    }

    /// Like [`start`](Self::start) but seeds the served snapshot with `initial`
    /// (handy for tests that want a non-empty body before the first publish).
    pub fn start_with(addr: impl ToSocketAddrs, initial: Metrics) -> io::Result<Self> {
        // Resolve to a concrete address so `tiny_http` (which takes a single
        // `SocketAddr`) gets one and bind errors surface as `io::Error`.
        let sock = addr
            .to_socket_addrs()?
            .next()
            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "no socket address"))?;
        // `tiny_http::Server::http` returns a boxed error; normalise to io::Error.
        let server = tiny_http::Server::http(sock).map_err(|e| io::Error::other(e.to_string()))?;
        let server = Arc::new(server);
        let handle = MetricsHandle {
            inner: Arc::new(Mutex::new(initial)),
        };

        let thread = std::thread::Builder::new()
            .name("bifrost-metrics-http".into())
            .spawn({
                let server = Arc::clone(&server);
                let handle = handle.clone();
                move || serve_loop(&server, &handle)
            })?;

        Ok(Self {
            handle,
            server,
            thread: Some(thread),
        })
    }

    /// The address actually bound (useful when `:0` picked an ephemeral port).
    pub fn local_addr(&self) -> io::Result<std::net::SocketAddr> {
        match self.server.server_addr() {
            tiny_http::ListenAddr::IP(addr) => Ok(addr),
            // Unix-socket variant isn't used here; report it rather than panic.
            other => Err(io::Error::other(format!(
                "non-IP listen address: {other:?}"
            ))),
        }
    }

    /// A handle to the shared snapshot cell. Clone it freely; call
    /// [`MetricsHandle::publish`] after each round so scrapes see fresh counters.
    pub fn handle(&self) -> MetricsHandle {
        self.handle.clone()
    }
}

impl Drop for MetricsServer {
    fn drop(&mut self) {
        // Unblock the accept loop so the thread can observe the shutdown and
        // exit, then join it. `unblock` makes the blocking `recv()` return.
        self.server.unblock();
        if let Some(t) = self.thread.take() {
            let _ = t.join();
        }
    }
}

/// The accept loop: answer each request and exit when the server is unblocked
/// (drop) — `incoming_requests` ends once `unblock` has been called.
fn serve_loop(server: &tiny_http::Server, handle: &MetricsHandle) {
    for request in server.incoming_requests() {
        respond(request, handle);
    }
}

/// Answer one request: `GET /metrics` → `200` + Prometheus text; anything else
/// → `404`. Errors writing the response are ignored (a dropped client is not
/// the server's problem; the loop carries on).
fn respond(request: tiny_http::Request, handle: &MetricsHandle) {
    if request.url() == "/metrics" {
        let body = handle.render();
        let header = tiny_http::Header::from_bytes(&b"Content-Type"[..], PROMETHEUS_CONTENT_TYPE)
            .expect("static content-type header is valid");
        let response = tiny_http::Response::from_string(body).with_header(header);
        let _ = request.respond(response);
    } else {
        let response = tiny_http::Response::from_string("not found").with_status_code(404);
        let _ = request.respond(response);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::{Read, Write};
    use std::net::TcpStream;

    /// A minimal blocking HTTP/1.0 GET — no http-client dependency. Returns the
    /// raw response (status line + headers + body) so the test can assert on
    /// both the status line and the rendered metrics body.
    fn http_get(addr: std::net::SocketAddr, path: &str) -> io::Result<String> {
        let mut stream = TcpStream::connect(addr)?;
        // HTTP/1.0 so the server closes the connection after the body and our
        // read-to-end terminates without chunked/keep-alive bookkeeping.
        write!(stream, "GET {path} HTTP/1.0\r\nHost: localhost\r\n\r\n")?;
        stream.flush()?;
        let mut buf = String::new();
        stream.read_to_string(&mut buf)?;
        Ok(buf)
    }

    #[test]
    fn get_metrics_serves_rendered_counters() {
        // Seed a non-zero snapshot so the body has a counter to assert on, then
        // start on an ephemeral port (`:0`) and GET `/metrics`.
        let metrics = Metrics {
            rounds_total: 7,
            ..Metrics::default()
        };
        let server = MetricsServer::start_with("127.0.0.1:0", metrics).expect("start server");
        let addr = server.local_addr().expect("local addr");

        let resp = http_get(addr, "/metrics").expect("GET /metrics");
        // tiny_http echoes the request's HTTP version in the status line, so we
        // match the version-agnostic status code rather than `HTTP/1.1`.
        let status = resp.lines().next().unwrap_or_default();
        assert!(status.contains(" 200 "), "200 status, got:\n{resp}");
        assert!(
            resp.contains("text/plain; version=0.0.4"),
            "prometheus content type, got:\n{resp}"
        );
        // The body carries the live counter rendered by `render_prometheus`.
        assert!(
            resp.contains("bifrost_rounds_total 7"),
            "live counter, got:\n{resp}"
        );
        assert!(
            resp.contains("# TYPE bifrost_rounds_total counter"),
            "typed, got:\n{resp}"
        );
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "metrics_http::get_metrics_serves_rendered_counters",
            status.contains(" 200 ")
                && resp.contains("bifrost_rounds_total 7")
                && resp.contains("text/plain; version=0.0.4"),
            "GET /metrics → 200, prometheus content-type, live counter=7",
        );
    }

    #[test]
    fn publish_updates_the_served_snapshot() {
        let server = MetricsServer::start("127.0.0.1:0").expect("start server");
        let addr = server.local_addr().expect("local addr");

        // Empty before any publish.
        let resp = http_get(addr, "/metrics").expect("GET");
        assert!(
            resp.contains("bifrost_rounds_total 0"),
            "starts empty, got:\n{resp}"
        );

        // Publish a fresh snapshot; the next scrape reflects it.
        server.handle().publish(&Metrics {
            rounds_total: 3,
            rows_ingested: 42,
            ..Default::default()
        });
        let resp = http_get(addr, "/metrics").expect("GET");
        assert!(
            resp.contains("bifrost_rounds_total 3"),
            "published rounds, got:\n{resp}"
        );
        assert!(
            resp.contains("bifrost_rows_ingested 42"),
            "published rows, got:\n{resp}"
        );
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "metrics_http::publish_updates_the_served_snapshot",
            resp.contains("bifrost_rounds_total 3") && resp.contains("bifrost_rows_ingested 42"),
            "publish() reflected on next scrape: rounds=3, rows=42",
        );
    }

    #[test]
    fn unknown_path_is_404() {
        let server = MetricsServer::start("127.0.0.1:0").expect("start server");
        let addr = server.local_addr().expect("local addr");
        let resp = http_get(addr, "/nope").expect("GET");
        let status = resp.lines().next().unwrap_or_default();
        assert!(
            status.contains(" 404 "),
            "404 for unknown path, got:\n{resp}"
        );
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "metrics_http::unknown_path_is_404",
            status.contains(" 404 "),
            "unknown path → 404",
        );
    }
}