krabka_logfmt/
lib.rs

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

krabka_logfmt/
lib.rs
