From e5abf9f81244246552c42358b4d5b14e5a13311f Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Thu, 22 Oct 2026 11:03:33 +0000 Subject: [PATCH 1/3] Write per-VM log through a tracing layer Every per-VM event was logged twice, once through tracing and once as a hand-formatted line for the per-VM file, and events logged elsewhere, such as the threshold engine's scaling decisions, never reached the per-VM file at all. Add a tracing layer that copies every event carrying a vm field to that VM's file, without the vm field, and log per-VM events only once. Status and lifecycle records use a dedicated per_vm target that the main log filters out, so they still only go to the per-VM files. The layer has its own INFO filter, so RUST_LOG no longer affects the per-VM files. Use vm consistently for the VM ID field, instead of id or uuid, so that those events are routed too. AI-generated: everything Signed-off-by: Thanos Makatos --- README.md | 4 +- src/backends/qemu/client.rs | 12 +- src/backends/qemu/libvirt.rs | 2 +- src/backends/qemu/mod.rs | 4 +- src/controller.rs | 255 +++++++++++----------------------- src/engines/threshold.rs | 2 +- src/lib.rs | 1 + src/main.rs | 58 +++----- src/per_vm_log.rs | 262 +++++++++++++++++++++++++++++++++++ 9 files changed, 377 insertions(+), 223 deletions(-) create mode 100644 src/per_vm_log.rs diff --git a/README.md b/README.md index 0329432a..e4ece764 100644 --- a/README.md +++ b/README.md @@ -205,7 +205,9 @@ A dash (`-`) means no sample is available for that field. The per-VM log also records `event=vm_added` when the daemon starts tracking the VM and `event=vm_removed reason=` when it -stops. +stops. Every other log event that names the VM in a `vm` field, such as scaling +decisions, blocked or failed scale operations, and backend errors, is copied to +the VM's file as its message followed by its remaining `key=value` fields. ### Aggregate line diff --git a/src/backends/qemu/client.rs b/src/backends/qemu/client.rs index cd72439a..5dfb862b 100644 --- a/src/backends/qemu/client.rs +++ b/src/backends/qemu/client.rs @@ -103,7 +103,7 @@ impl QemuInstanceClient { if let Err(del_err) = self.client.del_io_thread(&new_id).await { tracing::debug!( target: "qemu", - uuid = %self.uuid, + vm = %self.uuid, iothread = %new_id, "rollback object-del after set-mapping failure: {del_err}" ); @@ -133,7 +133,7 @@ impl QemuInstanceClient { if let Err(e) = self.client.del_io_thread(&doomed).await { tracing::warn!( target: "qemu", - uuid = %self.uuid, + vm = %self.uuid, iothread = %doomed, error = ?e, "object-del failed; will retry next tick" @@ -159,7 +159,7 @@ impl QemuInstanceClient { if thread_count == 1 { tracing::debug!( target: "qemu", - uuid = %self.uuid, + vm = %self.uuid, "using minimal fallback snapshot without cached topology" ); } @@ -170,7 +170,7 @@ impl QemuInstanceClient { Err(e) => { tracing::debug!( target: "qemu", - uuid = %self.uuid, + vm = %self.uuid, error = ?e, "query-blockstats failed" ); @@ -239,7 +239,7 @@ impl InstanceClient for QemuInstanceClient { if should_warn { tracing::warn!( target: "qemu", - uuid = %self.uuid, + vm = %self.uuid, "QMP command x-query-prometheus-metrics unsupported; using fallback snapshot" ); } @@ -305,7 +305,7 @@ impl InstanceClient for QemuInstanceClient { Err(e) => { tracing::debug!( target: "qemu", - uuid = %self.uuid, + vm = %self.uuid, error = ?e, "query-blockstats failed" ); diff --git a/src/backends/qemu/libvirt.rs b/src/backends/qemu/libvirt.rs index 4f054cf7..99d8c292 100644 --- a/src/backends/qemu/libvirt.rs +++ b/src/backends/qemu/libvirt.rs @@ -108,7 +108,7 @@ impl LibvirtConn { Err(e) => { tracing::debug!( target: "qemu", - uuid = %uuid, + vm = %uuid, error = ?e, "libvirt: skip vm (xml fetch)" ); diff --git a/src/backends/qemu/mod.rs b/src/backends/qemu/mod.rs index 51265dc0..a840953a 100644 --- a/src/backends/qemu/mod.rs +++ b/src/backends/qemu/mod.rs @@ -186,7 +186,7 @@ impl Backend for QemuBackend { Ok(_) => { tracing::warn!( target: "qemu", - %uuid, + vm = %uuid, "libvirt reported zero maximum vCPUs; skipping VM" ); continue; @@ -194,7 +194,7 @@ impl Backend for QemuBackend { Err(error) => { tracing::warn!( target: "qemu", - %uuid, + vm = %uuid, %error, "failed to read maximum vCPU count; skipping VM" ); diff --git a/src/controller.rs b/src/controller.rs index 327404b1..920a1771 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -7,10 +7,8 @@ use std::{ collections::{HashMap, HashSet}, - fs::{OpenOptions, create_dir_all}, - io::Write, sync::Arc, - time::{Duration, Instant, SystemTime}, + time::{Duration, Instant}, }; use futures_util::future::join_all; @@ -28,7 +26,6 @@ use crate::{ }, instance::{CgroupError, Instance, InstanceStatus, SnapshotLatency}, state::{StateError, VmOwnership, VmStateStore}, - util::Path, }; /// Wire-facing container for the D-Bus `GetSnapshot` reply. #[derive(Debug, Clone, Default, Serialize, Deserialize)] @@ -182,7 +179,6 @@ const fn from_mins(minutes: u64) -> Duration { /// Rolling windows displayed in uptime-style status fields. const STATUS_WINDOWS: [Duration; 3] = [from_mins(1), from_mins(5), from_mins(15)]; -const PER_VM_LOG_DIR: &str = "/var/log/io-thread-controller/vms"; /// Host-wide cumulative CPU counters from `/proc/stat`. #[derive(Debug, Default, Clone, Copy)] @@ -211,10 +207,6 @@ pub struct Controller { /// and decide. // FIXME Instance probably doesn't need to be an Arc. pub instances: HashMap>, - /// Per-VM log file destination when per-VM main log is disabled. - per_vm_log_dir: Path, - /// VM IDs for which per-VM log I/O failures were already reported. - per_vm_log_io_error_warned: std::sync::Mutex>, /// Monotonically increasing tick sequence. pub tick_index: u64, /// Previous host CPU sample. @@ -236,8 +228,6 @@ impl Controller { vm_state_store, vm_ownership, instances: HashMap::new(), - per_vm_log_dir: Path::from(PER_VM_LOG_DIR), - per_vm_log_io_error_warned: std::sync::Mutex::new(HashSet::new()), tick_index: 0, last_host_cpu: None, host_cpu_util: 0.0, @@ -267,7 +257,7 @@ impl Controller { status.ownership_classification = classification; drop(status); self.engine.on_instance_added(&instance).await; - self.log_per_vm_line(&instance.id, "event=vm_added"); + tracing::info!(target: per_vm_log::TARGET, vm = %instance.id, event = "vm_added"); self.instances.insert(instance.id.clone(), instance); added += 1; } @@ -283,7 +273,12 @@ impl Controller { if let Some(instance) = self.instances.remove(&id) { instance.client.close().await; self.engine.on_instance_removed(&id).await; - self.log_per_vm_line(&id, "event=vm_removed reason=not_discovered"); + tracing::info!( + target: per_vm_log::TARGET, + vm = %id, + event = "vm_removed", + reason = "not_discovered" + ); } } Ok((added, removed)) @@ -573,7 +568,12 @@ impl Controller { { self.instances.remove(&id); self.engine.on_instance_removed(&id).await; - self.log_per_vm_line(&id, "event=vm_removed reason=refresh_failed"); + tracing::info!( + target: per_vm_log::TARGET, + vm = %id, + event = "vm_removed", + reason = "refresh_failed" + ); // FIXME IIUC a failed instance gets dropped but we don't call // close() } @@ -688,10 +688,6 @@ impl Controller { if sticky { self.report_blocked_scale(&instance.id, action, BlockedReason::ManualOverride) .await; - self.log_per_vm_line( - &instance.id, - &format!("event=scale_blocked reason=manual_override action={action}"), - ); tracing::info!( target: "controller", event = "scale_blocked", @@ -705,10 +701,6 @@ impl Controller { // FIXME why would an engine even consider an unmanaged VM? self.report_blocked_scale(&instance.id, action, BlockedReason::UnmanagedVm) .await; - self.log_per_vm_line( - &instance.id, - &format!("event=scale_blocked reason=unmanaged_vm action={action}"), - ); tracing::info!( target: "controller", event = "scale_blocked", @@ -721,16 +713,12 @@ impl Controller { if target < self.cfg.min_thread_count { self.report_blocked_scale(instance_id, action, BlockedReason::TargetBelowMinimum) .await; - self.log_per_vm_line( - &instance.id, - &format!( - "event=scale_blocked reason=target_below_minimum action={action} target={} minimum={}", - target, self.cfg.min_thread_count - ), - ); tracing::info!( target: "controller", - id = %instance.id, + event = "scale_blocked", + reason = "target_below_minimum", + vm = %instance.id, + action = %action, target, minimum = self.cfg.min_thread_count, "automatic scaling suppressed by controller minimum" @@ -740,16 +728,12 @@ impl Controller { if target > self.cfg.max_thread_count { self.report_blocked_scale(instance_id, action, BlockedReason::TargetExceedsMaximum) .await; - self.log_per_vm_line( - &instance.id, - &format!( - "event=scale_blocked reason=target_exceeds_maximum action={action} target={} maximum={}", - target, self.cfg.max_thread_count - ), - ); tracing::info!( target: "controller", - id = %instance.id, + event = "scale_blocked", + reason = "target_exceeds_maximum", + vm = %instance.id, + action = %action, target, maximum = self.cfg.max_thread_count, "automatic scaling suppressed by controller maximum" @@ -759,16 +743,12 @@ impl Controller { if target > vcpu_count { self.report_blocked_scale(instance_id, action, BlockedReason::TargetExceedsVcpuCount) .await; - self.log_per_vm_line( - &instance.id, - &format!( - "event=scale_blocked reason=target_exceeds_vcpu_count action={action} target={} vcpus={}", - target, vcpu_count - ), - ); tracing::info!( target: "controller", - id = %instance.id, + event = "scale_blocked", + reason = "target_exceeds_vcpu_count", + vm = %instance.id, + action = %action, target, vcpus = vcpu_count, "automatic scaling suppressed by vCPU cap" @@ -780,13 +760,12 @@ impl Controller { { self.report_blocked_scale(instance_id, action, BlockedReason::Cooldown) .await; - self.log_per_vm_line( - &instance.id, - &format!("event=scale_blocked reason=cooldown action={action}"), - ); tracing::info!( target: "controller", - id = %instance.id, + event = "scale_blocked", + reason = "cooldown", + vm = %instance.id, + action = %action, "automatic scaling suppressed by cooldown" ); return Ok(false); @@ -801,16 +780,12 @@ impl Controller { { self.report_blocked_scale(instance_id, action, BlockedReason::HostCpuCeiling) .await; - self.log_per_vm_line( - &instance.id, - &format!( - "event=scale_blocked reason=host_cpu_ceiling action={action} host_cpu={} ceiling={}", - self.host_cpu_util, self.cfg.host_cpu_scale_up_ceiling - ), - ); tracing::info!( target: "controller", - id = %instance.id, + event = "scale_blocked", + reason = "host_cpu_ceiling", + vm = %instance.id, + action = %action, host_cpu = self.host_cpu_util, ceiling = self.cfg.host_cpu_scale_up_ceiling, "automatic scaling suppressed by host CPU guard" @@ -819,12 +794,9 @@ impl Controller { } if self.cfg.dry_run { - self.log_per_vm_line( - &instance.id, - &format!("event=scale_dry_run action={action} target={target}"), - ); tracing::info!( target: "controller", + event = "scale_dry_run", vm = %instance.id, action = %action, target, @@ -885,13 +857,6 @@ impl Controller { prev_thread_count = previous_count, prev_io_count_total = previous_io_count ); - self.log_per_vm_line( - &instance.id, - &format!( - "event=scale_applied action={action} target={} prev_thread_count={} prev_io_count_total={}", - target, previous_count, previous_io_count - ), - ); Ok(true) } Err(error) => { @@ -907,13 +872,6 @@ impl Controller { target, error = %error_text ); - self.log_per_vm_line( - &instance.id, - &format!( - "event=scale_failed action={action} target={} error={}", - target, error_text - ), - ); Ok(false) } } @@ -950,7 +908,7 @@ impl Controller { if self.cfg.enable_per_vm_status_line && self.cfg.enable_per_vm_main_log { Self::emit_instance_status_main(instance, &status, iops_windows); } else if self.cfg.enable_per_vm_status_line { - self.emit_instance_status_file(instance, &status, iops_windows); + Self::emit_instance_status_file(instance, &status, iops_windows); } if self.cfg.enable_per_vm_main_log { Self::emit_instance_status_main(instance, &status, iops_windows); @@ -997,14 +955,15 @@ impl Controller { } fn emit_instance_status_file( - &self, instance: &Instance, status: &InstanceStatus, iops_windows: [Option; 3], ) { - self.log_per_vm_line( - &instance.id, - &Self::format_instance_status_line(status, iops_windows), + tracing::info!( + target: per_vm_log::TARGET, + vm = %instance.id, + "{}", + Self::format_instance_status_line(status, iops_windows) ); } @@ -1029,88 +988,6 @@ impl Controller { write_latency_us = %format_latency_cell(status.write_latency_us, "/") ); } - - fn should_log_per_vm_file(&self) -> bool { - self.cfg.enable_per_vm_status_line && !self.cfg.enable_per_vm_main_log - } - - fn per_vm_log_path(&self, vm: &str) -> Path { - let name: String = vm - .chars() - .map(|ch| { - if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.') { - ch - } else { - '_' - } - }) - .collect(); - Path::new(&self.per_vm_log_dir.join(format!("{name}.log"))) - } - - fn log_per_vm_line(&self, vm: &str, line: &str) { - if !self.should_log_per_vm_file() { - return; - } - let path = self.per_vm_log_path(vm); - let path_text = path.display().to_string(); - if let Some(parent) = path.parent() - && let Err(error) = create_dir_all(parent) - { - let mut warned = self - .per_vm_log_io_error_warned - .lock() - .expect("lock poisoned"); - if warned.insert(vm.to_string()) { - tracing::warn!( - target: "controller", - vm, - path = %path_text, - %error, - "failed to create per-VM log directory" - ); - } - return; - } - match OpenOptions::new().create(true).append(true).open(&path) { - Ok(mut file) => { - if let Err(error) = writeln!( - file, - "{} {line}", - humantime::format_rfc3339_millis(SystemTime::now()) - ) { - let mut warned = self - .per_vm_log_io_error_warned - .lock() - .expect("lock poisoned"); - if warned.insert(vm.to_string()) { - tracing::warn!( - target: "controller", - vm, - path = %path_text, - %error, - "failed to write per-VM log" - ); - } - } - } - Err(error) => { - let mut warned = self - .per_vm_log_io_error_warned - .lock() - .expect("lock poisoned"); - if warned.insert(vm.to_string()) { - tracing::warn!( - target: "controller", - vm, - path = %path_text, - %error, - "failed to open per-VM log" - ); - } - } - } - } } /// Read CPU, I/O, and memory pressure; unavailable values become zero. @@ -1202,8 +1079,10 @@ mod tests { backends::BackendClientError, engines::{EngineTickContext, ScalingEngine}, instance::{InstanceClient, ThreadPoolSnapshot}, + per_vm_log::PerVmLog, util::Path, }; + use tracing_subscriber::layer::SubscriberExt; #[cfg(feature = "threshold-engine")] use crate::engines::threshold::{ThresholdConfig, ThresholdEngine}; @@ -1265,6 +1144,7 @@ mod tests { controller .instances .insert(instance.id.clone(), Arc::clone(&instance)); + let (per_vm_log, _log_dir, _guard) = capture_per_vm_log(); controller .apply_engine_decision(&instance.id, ScaleAction::Up(4)) @@ -1294,6 +1174,16 @@ mod tests { .await .unwrap(); assert_eq!(target.load(Ordering::Relaxed), 3); + + let events = per_vm_log_events(&per_vm_log, &instance.id); + assert!(events[0].starts_with("event=scale_applied "), "{events:?}"); + assert!( + events[1].starts_with( + "automatic scaling suppressed by vCPU cap event=scale_blocked \ + reason=target_exceeds_vcpu_count " + ) && events[1].ends_with(" target=5 vcpus=4"), + "{events:?}" + ); } // FIXME This test seems like it mixes basic ScaleAction unit tests and more @@ -1687,8 +1577,7 @@ mod tests { }), ) .unwrap(); - let log_dir = tempfile::tempdir().unwrap(); - controller.per_vm_log_dir = Path::new(log_dir.path()); + let (per_vm_log, _log_dir, _guard) = capture_per_vm_log(); let closed = Arc::new(AtomicUsize::new(0)); let first = instance("vm-a", 1, Arc::clone(&closed)); @@ -1712,19 +1601,34 @@ mod tests { assert_eq!(removed.load(Ordering::Relaxed), 1); assert_eq!(evaluated.load(Ordering::Relaxed), 0); assert_eq!( - per_vm_log_events(&controller, "vm-a"), + per_vm_log_events(&per_vm_log, "vm-a"), ["event=vm_added", "event=vm_removed reason=not_discovered"] ); } - /// Return the `event=` records of one VM's per-VM log, without - /// timestamps. - fn per_vm_log_events(controller: &Controller, vm: &str) -> Vec { - std::fs::read_to_string(controller.per_vm_log_path(vm)) + /// Route this thread's per-VM events to files in a temporary directory. + fn capture_per_vm_log() -> ( + PerVmLog, + tempfile::TempDir, + tracing::subscriber::DefaultGuard, + ) { + let dir = tempfile::tempdir().unwrap(); + let per_vm_log = PerVmLog::default(); + per_vm_log.set_dir(Some(Path::new(dir.path()))); + let guard = tracing::subscriber::set_default( + tracing_subscriber::registry().with(per_vm_log.clone()), + ); + (per_vm_log, dir, guard) + } + + /// Return the records of one VM's per-VM log that carry an `event` + /// field, without timestamps. + fn per_vm_log_events(per_vm_log: &PerVmLog, vm: &str) -> Vec { + std::fs::read_to_string(per_vm_log.path(vm).unwrap()) .unwrap() .lines() .filter_map(|line| line.split_once(' ').map(|(_, record)| record)) - .filter(|record| record.starts_with("event=")) + .filter(|record| record.contains("event=")) .map(str::to_string) .collect() } @@ -1750,8 +1654,7 @@ mod tests { }), ) .unwrap(); - let log_dir = tempfile::tempdir().unwrap(); - controller.per_vm_log_dir = Path::new(log_dir.path()); + let (per_vm_log, _log_dir, _guard) = capture_per_vm_log(); let closed = Arc::new(AtomicUsize::new(0)); let ok = instance("ok", 4, Arc::clone(&closed)); @@ -1774,9 +1677,9 @@ mod tests { ); assert_eq!(evaluated.load(Ordering::Relaxed), 1); assert_eq!(removed.load(Ordering::Relaxed), 1); - assert_eq!(per_vm_log_events(&controller, "ok"), ["event=vm_added"]); + assert_eq!(per_vm_log_events(&per_vm_log, "ok"), ["event=vm_added"]); assert_eq!( - per_vm_log_events(&controller, "bad"), + per_vm_log_events(&per_vm_log, "bad"), ["event=vm_added", "event=vm_removed reason=refresh_failed"] ); } diff --git a/src/engines/threshold.rs b/src/engines/threshold.rs index 045bc2f8..3102812b 100644 --- a/src/engines/threshold.rs +++ b/src/engines/threshold.rs @@ -751,7 +751,7 @@ impl ScalingEngine for ThresholdEngine { if self.cfg.block_scale_up_when_cgroup_throttled && throttled_recently { tracing::info!( target: "controller", - id = %instance.id, + vm = %instance.id, "scale-up suppressed by cgroup CPU throttling" ); return ScaleAction::None; diff --git a/src/lib.rs b/src/lib.rs index ebc08606..252818b3 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -12,6 +12,7 @@ pub mod daemon; pub mod dbus; pub mod engines; pub mod instance; +pub mod per_vm_log; pub mod rolling; pub mod state; pub mod util; diff --git a/src/main.rs b/src/main.rs index 06aaccb9..feaff3f0 100644 --- a/src/main.rs +++ b/src/main.rs @@ -13,10 +13,12 @@ use io_thread_controller::{ backends::registered_backends, config::{Config, ConfigError, dump_default_config, load_json_with_includes, validate_config}, daemon::{DaemonError, VERSION, run}, + per_vm_log::{self, PER_VM_LOG_DIR, PerVmLog}, util::Path, }; use thiserror::Error; use tracing::{debug, info, instrument}; +use tracing_subscriber::{Layer, filter::LevelFilter, fmt, layer::SubscriberExt}; #[derive(Error, Debug)] enum IoThreadControllerError { @@ -104,7 +106,8 @@ async fn main() -> Result<(), IoThreadControllerError> { } let matches = command.get_matches(); let cli = Cli::from_arg_matches(&matches)?; - init_logging(&cli.log_level, cli.log_style); + let per_vm_log = PerVmLog::default(); + init_logging(&cli.log_level, cli.log_style, per_vm_log.clone()); if cli.dump_config { println!("{}", dump_default_config()); @@ -138,6 +141,10 @@ async fn main() -> Result<(), IoThreadControllerError> { cfg.dry_run = true; } validate_config(&cfg)?; + per_vm_log.set_dir( + (cfg.enable_per_vm_status_line && !cfg.enable_per_vm_main_log) + .then(|| Path::from(PER_VM_LOG_DIR)), + ); let backends = registered_backends(&cfg)?; run(cfg, backends).await?; Ok(()) @@ -154,7 +161,7 @@ fn load_daemon_config(path: &Path) -> Result { } /// Install the process-wide tracing subscriber. -fn init_logging(filter: &str, style: LogStyle) { +fn init_logging(filter: &str, style: LogStyle, per_vm_log: PerVmLog) { // Auto-detection: systemd sets `JOURNAL_STREAM=:` // on every service invocation whose stderr is journald. // Presence is enough; the value itself is only useful when @@ -173,38 +180,17 @@ fn init_logging(filter: &str, style: LogStyle) { } } }; - let env_filter = tracing_subscriber::EnvFilter::try_new(filter) - .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")); - // `tracing_subscriber::fmt` returns two different concrete - // subscriber types depending on whether `.without_time()` - // is chained, so we build and install each branch in its - // own scope rather than fighting the type system to - // unify them. - match effective_style { - LogStyle::Systemd => { - let subscriber = tracing_subscriber::fmt() - // Keep target so downstream routing can separate - // per-VM status (`status`) from controller logs. - .with_target(true) - .without_time() - .with_env_filter(env_filter) - .finish(); - let _ = tracing::subscriber::set_global_default(subscriber); - } - LogStyle::Json => { - let subscriber = tracing_subscriber::fmt() - .json() - .with_target(true) - .with_env_filter(env_filter) - .finish(); - let _ = tracing::subscriber::set_global_default(subscriber); - } - LogStyle::Human | LogStyle::Auto => { - let subscriber = tracing_subscriber::fmt() - .with_target(true) - .with_env_filter(env_filter) - .finish(); - let _ = tracing::subscriber::set_global_default(subscriber); - } - } + let main_log = match effective_style { + // Keep target so downstream routing can separate per-VM status + // (`status`) from controller logs. + LogStyle::Systemd => fmt::layer().with_target(true).without_time().boxed(), + LogStyle::Json => fmt::layer().json().with_target(true).boxed(), + LogStyle::Human | LogStyle::Auto => fmt::layer().with_target(true).boxed(), + }; + // Per-VM files have their own filter so that RUST_LOG, which configures + // the main log, does not drop their status records. + let subscriber = tracing_subscriber::registry() + .with(main_log.with_filter(per_vm_log::main_log_filter(filter))) + .with(per_vm_log.with_filter(LevelFilter::INFO)); + let _ = tracing::subscriber::set_global_default(subscriber); } diff --git a/src/per_vm_log.rs b/src/per_vm_log.rs new file mode 100644 index 00000000..7ea41822 --- /dev/null +++ b/src/per_vm_log.rs @@ -0,0 +1,262 @@ +// SPDX-License-Identifier: BSD-3-Clause +// Copyright (c) 2026 Nutanix, Inc. +// +// Author: Thanos Makatos + +//! Per-VM log files. +//! +//! [`PerVmLog`] is a [`tracing_subscriber::Layer`] that appends every event +//! carrying a `vm` field to that VM's own log file, without repeating the `vm` +//! field. Events on [`TARGET`] belong only in these files and are filtered out +//! of the main log. + +use std::{ + collections::HashSet, + fmt::{self, Write as _}, + fs::{OpenOptions, create_dir_all}, + io::Write as _, + path::PathBuf, + sync::{Arc, Mutex, RwLock}, + time::SystemTime, +}; + +use tracing::{ + Event, Subscriber, + field::{Field, Visit}, +}; +use tracing_subscriber::{EnvFilter, Layer, layer::Context}; + +use crate::util::Path; + +/// Default per-VM log directory. +pub const PER_VM_LOG_DIR: &str = "/var/log/io-thread-controller/vms"; + +/// Target for events that belong only in per-VM log files. +pub const TARGET: &str = "per_vm"; + +/// Build the main log's filter from `directives`, excluding [`TARGET`]. +/// +/// Invalid directives fall back to `info`. +pub fn main_log_filter(directives: &str) -> EnvFilter { + EnvFilter::try_new(directives) + .unwrap_or_else(|_| EnvFilter::new("info")) + .add_directive(format!("{TARGET}=off").parse().expect("valid directive")) +} + +/// Layer writing `vm`-tagged events to per-VM files; clones share state. +#[derive(Clone, Default)] +pub struct PerVmLog { + inner: Arc, +} + +#[derive(Default)] +struct Inner { + /// Destination directory; `None` disables per-VM files. + dir: RwLock>, + /// VM IDs for which per-VM log I/O failures were already reported. + warned: Mutex>, +} + +impl PerVmLog { + /// Write per-VM files under `dir`, or stop writing them with `None`. + pub fn set_dir(&self, dir: Option) { + *self.inner.dir.write().expect("lock poisoned") = dir; + } + + /// Return the log file of `vm`, or `None` while per-VM files are disabled. + pub fn path(&self, vm: &str) -> Option { + let name: String = vm + .chars() + .map(|ch| { + if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.') { + ch + } else { + '_' + } + }) + .collect(); + self.inner + .dir + .read() + .expect("lock poisoned") + .as_ref() + .map(|dir| dir.join(format!("{name}.log"))) + } + + fn append(&self, vm: &str, line: &str) { + let Some(path) = self.path(vm) else { + return; + }; + let result = path.parent().map_or(Ok(()), create_dir_all).and_then(|()| { + let mut file = OpenOptions::new().create(true).append(true).open(&path)?; + writeln!( + file, + "{} {line}", + humantime::format_rfc3339_millis(SystemTime::now()) + ) + }); + if let Err(error) = result + && self + .inner + .warned + .lock() + .expect("lock poisoned") + .insert(vm.to_string()) + { + // Reporting through tracing from inside a layer would re-enter the + // subscriber while it is dispatching this event. + eprintln!( + "WARN failed to write per-VM log path={} error={error}", + path.display() + ); + } + } +} + +impl Layer for PerVmLog { + fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) { + if self.inner.dir.read().expect("lock poisoned").is_none() { + return; + } + let mut record = Record::default(); + event.record(&mut record); + if let Some(vm) = record.vm.take() { + self.append(&vm, &record.line()); + } + } +} + +/// One event split into its VM, message, and remaining `key=value` fields. +#[derive(Default)] +struct Record { + vm: Option, + message: Option, + fields: String, +} + +impl Record { + fn record(&mut self, field: &Field, value: String) { + match field.name() { + "vm" => self.vm = Some(value), + "message" => self.message = Some(value), + name => { + if !self.fields.is_empty() { + self.fields.push(' '); + } + let _ = write!(self.fields, "{name}={value}"); + } + } + } + + /// Render the message, if any, followed by the remaining fields. + fn line(self) -> String { + match self.message { + Some(message) if self.fields.is_empty() => message, + Some(message) => format!("{message} {}", self.fields), + None => self.fields, + } + } +} + +impl Visit for Record { + fn record_str(&mut self, field: &Field, value: &str) { + self.record(field, value.to_string()); + } + + fn record_debug(&mut self, field: &Field, value: &dyn fmt::Debug) { + self.record(field, format!("{value:?}")); + } +} + +#[cfg(test)] +mod tests { + use tracing_subscriber::{filter::LevelFilter, layer::SubscriberExt}; + + use super::*; + + /// Test that only `vm`-tagged events reach the per-VM file, rendered + /// as message then fields and without the `vm` field. + #[test] + fn test_writes_vm_tagged_events_without_vm_field() { + let dir = tempfile::tempdir().unwrap(); + let log = PerVmLog::default(); + log.set_dir(Some(Path::new(dir.path()))); + let subscriber = tracing_subscriber::registry().with(log.clone()); + tracing::subscriber::with_default(subscriber, || { + tracing::info!(target: TARGET, vm = "vm-a", "status=1 2 3"); + tracing::warn!(target: "controller", event = "scale_failed", vm = %"vm-a", target = 4, "oops"); + tracing::info!(target: "controller", event = "vm_added", vm = "vm-a"); + tracing::info!(target: "controller", "no VM here"); + }); + + let contents = std::fs::read_to_string(log.path("vm-a").unwrap()).unwrap(); + let records: Vec<_> = contents + .lines() + .map(|line| line.split_once(' ').unwrap().1) + .collect(); + assert_eq!( + records, + [ + "status=1 2 3", + "oops event=scale_failed target=4", + "event=vm_added" + ] + ); + assert_eq!(std::fs::read_dir(dir.path()).unwrap().count(), 1); + } + + #[derive(Clone, Default)] + struct Buffer(Arc>>); + + impl std::io::Write for Buffer { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + /// Test that per-VM-only events stay out of the main log, and that the + /// main log's level does not drop them from per-VM files. + #[test] + fn test_per_vm_target_bypasses_main_log() { + let dir = tempfile::tempdir().unwrap(); + let log = PerVmLog::default(); + log.set_dir(Some(Path::new(dir.path()))); + let main_log = Buffer::default(); + let writer = main_log.clone(); + let subscriber = tracing_subscriber::registry() + .with( + tracing_subscriber::fmt::layer() + .with_writer(move || writer.clone()) + .with_filter(main_log_filter("warn")), + ) + .with(log.clone().with_filter(LevelFilter::INFO)); + tracing::subscriber::with_default(subscriber, || { + tracing::info!(target: TARGET, vm = "vm-a", "status=1"); + tracing::warn!(target: "controller", vm = "vm-a", "refresh failed"); + }); + + let main_log = String::from_utf8(main_log.0.lock().unwrap().clone()).unwrap(); + assert!(!main_log.contains("status=1"), "{main_log}"); + assert!(main_log.contains("refresh failed"), "{main_log}"); + let contents = std::fs::read_to_string(log.path("vm-a").unwrap()).unwrap(); + assert_eq!(contents.lines().count(), 2, "{contents}"); + } + + /// Test that nothing is written while per-VM files are disabled. + #[test] + fn test_disabled_writes_nothing() { + let dir = tempfile::tempdir().unwrap(); + let log = PerVmLog::default(); + let subscriber = tracing_subscriber::registry().with(log.clone()); + tracing::subscriber::with_default(subscriber, || { + tracing::info!(target: TARGET, vm = "vm-a", "status=1"); + }); + assert!(log.path("vm-a").is_none()); + assert_eq!(std::fs::read_dir(dir.path()).unwrap().count(), 0); + } +} From 6d52160a6d1cf696c4676070bca9df9977842039 Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Thu, 22 Oct 2026 13:59:34 +0000 Subject: [PATCH 2/3] fixup! Write per-VM log through a tracing layer Signed-off-by: Thanos Makatos --- src/controller.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/controller.rs b/src/controller.rs index 920a1771..1485f28c 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -25,6 +25,7 @@ use crate::{ AppliedOutcome, BlockedReason, EngineTickContext, PsiSample, ScaleAction, ScalingEngine, }, instance::{CgroupError, Instance, InstanceStatus, SnapshotLatency}, + per_vm_log, state::{StateError, VmOwnership, VmStateStore}, }; /// Wire-facing container for the D-Bus `GetSnapshot` reply. From d6d9bf32911f79b7360291a5e5f99a744919577e Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Mon, 26 Oct 2026 17:58:35 +0000 Subject: [PATCH 3/3] fixup! Write per-VM log through a tracing layer Signed-off-by: Thanos Makatos --- src/controller.rs | 8 ++++---- src/per_vm_log.rs | 14 +++++++------- 2 files changed, 11 insertions(+), 11 deletions(-) diff --git a/src/controller.rs b/src/controller.rs index 1485f28c..8eeaf0c0 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -258,7 +258,7 @@ impl Controller { status.ownership_classification = classification; drop(status); self.engine.on_instance_added(&instance).await; - tracing::info!(target: per_vm_log::TARGET, vm = %instance.id, event = "vm_added"); + tracing::info!(target: per_vm_log::VM, vm = %instance.id, event = "vm_added"); self.instances.insert(instance.id.clone(), instance); added += 1; } @@ -275,7 +275,7 @@ impl Controller { instance.client.close().await; self.engine.on_instance_removed(&id).await; tracing::info!( - target: per_vm_log::TARGET, + target: per_vm_log::VM, vm = %id, event = "vm_removed", reason = "not_discovered" @@ -570,7 +570,7 @@ impl Controller { self.instances.remove(&id); self.engine.on_instance_removed(&id).await; tracing::info!( - target: per_vm_log::TARGET, + target: per_vm_log::VM, vm = %id, event = "vm_removed", reason = "refresh_failed" @@ -961,7 +961,7 @@ impl Controller { iops_windows: [Option; 3], ) { tracing::info!( - target: per_vm_log::TARGET, + target: per_vm_log::VM, vm = %instance.id, "{}", Self::format_instance_status_line(status, iops_windows) diff --git a/src/per_vm_log.rs b/src/per_vm_log.rs index 7ea41822..e2ffb264 100644 --- a/src/per_vm_log.rs +++ b/src/per_vm_log.rs @@ -7,7 +7,7 @@ //! //! [`PerVmLog`] is a [`tracing_subscriber::Layer`] that appends every event //! carrying a `vm` field to that VM's own log file, without repeating the `vm` -//! field. Events on [`TARGET`] belong only in these files and are filtered out +//! field. Events on [`VM`] belong only in these files and are filtered out //! of the main log. use std::{ @@ -32,15 +32,15 @@ use crate::util::Path; pub const PER_VM_LOG_DIR: &str = "/var/log/io-thread-controller/vms"; /// Target for events that belong only in per-VM log files. -pub const TARGET: &str = "per_vm"; +pub const VM: &str = "per_vm"; -/// Build the main log's filter from `directives`, excluding [`TARGET`]. +/// Build the main log's filter from `directives`, excluding [`VM`]. /// /// Invalid directives fall back to `info`. pub fn main_log_filter(directives: &str) -> EnvFilter { EnvFilter::try_new(directives) .unwrap_or_else(|_| EnvFilter::new("info")) - .add_directive(format!("{TARGET}=off").parse().expect("valid directive")) + .add_directive(format!("{VM}=off").parse().expect("valid directive")) } /// Layer writing `vm`-tagged events to per-VM files; clones share state. @@ -183,7 +183,7 @@ mod tests { log.set_dir(Some(Path::new(dir.path()))); let subscriber = tracing_subscriber::registry().with(log.clone()); tracing::subscriber::with_default(subscriber, || { - tracing::info!(target: TARGET, vm = "vm-a", "status=1 2 3"); + tracing::info!(target: VM, vm = "vm-a", "status=1 2 3"); tracing::warn!(target: "controller", event = "scale_failed", vm = %"vm-a", target = 4, "oops"); tracing::info!(target: "controller", event = "vm_added", vm = "vm-a"); tracing::info!(target: "controller", "no VM here"); @@ -236,7 +236,7 @@ mod tests { ) .with(log.clone().with_filter(LevelFilter::INFO)); tracing::subscriber::with_default(subscriber, || { - tracing::info!(target: TARGET, vm = "vm-a", "status=1"); + tracing::info!(target: VM, vm = "vm-a", "status=1"); tracing::warn!(target: "controller", vm = "vm-a", "refresh failed"); }); @@ -254,7 +254,7 @@ mod tests { let log = PerVmLog::default(); let subscriber = tracing_subscriber::registry().with(log.clone()); tracing::subscriber::with_default(subscriber, || { - tracing::info!(target: TARGET, vm = "vm-a", "status=1"); + tracing::info!(target: VM, vm = "vm-a", "status=1"); }); assert!(log.path("vm-a").is_none()); assert_eq!(std::fs::read_dir(dir.path()).unwrap().count(), 0);