From 77448846b987432f9495af82528a0094850d3fea Mon Sep 17 00:00:00 2001 From: Cappy Ishihara Date: Sun, 16 Aug 2026 22:08:03 +0700 Subject: [PATCH 1/2] add console spooling --- justfile | 4 +- odorobo/src/actors/http_actor.rs | 20 ++- odorobo/src/actors/scheduler_actor.rs | 23 +++- odorobo/src/ch_driver/actor.rs | 183 +++++++++++++++++++++++++- odorobo/src/ch_driver/instance.rs | 19 ++- odorobo/src/http_api/vms.rs | 16 ++- odorobo/src/messages/vm.rs | 13 +- 7 files changed, 262 insertions(+), 16 deletions(-) diff --git a/justfile b/justfile index e186dd0..b87e18e 100644 --- a/justfile +++ b/justfile @@ -21,8 +21,8 @@ build_cli: build_debug: cargo build -p odorobo -debug: build_debug - sudo target/debug/odorobo +debug *args: build_debug + sudo target/debug/odorobo {{ args }} install: install_unit install_agent install_ctl diff --git a/odorobo/src/actors/http_actor.rs b/odorobo/src/actors/http_actor.rs index 56a39a3..dcb0194 100644 --- a/odorobo/src/actors/http_actor.rs +++ b/odorobo/src/actors/http_actor.rs @@ -1,6 +1,6 @@ use crate::messages::vm::{ - AgentListVMs, AgentListVMsReply, CreateVM, CreateVMReply, DeleteVM, DeleteVMReply, ShutdownVM, - ShutdownVMReply, + AgentListVMs, AgentListVMsReply, CreateVM, CreateVMReply, DeleteVM, DeleteVMReply, + GetConsoleHistory, GetConsoleHistoryReply, ShutdownVM, ShutdownVMReply, }; use kameo::prelude::*; use stable_eyre::{ @@ -69,6 +69,22 @@ impl Message for HTTPActor { } } +impl Message for HTTPActor { + type Reply = Result; + + async fn handle( + &mut self, + msg: GetConsoleHistory, + _ctx: &mut Context, + ) -> Self::Reply { + self.scheduler + .ask(msg) + .await + .map_err(|err| eyre!(err.to_string())) + .wrap_err("failed to retrieve VM console history via scheduler") + } +} + impl Message for HTTPActor { type Reply = Result; diff --git a/odorobo/src/actors/scheduler_actor.rs b/odorobo/src/actors/scheduler_actor.rs index ee4ef59..132c11e 100644 --- a/odorobo/src/actors/scheduler_actor.rs +++ b/odorobo/src/actors/scheduler_actor.rs @@ -4,8 +4,9 @@ use crate::actors::agent_actor::AgentActor; use crate::ch_driver::actor::VMActor; use crate::messages::agent::{AgentStatus, GetAgentStatus}; use crate::messages::vm::{ - AgentListVMs, AgentListVMsReply, CreateVM, CreateVMReply, DeleteVM, DeleteVMReply, GetVMInfo, - GetVMInfoReply, ShutdownVM, ShutdownVMReply, + AgentListVMs, AgentListVMsReply, CreateVM, CreateVMReply, DeleteVM, DeleteVMReply, + GetConsoleHistory, GetConsoleHistoryReply, GetVMInfo, GetVMInfoReply, ShutdownVM, + ShutdownVMReply, }; use crate::messages::{Ping, Pong}; use crate::utils::actor_cache::ActorCache; @@ -268,6 +269,24 @@ impl Message for SchedulerActor { } } +impl Message for SchedulerActor { + type Reply = Result; + + async fn handle( + &mut self, + msg: GetConsoleHistory, + _ctx: &mut Context, + ) -> Self::Reply { + let vm = RemoteActorRef::::lookup(vm_actor_id(msg.vmid)).await?; + tracing::trace!(?vm, vmid = %msg.vmid, "GetConsoleHistory"); + if let Some(vm) = vm { + Ok(vm.ask(&msg).await?) + } else { + Err(eyre!("VM not found")) + } + } +} + impl Message for SchedulerActor { type Reply = Result; diff --git a/odorobo/src/ch_driver/actor.rs b/odorobo/src/ch_driver/actor.rs index da985a5..9b26f47 100644 --- a/odorobo/src/ch_driver/actor.rs +++ b/odorobo/src/ch_driver/actor.rs @@ -1,6 +1,8 @@ +use std::{collections::VecDeque, sync::Arc}; + use crate::messages::vm::{ - DeleteVM, GetVMInfo, GetVMInfoReply, MigrateVMReceive, MigrateVMReceiveReply, PrepMigration, - ShutdownVM, + DeleteVM, GetConsoleHistory, GetConsoleHistoryReply, GetVMInfo, GetVMInfoReply, + MigrateVMReceive, MigrateVMReceiveReply, PrepMigration, ShutdownVM, }; use crate::{ch_driver::VMInstance, types::VirtualMachine}; use cloud_hypervisor_client::models::{ @@ -9,7 +11,12 @@ use cloud_hypervisor_client::models::{ use kameo::prelude::*; use serde::{Deserialize, Serialize}; use stable_eyre::{Report, Result}; -use tokio::task::JoinHandle; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::{UnixStream, unix::OwnedWriteHalf}, + sync::{Mutex, broadcast}, + task::JoinHandle, +}; use tracing::{debug, error, info, trace, warn}; /// A migration state that holds the listening address and VM config for a migration, @@ -21,6 +28,156 @@ pub struct MigrationState { pub migration_task: Option>, } +const CONSOLE_SPOOL_SIZE: usize = 1024 * 1024; + +/// Bounded serial-console history shared with the task draining the CH socket. +#[derive(Clone)] +pub struct Console { + inner: Arc>, + output: broadcast::Sender>, + writer: Arc>>, +} + +impl Default for Console { + fn default() -> Self { + let (output, _) = broadcast::channel(256); + Self { + inner: Arc::new(Mutex::new(ConsoleBuffer::default())), + output, + writer: Arc::new(Mutex::new(None)), + } + } +} + +#[derive(Default)] +struct ConsoleBuffer { + ring: VecDeque>, + len: usize, +} + +impl Console { + /// Attach to a Cloud Hypervisor serial socket and start spooling its output. + pub async fn attach(socket_path: std::path::PathBuf) -> Result { + let stream = UnixStream::connect(&socket_path).await.map_err(|err| { + Report::msg(format!( + "failed to attach console spool to {}: {err}", + socket_path.display() + )) + })?; + let (mut reader, writer) = stream.into_split(); + let (output, _) = broadcast::channel(256); + let console = Self { + inner: Arc::new(Mutex::new(ConsoleBuffer::default())), + output, + writer: Arc::new(Mutex::new(Some(writer))), + }; + let spool = console.clone(); + tokio::spawn(async move { + let mut buffer = [0_u8; 16 * 1024]; + loop { + match reader.read(&mut buffer).await { + Ok(0) => { + debug!("serial console closed"); + break; + } + Ok(read) => spool.push(buffer[..read].to_vec()).await, + Err(err) => { + warn!(?err, "serial console spool stopped reading"); + break; + } + } + } + }); + Ok(console) + } + + async fn push(&self, chunk: Vec) { + trace!( + bytes = chunk.len(), + output = %String::from_utf8_lossy(&chunk), + "serial console output received" + ); + let _subscribers = self.output.send(chunk.clone()); + let chunk = if chunk.len() > CONSOLE_SPOOL_SIZE { + chunk[chunk.len().saturating_sub(CONSOLE_SPOOL_SIZE)..].to_vec() + } else { + chunk + }; + { + let mut buffer = self.inner.lock().await; + buffer.len = buffer.len.saturating_add(chunk.len()); + buffer.ring.push_back(chunk); + while buffer.len > CONSOLE_SPOOL_SIZE { + let excess = buffer.len.saturating_sub(CONSOLE_SPOOL_SIZE); + if let Some(oldest) = buffer.ring.pop_front() { + if oldest.len() > excess { + buffer.len = buffer.len.saturating_sub(excess); + buffer.ring.push_front(oldest[excess..].to_vec()); + } else { + buffer.len = buffer.len.saturating_sub(oldest.len()); + } + } else { + buffer.len = 0; + break; + } + } + drop(buffer); + } + } + + /// Subscribe to live serial output. Chunks are broadcast without replay. + pub fn subscribe(&self) -> broadcast::Receiver> { + self.output.subscribe() + } + + /// Write input bytes to the guest serial console. + pub async fn write_input(&self, input: &[u8]) -> Result<()> { + { + let mut writer_guard = self.writer.lock().await; + let writer = writer_guard + .as_mut() + .ok_or_else(|| Report::msg("console is not attached"))?; + let result = writer + .write_all(input) + .await + .map_err(|err| Report::msg(format!("failed to write to serial console: {err}"))); + drop(writer_guard); + result + } + } + + /// Return the currently retained serial output, oldest bytes first. + pub async fn history(&self) -> Vec { + let mut history = Vec::new(); + { + let buffer = self.inner.lock().await; + history.reserve(buffer.len); + for chunk in &buffer.ring { + history.extend_from_slice(chunk); + } + drop(buffer); + }; + history + } +} + +#[cfg(test)] +mod tests { + use super::{CONSOLE_SPOOL_SIZE, Console}; + + #[tokio::test] + async fn console_history_is_bounded_to_one_megabyte() { + let console = Console::default(); + console.push(vec![b'a'; CONSOLE_SPOOL_SIZE]).await; + console.push(b"tail".to_vec()).await; + + let history = console.history().await; + assert_eq!(history.len(), CONSOLE_SPOOL_SIZE); + assert_eq!(&history[..4], b"aaaa"); + assert_eq!(&history[CONSOLE_SPOOL_SIZE - 4..], b"tail"); + } +} + #[derive(Clone, Debug, Serialize, Deserialize)] pub struct MigrationFinished; @@ -30,6 +187,7 @@ pub struct VMActor { /// path to the Cloud Hypervisor socket, in /run/odorobo/vms//ch.sock pub vm_instance: VMInstance, pub migration_state: Option, + pub console: Console, } impl Actor for VMActor { @@ -42,6 +200,9 @@ impl Actor for VMActor { let mut vminstance = VMInstance::spawn(&vmid.to_string(), vm_config.map(VmConfig::from), None).await?; + // attach console on startup for spooling + let console = Console::attach(vminstance.console_socket_path()).await?; + // Take the child process out so we can watch for unexpected death. // destroy() handles a missing child_process gracefully. if let Some(mut child_process) = vminstance.take_child_process() { @@ -72,6 +233,7 @@ impl Actor for VMActor { vmid, vm_instance: vminstance, migration_state: None, + console, }) } @@ -149,6 +311,21 @@ impl From for VMInstance { } } +#[remote_message] +impl Message for VMActor { + type Reply = GetConsoleHistoryReply; + + async fn handle( + &mut self, + _msg: GetConsoleHistory, + _ctx: &mut Context, + ) -> Self::Reply { + GetConsoleHistoryReply { + history: self.console.history().await, + } + } +} + #[remote_message] impl Message for VMActor { type Reply = GetVMInfoReply; diff --git a/odorobo/src/ch_driver/instance.rs b/odorobo/src/ch_driver/instance.rs index 29f0f97..c3599f4 100644 --- a/odorobo/src/ch_driver/instance.rs +++ b/odorobo/src/ch_driver/instance.rs @@ -289,15 +289,24 @@ impl VMInstance { /// Returns the PTY path for this VM's serial console by querying the CH API. #[tracing::instrument] pub async fn console_path(&self) -> Result { - trace!("Getting console PTY path from CH API"); + trace!("Getting console path from CH API"); let info = self.info().await?; - let path = - info.config.serial.and_then(|s| s.file).ok_or_else(|| { - eyre!("No serial console PTY path available for {}", self.vm_id()) - })?; + let serial = info + .config + .serial + .ok_or_else(|| eyre!("No serial console configured for {}", self.vm_id()))?; + let path = serial + .file + .or(serial.socket) + .ok_or_else(|| eyre!("No serial console path available for {}", self.vm_id()))?; Ok(PathBuf::from(path)) } + /// Returns the configured UNIX socket path for this VM's serial console. + pub fn console_socket_path(&self) -> PathBuf { + self.runtime_dir().join("console.sock") + } + /// Opens the PTY console device for this VM and returns a connected stream. #[tracing::instrument] pub async fn open_console(&self) -> Result { diff --git a/odorobo/src/http_api/vms.rs b/odorobo/src/http_api/vms.rs index 329244d..9a65d0e 100644 --- a/odorobo/src/http_api/vms.rs +++ b/odorobo/src/http_api/vms.rs @@ -1,5 +1,5 @@ //! VM management API handlers. -use crate::messages::vm::{AgentListVMs, DeleteVM, ShutdownVM}; +use crate::messages::vm::{AgentListVMs, DeleteVM, GetConsoleHistory, ShutdownVM}; use crate::{ actors::http_actor::HTTPActor, messages::vm::CreateVM, @@ -13,6 +13,7 @@ use aide::axum::{ use axum::{ Json, extract::{Path, State}, + http::header, }; use kameo::actor::ActorRef; @@ -24,6 +25,7 @@ pub fn router() -> ApiRouter> { .api_route("/{vmid}", patch(update_vm)) .api_route("/{vmid}", delete(delete_vm)) .api_route("/{vmid}/shutdown", put(shutdown_vm)) + .api_route("/{vmid}/console/history", get(console_history)) } async fn list_vms( @@ -77,6 +79,18 @@ async fn shutdown_vm( Ok(Json(())) } +async fn console_history( + State(state): State>, + Path(VmId(vmid)): Path, +) -> Result { + let reply = state.ask(GetConsoleHistory { vmid }).await?; + + Ok(( + [(header::CONTENT_TYPE, "application/octet-stream")], + reply.history, + )) +} + /// Update an existing VM's configuration (e.g. resize, change resources, etc.) /// /// todo: make new schema for update request that allows partial updates diff --git a/odorobo/src/messages/vm.rs b/odorobo/src/messages/vm.rs index c46d9b2..568b9f8 100644 --- a/odorobo/src/messages/vm.rs +++ b/odorobo/src/messages/vm.rs @@ -91,7 +91,7 @@ pub struct AgentListVMsReply { } /// Get VM info -#[derive(Serialize, Deserialize, Debug)] +#[derive(Serialize, Deserialize, Debug, Clone)] pub struct GetVMInfo { pub vmid: Option, } @@ -101,3 +101,14 @@ pub struct GetVMInfoReply { pub vmid: Ulid, pub config: Option, } + +/// Retrieve the retained serial-console output for a VM. +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct GetConsoleHistory { + pub vmid: Ulid, +} + +#[derive(Serialize, Deserialize, Reply, Debug, Clone)] +pub struct GetConsoleHistoryReply { + pub history: Vec, +} From d21534a6b10cdd09b4cb9a2e787abcf22f10cb29 Mon Sep 17 00:00:00 2001 From: Cappy Ishihara Date: Sun, 16 Aug 2026 23:13:02 +0700 Subject: [PATCH 2/2] implement message type for sending input --- odorobo/src/actors/scheduler_actor.rs | 22 ++++++++++++++++++-- odorobo/src/ch_driver/actor.rs | 29 ++++++++++++++++++++++++++- odorobo/src/messages/vm.rs | 13 ++++++++++++ 3 files changed, 61 insertions(+), 3 deletions(-) diff --git a/odorobo/src/actors/scheduler_actor.rs b/odorobo/src/actors/scheduler_actor.rs index 132c11e..8d5ff49 100644 --- a/odorobo/src/actors/scheduler_actor.rs +++ b/odorobo/src/actors/scheduler_actor.rs @@ -5,8 +5,8 @@ use crate::ch_driver::actor::VMActor; use crate::messages::agent::{AgentStatus, GetAgentStatus}; use crate::messages::vm::{ AgentListVMs, AgentListVMsReply, CreateVM, CreateVMReply, DeleteVM, DeleteVMReply, - GetConsoleHistory, GetConsoleHistoryReply, GetVMInfo, GetVMInfoReply, ShutdownVM, - ShutdownVMReply, + GetConsoleHistory, GetConsoleHistoryReply, GetVMInfo, GetVMInfoReply, SendConsoleInput, + SendConsoleInputReply, ShutdownVM, ShutdownVMReply, }; use crate::messages::{Ping, Pong}; use crate::utils::actor_cache::ActorCache; @@ -287,6 +287,24 @@ impl Message for SchedulerActor { } } +impl Message for SchedulerActor { + type Reply = Result; + + async fn handle( + &mut self, + msg: SendConsoleInput, + _ctx: &mut Context, + ) -> Self::Reply { + let vm = RemoteActorRef::::lookup(vm_actor_id(msg.vmid)).await?; + tracing::trace!(?vm, vmid = %msg.vmid, bytes = msg.input.len(), "SendConsoleInput"); + if let Some(vm) = vm { + Ok(vm.ask(&msg).await?) + } else { + Err(eyre!("VM not found")) + } + } +} + impl Message for SchedulerActor { type Reply = Result; diff --git a/odorobo/src/ch_driver/actor.rs b/odorobo/src/ch_driver/actor.rs index 9b26f47..eebdcaf 100644 --- a/odorobo/src/ch_driver/actor.rs +++ b/odorobo/src/ch_driver/actor.rs @@ -2,7 +2,8 @@ use std::{collections::VecDeque, sync::Arc}; use crate::messages::vm::{ DeleteVM, GetConsoleHistory, GetConsoleHistoryReply, GetVMInfo, GetVMInfoReply, - MigrateVMReceive, MigrateVMReceiveReply, PrepMigration, ShutdownVM, + MigrateVMReceive, MigrateVMReceiveReply, PrepMigration, SendConsoleInput, + SendConsoleInputReply, ShutdownVM, }; use crate::{ch_driver::VMInstance, types::VirtualMachine}; use cloud_hypervisor_client::models::{ @@ -326,6 +327,32 @@ impl Message for VMActor { } } +#[remote_message] +impl Message for VMActor { + type Reply = SendConsoleInputReply; + + async fn handle( + &mut self, + msg: SendConsoleInput, + _ctx: &mut Context, + ) -> Self::Reply { + let written = msg.input.len(); + match self.console.write_input(&msg.input).await { + Ok(()) => SendConsoleInputReply { + written, + error: None, + }, + Err(err) => { + error!(vmid = %self.vmid, ?err, "failed to write to serial console"); + SendConsoleInputReply { + written: 0, + error: Some(err.to_string()), + } + } + } + } +} + #[remote_message] impl Message for VMActor { type Reply = GetVMInfoReply; diff --git a/odorobo/src/messages/vm.rs b/odorobo/src/messages/vm.rs index 568b9f8..dffe4a5 100644 --- a/odorobo/src/messages/vm.rs +++ b/odorobo/src/messages/vm.rs @@ -112,3 +112,16 @@ pub struct GetConsoleHistory { pub struct GetConsoleHistoryReply { pub history: Vec, } + +/// Send raw input bytes to a VM's serial console. +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct SendConsoleInput { + pub vmid: Ulid, + pub input: Vec, +} + +#[derive(Serialize, Deserialize, Reply, Debug, Clone)] +pub struct SendConsoleInputReply { + pub written: usize, + pub error: Option, +}