vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(#1116): stream nixos-container create/update output live into build_logs

Browse source
This commit is contained in:
damocles 2026-06-03 10:17:44 +02:00
commit 68451eb205
4 changed files with 335 additions and 46 deletions
Download patch file Download diff file Expand all files Collapse all files

98
hive-c0re/src/lifecycle.rs
View file

@ -1207,14 +1207,13 @@ async fn set_nspawn_flags(
}
/// Execute a container operation via hive-priv and integrate with
/// build_logs.sqlite. hive-priv runs as root and logs output to
/// journald as it arrives; this function captures the final stdout +
/// stderr into build_logs for the dashboard after the operation
/// completes. For `create` and `update` (the long-running ops)
/// hive-priv already logs each line to its own journald stream —
/// streaming into build_logs is deferred to a follow-up that adds a
/// streaming mode to the priv protocol.
/// build_logs.sqlite. hive-priv runs as root and forwards output lines
/// to hive-c0re in real time via the streaming priv protocol. Each line
/// is appended to the build-log row as it arrives, so the dashboard
/// shows live progress during long `nixos-container create` / `update` runs.
async fn priv_run(kind: &str, name: &str) -> Result<()> {
use hive_sh4re::priv_proto::PrivStream;
let container = container_name(name);
let cmdline = format!("nixos-container {kind} {container}");
@ -1227,35 +1226,78 @@ async fn priv_run(kind: &str, name: &str) -> Result<()> {
.ok()
});
let result: Result<(String, String)> = match kind {
"create" => crate::priv_client::create_container(name).await,
"update" => crate::priv_client::update_container(name).await,
"start" => crate::priv_client::start_container(name).await.map(|()| (String::new(), String::new())),
"stop" => crate::priv_client::stop_container(name).await.map(|()| (String::new(), String::new())),
"kill" => crate::priv_client::kill_container(name).await.map(|()| (String::new(), String::new())),
"destroy" => crate::priv_client::destroy_container(name).await.map(|()| (String::new(), String::new())),
// For long-running ops use the streaming protocol so build_logs
// receives lines in real time rather than as a batch at completion.
let result: Result<()> = match kind {
"create" => {
let h = logs.clone();
let id = log_id;
crate::priv_client::create_container_streaming(name, move |stream, line| {
match stream {
PrivStream::Stdout => {
tracing::info!(target: "nixos-container", cmdline = %cmdline, "{line}");
if let (Some(h), Some(id)) = (&h, id) {
h.append_stdout(id, line);
}
}
PrivStream::Stderr => {
tracing::warn!(target: "nixos-container", cmdline = %cmdline, "{line}");
if let (Some(h), Some(id)) = (&h, id) {
h.append_stderr(id, line);
}
}
}
})
.await
}
"update" => {
let h = logs.clone();
let id = log_id;
crate::priv_client::update_container_streaming(name, move |stream, line| {
match stream {
PrivStream::Stdout => {
tracing::info!(target: "nixos-container", cmdline = %cmdline, "{line}");
if let (Some(h), Some(id)) = (&h, id) {
h.append_stdout(id, line);
}
}
PrivStream::Stderr => {
tracing::warn!(target: "nixos-container", cmdline = %cmdline, "{line}");
if let (Some(h), Some(id)) = (&h, id) {
h.append_stderr(id, line);
}
}
}
})
.await
}
"start" => crate::priv_client::start_container(name).await,
"stop" => crate::priv_client::stop_container(name).await,
"kill" => crate::priv_client::kill_container(name).await,
"destroy" => crate::priv_client::destroy_container(name).await,
other => Err(anyhow::anyhow!("unknown container op: {other}")),
};
let ok = result.is_ok();
let succeeded = result.is_ok();
if let (Some(h), Some(id)) = (&logs, log_id) {
if let Ok((ref stdout, ref stderr)) = result {
for line in stdout.lines() {
tracing::info!(target: "nixos-container", cmdline = %cmdline, "{line}");
h.append_stdout(id, line);
}
for line in stderr.lines() {
tracing::warn!(target: "nixos-container", cmdline = %cmdline, "{line}");
h.append_stderr(id, line);
}
}
h.finish(id, if ok { crate::build_logs::BuildStatus::Ok } else { crate::build_logs::BuildStatus::Fail });
h.finish(
id,
if succeeded {
crate::build_logs::BuildStatus::Ok
} else {
crate::build_logs::BuildStatus::Fail
},
);
}
match result {
Ok(_) => Ok(()),
Ok(()) => Ok(()),
Err(e) => {
let journal = if kind == "update" { container_journal_tail(&container).await } else { String::new() };
let journal = if kind == "update" {
container_journal_tail(&container).await
} else {
String::new()
};
match log_id {
Some(id) => bail!("{e:#}; see build log #{id}{journal}"),
None => bail!("{e:#}{journal}"),

80
hive-c0re/src/priv_client.rs
View file

@ -8,12 +8,14 @@
use anyhow::{Context as _, Result, bail};
use hive_sh4re::priv_proto::{
BindMount, JournalOutput, NetworkIsolation, PRIV_SOCK, PrivRequest, PrivResponse,
BindMount, JournalOutput, NetworkIsolation, PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse,
PrivStream,
};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::UnixStream;
/// Send a single request to `hive-priv` and return the response.
/// For streaming ops use `call_streaming` instead.
pub async fn call(req: &PrivRequest) -> Result<PrivResponse> {
let mut stream = UnixStream::connect(PRIV_SOCK)
.await
@ -29,7 +31,46 @@ pub async fn call(req: &PrivRequest) -> Result<PrivResponse> {
.read_line(&mut resp_line)
.await
.context("read response from hive-priv")?;
serde_json::from_str(&resp_line).context("parse PrivResponse")
// New hive-priv sends `PrivEvent::Done(PrivResponse)` (untagged, wire-
// identical to bare `PrivResponse`) — deserialise as `PrivEvent` to
// handle both the old bare format and the new tagged format.
match serde_json::from_str::<PrivEvent>(&resp_line).context("parse PrivResponse")? {
PrivEvent::Done(resp) => Ok(resp),
PrivEvent::Line(_) => bail!("unexpected stream line from non-streaming priv op"),
}
}
/// Send a streaming request to `hive-priv`, calling `on_line` for each
/// `PrivEvent::Line` as it arrives, then returning the terminal
/// `PrivResponse`. Used for long-running ops (`create` / `update`).
pub async fn call_streaming(
req: &PrivRequest,
mut on_line: impl FnMut(PrivStream, &str),
) -> Result<PrivResponse> {
let mut stream = UnixStream::connect(PRIV_SOCK)
.await
.context("connect to hive-priv socket (streaming)")?;
let line = serde_json::to_string(req).context("serialise PrivRequest")? + "\n";
stream
.write_all(line.as_bytes())
.await
.context("send request to hive-priv")?;
stream.shutdown().await.context("shutdown write half")?;
let mut reader = BufReader::new(stream);
loop {
let mut event_line = String::new();
reader
.read_line(&mut event_line)
.await
.context("read event from hive-priv")?;
if event_line.is_empty() {
bail!("hive-priv closed connection before sending Done event");
}
match serde_json::from_str::<PrivEvent>(&event_line).context("parse PrivEvent")? {
PrivEvent::Line(l) => on_line(l.stream, &l.data),
PrivEvent::Done(resp) => return Ok(resp),
}
}
}
pub async fn start_container(name: &str) -> Result<()> {
@ -57,20 +98,55 @@ pub async fn update_container(name: &str) -> Result<(String, String)> {
check(
call(&PrivRequest::UpdateContainer {
name: name.to_owned(),
stream: false,
})
.await?,
)
}
/// Streaming variant: forward stdout/stderr lines to `on_line` as they
/// arrive. Returns `Ok(())` on success; the callback is responsible for
/// appending lines to build_logs or otherwise capturing the output.
pub async fn update_container_streaming(
name: &str,
on_line: impl FnMut(PrivStream, &str),
) -> Result<()> {
ok(call_streaming(
&PrivRequest::UpdateContainer {
name: name.to_owned(),
stream: true,
},
on_line,
)
.await?)
}
pub async fn create_container(name: &str) -> Result<(String, String)> {
check(
call(&PrivRequest::CreateContainer {
name: name.to_owned(),
stream: false,
})
.await?,
)
}
/// Streaming variant: forward stdout/stderr lines to `on_line` as they
/// arrive. Returns `Ok(())` on success.
pub async fn create_container_streaming(
name: &str,
on_line: impl FnMut(PrivStream, &str),
) -> Result<()> {
ok(call_streaming(
&PrivRequest::CreateContainer {
name: name.to_owned(),
stream: true,
},
on_line,
)
.await?)
}
pub async fn destroy_container(name: &str) -> Result<()> {
ok(call(&PrivRequest::DestroyContainer {
name: name.to_owned(),

146
hive-priv/src/main.rs
View file

@ -22,9 +22,10 @@ use std::path::{Path, PathBuf};
use anyhow::{Context as _, Result, bail};
use hive_sh4re::priv_proto::{
AGENT_PREFIX, BindMount, JournalOutput, MANAGER_NAME, META_DIR, NetworkIsolation, PRIV_SOCK,
PrivRequest, PrivResponse, SIBLING_CONTAINERS,
PrivEvent, PrivRequest, PrivResponse, PrivStream, PrivStreamLine, SIBLING_CONTAINERS,
};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::unix::OwnedWriteHalf;
use tokio::net::{UnixListener, UnixStream};
use tokio::process::Command;
@ -100,8 +101,12 @@ async fn handle(stream: UnixStream) {
let (reader, mut writer) = stream.into_split();
let mut lines = BufReader::new(reader).lines();
while let Ok(Some(line)) = lines.next_line().await {
let resp = dispatch(&line).await;
let mut json = serde_json::to_string(&resp).unwrap_or_else(|e| {
let resp = dispatch(&line, &mut writer).await;
// Write the terminal PrivResponse as a PrivEvent::Done. Wire-identical
// to a bare PrivResponse (untagged), so old hive-c0re callers that
// deserialise directly to PrivResponse continue to work.
let event = PrivEvent::Done(resp);
let mut json = serde_json::to_string(&event).unwrap_or_else(|e| {
format!("{{\"ok\":false,\"stdout\":\"\",\"stderr\":\"\",\"error\":\"serialise failed: {e}\"}}")
});
json.push('\n');
@ -112,9 +117,9 @@ async fn handle(stream: UnixStream) {
}
}
async fn dispatch(line: &str) -> PrivResponse {
async fn dispatch(line: &str, writer: &mut OwnedWriteHalf) -> PrivResponse {
match serde_json::from_str::<PrivRequest>(line) {
Ok(req) => match exec(req).await {
Ok(req) => match exec(req, writer).await {
Ok((stdout, stderr)) => PrivResponse {
ok: true,
stdout,
@ -137,8 +142,26 @@ async fn dispatch(line: &str) -> PrivResponse {
}
}
/// Write one `PrivEvent::Line` to the client. Best-effort: a write
/// failure is logged but doesn't abort the running subprocess.
async fn write_line_event(writer: &mut OwnedWriteHalf, stream: PrivStream, data: &str) {
let event = PrivEvent::Line(PrivStreamLine {
stream,
data: data.to_owned(),
});
if let Ok(mut json) = serde_json::to_string(&event) {
json.push('\n');
if let Err(e) = writer.write_all(json.as_bytes()).await {
tracing::warn!(error = %e, "write_line_event: write failed");
}
}
}
/// Execute a validated `PrivRequest`. Returns `(stdout, stderr)` on success.
async fn exec(req: PrivRequest) -> Result<(String, String)> {
/// For streaming ops (`CreateContainer`/`UpdateContainer` with `stream: true`)
/// output lines are forwarded to `writer` as `PrivEvent::Line` messages and
/// the returned strings are empty.
async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String, String)> {
match req {
PrivRequest::StartContainer { ref name } => {
validate_container_name(name)?;
@ -155,28 +178,36 @@ async fn exec(req: PrivRequest) -> Result<(String, String)> {
container_run(&["kill", &container_system_name(name)]).await
}
PrivRequest::UpdateContainer { ref name } => {
PrivRequest::UpdateContainer { ref name, stream } => {
validate_container_name(name)?;
let flake_ref = agent_flake_ref(name);
container_run(&[
let args = [
"update",
&container_system_name(name),
"--flake",
&flake_ref,
])
.await
];
if stream {
container_run_streaming(&args, writer).await
} else {
container_run(&args).await
}
}
PrivRequest::CreateContainer { ref name } => {
PrivRequest::CreateContainer { ref name, stream } => {
validate_container_name(name)?;
let flake_ref = agent_flake_ref(name);
container_run(&[
let args = [
"create",
&container_system_name(name),
"--flake",
&flake_ref,
])
.await
];
if stream {
container_run_streaming(&args, writer).await
} else {
container_run(&args).await
}
}
PrivRequest::DestroyContainer { ref name } => {
@ -332,6 +363,93 @@ async fn container_run(args: &[&str]) -> Result<(String, String)> {
Ok((stdout, stderr))
}
/// Invoke `nixos-container` with the given args and forward output lines
/// to the caller as `PrivEvent::Line` messages in real time, logging each
/// line to journald as it arrives. Returns `(String::new(), String::new())`
/// on success (all output was streamed); the error string includes stderr
/// tail on failure.
async fn container_run_streaming(
args: &[&str],
writer: &mut OwnedWriteHalf,
) -> Result<(String, String)> {
use tokio::io::AsyncBufReadExt as _;
use tokio::process::Command;
let mut child = Command::new("nixos-container")
.args(args)
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped())
.spawn()
.context("invoke nixos-container (streaming)")?;
let stdout = child.stdout.take().expect("stdout piped");
let stderr = child.stderr.take().expect("stderr piped");
let mut stdout_lines = BufReader::new(stdout).lines();
let mut stderr_lines = BufReader::new(stderr).lines();
// Collect stderr for the error message; stream both to the client.
let mut stderr_buf = String::new();
// Drive stdout and stderr concurrently. `tokio::select!` interleaves
// them without bias — both streams drain at roughly the same rate
// as the subprocess produces output.
loop {
tokio::select! {
line = stdout_lines.next_line() => {
match line {
Ok(Some(l)) => {
tracing::info!(target: "nixos-container", "{l}");
write_line_event(writer, PrivStream::Stdout, &l).await;
}
Ok(None) => break,
Err(e) => {
tracing::warn!(error = %e, "nixos-container stdout read error");
break;
}
}
}
line = stderr_lines.next_line() => {
match line {
Ok(Some(l)) => {
tracing::warn!(target: "nixos-container", "{l}");
write_line_event(writer, PrivStream::Stderr, &l).await;
if !stderr_buf.is_empty() {
stderr_buf.push('\n');
}
stderr_buf.push_str(&l);
}
Ok(None) => {}
Err(e) => {
tracing::warn!(error = %e, "nixos-container stderr read error");
}
}
}
}
}
// Drain any remaining stderr after stdout closed.
while let Ok(Some(l)) = stderr_lines.next_line().await {
tracing::warn!(target: "nixos-container", "{l}");
write_line_event(writer, PrivStream::Stderr, &l).await;
if !stderr_buf.is_empty() {
stderr_buf.push('\n');
}
stderr_buf.push_str(&l);
}
let status = child.wait().await.context("wait nixos-container")?;
if !status.success() {
bail!(
"nixos-container {} failed ({}): {}",
args.join(" "),
status,
stderr_buf.lines().last().unwrap_or("").trim()
);
}
Ok((String::new(), String::new()))
}
/// Read a container's journal as root via `journalctl -M`. Returns
/// `(stdout, stderr)`. Unlike `container_run` a non-zero exit is *not* a
/// hard error — journalctl's own diagnostic (folded into `stderr` with

57
hive-sh4re/src/priv_proto.rs
View file

@ -87,11 +87,23 @@ pub enum PrivRequest {
/// `nixos-container update <name> --flake <flake_ref>`
/// The flake ref is derived from `name` by hive-priv.
UpdateContainer { name: String },
/// When `stream` is `true`, hive-priv sends `PrivEvent::Line` messages
/// as the process runs, then a terminal `PrivEvent::Done`.
UpdateContainer {
name: String,
#[serde(default)]
stream: bool,
},
/// `nixos-container create <name> --flake <flake_ref>`
/// The flake ref is derived from `name` by hive-priv.
CreateContainer { name: String },
/// When `stream` is `true`, hive-priv sends `PrivEvent::Line` messages
/// as the process runs, then a terminal `PrivEvent::Done`.
CreateContainer {
name: String,
#[serde(default)]
stream: bool,
},
/// `nixos-container destroy <name>`
DestroyContainer { name: String },
@ -199,3 +211,44 @@ pub struct PrivResponse {
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
}
// ---------------------------------------------------------------------------
// Streaming protocol
// ---------------------------------------------------------------------------
/// Which output stream a `PrivStreamLine` came from.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PrivStream {
Stdout,
Stderr,
}
/// A single line forwarded from a streaming privileged operation.
/// Wire shape: `{"stream":"stdout","data":"..."}` — distinct from
/// `PrivResponse` (which has `ok` but not `stream`/`data`) so that
/// `PrivEvent` can disambiguate with `#[serde(untagged)]`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PrivStreamLine {
pub stream: PrivStream,
pub data: String,
}
/// An event in the streaming protocol used by long-running priv ops.
///
/// Wire format (multiple JSON lines over one connection):
/// - Zero or more `Line` events as the subprocess runs.
/// - One terminal `Done` event carrying the final status.
///
/// Non-streaming ops (and old hive-priv) send exactly one `Done` line,
/// which is wire-identical to a bare `PrivResponse` — so old `call()`
/// callers that deserialise straight to `PrivResponse` continue to work
/// with new hive-priv's terminal event.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(untagged)]
pub enum PrivEvent {
/// A line of output from the running subprocess.
Line(PrivStreamLine),
/// Terminal event: the operation has finished.
Done(PrivResponse),
}