gunnar-common 3.0.0

//! **The log tap**: every `INFO`+ line gunnar logs, handed to ONE in-process
//! reader as a value — the console dashboard (`gunnar_server::dashboard`).
//!
//! # Why a tracing layer and not a second event bus
//!
//! The things a human at the appliance's screen wants to see — a login, a
//! push, a clone, an admin call, a DHCP lease — are ALREADY said, once each,
//! as structured `tracing` events: `ssh.auth.accepted` in `gunnar-ssh`,
//! `git.transfer.finished` in the same crate, `grpc.call` in `gunnar-grpc`,
//! `DHCP ACK: lease taken` in `gunnar-server::dhcp`. Four crates, none of
//! which may depend on the server crate that draws the screen. A `publish()`
//! at each site would be a second path to the same facts, kept in step by
//! hand (one implementation), and a dependency edge in the wrong direction.
//!
//! So the screen reads the log. This layer sits in the subscriber
//! [`crate::logging::init`] installs, **after** the reloadable filter (so an
//! operator's `RUST_LOG`/`SIGUSR1` rung governs it exactly as it governs
//! stderr) and forwards each event as a [`LogLine`] — level, target, message,
//! the event's own fields, and the fields of every span it sits in (that is
//! where `repo`, `principal` and `peer` live: they are recorded on the
//! `ssh.connection` span once, not repeated on every line).
//!
//! # It can never slow or stop the server
//!
//! One bounded `sync_channel` of [`CAPACITY`] lines, `try_send` only: a reader
//! that is slow, or absent — every developer box, every container, every
//! `gunnar-solo` — costs the logging path one failed non-blocking send per
//! line and nothing else. Lines the buffer cannot hold are dropped and
//! counted ([`dropped`]); the buffer exists so the lines logged BEFORE the
//! dashboard starts (the DHCP lease is taken before `serve` binds anything)
//! are still there when it does.
//!
//! [`take_receiver`] hands the receiving end out **once**. There is one
//! screen; a second taker gets `None` and must not pretend to have one.

use std::fmt::Write as _;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::mpsc::{sync_channel, Receiver, SyncSender, TrySendError};
use std::sync::{Mutex, OnceLock};

use tracing::field::{Field, Visit};
use tracing::{Event, Level, Subscriber};
use tracing_subscriber::layer::{Context, Layer};
use tracing_subscriber::registry::LookupSpan;

/// How many lines wait for the reader. 256 × a few hundred bytes: bounded,
/// preallocated by the channel, never grows.
pub const CAPACITY: usize = 256;

