From 8cc7f90c98cd6bd15e5e79044bbe96ee8c15f4c1 Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 21 Sep 2026 00:20:33 +0200 Subject: [PATCH] log: send records natively to journald, keep stdout off-unit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- CLAUDE.md | 8 ++ Cargo.lock | 27 ++++- Cargo.toml | 16 +++ hive-agent/Cargo.toml | 2 +- hive-agent/src/main.rs | 11 +- hive-c0re/Cargo.toml | 2 +- hive-c0re/src/main.rs | 11 +- hive-log/Cargo.toml | 14 +++ hive-log/README.md | 49 +++++++++ hive-log/src/lib.rs | 199 +++++++++++++++++++++++++++++++++++ swarm-controller/Cargo.toml | 2 +- swarm-controller/src/main.rs | 11 +- 12 files changed, 316 insertions(+), 36 deletions(-) create mode 100644 hive-log/Cargo.toml create mode 100644 hive-log/README.md create mode 100644 hive-log/src/lib.rs diff --git a/CLAUDE.md b/CLAUDE.md index c900c2df..5ce16029 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -72,6 +72,14 @@ hand-maintained per-file tree drifts out of sync with the code. remix. `GraphWire::wire_snapshot` is blanket-implemented for any `Graph` whose parameters implement both — so a payload that has never said how it displays cannot reach a viewer at all. +- **`hive-log/`** — the one log-sink chooser, called by `hive-c0re`, + `hive-agent` and `swarm-controller` and by nothing else. `init()` builds + the `EnvFilter` those binaries always built, then installs **exactly + one** sink: the native journald layer when an `fstat` of stdout matches + the device and inode numbers systemd exports in `$JOURNAL_STREAM`, the + `fmt` stdout layer otherwise. Under a unit stdout already **is** the + journal, so both layers together store every record twice. Deliberately + single-purpose — do not grow it into a utility crate. - **`hive-screen-mcp/`** — stdio MCP bridge for GUI agents (`hyperhive.gui.enable`): `screenshot` via `grim`, `type_text` / `key_press` via `wtype` (Wayland virtual-keyboard protocol), and diff --git a/Cargo.lock b/Cargo.lock index d0371900..e7bd31a4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1722,6 +1722,7 @@ dependencies = [ "hive-agent-sock", "hive-claude", "hive-core-agent-sock", + "hive-log", "hive-sh4re", "hive-sock-client", "http-body-util", @@ -1743,7 +1744,6 @@ dependencies = [ "tokio-stream", "tower-http 0.7.0", "tracing", - "tracing-subscriber", ] [[package]] @@ -1821,6 +1821,7 @@ dependencies = [ "hive-host-sock", "hive-jobq", "hive-jobq-wire", + "hive-log", "hive-priv-sock", "hive-sh4re", "hive-sock-client", @@ -1846,7 +1847,6 @@ dependencies = [ "tokio", "tokio-stream", "tracing", - "tracing-subscriber", "url", "utoipa", "utoipa-axum", @@ -1962,6 +1962,16 @@ dependencies = [ "utoipa", ] +[[package]] +name = "hive-log" +version = "0.1.0" +dependencies = [ + "rustix", + "tracing", + "tracing-journald", + "tracing-subscriber", +] + [[package]] name = "hive-matrix-mcp" version = "0.1.0" @@ -4886,6 +4896,7 @@ dependencies = [ "hive-jobq", "hive-jobq-metrics", "hive-jobq-wire", + "hive-log", "hive-types", "hmac 0.13.0", "http", @@ -4906,7 +4917,6 @@ dependencies = [ "time", "tokio", "tracing", - "tracing-subscriber", "url", "utoipa", "utoipa-axum", @@ -5465,6 +5475,17 @@ dependencies = [ "valuable", ] +[[package]] +name = "tracing-journald" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d3a81ed245bfb62592b1e2bc153e77656d94ee6a0497683a65a12ccaf2438d0" +dependencies = [ + "libc", + "tracing-core", + "tracing-subscriber", +] + [[package]] name = "tracing-log" version = "0.2.0" diff --git a/Cargo.toml b/Cargo.toml index a4baf59e..28c22610 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,6 +14,7 @@ members = [ "hive-jobq", "hive-jobq-metrics", "hive-jobq-wire", + "hive-log", "hive-matrix-mcp", "hive-metric", "hive-priv", @@ -86,6 +87,7 @@ hive-agent-sock = { path = "hive-agent-sock" } hive-jobq = { path = "hive-jobq" } hive-jobq-metrics = { path = "hive-jobq-metrics" } hive-jobq-wire = { path = "hive-jobq-wire" } +hive-log = { path = "hive-log" } hive-core-agent-sock = { path = "hive-core-agent-sock" } hive-claude = "0.1.1" hive-host-sock = { path = "hive-host-sock" } @@ -147,6 +149,20 @@ tokio = { version = "1", features = [ tokio-stream = { version = "0.1", features = ["sync"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } +# The journald sink `hive-log` installs when the process runs under a +# systemd unit. Tokio-maintained; its whole dependency set is `libc` + +# `tracing-core` + `tracing-subscriber`, since it speaks the native journal +# protocol over a `UnixDatagram` and links no C journal library. +tracing-journald = "0.3.2" +# `fs::fstat` — the safe wrapper `hive-log` compares a descriptor's device +# and inode numbers against `$JOURNAL_STREAM` with. Already in the lock as +# a transitive dependency; direct here so that comparison needs no +# hand-written `unsafe libc::fstat` call. Only the `fs` API module is asked +# for: rustix gates each one behind its own feature. +rustix = { version = "1.1.4", default-features = false, features = [ + "std", + "fs", +] } reqwest = { version = "0.13", default-features = false, features = [ # RFC 7662 introspection posts an urlencoded body; without this, # `.form()` does not exist and the alternative is percent-encoding a diff --git a/hive-agent/Cargo.toml b/hive-agent/Cargo.toml index 14478708..b928b7eb 100644 --- a/hive-agent/Cargo.toml +++ b/hive-agent/Cargo.toml @@ -23,6 +23,7 @@ clap.workspace = true hive-claude.workspace = true hive-agent-sock.workspace = true hive-core-agent-sock.workspace = true +hive-log.workspace = true hive-sh4re.workspace = true hive-sock-client.workspace = true libc.workspace = true @@ -42,7 +43,6 @@ tokio.workspace = true tokio-stream.workspace = true tower-http.workspace = true tracing.workspace = true -tracing-subscriber.workspace = true [dev-dependencies] tempfile = "3" diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index a1537602..3b1e2e8c 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -94,16 +94,7 @@ struct Cli { #[tokio::main] async fn main() -> Result<()> { - tracing_subscriber::fmt() - .with_env_filter( - tracing_subscriber::EnvFilter::try_from_default_env() - .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), - ) - // This is a systemd-managed daemon — stdout always goes to journald, - // never a human terminal, and journald doesn't strip ANSI escapes: - // they land in victorialogs as raw byte-array spam otherwise. - .with_ansi(false) - .init(); + hive_log::init(); let cli = Cli::parse(); serve_main::(&cli.socket, cli.poll_ms).await diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index b747e009..f8a37fe5 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -38,6 +38,7 @@ hive-sh4re.workspace = true hive-host-sock.workspace = true hive-jobq.workspace = true hive-jobq-wire.workspace = true +hive-log.workspace = true hive-priv-sock.workspace = true hive-agent-sock.workspace = true hive-sock-client.workspace = true @@ -59,7 +60,6 @@ swarm-secret-client.workspace = true tokio.workspace = true tokio-stream.workspace = true tracing.workspace = true -tracing-subscriber.workspace = true problem_details = { version = "0.9.0", features = ["axum"] } utoipa.workspace = true utoipa-axum.workspace = true diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 58749694..13c29b58 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -143,16 +143,7 @@ enum Cmd { #[tokio::main] async fn main() -> Result<()> { - tracing_subscriber::fmt() - .with_env_filter( - tracing_subscriber::EnvFilter::try_from_default_env() - .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), - ) - // This is a systemd-managed daemon — stdout always goes to journald, - // never a human terminal, and journald doesn't strip ANSI escapes: - // they land in victorialogs as raw byte-array spam otherwise. - .with_ansi(false) - .init(); + hive_log::init(); let cli = Cli::parse(); match cli.cmd { diff --git a/hive-log/Cargo.toml b/hive-log/Cargo.toml new file mode 100644 index 00000000..b3f06516 --- /dev/null +++ b/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 } diff --git a/hive-log/README.md b/hive-log/README.md new file mode 100644 index 00000000..cfb74026 --- /dev/null +++ b/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. diff --git a/hive-log/src/lib.rs b/hive-log/src/lib.rs new file mode 100644 index 00000000..031e760a --- /dev/null +++ b/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); + } +} diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index 62d2777e..c64eea31 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -38,6 +38,7 @@ problem_details = { version = "0.9.0", features = ["axum"] } # same shape `hive-c0re/src/job_queue/scheduler.rs` uses over its own graph. hive-jobq.workspace = true hive-jobq-wire.workspace = true +hive-log.workspace = true # The jobq-rollup OTEL exporter, wired up in `main` via # `hive_jobq_metrics::spawn_exporter` — moved to its own crate (rather than # living here as `jobq_metrics.rs`) specifically so a future second caller @@ -115,7 +116,6 @@ bytes.workspace = true http.workspace = true tokio.workspace = true tracing.workspace = true -tracing-subscriber.workspace = true url.workspace = true utoipa.workspace = true utoipa-axum.workspace = true diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 8f673991..69fe1cad 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -1669,16 +1669,7 @@ fn keep_forge_for_state( #[tokio::main] async fn main() -> Result<()> { - tracing_subscriber::fmt() - .with_env_filter( - tracing_subscriber::EnvFilter::try_from_default_env() - .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), - ) - // This is a systemd-managed daemon — stdout always goes to journald, - // never a human terminal, and journald doesn't strip ANSI escapes: - // they land in victorialogs as raw byte-array spam otherwise. - .with_ansi(false) - .init(); + hive_log::init(); let path = socket_path();