log: send records natively to journald, keep stdout off-unit

A record written to stdout carries no priority, so journald files the
whole stream at one level and the swarm log store shows `info` whatever
level `tracing` gave it. Under a systemd unit the process's stdout
already *is* the journal, so the fix is to speak the journal protocol
directly and let each record carry its own severity.

New `hive-log` crate holds the one sink chooser, called by `hive-c0re`,
`hive-agent` and `swarm-controller`. It builds the same `EnvFilter`
those binaries always built, then installs exactly one layer — never
both, since a journald layer stacked on the `fmt` layer under a unit
stores every record twice.

The choice is an fstat compare, not a presence test: a child inherits
`$JOURNAL_STREAM` even when its own stdout was redirected elsewhere, so
the variable existing proves nothing. The crate parses `dev:inode` out
of it and compares both numbers against an fstat of stdout, the
descriptor the `fmt` layer writes to by default. No match, unset, or
unparseable takes the `fmt` branch. A journald layer that fails to
construct despite a match falls back to `fmt` and warns through it —
a process must never fail to start because of its logger.

hive-log/Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+name = "hive-log"
+edition.workspace = true
+version.workspace = true
+readme = "README.md"
+
+[lints]
+workspace = true
+
+[dependencies]
+rustix = { workspace = true }
+tracing = { workspace = true }
+tracing-journald = { workspace = true }
+tracing-subscriber = { workspace = true }

hive-log/README.md

@@ -0,0 +1,49 @@
+# hive-log
+
+One shared log-sink chooser for the hyperhive daemons: `hive_log::init()`
+builds the `EnvFilter` every binary built before, then installs **exactly
+one** sink.
+
+## Why one sink and not both
+
+Under a systemd unit the process's stdout already **is** the journal.
+Adding a journald layer on top of the `fmt` layer therefore stores every
+record twice — once as the native journal entry, once as the text line
+`fmt` wrote to a descriptor that leads to the same journal.
+
+Writing to stdout also throws the severity away. A stdout line carries no
+priority field, so journald files the whole stream at one level and the
+swarm log store shows `info` whatever level `tracing` gave the record.
+Logging natively carries the level across. That is the reason to prefer
+journald where a journal exists, and the reason never to run both layers.
+
+## How the choice happens
+
+systemd sets `$JOURNAL_STREAM` to the device and inode numbers of the
+journal-connected descriptor, in decimal, separated by a colon.
+
+Checking that the variable exists is **not** enough. A child process
+inherits `$JOURNAL_STREAM` from its parent even when the parent redirected
+that child's stdout somewhere else, so presence alone reports a journal
+that the process does not actually write to. `hive-log` instead stats
+stdout — the descriptor the `fmt` layer writes to by default, and the one
+a journald layer would double up on — and compares both numbers against
+the parsed pair.
+
+A match installs the native journald layer alone. No match, an unset
+variable, or a value that does not parse installs the `fmt` layer alone.
+
+## When the journald layer fails to build
+
+`tracing_journald::layer()` validates the journal socket up front and
+returns an `io::Result`, so a missing journal surfaces as an error rather
+than a silent no-op. On that error `hive-log` installs the `fmt` layer and
+emits one warning through it. A process must never fail to start because
+of its logger, and a logger that drops every record in silence is worse
+than a degraded one that says so.
+
+## Scope
+
+The sink chooser and its tests. Nothing else belongs here: the crate
+exists so three binaries share one copy of this decision, not as a place
+for general utilities.

hive-log/src/lib.rs