/// One logged line, as values.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LogLine {
    pub level: Level,
    /// The `target:` — `gunnar::ssh`, `gunnar::git`, `gunnar::audit`, or the
    /// module path when none was named.
    pub target: &'static str,
    /// The format-string message; empty when the event had only fields.
    pub message: String,
    /// The event's own fields, in declaration order, values rendered with
    /// `Debug`/`Display` the way `fmt` would (strings unquoted).
    pub fields: Vec<(&'static str, String)>,
    /// The fields of the spans this event sits in, outermost first. A field
    /// recorded later via `Span::record` replaces the earlier value.
    pub span_fields: Vec<(&'static str, String)>,
}

impl LogLine {
    /// The value of an event field by name.
    pub fn field(&self, name: &str) -> Option<&str> {
        self.fields
            .iter()
            .find(|(k, _)| *k == name)
            .map(|(_, v)| v.as_str())
    }

    /// A field, from the event first and then from the innermost span that
    /// carries it — `repo` and `principal` are span fields on a transfer,
    /// event fields on a gRPC call, and a reader should not have to know.
    pub fn get(&self, name: &str) -> Option<&str> {
        self.field(name).or_else(|| {
            self.span_fields
                .iter()
                .rev()
                .find(|(k, _)| *k == name)
                .map(|(_, v)| v.as_str())
        })
    }

    /// The structured `event = "…"` name, when the line has one.
    pub fn event(&self) -> Option<&str> {
        self.field("event")
    }
}

struct Tap {
    tx: SyncSender<LogLine>,
}

static TAP: OnceLock<Tap> = OnceLock::new();
static RECEIVER: OnceLock<Mutex<Option<Receiver<LogLine>>>> = OnceLock::new();
static DROPPED: AtomicU64 = AtomicU64::new(0);

/// The layer. Constructing it creates the channel; a second construction
/// (a test binary that installs its own subscriber) reuses the same one.
pub struct TapLayer;

impl TapLayer {
    pub fn new() -> Self {
        TAP.get_or_init(|| {
            let (tx, rx) = sync_channel(CAPACITY);
            let _ = RECEIVER.set(Mutex::new(Some(rx)));
            Tap { tx }
        });
        TapLayer
    }
}

impl Default for TapLayer {
    fn default() -> Self {
        Self::new()
    }
}

/// The reading end, once. `None` when already taken, or when no
/// [`TapLayer`] was ever installed.
pub fn take_receiver() -> Option<Receiver<LogLine>> {
    RECEIVER.get()?.lock().ok()?.take()
}

/// Lines the buffer could not hold since start.
pub fn dropped() -> u64 {
    DROPPED.load(Ordering::Relaxed)
}

/// Push one line from outside the tracing pipeline — the dashboard's own
/// notes (a lease it was handed at start) ride the same channel.
pub fn push(line: LogLine) {
    if let Some(tap) = TAP.get() {
        offer(&tap.tx, line);
    }
}

fn offer(tx: &SyncSender<LogLine>, line: LogLine) {
    match tx.try_send(line) {
        Ok(()) => {}
        Err(TrySendError::Full(_)) | Err(TrySendError::Disconnected(_)) => {
            DROPPED.fetch_add(1, Ordering::Relaxed);
        }
    }
}

/// The fields a span was created with plus what `record` added later, kept
/// in the registry's per-span extensions.
#[derive(Default)]
struct SpanFields(Vec<(&'static str, String)>);

impl SpanFields {
    fn put(&mut self, name: &'static str, value: String) {
        if let Some(slot) = self.0.iter_mut().find(|(k, _)| *k == name) {
            slot.1 = value;
        } else {
            self.0.push((name, value));
        }
    }
}

/// Renders field values the way `fmt` does: strings bare, everything else
/// through `Debug` (which is what `%`/`?` sigils already produced).
struct Collect<'a> {
    message: &'a mut String,
    fields: &'a mut Vec<(&'static str, String)>,
}

impl Visit for Collect<'_> {
    fn record_str(&mut self, field: &Field, value: &str) {
        if field.name() == "message" {
            self.message.push_str(value);
        } else {
            self.fields.push((field.name(), value.to_owned()));
        }
    }

    fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
        if field.name() == "message" {
            let _ = write!(self.message, "{value:?}");
        } else {
            self.fields.push((field.name(), format!("{value:?}")));
        }
    }
}

