conmonrs/
journal.rs

1//! A client for the native journal protocol
2//! (<https://systemd.io/JOURNAL_NATIVE_PROTOCOL/>) and a `tracing` writer
3//! which forwards the server logs to journald.
4
5use nix::{
6    errno::Errno,
7    fcntl::{FcntlArg, SealFlag, fcntl},
8    sys::{
9        memfd::{MFdFlags, memfd_create},
10        prctl,
11        socket::{ControlMessage, MsgFlags, UnixAddr, sendmsg, setsockopt, sockopt},
12    },
13    unistd::{Pid, getpid, gettid},
14};
15use std::{
16    fmt::Display,
17    fs::File,
18    io::{self, ErrorKind, Write},
19    os::{
20        fd::{AsRawFd, OwnedFd},
21        unix::net::UnixDatagram,
22    },
23    panic::{self, PanicHookInfo},
24    path::Path,
25    sync::OnceLock,
26    time::{Duration, Instant},
27};
28use tracing::{Level, Metadata};
29use tracing_subscriber::fmt::writer::MakeWriter;
30
31/// The datagram socket journald receives native protocol entries on.
32pub(crate) const SOCKET_PATH: &str = "/run/systemd/journal/socket";
33
34/// The SYSLOG_IDENTIFIER of the entries of the server, so that
35/// `journalctl -t conmonrs` finds them, and of containers with neither a
36/// name nor an ID.
37pub(crate) const IDENTIFIER: &str = "conmonrs";
38
39/// The syslog priorities used for the `PRIORITY` field.
40#[derive(Clone, Copy, Debug, Eq, PartialEq)]
41pub(crate) enum Priority {
42    Error = 3,
43    Warning = 4,
44    Notice = 5,
45    Info = 6,
46    Debug = 7,
47}
48
49/// A journal entry, encoded for the native protocol.
50#[derive(Debug, Default)]
51pub(crate) struct Entry(Vec<u8>);
52
53impl Entry {
54    /// An entry with the `PRIORITY` and `MESSAGE` fields.
55    pub(crate) fn new(priority: Priority, message: impl AsRef<[u8]>) -> Self {
56        let mut entry = Self::default();
57        entry.reset(priority, message);
58        entry
59    }
60
61    /// Start over with the `PRIORITY` and `MESSAGE` fields, keeping the
62    /// allocation.
63    pub(crate) fn reset(&mut self, priority: Priority, message: impl AsRef<[u8]>) -> &mut Self {
64        self.0.clear();
65        self.field("PRIORITY", [b'0' + priority as u8])
66            .field("MESSAGE", message)
67    }
68
69    /// Append a field. The key must be a valid journal field name: upper case
70    /// letters, digits and underscores, not starting with a digit or an
71    /// underscore.
72    pub(crate) fn field(&mut self, key: &str, value: impl AsRef<[u8]>) -> &mut Self {
73        debug_assert!(Self::valid_key(key), "invalid journal field name {key}");
74        let value = value.as_ref();
75        self.0.extend_from_slice(key.as_bytes());
76        // `KEY=value\n` cannot carry a newline in the value, those values get
77        // their length as 64 bit little endian integer instead.
78        if value.contains(&b'\n') {
79            self.0.push(b'\n');
80            self.0
81                .extend_from_slice(&u64::try_from(value.len()).unwrap_or(u64::MAX).to_le_bytes());
82        } else {
83            self.0.push(b'=');
84        }
85        self.0.extend_from_slice(value);
86        self.0.push(b'\n');
87        self
88    }
89
90    /// Whether the key is a valid journal field name, see [`Entry::field`],
91    /// at most 64 characters.
92    fn valid_key(key: &str) -> bool {
93        (1..=64).contains(&key.len())
94            && !key.starts_with(|c: char| c == '_' || c.is_ascii_digit())
95            && key
96                .chars()
97                .all(|c| c.is_ascii_uppercase() || c.is_ascii_digit() || c == '_')
98    }
99
100    /// Append the already encoded fields of another entry.
101    pub(crate) fn extend(&mut self, fields: &Entry) -> &mut Self {
102        self.0.extend_from_slice(&fields.0);
103        self
104    }
105
106    /// The encoded fields, as they get sent.
107    pub(crate) fn as_bytes(&self) -> &[u8] {
108        &self.0
109    }
110}
111
112/// How sending waits for a journal which does not keep up, for example while
113/// it restarts or its disk is slow. The modes use separate sockets, and the
114/// server logs one of their own, see [`SERVER_SOCKET`].
115#[derive(Clone, Copy, Debug, Eq, PartialEq)]
116pub(crate) enum Wait {
117    /// Never block, the entry fails with `ErrorKind::WouldBlock`. For the
118    /// server logs, which tracing writes from the runtime threads, and the
119    /// container logs as long as the journal takes them right away.
120    Never,
121    /// Each attempt blocks for at most `SEND_TIMEOUT`, then fails with
122    /// `ErrorKind::TimedOut`, one which a signal interrupted gets retried
123    /// until the instant. For the container logs which the journal did not
124    /// take right away, which get sent from the blocking pool then and
125    /// should not be dropped on a short stall.
126    Bounded(Instant),
127}
128
129/// The longest an attempt of `Wait::Bounded` blocks for one entry.
130const SEND_TIMEOUT: Duration = Duration::from_secs(1);
131
132/// The send buffer systemd uses for its journal clients, so that bursts and
133/// entries up to that size do not need a memfd.
134const SEND_BUFFER_SIZE: usize = 8 * 1024 * 1024;
135
136/// The sockets of the container output, by mode.
137static NEVER_SOCKET: OnceLock<UnixDatagram> = OnceLock::new();
138static BOUNDED_SOCKET: OnceLock<UnixDatagram> = OnceLock::new();
139
140/// The socket of the server logs. Entries take up the send buffer of the
141/// socket they got sent on until the journal received them, so container
142/// output which the journal does not take fast enough must not fill the one
143/// of the server logs, they would be dropped otherwise.
144static SERVER_SOCKET: OnceLock<UnixDatagram> = OnceLock::new();
145
146/// Send an entry of the container output to the journal socket at a path.
147pub(crate) fn send_at(wait: Wait, path: &Path, entry: &Entry) -> io::Result<()> {
148    let cell = match wait {
149        Wait::Never => &NEVER_SOCKET,
150        Wait::Bounded(_) => &BOUNDED_SOCKET,
151    };
152    send_on(socket(cell, wait)?, wait, path, entry)
153}
154
155/// Send an entry on a socket of the mode, see [`new_socket`].
156fn send_on(socket: &UnixDatagram, wait: Wait, path: &Path, entry: &Entry) -> io::Result<()> {
157    let retry_until = match wait {
158        Wait::Never => None,
159        Wait::Bounded(until) => Some(until),
160    };
161    match send_to(socket, path, entry.as_bytes(), retry_until) {
162        Err(e) if retry_until.is_some() && e.kind() == ErrorKind::WouldBlock => {
163            Err(io::Error::new(
164                ErrorKind::TimedOut,
165                format!("journal did not take the entry within an attempt of {SEND_TIMEOUT:?}"),
166            ))
167        }
168        result => result,
169    }
170}
171
172/// The socket kept in `cell`, created for the mode on first use. A failed
173/// creation is not stored, the next entry tries again.
174fn socket(cell: &'static OnceLock<UnixDatagram>, wait: Wait) -> io::Result<&'static UnixDatagram> {
175    if let Some(socket) = cell.get() {
176        return Ok(socket);
177    }
178    let socket = new_socket(wait)?;
179    Ok(cell.get_or_init(|| socket))
180}
181
182/// Create a socket which sends the way the mode tells.
183fn new_socket(wait: Wait) -> io::Result<UnixDatagram> {
184    let socket = UnixDatagram::unbound()?;
185    match wait {
186        Wait::Never => socket.set_nonblocking(true)?,
187        Wait::Bounded(_) => socket.set_write_timeout(Some(SEND_TIMEOUT))?,
188    }
189    // Best effort like systemd: forcing needs CAP_NET_ADMIN, without it the
190    // kernel caps the size at net.core.wmem_max.
191    if setsockopt(&socket, sockopt::SndBufForce, &SEND_BUFFER_SIZE).is_err() {
192        setsockopt(&socket, sockopt::SndBuf, &SEND_BUFFER_SIZE).ok();
193    }
194    Ok(socket)
195}
196
197/// Send the data as one datagram, or as a memfd if it does not fit.
198/// A signal, like the SIGCHLD of every exited child, interrupts a send which
199/// waits, the kernel does not restart one with a timeout, so it gets retried
200/// until `retry_until`, if any.
201fn send_to(
202    socket: &UnixDatagram,
203    path: &Path,
204    data: &[u8],
205    retry_until: Option<Instant>,
206) -> io::Result<()> {
207    let retry = || retry_until.is_none_or(|until| Instant::now() < until);
208    loop {
209        match socket.send_to(data, path) {
210            Ok(_) => return Ok(()),
211            Err(e) if e.kind() == ErrorKind::Interrupted && retry() => {}
212            Err(e)
213                if matches!(
214                    e.raw_os_error().map(Errno::from_raw),
215                    Some(Errno::EMSGSIZE | Errno::ENOBUFS)
216                ) =>
217            {
218                return send_memfd(socket, path, data, retry);
219            }
220            Err(e) => return Err(e),
221        }
222    }
223}
224
225/// Whether an error means that there is no journal to send to: no socket
226/// file, or no one listening on it.
227pub(crate) fn unavailable(error: &io::Error) -> bool {
228    matches!(
229        error.kind(),
230        ErrorKind::NotFound | ErrorKind::ConnectionRefused
231    )
232}
233
234/// Send the data in a sealed memfd, the only kind journald accepts, as
235/// `SCM_RIGHTS` of an otherwise empty datagram.
236fn send_memfd(
237    socket: &UnixDatagram,
238    path: &Path,
239    data: &[u8],
240    retry: impl Fn() -> bool,
241) -> io::Result<()> {
242    let mut file = File::from(create_memfd()?);
243    file.write_all(data)?;
244    fcntl(
245        &file,
246        FcntlArg::F_ADD_SEALS(
247            SealFlag::F_SEAL_SHRINK
248                | SealFlag::F_SEAL_GROW
249                | SealFlag::F_SEAL_WRITE
250                | SealFlag::F_SEAL_SEAL,
251        ),
252    )?;
253    let addr = UnixAddr::new(path)?;
254    loop {
255        match sendmsg(
256            socket.as_raw_fd(),
257            &[],
258            &[ControlMessage::ScmRights(&[file.as_raw_fd()])],
259            MsgFlags::empty(),
260            Some(&addr),
261        ) {
262            Err(Errno::EINTR) if retry() => {}
263            res => return res.map(drop).map_err(Into::into),
264        }
265    }
266}
267
268/// Create a memfd which allows sealing. Like systemd, ask for a
269/// non-executable one first, kernels before 6.3 reject that flag.
270fn create_memfd() -> nix::Result<OwnedFd> {
271    let flags = MFdFlags::MFD_CLOEXEC | MFdFlags::MFD_ALLOW_SEALING;
272    let noexec = MFdFlags::from_bits_retain(libc::MFD_NOEXEC_SEAL);
273    match memfd_create(c"conmonrs-journal", flags | noexec) {
274        Err(Errno::EINVAL) => memfd_create(c"conmonrs-journal", flags),
275        result => result,
276    }
277}
278
279/// Writes every log line as one journal entry with the priority derived from
280/// the tracing level.
281#[derive(Default)]
282pub(crate) struct Journal {
283    priority: Option<Priority>,
284}
285
286impl Journal {
287    const DEFAULT_PRIORITY: Priority = Priority::Notice;
288
289    /// A writer with the priority of errors, for the errors of the server
290    /// from before its logging got set up.
291    pub(crate) fn error() -> Self {
292        Self {
293            priority: Some(Priority::Error),
294        }
295    }
296
297    /// The entry of a formatted event. The journal separates entries itself,
298    /// so the line ending of the event is dropped.
299    fn entry(&self, line: &[u8]) -> Entry {
300        let mut entry = Entry::new(
301            self.priority.unwrap_or(Self::DEFAULT_PRIORITY),
302            line.strip_suffix(b"\n").unwrap_or(line),
303        );
304        entry.field("SYSLOG_IDENTIFIER", IDENTIFIER);
305        entry
306    }
307
308    fn priority_for(level: Level) -> Priority {
309        match level {
310            Level::ERROR => Priority::Error,
311            Level::WARN => Priority::Warning,
312            Level::INFO => Priority::Info,
313            Level::DEBUG | Level::TRACE => Priority::Debug,
314        }
315    }
316}
317
318impl Journal {
319    /// Send without blocking the runtime thread of the event. A system
320    /// without journald is not an error, like with `sd_journal_send`, and a
321    /// journal which does not keep up loses the line, otherwise every event
322    /// would report an error.
323    fn send_at(&self, path: &Path, line: &[u8]) -> io::Result<()> {
324        let socket = socket(&SERVER_SOCKET, Wait::Never)?;
325        match send_on(socket, Wait::Never, path, &self.entry(line)) {
326            Err(e) if unavailable(&e) || e.kind() == ErrorKind::WouldBlock => Ok(()),
327            result => result,
328        }
329    }
330}
331
332/// Send the message of every panic to the journal before the default hook
333/// prints it. The release builds abort on a panic, and the forked server
334/// has `/dev/null` as its stderr unless it logs to stdout, so the message
335/// would get lost. It bypasses tracing, as its subscriber may be what
336/// panicked.
337pub(crate) fn log_panics() {
338    let default_hook = panic::take_hook();
339    panic::set_hook(Box::new(move |info: &PanicHookInfo<'_>| {
340        // Nothing to report it to if this fails, the default hook follows.
341        let _ = log_panic_at(
342            Path::new(SOCKET_PATH),
343            thread_name().as_deref(),
344            gettid(),
345            info,
346        );
347        default_hook(info);
348    }));
349}
350
351/// The name of the calling thread, without `thread::current`, which panics
352/// once the local data of the thread is gone, like in a destructor of it
353/// while the thread exits. A panic in the hook aborts without any message.
354/// The kernel keeps the names of the named threads, at most 15 bytes of
355/// them, the others have the one of the thread which started them. The
356/// main thread has the name of the process there, it gets `main` like
357/// from the default hook.
358fn thread_name() -> Option<String> {
359    if gettid() == getpid() {
360        return Some("main".into());
361    }
362    let name = prctl::get_name().ok()?;
363    Some(name.to_string_lossy().into_owned())
364}
365
366/// Send the message of a panic to the journal socket at a path, with the
367/// name and the ID of its thread like the default hook, and the location
368/// the `Display` of the `PanicHookInfo` includes. The process aborts right
369/// after, so the message waits for a journal which does not take it right
370/// away, on a socket of its own, as the send buffer of the server logs may
371/// be full. Without one, like at the limit of open files, which may be
372/// what panicked, it goes to the socket of the server logs without
373/// waiting.
374fn log_panic_at(path: &Path, thread: Option<&str>, id: Pid, info: &dyn Display) -> io::Result<()> {
375    let message = format!("thread '{}' ({id}) {info}", thread.unwrap_or("<unnamed>"));
376    let wait = Wait::Bounded(Instant::now() + SEND_TIMEOUT);
377    match new_socket(wait) {
378        Ok(socket) => send_on(
379            &socket,
380            wait,
381            path,
382            &Journal::error().entry(message.as_bytes()),
383        ),
384        Err(_) => Journal::error().send_at(path, message.as_bytes()),
385    }
386}
387
388impl Write for Journal {
389    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
390        self.send_at(Path::new(SOCKET_PATH), buf)?;
391        Ok(buf.len())
392    }
393
394    fn flush(&mut self) -> io::Result<()> {
395        Ok(())
396    }
397}
398
399impl<'a> MakeWriter<'a> for Journal {
400    type Writer = Journal;
401
402    fn make_writer(&'a self) -> Self::Writer {
403        Journal::default()
404    }
405
406    fn make_writer_for(&'a self, meta: &Metadata<'_>) -> Self::Writer {
407        Journal {
408            priority: Some(Self::priority_for(*meta.level())),
409        }
410    }
411}
412
413/// The entry point of the fuzz target.
414#[cfg(any(test, feature = "fuzzing"))]
415pub(crate) mod fuzzing;
416
417/// What the tests and the fuzz target share.
418#[cfg(any(test, feature = "fuzzing"))]
419pub(crate) mod testing;
420
421#[cfg(test)]
422pub(crate) mod tests;
conmonrs/journal.rs

conmonrs/
journal.rs
