gunnar_common/
logtap.rs

1//! **The log tap**: every `INFO`+ line gunnar logs, handed to ONE in-process
2//! reader as a value — the console dashboard (`gunnar_server::dashboard`).
3//!
4//! # Why a tracing layer and not a second event bus
5//!
6//! The things a human at the appliance's screen wants to see — a login, a
7//! push, a clone, an admin call, a DHCP lease — are ALREADY said, once each,
8//! as structured `tracing` events: `ssh.auth.accepted` in `gunnar-ssh`,
9//! `git.transfer.finished` in the same crate, `grpc.call` in `gunnar-grpc`,
10//! `DHCP ACK: lease taken` in `gunnar-server::dhcp`. Four crates, none of
11//! which may depend on the server crate that draws the screen. A `publish()`
12//! at each site would be a second path to the same facts, kept in step by
13//! hand (one implementation), and a dependency edge in the wrong direction.
14//!
15//! So the screen reads the log. This layer sits in the subscriber
16//! [`crate::logging::init`] installs, **after** the reloadable filter (so an
17//! operator's `RUST_LOG`/`SIGUSR1` rung governs it exactly as it governs
18//! stderr) and forwards each event as a [`LogLine`] — level, target, message,
19//! the event's own fields, and the fields of every span it sits in (that is
20//! where `repo`, `principal` and `peer` live: they are recorded on the
21//! `ssh.connection` span once, not repeated on every line).
22//!
23//! # It can never slow or stop the server
24//!
25//! One bounded `sync_channel` of [`CAPACITY`] lines, `try_send` only: a reader
26//! that is slow, or absent — every developer box, every container, every
27//! `gunnar-solo` — costs the logging path one failed non-blocking send per
28//! line and nothing else. Lines the buffer cannot hold are dropped and
29//! counted ([`dropped`]); the buffer exists so the lines logged BEFORE the
30//! dashboard starts (the DHCP lease is taken before `serve` binds anything)
31//! are still there when it does.
32//!
33//! [`take_receiver`] hands the receiving end out **once**. There is one
34//! screen; a second taker gets `None` and must not pretend to have one.
35
36use std::fmt::Write as _;
37use std::sync::atomic::{AtomicU64, Ordering};
38use std::sync::mpsc::{sync_channel, Receiver, SyncSender, TrySendError};
39use std::sync::{Mutex, OnceLock};
40
41use tracing::field::{Field, Visit};
42use tracing::{Event, Level, Subscriber};
43use tracing_subscriber::layer::{Context, Layer};
44use tracing_subscriber::registry::LookupSpan;
45
46/// How many lines wait for the reader. 256 × a few hundred bytes: bounded,
47/// preallocated by the channel, never grows.
48pub const CAPACITY: usize = 256;
49
50/// One logged line, as values.
51#[derive(Debug, Clone, PartialEq, Eq)]
52pub struct LogLine {
53    pub level: Level,
54    /// The `target:` — `gunnar::ssh`, `gunnar::git`, `gunnar::audit`, or the
55    /// module path when none was named.
56    pub target: &'static str,
57    /// The format-string message; empty when the event had only fields.
58    pub message: String,
59    /// The event's own fields, in declaration order, values rendered with
60    /// `Debug`/`Display` the way `fmt` would (strings unquoted).
61    pub fields: Vec<(&'static str, String)>,
62    /// The fields of the spans this event sits in, outermost first. A field
63    /// recorded later via `Span::record` replaces the earlier value.
64    pub span_fields: Vec<(&'static str, String)>,
65}
66
67impl LogLine {
68    /// The value of an event field by name.
69    pub fn field(&self, name: &str) -> Option<&str> {
70        self.fields
71            .iter()
72            .find(|(k, _)| *k == name)
73            .map(|(_, v)| v.as_str())
74    }
75
76    /// A field, from the event first and then from the innermost span that
77    /// carries it — `repo` and `principal` are span fields on a transfer,
78    /// event fields on a gRPC call, and a reader should not have to know.
79    pub fn get(&self, name: &str) -> Option<&str> {
80        self.field(name).or_else(|| {
81            self.span_fields
82                .iter()
83                .rev()
84                .find(|(k, _)| *k == name)
85                .map(|(_, v)| v.as_str())
86        })
87    }
88
89    /// The structured `event = "…"` name, when the line has one.
90    pub fn event(&self) -> Option<&str> {
91        self.field("event")
92    }
93}
94
95struct Tap {
96    tx: SyncSender<LogLine>,
97}
98
99static TAP: OnceLock<Tap> = OnceLock::new();
100static RECEIVER: OnceLock<Mutex<Option<Receiver<LogLine>>>> = OnceLock::new();
101static DROPPED: AtomicU64 = AtomicU64::new(0);
102
103/// The layer. Constructing it creates the channel; a second construction
104/// (a test binary that installs its own subscriber) reuses the same one.
105pub struct TapLayer;
106
107impl TapLayer {
108    pub fn new() -> Self {
109        TAP.get_or_init(|| {
110            let (tx, rx) = sync_channel(CAPACITY);
111            let _ = RECEIVER.set(Mutex::new(Some(rx)));
112            Tap { tx }
113        });
114        TapLayer
115    }
116}
117
118impl Default for TapLayer {
119    fn default() -> Self {
120        Self::new()
121    }
122}
123
124/// The reading end, once. `None` when already taken, or when no
125/// [`TapLayer`] was ever installed.
126pub fn take_receiver() -> Option<Receiver<LogLine>> {
127    RECEIVER.get()?.lock().ok()?.take()
128}
129
130/// Lines the buffer could not hold since start.
131pub fn dropped() -> u64 {
132    DROPPED.load(Ordering::Relaxed)
133}
134
135/// Push one line from outside the tracing pipeline — the dashboard's own
136/// notes (a lease it was handed at start) ride the same channel.
137pub fn push(line: LogLine) {
138    if let Some(tap) = TAP.get() {
139        offer(&tap.tx, line);
140    }
141}
142
143fn offer(tx: &SyncSender<LogLine>, line: LogLine) {
144    match tx.try_send(line) {
145        Ok(()) => {}
146        Err(TrySendError::Full(_)) | Err(TrySendError::Disconnected(_)) => {
147            DROPPED.fetch_add(1, Ordering::Relaxed);
148        }
149    }
150}
151
152/// The fields a span was created with plus what `record` added later, kept
153/// in the registry's per-span extensions.
154#[derive(Default)]
155struct SpanFields(Vec<(&'static str, String)>);
156
157impl SpanFields {
158    fn put(&mut self, name: &'static str, value: String) {
159        if let Some(slot) = self.0.iter_mut().find(|(k, _)| *k == name) {
160            slot.1 = value;
161        } else {
162            self.0.push((name, value));
163        }
164    }
165}
166
167/// Renders field values the way `fmt` does: strings bare, everything else
168/// through `Debug` (which is what `%`/`?` sigils already produced).
169struct Collect<'a> {
170    message: &'a mut String,
171    fields: &'a mut Vec<(&'static str, String)>,
172}
173
174impl Visit for Collect<'_> {
175    fn record_str(&mut self, field: &Field, value: &str) {
176        if field.name() == "message" {
177            self.message.push_str(value);
178        } else {
179            self.fields.push((field.name(), value.to_owned()));
180        }
181    }
182
183    fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
184        if field.name() == "message" {
185            let _ = write!(self.message, "{value:?}");
186        } else {
187            self.fields.push((field.name(), format!("{value:?}")));
188        }
189    }
190}
191
192impl<S> Layer<S> for TapLayer
193where
194    S: Subscriber + for<'a> LookupSpan<'a>,
195{
196    fn on_new_span(
197        &self,
198        attrs: &tracing::span::Attributes<'_>,
199        id: &tracing::span::Id,
200        ctx: Context<'_, S>,
201    ) {
202        let Some(span) = ctx.span(id) else { return };
203        let mut fields = SpanFields::default();
204        let mut message = String::new();
205        attrs.record(&mut Collect {
206            message: &mut message,
207            fields: &mut fields.0,
208        });
209        span.extensions_mut().insert(fields);
210    }
211
212    fn on_record(
213        &self,
214        id: &tracing::span::Id,
215        values: &tracing::span::Record<'_>,
216        ctx: Context<'_, S>,
217    ) {
218        let Some(span) = ctx.span(id) else { return };
219        let mut fresh = Vec::new();
220        let mut message = String::new();
221        values.record(&mut Collect {
222            message: &mut message,
223            fields: &mut fresh,
224        });
225        let mut ext = span.extensions_mut();
226        let fields = ext.get_mut::<SpanFields>();
227        match fields {
228            Some(fields) => fresh.into_iter().for_each(|(k, v)| fields.put(k, v)),
229            None => ext.insert(SpanFields(fresh)),
230        }
231    }
232
233    fn on_event(&self, event: &Event<'_>, ctx: Context<'_, S>) {
234        let Some(tap) = TAP.get() else { return };
235        // INFO and louder only: DEBUG/TRACE is a firehose the screen never
236        // wants, and cheapest to refuse before allocating anything.
237        if *event.metadata().level() > Level::INFO {
238            return;
239        }
240        let mut message = String::new();
241        let mut fields = Vec::new();
242        event.record(&mut Collect {
243            message: &mut message,
244            fields: &mut fields,
245        });
246        let mut span_fields: Vec<(&'static str, String)> = Vec::new();
247        if let Some(scope) = ctx.event_scope(event) {
248            for span in scope.from_root() {
249                if let Some(sf) = span.extensions().get::<SpanFields>() {
250                    for (k, v) in &sf.0 {
251                        if let Some(slot) = span_fields.iter_mut().find(|(n, _)| n == k) {
252                            slot.1 = v.clone();
253                        } else {
254                            span_fields.push((k, v.clone()));
255                        }
256                    }
257                }
258            }
259        }
260        offer(
261            &tap.tx,
262            LogLine {
263                level: *event.metadata().level(),
264                target: event.metadata().target(),
265                message,
266                fields,
267                span_fields,
268            },
269        );
270    }
271}
272
273#[cfg(test)]
274mod tests {
275    use super::*;
276    use tracing_subscriber::layer::SubscriberExt as _;
277
278    /// One subscriber for the whole test module: the tap is a process-wide
279    /// singleton, so the tests share it — and run one at a time, because a
280    /// test draining the channel while another floods it would make the
281    /// drop count mean nothing.
282    static ONE_AT_A_TIME: std::sync::Mutex<()> = std::sync::Mutex::new(());
283
284    fn with_tap<R>(f: impl FnOnce() -> R) -> R {
285        let _serial = ONE_AT_A_TIME.lock().unwrap_or_else(|e| e.into_inner());
286        let subscriber = tracing_subscriber::registry().with(TapLayer::new());
287        tracing::subscriber::with_default(subscriber, f)
288    }
289
290    #[test]
291    fn an_event_arrives_with_its_fields_and_the_span_fields_around_it() {
292        let rx = with_tap(|| {
293            let rx = take_receiver();
294            // The flood test may have filled the buffer first: empty it so
295            // this test's line has a slot.
296            if let Some(rx) = &rx {
297                for _ in rx.try_iter() {}
298            }
299            let span = tracing::info_span!(
300                "ssh.connection",
301                repo = "ns/r",
302                principal = tracing::field::Empty
303            );
304            span.record("principal", "alice");
305            let _e = span.enter();
306            tracing::info!(target: "tap-test", event = "ssh.auth.accepted", fingerprint = "SHA256:x", "publickey accepted");
307            tracing::debug!(target: "tap-test", "never forwarded");
308            rx.map(|rx| rx.try_iter().collect::<Vec<_>>())
309        });
310        let Some(lines) = rx else { return };
311        let rx = lines.into_iter();
312        let mut rx = rx;
313        let line = rx
314            .by_ref()
315            .find(|l| l.target == "tap-test")
316            .expect("the info line was tapped");
317        assert_eq!(line.level, Level::INFO);
318        assert_eq!(line.message, "publickey accepted");
319        assert_eq!(line.event(), Some("ssh.auth.accepted"));
320        assert_eq!(line.field("fingerprint"), Some("SHA256:x"));
321        assert_eq!(line.get("repo"), Some("ns/r"));
322        assert_eq!(
323            line.get("principal"),
324            Some("alice"),
325            "a recorded-later span field is seen"
326        );
327        assert!(
328            rx.all(|l| l.target != "tap-test" || l.message != "never forwarded"),
329            "DEBUG is refused before the channel"
330        );
331    }
332
333    #[test]
334    fn a_full_buffer_drops_and_counts_rather_than_blocking() {
335        let before = dropped();
336        // Whether or not the receiver was taken, pushing far past CAPACITY
337        // must return promptly; the count says the surplus went nowhere.
338        with_tap(|| {
339            for i in 0..(CAPACITY * 2) {
340                push(LogLine {
341                    level: Level::INFO,
342                    target: "tap-flood",
343                    message: format!("{i}"),
344                    fields: vec![],
345                    span_fields: vec![],
346                });
347            }
348        });
349        assert!(
350            dropped() >= before + CAPACITY as u64,
351            "the surplus was counted, not waited for"
352        );
353    }
354}
gunnar_common/logtap.rs

gunnar_common/
logtap.rs