impl<S> Layer<S> for TapLayer
where
    S: Subscriber + for<'a> LookupSpan<'a>,
{
    fn on_new_span(
        &self,
        attrs: &tracing::span::Attributes<'_>,
        id: &tracing::span::Id,
        ctx: Context<'_, S>,
    ) {
        let Some(span) = ctx.span(id) else { return };
        let mut fields = SpanFields::default();
        let mut message = String::new();
        attrs.record(&mut Collect {
            message: &mut message,
            fields: &mut fields.0,
        });
        span.extensions_mut().insert(fields);
    }

    fn on_record(
        &self,
        id: &tracing::span::Id,
        values: &tracing::span::Record<'_>,
        ctx: Context<'_, S>,
    ) {
        let Some(span) = ctx.span(id) else { return };
        let mut fresh = Vec::new();
        let mut message = String::new();
        values.record(&mut Collect {
            message: &mut message,
            fields: &mut fresh,
        });
        let mut ext = span.extensions_mut();
        let fields = ext.get_mut::<SpanFields>();
        match fields {
            Some(fields) => fresh.into_iter().for_each(|(k, v)| fields.put(k, v)),
            None => ext.insert(SpanFields(fresh)),
        }
    }

    fn on_event(&self, event: &Event<'_>, ctx: Context<'_, S>) {
        let Some(tap) = TAP.get() else { return };
        // INFO and louder only: DEBUG/TRACE is a firehose the screen never
        // wants, and cheapest to refuse before allocating anything.
        if *event.metadata().level() > Level::INFO {
            return;
        }
        let mut message = String::new();
        let mut fields = Vec::new();
        event.record(&mut Collect {
            message: &mut message,
            fields: &mut fields,
        });
        let mut span_fields: Vec<(&'static str, String)> = Vec::new();
        if let Some(scope) = ctx.event_scope(event) {
            for span in scope.from_root() {
                if let Some(sf) = span.extensions().get::<SpanFields>() {
                    for (k, v) in &sf.0 {
                        if let Some(slot) = span_fields.iter_mut().find(|(n, _)| n == k) {
                            slot.1 = v.clone();
                        } else {
                            span_fields.push((k, v.clone()));
                        }
                    }
                }
            }
        }
        offer(
            &tap.tx,
            LogLine {
                level: *event.metadata().level(),
                target: event.metadata().target(),
                message,
                fields,
                span_fields,
            },
        );
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tracing_subscriber::layer::SubscriberExt as _;

    /// One subscriber for the whole test module: the tap is a process-wide
    /// singleton, so the tests share it — and run one at a time, because a
    /// test draining the channel while another floods it would make the
    /// drop count mean nothing.
    static ONE_AT_A_TIME: std::sync::Mutex<()> = std::sync::Mutex::new(());

    fn with_tap<R>(f: impl FnOnce() -> R) -> R {
        let _serial = ONE_AT_A_TIME.lock().unwrap_or_else(|e| e.into_inner());
        let subscriber = tracing_subscriber::registry().with(TapLayer::new());
        tracing::subscriber::with_default(subscriber, f)
    }

    #[test]
    fn an_event_arrives_with_its_fields_and_the_span_fields_around_it() {
        let rx = with_tap(|| {
            let rx = take_receiver();
            // The flood test may have filled the buffer first: empty it so
            // this test's line has a slot.
            if let Some(rx) = &rx {
                for _ in rx.try_iter() {}
            }
            let span = tracing::info_span!(
                "ssh.connection",
                repo = "ns/r",
                principal = tracing::field::Empty
            );
            span.record("principal", "alice");
            let _e = span.enter();
            tracing::info!(target: "tap-test", event = "ssh.auth.accepted", fingerprint = "SHA256:x", "publickey accepted");
            tracing::debug!(target: "tap-test", "never forwarded");
            rx.map(|rx| rx.try_iter().collect::<Vec<_>>())
        });
        let Some(lines) = rx else { return };
        let rx = lines.into_iter();
        let mut rx = rx;
        let line = rx
            .by_ref()
            .find(|l| l.target == "tap-test")
            .expect("the info line was tapped");
        assert_eq!(line.level, Level::INFO);
        assert_eq!(line.message, "publickey accepted");
        assert_eq!(line.event(), Some("ssh.auth.accepted"));
        assert_eq!(line.field("fingerprint"), Some("SHA256:x"));
        assert_eq!(line.get("repo"), Some("ns/r"));
        assert_eq!(
            line.get("principal"),
            Some("alice"),
            "a recorded-later span field is seen"
        );
        assert!(
            rx.all(|l| l.target != "tap-test" || l.message != "never forwarded"),
            "DEBUG is refused before the channel"
        );
    }

    #[test]
    fn a_full_buffer_drops_and_counts_rather_than_blocking() {
        let before = dropped();
        // Whether or not the receiver was taken, pushing far past CAPACITY
        // must return promptly; the count says the surplus went nowhere.
        with_tap(|| {
            for i in 0..(CAPACITY * 2) {
                push(LogLine {
                    level: Level::INFO,
                    target: "tap-flood",
                    message: format!("{i}"),
                    fields: vec![],
                    span_fields: vec![],
                });
            }
        });
        assert!(
            dropped() >= before + CAPACITY as u64,
            "the surplus was counted, not waited for"
        );
    }
}