@@ -0,0 +1,199 @@
+//! Log-sink selection for the hyperhive daemons: exactly one sink per
+//! process, chosen at startup.
+//!
+//! Under a systemd unit the process's stdout already *is* the journal, so
+//! installing a journald layer on top of the `fmt` layer stores every
+//! record twice. There the daemon logs natively instead, which is the
+//! only way the journal learns a record's severity: a line written to
+//! stdout carries no priority, so the log store files it all as `info`
+//! whatever level `tracing` gave it. Off a unit — by hand, or in a test —
+//! the ordinary `fmt` stdout layer keeps the output human-readable.
+//!
+//! systemd sets `JOURNAL_STREAM` to the device and inode numbers of the
+//! journal-connected descriptor, in decimal, separated by a colon.
+//! Testing that the variable merely exists does not work: a child process
+//! inherits it even when its own stdout points somewhere else. So this
+//! crate stats the descriptor the `fmt` layer would write to and compares
+//! both numbers against the parsed pair.
+
+use std::io;
+use std::os::fd::{AsFd, BorrowedFd};
+
+use tracing_subscriber::EnvFilter;
+use tracing_subscriber::layer::SubscriberExt as _;
+use tracing_subscriber::util::SubscriberInitExt as _;
+
+/// The environment variable systemd sets on a journal-connected stream.
+const JOURNAL_STREAM: &str = "JOURNAL_STREAM";
+
+/// Installs the process-wide subscriber with whichever single sink fits
+/// the context.
+///
+/// The filter is the one every caller built before this crate existed:
+/// `RUST_LOG` when it parses, `info` otherwise. The `fmt` layer turns ANSI
+/// escapes off, because neither sink is a human terminal and journald does
+/// not strip escapes, so they reach the log store as raw byte-array spam
+/// otherwise.
+pub fn init() {
+    let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
+
+    if !stdout_is_journal() {
+        init_fmt(filter);
+        return;
+    }
+
+    match tracing_journald::layer() {
+        Ok(layer) => tracing_subscriber::registry()
+            .with(filter)
+            .with(layer.with_priority_mappings(priority_mappings()))
+            .init(),
+        // A process must never fail to start because of its logger, and a
+        // logger that silently drops every record is worse than a degraded
+        // one. Take the stdout layer and say so, loudly, through it.
+        Err(err) => {
+            init_fmt(filter);
+            tracing::warn!(
+                error = %err,
+                "journald sink unavailable under a journal-connected stdout; \
+                 falling back to the stdout layer, so records reach the log \
+                 store without a severity"
+            );
+        }
+    }
+}
+
+/// Maps each `tracing` level onto the journal priority of the same name.
+///
+/// `tracing-journald` defaults to a one-notch shift — `INFO` to `Notice`,
+/// `DEBUG` to `Informational` — which would rename every level on the way
+/// to the log store. Naming the mappings keeps an `info!` call readable as
+/// `info` there, which is the whole point of sending the severity at all.
+fn priority_mappings() -> tracing_journald::PriorityMappings {
+    use tracing_journald::Priority;
+
+    tracing_journald::PriorityMappings {
+        error: Priority::Error,
+        warn: Priority::Warning,
+        info: Priority::Informational,
+        debug: Priority::Debug,
+        trace: Priority::Debug,
+    }
+}
+
+fn init_fmt(filter: EnvFilter) {
+    tracing_subscriber::registry()
+        .with(filter)
+        .with(tracing_subscriber::fmt::layer().with_ansi(false))
+        .init();
+}
+
+/// Reports whether stdout is the descriptor systemd named in
+/// `JOURNAL_STREAM`.
+///
+/// Stdout, not stderr: the `fmt` layer writes there by default and none of
+/// the callers change that, so stdout is the descriptor a second journald
+/// layer would double up on.
+fn stdout_is_journal() -> bool {
+    let raw = std::env::var(JOURNAL_STREAM).ok();
+    fd_is_journal_stream(raw.as_deref(), io::stdout().as_fd())
+}
+
+/// Parses `dev:inode` out of a `JOURNAL_STREAM` value.
+///
+/// Returns `None` when the value is absent, carries no colon, or holds a
+/// part that is not a decimal number.
+fn parse_journal_stream(raw: Option<&str>) -> Option<(u64, u64)> {
+    let (dev, inode) = raw?.split_once(':')?;
+    Some((dev.parse().ok()?, inode.parse().ok()?))
+}
+
+/// Compares a parsed `JOURNAL_STREAM` value against a live descriptor.
+///
+/// Returns `false` for anything `parse_journal_stream` rejects, for a
+/// descriptor that cannot be stated, and — the case presence alone misses
+/// — for a well-formed pair naming some *other* descriptor, which is what
+/// a child sees once its parent redirected its stdout.
+fn fd_is_journal_stream(raw: Option<&str>, fd: BorrowedFd<'_>) -> bool {
+    let Some((dev, inode)) = parse_journal_stream(raw) else {
+        return false;
+    };
+    let Ok(stat) = rustix::fs::fstat(fd) else {
+        return false;
+    };
+    stat.st_dev == dev && stat.st_ino == inode
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// The `dev:inode` pair systemd would export for this descriptor.
+    fn own_stream(fd: BorrowedFd<'_>) -> String {
+        let stat = rustix::fs::fstat(fd).expect("fstat of a live descriptor");
+        format!("{}:{}", stat.st_dev, stat.st_ino)
+    }
+
+    #[test]
+    fn unset_variable_is_not_a_journal_stream() {
+        let fd = io::stdout();
+        assert!(!fd_is_journal_stream(None, fd.as_fd()));
+    }
+
+    #[test]
+    fn malformed_values_are_not_a_journal_stream() {
+        let fd = io::stdout();
+        for raw in ["", "12", "8:", ":42", "eight:42", "8:forty-two", "8:42:7"] {
+            assert!(
+                !fd_is_journal_stream(Some(raw), fd.as_fd()),
+                "{raw:?} parsed as a journal stream"
+            );
+        }
+    }
+
+    #[test]
+    fn a_well_formed_pair_naming_another_fd_does_not_match() {
+        let fd = io::stdout();
+        let stat = rustix::fs::fstat(fd.as_fd()).expect("fstat of a live descriptor");
+        // What a child inherits after its parent redirected its stdout:
+        // the variable is set and parses, and names a descriptor that is
+        // no longer this one.
+        let other = format!("{}:{}", stat.st_dev, stat.st_ino.wrapping_add(1));
+        assert!(!fd_is_journal_stream(Some(&other), fd.as_fd()));
+
+        let other_dev = format!("{}:{}", stat.st_dev.wrapping_add(1), stat.st_ino);
+        assert!(!fd_is_journal_stream(Some(&other_dev), fd.as_fd()));
+    }
+
+    #[test]
+    fn a_pair_naming_this_fd_matches() {
+        let fd = io::stdout();
+        assert!(fd_is_journal_stream(
+            Some(&own_stream(fd.as_fd())),
+            fd.as_fd()
+        ));
+    }
+
+    /// The comparison reads the descriptor handed to it, not descriptor 1:
+    /// a pair naming one fd must not match a different one.
+    #[test]
+    fn the_compared_descriptor_is_the_one_passed_in() {
+        let out = io::stdout();
+        let err = io::stderr();
+        let out_stream = own_stream(out.as_fd());
+        let err_stream = own_stream(err.as_fd());
+        // Under a harness that gives both streams the same pipe there is
+        // nothing to tell apart, so this asserts only when they differ.
+        if out_stream != err_stream {
+            assert!(!fd_is_journal_stream(Some(&err_stream), out.as_fd()));
+            assert!(fd_is_journal_stream(Some(&err_stream), err.as_fd()));
+        }
+    }
+
+    #[test]
+    fn parsing_keeps_both_numbers_in_order() {
+        assert_eq!(parse_journal_stream(Some("8:4242")), Some((8, 4242)));
+        assert_eq!(parse_journal_stream(Some("0:0")), Some((0, 0)));
+        assert_eq!(parse_journal_stream(None), None);
+        assert_eq!(parse_journal_stream(Some("no-colon")), None);
+    }
+}