crabka_logfmt/
lib.rs

1//! Structured-JSON `tracing` log formatter shared across Crabka services.
2//!
3//! Every Crabka service (broker, gateway, operator, schema-registry) installs
4//! this formatter on its stdout `fmt` layer so container log collectors ingest
5//! each line as fields rather than ANSI-coloured human text. The output is
6//! shaped for Google Cloud Logging (GKE), whose agent parses stdout JSON and
7//! recognises a handful of special fields:
8//!
9//! - `severity` — mapped from the `tracing` level (`WARN` → `WARNING`,
10//!   `TRACE` → `DEBUG`); sets the log entry's `LogSeverity`.
11//! - `message` — the event message, flattened to the top level so it becomes
12//!   the entry's summary line.
13//! - `timestamp` — RFC3339 UTC, recognised as the entry timestamp.
14//!
15//! Everything else (the event `target` and all event fields) is emitted at the
16//! top level too, so a line looks like:
17//!
18//! ```json
19//! {"timestamp":"2026-06-13T05:55:09.951788Z","severity":"INFO","target":"crabka_broker::network::dispatch","message":"connection opened","listener":"PLAIN","sasl":false}
20//! ```
21//!
22//! The JSON formatter never writes ANSI escape codes, so logs stay clean in
23//! non-TTY environments (the default `tracing_subscriber` `fmt` layer emits
24//! ANSI colours regardless of whether stdout is a terminal — the bug this
25//! crate exists to avoid).
26
27#![forbid(unsafe_code)]
28
29use std::fmt;
30
31use serde_json::{Map, Value};
32use tracing::field::{Field, Visit};
33use tracing::{Event, Level, Subscriber};
34use tracing_subscriber::fmt::format::Writer;
35use tracing_subscriber::fmt::time::{FormatTime, SystemTime};
36use tracing_subscriber::fmt::{FmtContext, FormatEvent, FormatFields, MakeWriter};
37use tracing_subscriber::registry::LookupSpan;
38use tracing_subscriber::{EnvFilter, Layer, Registry};
39
40/// Map a `tracing` [`Level`] to a Google Cloud Logging `LogSeverity` string.
41///
42/// `WARN` becomes `WARNING` (Cloud Logging's spelling) and `TRACE` floors to
43/// `DEBUG` (Cloud Logging has no finer level). See
44/// <https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry#LogSeverity>.
45fn severity(level: Level) -> &'static str {
46    match level {
47        Level::ERROR => "ERROR",
48        Level::WARN => "WARNING",
49        Level::INFO => "INFO",
50        Level::DEBUG | Level::TRACE => "DEBUG",
51    }
52}
53
54/// Collects `tracing` event fields into a JSON object, preserving primitive
55/// types and stringifying everything else via `Debug`.
56#[derive(Default)]
57struct JsonVisitor {
58    fields: Map<String, Value>,
59}
60
61impl JsonVisitor {
62    fn insert(&mut self, field: &Field, value: Value) {
63        self.fields.insert(field.name().to_owned(), value);
64    }
65}
66
67impl Visit for JsonVisitor {
68    fn record_str(&mut self, field: &Field, value: &str) {
69        self.insert(field, Value::from(value));
70    }
71
72    fn record_bool(&mut self, field: &Field, value: bool) {
73        self.insert(field, Value::from(value));
74    }
75
76    fn record_i64(&mut self, field: &Field, value: i64) {
77        self.insert(field, Value::from(value));
78    }
79
80    fn record_u64(&mut self, field: &Field, value: u64) {
81        self.insert(field, Value::from(value));
82    }
83
84    fn record_f64(&mut self, field: &Field, value: f64) {
85        self.insert(field, Value::from(value));
86    }
87
88    fn record_debug(&mut self, field: &Field, value: &dyn fmt::Debug) {
89        // The event message is recorded here as a `fmt::Arguments`, whose
90        // `Debug` impl is its `Display` output — so `message` is a clean string,
91        // not a quoted one.
92        self.insert(field, Value::from(format!("{value:?}")));
93    }
94}
95
96/// `tracing_subscriber` event formatter emitting one Cloud Logging-friendly
97/// JSON object per line. See the crate-level docs for the output shape.
98#[derive(Debug, Clone, Copy, Default)]
99pub struct CloudLogging;
100
101impl<S, N> FormatEvent<S, N> for CloudLogging
102where
103    S: Subscriber + for<'a> LookupSpan<'a>,
104    N: for<'a> FormatFields<'a> + 'static,
105{
106    fn format_event(
107        &self,
108        _ctx: &FmtContext<'_, S, N>,
109        mut writer: Writer<'_>,
110        event: &Event<'_>,
111    ) -> fmt::Result {
112        let meta = event.metadata();
113        let mut obj = Map::new();
114
115        // Reuse `tracing_subscriber`'s default timer so the timestamp format
116        // (RFC3339 UTC, microsecond precision) matches the rest of the
117        // ecosystem exactly.
118        let mut timestamp = String::new();
119        {
120            let mut tw = Writer::new(&mut timestamp);
121            SystemTime.format_time(&mut tw)?;
122        }
123        obj.insert("timestamp".to_owned(), Value::from(timestamp));
124        obj.insert("severity".to_owned(), Value::from(severity(*meta.level())));
125        obj.insert("target".to_owned(), Value::from(meta.target()));
126
127        // Flatten event fields (including `message`) to the top level. `entry`
128        // keeps the reserved keys above from being clobbered by a same-named
129        // user field.
130        let mut visitor = JsonVisitor::default();
131        event.record(&mut visitor);
132        for (key, value) in visitor.fields {
133            obj.entry(key).or_insert(value);
134        }
135
136        let line = serde_json::to_string(&Value::Object(obj)).map_err(|_| fmt::Error)?;
137        writeln!(writer, "{line}")
138    }
139}
140
141/// Build a stdout `fmt` layer that emits [`CloudLogging`] JSON, filtered by
142/// `filter`. `make_writer` is the sink — production passes `std::io::stdout`;
143/// tests pass a capturing buffer.
144///
145/// Returns a boxed layer over a [`Registry`] so call sites compose it with
146/// `tracing_subscriber::registry().with(...)`.
147#[must_use]
148pub fn layer<W>(filter: EnvFilter, make_writer: W) -> Box<dyn Layer<Registry> + Send + Sync>
149where
150    W: for<'writer> MakeWriter<'writer> + Send + Sync + 'static,
151{
152    tracing_subscriber::fmt::layer()
153        .event_format(CloudLogging)
154        .with_writer(make_writer)
155        .with_filter(filter)
156        .boxed()
157}
158
159#[cfg(test)]
160mod tests {
161    use super::*;
162    use assert2::assert;
163    use std::io::Write;
164    use std::sync::{Arc, Mutex};
165    use tracing_subscriber::layer::SubscriberExt as _;
166
167    #[derive(Clone)]
168    struct Buf(Arc<Mutex<Vec<u8>>>);
169
170    impl Write for Buf {
171        fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
172            self.0.lock().unwrap().extend_from_slice(bytes);
173            Ok(bytes.len())
174        }
175        fn flush(&mut self) -> std::io::Result<()> {
176            Ok(())
177        }
178    }
179
180    impl<'a> MakeWriter<'a> for Buf {
181        type Writer = Buf;
182        fn make_writer(&'a self) -> Self::Writer {
183            self.clone()
184        }
185    }
186
187    /// Capture the JSON emitted for a single event by `emit`.
188    fn capture(emit: impl FnOnce()) -> serde_json::Value {
189        let buf = Arc::new(Mutex::new(Vec::new()));
190        let subscriber =
191            tracing_subscriber::registry().with(layer(EnvFilter::new("trace"), Buf(buf.clone())));
192        tracing::subscriber::with_default(subscriber, emit);
193        let out = String::from_utf8(buf.lock().unwrap().clone()).expect("utf8 log output");
194        // The crux of the fix: captured logs carry no ANSI escape sequences.
195        assert!(!out.contains('\u{1b}'));
196        let line = out.lines().next().expect("a log line");
197        serde_json::from_str(line).expect("each log line is valid JSON")
198    }
199
200    #[test]
201    fn emits_cloud_logging_json() {
202        let v = capture(|| {
203            tracing::info!(
204                listener = "PLAIN",
205                sasl = false,
206                port = 9092,
207                "connection opened"
208            );
209        });
210        assert!(v["severity"] == "INFO");
211        assert!(v["message"] == "connection opened");
212        assert!(v["listener"] == "PLAIN");
213        // Field types are preserved, not stringified.
214        assert!(v["sasl"] == false);
215        assert!(v["port"] == 9092);
216        assert!(v["target"].is_string());
217        assert!(v["timestamp"].as_str().is_some_and(|t| t.ends_with('Z')));
218        // `level` is replaced by Cloud Logging's `severity`.
219        assert!(v.get("level").is_none());
220    }
221
222    #[test]
223    fn maps_warn_to_warning_severity() {
224        let v = capture(|| tracing::warn!("disk almost full"));
225        assert!(v["severity"] == "WARNING");
226        assert!(v["message"] == "disk almost full");
227    }
228
229    #[test]
230    fn maps_trace_to_debug_severity() {
231        let v = capture(|| tracing::trace!("fine-grained detail"));
232        assert!(v["severity"] == "DEBUG");
233    }
234
235    #[test]
236    fn preserves_u64_and_f64_field_types() {
237        // The other record_* visitor paths are covered by emits_cloud_logging_json
238        // (str/bool/i64/debug); u64 and f64 have their own visitor methods, so
239        // exercise them too and confirm the values survive as native JSON types.
240        let v = capture(|| {
241            tracing::info!(offset = 42_u64, ratio = 0.5_f64, "metrics");
242        });
243        assert!(v["offset"] == 42_u64);
244        assert!(v["ratio"] == 0.5);
245    }
246}
crabka_logfmt/lib.rs

crabka_logfmt/
lib.rs
