tuwunel_core/log/
journald.rs

1//! Native systemd journal submission and field recording.
2//!
3//! The module routes formatted messages to the journal socket and records
4//! structured tracing fields for journal queries.
5
6#[cfg(unix)]
7use std::os::unix::net::UnixDatagram;
8use std::{
9	cell::RefCell,
10	env::args_os,
11	ffi::OsStr,
12	fmt::Debug,
13	io::{self, Write, stderr},
14	path::Path,
15};
16
17use tracing::{
18	Event, Level, Metadata, Subscriber,
19	field::{Field, Visit},
20	level_filters::LevelFilter,
21	span::{Attributes, Id, Record},
22	subscriber::Interest,
23};
24use tracing_subscriber::{
25	layer::{Context, Layer},
26	registry::LookupSpan,
27};
28
29#[cfg(not(unix))]
30use self::unsupported::UnixDatagram;
31use super::is_systemd_mode;
32use crate::{
33	Config, Result,
34	arrayvec::ArrayString,
35	err, format_small_string, implement,
36	smallstr::SmallString,
37	smallvec::SmallVec,
38	utils::{
39		math::{ExpectInto, Expected},
40		string::{Unquote, to_array_string},
41	},
42};
43
44#[cfg(test)]
45mod tests;
46
47/// Stand-in for the journald socket on targets outside the unix family.
48///
49/// Nothing constructs it, since `enabled()` is always false without systemd.
50#[cfg(not(unix))]
51mod unsupported {
52	use std::io;
53
54	use super::implement;
55
56	pub(super) struct UnixDatagram;
57
58	#[implement(UnixDatagram)]
59	pub(super) fn unbound() -> io::Result<Self> { Err(unavailable()) }
60
61	#[implement(UnixDatagram)]
62	pub(super) fn send_to(&self, _payload: &[u8], _path: &str) -> io::Result<usize> {
63		Err(unavailable())
64	}
65
66	fn unavailable() -> io::Error {
67		io::Error::new(io::ErrorKind::Unsupported, "journald requires unix datagram sockets")
68	}
69}
70
71/// Encoded journal fields; a longer run spills to the heap.
72type Buffer = SmallVec<[u8; 128]>;
73
74/// Syslog identifier tagging every entry.
75type Identifier = SmallString<[u8; 16]>;
76
77/// Formatted source line number.
78type CodeLine = ArrayString<10>;
79
80/// Journal name of a user field, including the prefix.
81type FieldName = SmallString<[u8; 32]>;
82
83/// Value of a user field recorded through its `Debug` implementation.
84type FieldValue = SmallString<[u8; 64]>;
85
86const SOCKET: &str = "/run/systemd/journal/socket";
87
88const IDENTIFIER: &str = "tuwunel";
89
90/// Keeps a user field from colliding with one of the journal's own.
91const PREFIX: &str = "F_";
92
93/// Journald rejects a longer field name.
94const NAME_MAX: usize = 64;
95
96/// Bounds the datagram well under the default socket send buffer.
97const PAYLOAD_MAX: usize = 128 * 1024;
98
99const LEN_PREFIX: usize = size_of::<u64>();
100
101/// Datagram connection to the journald submission socket.
102pub struct Journal {
103	socket: UnixDatagram,
104	identifier: Identifier,
105}
106
107/// One entry, submitted when the writer is dropped. Its payload accumulates
108/// in the thread's buffer, which the fields layer opened for this event.
109pub struct Entry<'a> {
110	journal: &'a Journal,
111	message: usize,
112}
113
114/// Records the fields of each event and span for the entry the wrapped
115/// console layer is about to write, leaving them queryable with
116/// `journalctl F_NAME=value`.
117pub struct Fields<L> {
118	inner: L,
119	submit: bool,
120}
121
122/// Encoded fields of one span, held in its registry extensions.
123struct SpanFields(Buffer);
124
125struct Visitor<'a> {
126	fields: &'a mut Buffer,
127}
128
129thread_local! {
130	/// Entry under construction on this thread, reused across events.
131	static PAYLOAD: RefCell<Buffer> = const { RefCell::new(Buffer::new_const()) };
132}
133
134impl Journal {
135	/// Opens the socket when journald is configured, keeping the console when
136	/// nothing is listening on it.
137	#[must_use]
138	pub fn open(config: &Config) -> Option<Self> {
139		enabled(config)
140			.then(Self::new)?
141			.inspect_err(|e| {
142				writeln!(stderr(), "{e}").ok();
143			})
144			.ok()
145	}
146
147	fn new() -> Result<Self> {
148		let journal = Self {
149			socket: UnixDatagram::unbound()?,
150			identifier: identifier(),
151		};
152
153		journal
154			.send(&[])
155			.map_err(|e| err!(Config("log_journald", "{SOCKET}: {e}.")))?;
156
157		Ok(journal)
158	}
159}
160
161/// Tags entries with the running executable's name.
162fn identifier() -> Identifier {
163	args_os()
164		.next()
165		.as_deref()
166		.map(Path::new)
167		.and_then(Path::file_name)
168		.and_then(OsStr::to_str)
169		.map_or_else(|| IDENTIFIER.into(), Into::into)
170}
171
172/// Opens an entry at the event's severity, carrying its source metadata after
173/// the fields recorded for it, and leaves the message field open.
174#[implement(Journal)]
175#[must_use]
176pub fn entry(&self, meta: &Metadata<'_>) -> Entry<'_> {
177	let message = PAYLOAD.with_borrow_mut(|payload| {
178		put(payload, "PRIORITY", &[priority(*meta.level())]);
179		put(payload, "SYSLOG_IDENTIFIER", self.identifier.as_bytes());
180		put(payload, "TARGET", meta.target().as_bytes());
181
182		if let Some(file) = meta.file() {
183			put(payload, "CODE_FILE", file.as_bytes());
184		}
185
186		if let Some(line) = meta.line() {
187			let line: CodeLine = to_array_string(line);
188
189			put(payload, "CODE_LINE", line.as_bytes());
190		}
191
192		payload.extend_from_slice(b"MESSAGE\n");
193		payload.extend_from_slice(&[0; LEN_PREFIX]);
194
195		payload.len()
196	});
197
198	Entry { journal: self, message }
199}
200
201#[implement(Journal)]
202fn send(&self, payload: &[u8]) -> io::Result<usize> { self.socket.send_to(payload, SOCKET) }
203
204impl Write for Entry<'_> {
205	fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
206		PAYLOAD.with_borrow_mut(|payload| payload.extend_from_slice(buf));
207
208		Ok(buf.len())
209	}
210
211	fn flush(&mut self) -> io::Result<()> { Ok(()) }
212}
213
214impl Drop for Entry<'_> {
215	fn drop(&mut self) {
216		PAYLOAD.with_borrow_mut(|payload| {
217			close(payload, self.message);
218
219			if self.journal.send(payload).is_err() {
220				fallback(&payload[self.message..]);
221			}
222
223			// A span event reaches the writer without passing the fields layer,
224			// so the buffer is released here rather than after recording.
225			payload.clear();
226		});
227	}
228}
229
230/// Writes a message the socket would not take to the console, which journald
231/// captures in turn under systemd.
232fn fallback(message: &[u8]) { stderr().write_all(message).ok(); }
233
234/// Seals the message field, trimming the formatter's trailing whitespace and
235/// truncating on a character boundary what the datagram cannot carry.
236#[expect(
237	clippy::little_endian_bytes,
238	reason = "the journal protocol specifies little-endian field lengths"
239)]
240fn close(payload: &mut Buffer, message: usize) {
241	let budget = PAYLOAD_MAX.saturating_sub(message);
242	let len = payload[message..].trim_ascii_end().len();
243	let len = (len > budget)
244		.then(|| boundary(&payload[message..], budget))
245		.unwrap_or(len);
246
247	let size: u64 = len.expect_into();
248
249	payload.truncate(message.expected_add(len));
250	payload[message.expected_sub(LEN_PREFIX)..message].copy_from_slice(&size.to_le_bytes());
251	payload.push(b'\n');
252}
253
254/// Steps back to the nearest character boundary at or below `len`; a
255/// continuation byte never begins a character.
256fn boundary(message: &[u8], len: usize) -> usize {
257	(0..=len)
258		.rev()
259		.find(|&i| {
260			message
261				.get(i)
262				.is_none_or(|byte| byte & 0b1100_0000 != 0b1000_0000)
263		})
264		.unwrap_or_default()
265}
266
267impl<L> Fields<L> {
268	/// Wraps a subscriber layer with optional native journal field recording.
269	///
270	/// Events always continue through the inner layer. Structured field
271	/// submission is enabled only when the supplied configuration selects the
272	/// journal.
273	pub fn new(inner: L, config: &Config) -> Self { Self { inner, submit: enabled(config) } }
274}
275
276/// Whether events are submitted to journald rather than written to the
277/// console, which journald otherwise captures at a single fixed priority.
278pub(super) fn enabled(config: &Config) -> bool { config.log_journald && is_systemd_mode() }
279
280impl<S, L> Layer<S> for Fields<L>
281where
282	S: Subscriber + for<'a> LookupSpan<'a>,
283	L: Layer<S>,
284{
285	#[inline]
286	fn on_layer(&mut self, subscriber: &mut S) { self.inner.on_layer(subscriber); }
287
288	#[inline]
289	fn register_callsite(&self, meta: &'static Metadata<'static>) -> Interest {
290		self.inner.register_callsite(meta)
291	}
292
293	#[inline]
294	fn enabled(&self, meta: &Metadata<'_>, ctx: Context<'_, S>) -> bool {
295		self.inner.enabled(meta, ctx)
296	}
297
298	#[inline]
299	fn max_level_hint(&self) -> Option<LevelFilter> { self.inner.max_level_hint() }
300
301	fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, ctx: Context<'_, S>) {
302		if self.submit {
303			record_span(attrs, id, &ctx);
304		}
305
306		self.inner.on_new_span(attrs, id, ctx);
307	}
308
309	fn on_record(&self, id: &Id, values: &Record<'_>, ctx: Context<'_, S>) {
310		if self.submit {
311			record_values(id, values, &ctx);
312		}
313
314		self.inner.on_record(id, values, ctx);
315	}
316
317	#[inline]
318	fn on_follows_from(&self, id: &Id, follows: &Id, ctx: Context<'_, S>) {
319		self.inner.on_follows_from(id, follows, ctx);
320	}
321
322	#[inline]
323	fn event_enabled(&self, event: &Event<'_>, ctx: Context<'_, S>) -> bool {
324		self.inner.event_enabled(event, ctx)
325	}
326
327	fn on_event(&self, event: &Event<'_>, ctx: Context<'_, S>) {
328		if self.submit {
329			record_event(event, &ctx);
330		}
331
332		self.inner.on_event(event, ctx);
333	}
334
335	#[inline]
336	fn on_enter(&self, id: &Id, ctx: Context<'_, S>) { self.inner.on_enter(id, ctx); }
337
338	#[inline]
339	fn on_exit(&self, id: &Id, ctx: Context<'_, S>) { self.inner.on_exit(id, ctx); }
340
341	#[inline]
342	fn on_close(&self, id: Id, ctx: Context<'_, S>) { self.inner.on_close(id, ctx); }
343
344	#[inline]
345	fn on_id_change(&self, old: &Id, new: &Id, ctx: Context<'_, S>) {
346		self.inner.on_id_change(old, new, ctx);
347	}
348}
349
350/// Encodes a span's fields once, for every event it later encloses.
351fn record_span<S>(attrs: &Attributes<'_>, id: &Id, ctx: &Context<'_, S>)
352where
353	S: Subscriber + for<'a> LookupSpan<'a>,
354{
355	let Some(span) = ctx.span(id) else {
356		return;
357	};
358
359	let mut fields = Buffer::new();
360
361	put(&mut fields, "SPAN_NAME", span.name().as_bytes());
362	attrs.record(&mut Visitor { fields: &mut fields });
363
364	span.extensions_mut().insert(SpanFields(fields));
365}
366
367/// Appends the fields of a span recorded after its creation.
368fn record_values<S>(id: &Id, values: &Record<'_>, ctx: &Context<'_, S>)
369where
370	S: Subscriber + for<'a> LookupSpan<'a>,
371{
372	let Some(span) = ctx.span(id) else {
373		return;
374	};
375
376	let mut extensions = span.extensions_mut();
377	let Some(SpanFields(fields)) = extensions.get_mut::<SpanFields>() else {
378		return;
379	};
380
381	values.record(&mut Visitor { fields });
382}
383
384/// Opens this thread's buffer with the event's fields and those of the spans
385/// enclosing it, for the entry the console writer appends to next.
386fn record_event<S>(event: &Event<'_>, ctx: &Context<'_, S>)
387where
388	S: Subscriber + for<'a> LookupSpan<'a>,
389{
390	PAYLOAD.with_borrow_mut(|payload| {
391		payload.clear();
392
393		extend_scope(payload, event, ctx);
394		event.record(&mut Visitor { fields: payload });
395	});
396}
397
398/// Appends the fields of every span enclosing the event, innermost first.
399fn extend_scope<S>(payload: &mut Buffer, event: &Event<'_>, ctx: &Context<'_, S>)
400where
401	S: Subscriber + for<'a> LookupSpan<'a>,
402{
403	ctx.event_scope(event)
404		.into_iter()
405		.flatten()
406		.for_each(|span| {
407			let extensions = span.extensions();
408
409			if let Some(SpanFields(fields)) = extensions.get::<SpanFields>() {
410				payload.extend_from_slice(fields);
411			}
412		});
413}
414
415impl Visit for Visitor<'_> {
416	fn record_str(&mut self, field: &Field, value: &str) { self.record(field, value.as_bytes()); }
417
418	fn record_debug(&mut self, field: &Field, value: &dyn Debug) {
419		let value: FieldValue = format_small_string!("{value:?}");
420
421		self.record(field, value.as_str().unquote_infallible().as_bytes());
422	}
423}
424
425/// The message is the entry's own `MESSAGE`, and an underscored name marks a
426/// field the console formatter also hides.
427#[implement(Visitor, params = "<'_>")]
428fn record(&mut self, field: &Field, value: &[u8]) {
429	let name = field.name();
430	let skip = name == "message" || name.starts_with('_') || self.fields.len() >= PAYLOAD_MAX;
431
432	if skip {
433		return;
434	}
435
436	put(self.fields, &sanitize(name), value);
437}
438
439/// Journald accepts only uppercase alphanumerics and underscores in a field
440/// name, and drops a field whose name it rejects.
441fn sanitize(name: &str) -> FieldName {
442	PREFIX
443		.chars()
444		.chain(
445			name.chars()
446				.map(|c| if matches!(c, '.' | '-') { '_' } else { c })
447				.filter(|c| *c == '_' || c.is_ascii_alphanumeric())
448				.map(|c| c.to_ascii_uppercase()),
449		)
450		.take(NAME_MAX)
451		.collect()
452}
453
454/// Appends a length-encoded field, which may carry any byte sequence.
455#[expect(
456	clippy::little_endian_bytes,
457	reason = "the journal protocol specifies little-endian field lengths"
458)]
459fn put(payload: &mut Buffer, name: &str, value: &[u8]) {
460	let len: u64 = value.len().expect_into();
461
462	payload.extend_from_slice(name.as_bytes());
463	payload.push(b'\n');
464	payload.extend_from_slice(&len.to_le_bytes());
465	payload.extend_from_slice(value);
466	payload.push(b'\n');
467}
468
469/// Maps a level onto the syslog severity code journald expects.
470const fn priority(level: Level) -> u8 {
471	match level {
472		| Level::ERROR => b'3',
473		| Level::WARN => b'4',
474		| Level::INFO => b'5',
475		| Level::DEBUG => b'6',
476		| Level::TRACE => b'7',
477	}
478}
tuwunel_core/log/journald.rs

tuwunel_core/log/
journald.rs
