diff --git a/CHANGELOG.md b/CHANGELOG.md index be6060d..d3e9c36 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,14 @@ Versioning](https://semver.org/spec/v2.0.0.html) once it reaches ## Unreleased +### Audit-log rotation + +The controller now reopens its audit log on `SIGHUP`, with reopen and request +writes serialized under the same writer lock so rotation does not drop or +split records. A successful reopen emits a `log_reopened` audit event. The +systemd unit exposes `systemctl reload`, and the packaged logrotate policy +uses rename/create plus reload instead of `copytruncate`. Closes #278. + ### MCP compatibility Constrain `forkd-mcp` to MCP `>=1.2,<2`: its FastMCP import is unavailable diff --git a/crates/forkd-controller/src/audit.rs b/crates/forkd-controller/src/audit.rs index 69a2a88..7d27dde 100644 --- a/crates/forkd-controller/src/audit.rs +++ b/crates/forkd-controller/src/audit.rs @@ -8,8 +8,8 @@ //! of request handling stays parallel. //! //! Designed to be tailed by an external log shipper (vector, fluentbit). -//! No rotation in-process — operators should plug in logrotate or run -//! the daemon under a journal that handles size caps. +//! Rotation is external: rename the file and send SIGHUP so the daemon +//! atomically reopens the configured path without dropping records. use anyhow::{Context, Result}; use axum::extract::Request; use axum::middleware::Next; @@ -56,6 +56,39 @@ impl AuditSink { &self.inner.path } + /// Flush the current audit file and reopen the configured path. + /// + /// The writer lock is held across flush, open, and replacement, so a + /// concurrent request writes wholly to either the rotated file or the + /// new file. If opening the new path fails, the old writer remains in + /// place and callers can retry on the next SIGHUP. + pub fn reopen(&self) -> Result<()> { + let mut writer = self.inner.writer.lock(); + writer + .flush() + .with_context(|| format!("flush audit log {}", self.inner.path.display()))?; + let file = OpenOptions::new() + .create(true) + .append(true) + .open(&self.inner.path) + .with_context(|| format!("reopen audit log {}", self.inner.path.display()))?; + let mut replacement = BufWriter::new(file); + writeln!( + replacement, + "{}", + json!({ + "ts": now_rfc3339(), + "event": "log_reopened", + }) + ) + .with_context(|| format!("write reopen event to {}", self.inner.path.display()))?; + replacement + .flush() + .with_context(|| format!("flush reopened audit log {}", self.inner.path.display()))?; + *writer = replacement; + Ok(()) + } + pub fn write(&self, line: serde_json::Value) { let mut w = self.inner.writer.lock(); if let Err(e) = writeln!(w, "{line}") { @@ -187,4 +220,46 @@ mod tests { assert!(lines.next().unwrap().contains("\"a\":2")); assert!(lines.next().is_none()); } + + #[cfg(unix)] + #[test] + fn audit_sink_reopens_after_external_rotation() { + let td = tempfile::TempDir::new().unwrap(); + let path = td.path().join("audit.log"); + let rotated = td.path().join("audit.log.1"); + let sink = AuditSink::open(&path).unwrap(); + + sink.write(json!({"before": true})); + std::fs::rename(&path, &rotated).unwrap(); + sink.reopen().unwrap(); + sink.write(json!({"after": true})); + + let old_contents = std::fs::read_to_string(rotated).unwrap(); + let new_contents = std::fs::read_to_string(path).unwrap(); + assert!(old_contents.contains("\"before\":true")); + assert!(!old_contents.contains("\"after\":true")); + assert!(new_contents.contains("\"event\":\"log_reopened\"")); + assert!(new_contents.contains("\"after\":true")); + assert!(!new_contents.contains("\"before\":true")); + } + + #[cfg(unix)] + #[test] + fn audit_sink_keeps_old_writer_when_reopen_fails() { + let td = tempfile::TempDir::new().unwrap(); + let path = td.path().join("audit.log"); + let rotated = td.path().join("audit.log.1"); + let sink = AuditSink::open(&path).unwrap(); + + sink.write(json!({"before": true})); + std::fs::rename(&path, &rotated).unwrap(); + std::fs::create_dir(&path).unwrap(); + + assert!(sink.reopen().is_err()); + sink.write(json!({"after_failed_reopen": true})); + + let old_contents = std::fs::read_to_string(rotated).unwrap(); + assert!(old_contents.contains("\"before\":true")); + assert!(old_contents.contains("\"after_failed_reopen\":true")); + } } diff --git a/crates/forkd-controller/src/lib.rs b/crates/forkd-controller/src/lib.rs index 684d98b..f07fbf2 100644 --- a/crates/forkd-controller/src/lib.rs +++ b/crates/forkd-controller/src/lib.rs @@ -81,7 +81,8 @@ fn unauthenticated_non_loopback(bind: SocketAddr, token_file: Option<&Path>) -> } /// Bring up the controller daemon. Blocks until the listener exits. -/// SIGTERM and SIGINT trigger a graceful shutdown. +/// SIGTERM and SIGINT trigger a graceful shutdown; SIGHUP reopens the +/// configured audit log after external rotation. pub async fn run_daemon(cfg: DaemonConfig) -> Result<()> { let registry = Registry::load_or_init(&cfg.state_file) .with_context(|| format!("load state from {}", cfg.state_file.display()))?; @@ -166,7 +167,7 @@ pub async fn run_daemon(cfg: DaemonConfig) -> Result<()> { // plus a Handle for cooperative shutdown that drains in-flight // requests up to a deadline. let handle = Handle::new(); - spawn_shutdown_signal(handle.clone()); + let _signal_task = spawn_signal_handler(handle.clone(), audit.clone()); let tls = match (&cfg.tls_cert, &cfg.tls_key) { (Some(c), Some(k)) => Some(load_tls(c, k).await?), @@ -207,30 +208,102 @@ async fn load_tls(cert: &Path, key: &Path) -> Result { .with_context(|| format!("load TLS cert {} / key {}", cert.display(), key.display())) } -fn spawn_shutdown_signal(handle: Handle) { - tokio::spawn(async move { - let ctrl_c = async { - let _ = tokio::signal::ctrl_c().await; - }; +struct SignalTask(tokio::task::JoinHandle<()>); - #[cfg(unix)] - let terminate = async { - if let Ok(mut sig) = - tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) - { - sig.recv().await; +impl Drop for SignalTask { + fn drop(&mut self) { + self.0.abort(); + } +} + +#[cfg(unix)] +fn spawn_signal_handler(handle: Handle, audit: AuditSink) -> SignalTask { + let mut interrupt = + match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt()) { + Ok(signal) => Some(signal), + Err(error) => { + tracing::error!(%error, "failed to install SIGINT handler"); + None } }; + let mut terminate = + match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) { + Ok(signal) => Some(signal), + Err(error) => { + tracing::error!(%error, "failed to install SIGTERM handler"); + None + } + }; + let mut hangup = match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::hangup()) { + Ok(signal) => Some(signal), + Err(error) => { + tracing::error!(%error, "failed to install SIGHUP handler"); + None + } + }; - #[cfg(not(unix))] - let terminate = std::future::pending::<()>(); + SignalTask(tokio::spawn(async move { + loop { + tokio::select! { + signal = recv_unix_signal(&mut interrupt) => { + if signal.is_none() { + tracing::error!("SIGINT signal stream closed"); + interrupt = None; + continue; + } + tracing::info!("received SIGINT, shutting down"); + break; + } + signal = recv_unix_signal(&mut terminate) => { + if signal.is_none() { + tracing::error!("SIGTERM signal stream closed"); + terminate = None; + continue; + } + tracing::info!("received SIGTERM, shutting down"); + break; + } + signal = recv_unix_signal(&mut hangup) => { + if signal.is_none() { + tracing::error!("SIGHUP signal stream closed"); + hangup = None; + continue; + } + match audit.reopen() { + Ok(()) => tracing::info!( + audit_log = %audit.path().display(), + "reopened audit log after SIGHUP" + ), + Err(error) => tracing::error!( + %error, + audit_log = %audit.path().display(), + "failed to reopen audit log after SIGHUP" + ), + } + } + } + } + handle.graceful_shutdown(Some(Duration::from_secs(30))); + })) +} - tokio::select! { - _ = ctrl_c => tracing::info!("received SIGINT, shutting down"), - _ = terminate => tracing::info!("received SIGTERM, shutting down"), +#[cfg(unix)] +async fn recv_unix_signal(signal: &mut Option) -> Option<()> { + match signal { + Some(signal) => signal.recv().await, + None => std::future::pending().await, + } +} + +#[cfg(not(unix))] +fn spawn_signal_handler(handle: Handle, _audit: AuditSink) -> SignalTask { + SignalTask(tokio::spawn(async move { + match tokio::signal::ctrl_c().await { + Ok(()) => tracing::info!("received interrupt, shutting down"), + Err(error) => tracing::error!(%error, "interrupt handler failed"), } handle.graceful_shutdown(Some(Duration::from_secs(30))); - }); + })) } /// Reject tokens that are empty, obvious placeholders, or below a minimum diff --git a/crates/forkd-controller/tests/http_integration.rs b/crates/forkd-controller/tests/http_integration.rs index c4eaec6..7d509c9 100644 --- a/crates/forkd-controller/tests/http_integration.rs +++ b/crates/forkd-controller/tests/http_integration.rs @@ -276,3 +276,54 @@ async fn end_to_end_audit_log_records_request() { } panic!("audit log never captured /version request"); } + +#[cfg(unix)] +#[tokio::test] +async fn end_to_end_sighup_reopens_rotated_audit_log() { + let d = TestDaemon::start().await; + let audit = d._td.path().join("audit.log"); + let rotated = d._td.path().join("audit.log.1"); + + let _ = reqwest::get(format!("{}/version", d.base)).await.unwrap(); + for _ in 0..50 { + if std::fs::read_to_string(&audit).is_ok_and(|contents| contents.contains("\"/version\"")) { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(std::fs::read_to_string(&audit) + .unwrap() + .contains("\"/version\"")); + + std::fs::rename(&audit, &rotated).unwrap(); + let signal_result = unsafe { libc::kill(libc::getpid(), libc::SIGHUP) }; + assert_eq!( + signal_result, + 0, + "send SIGHUP: {}", + std::io::Error::last_os_error() + ); + + for _ in 0..50 { + if audit.exists() { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(audit.exists(), "SIGHUP did not recreate the audit log path"); + + let _ = reqwest::get(format!("{}/metrics", d.base)).await.unwrap(); + for _ in 0..50 { + if std::fs::read_to_string(&audit).is_ok_and(|contents| contents.contains("\"/metrics\"")) { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + + let old_contents = std::fs::read_to_string(rotated).unwrap(); + let new_contents = std::fs::read_to_string(audit).unwrap(); + assert!(old_contents.contains("\"/version\"")); + assert!(!old_contents.contains("\"/metrics\"")); + assert!(new_contents.contains("\"event\":\"log_reopened\"")); + assert!(new_contents.contains("\"/metrics\"")); +} diff --git a/docs/RUNBOOK.md b/docs/RUNBOOK.md index 724e1af..c55e173 100644 --- a/docs/RUNBOOK.md +++ b/docs/RUNBOOK.md @@ -76,8 +76,15 @@ Suggested alerts: {"ts":"2026-05-12T07:12:34Z","method":"POST","path":"/v1/sandboxes","status":201,"latency_us":98342,"ua":"forkd-cli/0.1"} ``` -Rotate with `logrotate`. The daemon reopens the file on `SIGHUP` is -not yet implemented — for now, `systemctl restart` after a rotate. +Rotate with `logrotate`. The packaged policy renames the current file, +creates a new `0600` log, and runs `systemctl try-reload-or-restart +forkd-controller`. The service maps reload to `SIGHUP`; the daemon flushes +the old writer and atomically reopens the configured path without restarting +or dropping in-flight requests. The `try-` variant safely falls back to a +restart for an older installed service without reload support and does nothing +when the service is inactive. A successful reopen writes a `log_reopened` +event to the new file. For a manual rotation, send `systemctl reload +forkd-controller` after moving the file. --- diff --git a/packaging/arch/PKGBUILD b/packaging/arch/PKGBUILD index 00387a5..0c04f13 100644 --- a/packaging/arch/PKGBUILD +++ b/packaging/arch/PKGBUILD @@ -35,7 +35,7 @@ source_x86_64=( sha256sums=( '86f7b7013dff87a69534190a231c8b3545c7365d123728002089e2b91eb8f7b4' '2aab38bae2a1d975281806a3f9bdb86fc1b59e40664438f238887e7b9e888cae' - '55aeede89ec62af51910487d7c9d583632b3a1973f36c53680fe7febb5320271' + 'e7f34dfc8a32031ba897dd60452918ef8fa9118c1a17c51df8bb1b3764a92694' ) sha256sums_x86_64=( '417865e0d9bb3fcaf6bd0308dfe8d40a4bd282adefb62459dd76585d6bede7e4' @@ -86,9 +86,14 @@ check() { grep -Fqx ' delaycompress' "$srcdir/forkd.logrotate" grep -Fqx ' missingok' "$srcdir/forkd.logrotate" grep -Fqx ' notifempty' "$srcdir/forkd.logrotate" - grep -Fqx ' copytruncate' "$srcdir/forkd.logrotate" grep -Fqx ' su root root' "$srcdir/forkd.logrotate" grep -Fqx ' create 0600 root root' "$srcdir/forkd.logrotate" + grep -Fqx ' sharedscripts' "$srcdir/forkd.logrotate" + grep -Fqx ' postrotate' "$srcdir/forkd.logrotate" + grep -Fqx ' /usr/bin/systemctl try-reload-or-restart forkd-controller.service >/dev/null 2>&1 || true' \ + "$srcdir/forkd.logrotate" + grep -Fqx ' endscript' "$srcdir/forkd.logrotate" + ! grep -Fq 'copytruncate' "$srcdir/forkd.logrotate" } package() { diff --git a/packaging/arch/forkd.logrotate b/packaging/arch/forkd.logrotate index 4683b3c..747a5e6 100644 --- a/packaging/arch/forkd.logrotate +++ b/packaging/arch/forkd.logrotate @@ -5,7 +5,10 @@ delaycompress missingok notifempty - copytruncate su root root create 0600 root root + sharedscripts + postrotate + /usr/bin/systemctl try-reload-or-restart forkd-controller.service >/dev/null 2>&1 || true + endscript } diff --git a/packaging/systemd/forkd-controller.service b/packaging/systemd/forkd-controller.service index 2af5903..c29aed3 100644 --- a/packaging/systemd/forkd-controller.service +++ b/packaging/systemd/forkd-controller.service @@ -11,6 +11,7 @@ ExecStart=/usr/local/bin/forkd-controller serve \ --state /var/lib/forkd/state.json \ --audit-log /var/log/forkd/audit.log \ --token-file /etc/forkd/token +ExecReload=/bin/kill -HUP $MAINPID Restart=on-failure RestartSec=2s TimeoutStopSec=15s