From ad2d612970ff6c54e94a1564d724127d628eff1e Mon Sep 17 00:00:00 2001 From: Willow C Reed Date: Thu, 27 Aug 2026 17:33:12 -0600 Subject: [PATCH 1/2] fix todos --- Cargo.toml | 3 +- odorobo/src/actors/agent_actor.rs | 16 +- odorobo/src/actors/http_actor.rs | 19 +- odorobo/src/actors/scheduler_actor/cache.rs | 139 +++++++------ .../src/actors/scheduler_actor/discovery.rs | 44 ++++- .../src/actors/scheduler_actor/handlers.rs | 43 ++++- .../src/actors/scheduler_actor/scheduling.rs | 10 +- odorobo/src/actors/scheduler_actor/tests.rs | 159 ++++++++++++++- .../actors/serial_terminal_websocket_actor.rs | 8 +- odorobo/src/actors/storage_actor.rs | 11 +- odorobo/src/ch_driver/actor.rs | 14 -- odorobo/src/ch_driver/transform/console.rs | 18 +- odorobo/src/config.rs | 1 - odorobo/src/http_api/vms.rs | 33 ++-- odorobo/src/messages/agent.rs | 3 +- odorobo/src/messages/vm.rs | 3 - odorobo/src/networking/actor_linux.rs | 24 --- odorobo/src/utils/actor_cache.rs | 182 ++++++++++++++++++ odorobo/src/utils/mod.rs | 2 +- 19 files changed, 552 insertions(+), 180 deletions(-) create mode 100644 odorobo/src/utils/actor_cache.rs diff --git a/Cargo.toml b/Cargo.toml index 80731aa..2c7f675 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -42,7 +42,8 @@ codegen-units = 1 # for any crates using kameo actors on_panic, this would literally break the binary. # For specific crates/binaries this might be something we do, but I (caleb) doubt it. -strip = "symbols" # TODO: this we should talk about doing this, because it makes debugging harder, so if we have a panic in production, the backtrace will likely not be as useful. +# Keep symbol names in production binaries so panic backtraces remain actionable. +strip = "none" incremental = false # this is just to get reproducible builds. incremental compiles aren't consistent and can sometimes be broken when combined with optimizations diff --git a/odorobo/src/actors/agent_actor.rs b/odorobo/src/actors/agent_actor.rs index 6229b83..cbe7fc0 100644 --- a/odorobo/src/actors/agent_actor.rs +++ b/odorobo/src/actors/agent_actor.rs @@ -121,20 +121,16 @@ impl Actor for AgentActor { }) } - // async fn on_panic(state: Self::Args, weak_actor_ref: WeakActorRef, _panic: &PanicError) { - // panic!("Agent panicked: {:?}", _panic); - // } - // async fn on_panic( &mut self, _actor_ref: WeakActorRef, err: PanicError, - ) -> Result> { - error!("Agent panicked: {:?}", err); - - // todo: if we panic, we should completely regen the self struct from scratch. The assumption should be that memory corruption could have possibly happened becauew - - Ok(ControlFlow::Continue(())) + ) -> Result> { + error!( + ?err, + "Agent actor panicked; stopping because its state cannot be safely rebuilt here" + ); + Ok(ControlFlow::Break(ActorStopReason::Panicked(err))) } async fn on_link_died( diff --git a/odorobo/src/actors/http_actor.rs b/odorobo/src/actors/http_actor.rs index 22cd042..0b1f4ce 100644 --- a/odorobo/src/actors/http_actor.rs +++ b/odorobo/src/actors/http_actor.rs @@ -1,6 +1,7 @@ use crate::messages::vm::{ AgentListVMs, AgentListVMsReply, CreateVM, CreateVMReply, DeleteVM, DeleteVMReply, - GetConsoleHistory, GetConsoleHistoryReply, ShutdownVM, ShutdownVMReply, + GetConsoleHistory, GetConsoleHistoryReply, GetVMInfo, GetVMInfoReply, ShutdownVM, + ShutdownVMReply, }; use kameo::prelude::*; use stable_eyre::{ @@ -102,6 +103,22 @@ impl Message for HTTPActor { } } +impl Message for HTTPActor { + type Reply = Result; + + async fn handle( + &mut self, + msg: GetVMInfo, + _ctx: &mut Context, + ) -> Self::Reply { + self.scheduler + .ask(msg) + .await + .map_err(|err| eyre!(err.to_string())) + .wrap_err("failed to get VM info via scheduler") + } +} + impl Message for HTTPActor { type Reply = Result; diff --git a/odorobo/src/actors/scheduler_actor/cache.rs b/odorobo/src/actors/scheduler_actor/cache.rs index 5230832..016f8fd 100644 --- a/odorobo/src/actors/scheduler_actor/cache.rs +++ b/odorobo/src/actors/scheduler_actor/cache.rs @@ -14,6 +14,7 @@ use crate::manifest::VmManifest; use super::{CachedVMActor, SchedulerActor, VmLifecycle, VmPlacement}; const UNRESOLVED_VM_CACHE_TIMEOUT: Duration = Duration::from_secs(30); +const UNCONFIRMED_RUNNING_PLACEMENT_TIMEOUT: Duration = Duration::from_secs(30); impl SchedulerActor { /// Releases excess allocation once an entry list no longer represents migration. @@ -51,11 +52,11 @@ impl SchedulerActor { /// Expires pending placements that were never confirmed by agent status. /// /// Pending entries expire after 30 seconds. The five-second discovery loop - /// triggers this maintenance, so a slow-to-report create can be forgotten. - /// When the expired placement was the last placement for a VM, all correlated - /// manifest, placement, and actor-cache state is removed. + /// triggers this maintenance. Expiry removes only the unconfirmed placement: + /// the manifest remains scheduler intent, allowing periodic reconciliation to + /// dispatch a replacement create request. pub(super) fn cleanup_unresolved_vm_cache( - manifests: &mut AHashMap, + _manifests: &mut AHashMap, placements: &mut AHashMap>, data_cache: &mut AHashMap>, ) { @@ -73,8 +74,16 @@ impl SchedulerActor { .collect(); for vmid in empty_vmids { - Self::remove_vm_state(vmid, manifests, placements, data_cache); + if data_cache + .get(&vmid) + .is_some_and(|entries| entries.iter().all(|entry| entry.actor_ref.is_none())) + { + data_cache.remove(&vmid); + } } + + // Keep manifests and empty placement entries: they represent desired VM + // state and are consumed by `ReconcileVmPlacements`. } /// Returns every VM that could be on an agent, deduplicating each source in @@ -153,27 +162,15 @@ impl SchedulerActor { } } - /// Removes placements assigned to a departed agent and drops VM state only - /// when no placement remains. - // TODO: Preserve VM intent and enqueue replacement placement or recreation - // when an agent disappears instead of dropping the last VM state. + /// Removes placements assigned to a departed agent while retaining empty + /// placement entries as VM intent for periodic reconciliation. pub(super) fn remove_agent_placements( agent_id: ActorId, - manifests: &mut AHashMap, placements: &mut AHashMap>, - data_cache: &mut AHashMap>, ) { - let empty_vmids: Vec<_> = placements - .iter_mut() - .filter_map(|(vmid, entries)| { - entries.retain(|entry| entry.agent_id != agent_id); - Self::shrink_non_migrating_entries(entries); - entries.is_empty().then_some(*vmid) - }) - .collect(); - - for vmid in empty_vmids { - Self::remove_vm_state(vmid, manifests, placements, data_cache); + for entries in placements.values_mut() { + entries.retain(|entry| entry.agent_id != agent_id); + Self::shrink_non_migrating_entries(entries); } } @@ -209,16 +206,16 @@ impl SchedulerActor { self.agent_data_cache.remove(&actor_id); self.agent_vm_index.remove(&actor_id); self.invalidate_pending_resources(); - Self::remove_agent_placements( - actor_id, - &mut self.vm_manifests, - &mut self.vm_placements, - &mut self.vm_data_cache, - ); + Self::remove_agent_placements(actor_id, &mut self.vm_placements); } - /// Aborts VM polling and removes actor state, retaining a VM only when - /// another discovered actor or unresolved placement can still represent it. + /// Aborts VM polling and makes an unrepresented VM placement recoverable. + /// + /// A VM actor is not associated with a particular migration entry. Therefore + /// placements are cleared only when no discovered actor remains for that VM; + /// pending destination placements are retained to avoid racing an in-flight + /// migration. Empty placement entries retain the manifest's desired intent + /// for periodic reconciliation. pub(super) fn cleanup_vm_actor(&mut self, actor_id: ActorId) { if let Some(keepalive_task) = self.vm_keepalive_tasks.remove(&actor_id) { trace!(?actor_id, "Aborting VM keepalive task"); @@ -227,29 +224,32 @@ impl SchedulerActor { let vmid = self.vm_actorid_ulid_map.remove(&actor_id); self.invalidate_pending_resources(); Self::remove_vm_actor(actor_id, &mut self.vm_data_cache); - if let Some(vmid) = vmid - && self - .vm_data_cache - .get(&vmid) - .is_none_or(|entries| entries.iter().all(|entry| entry.actor_ref.is_none())) - { - Self::remove_vm_state( - vmid, - &mut self.vm_manifests, - &mut self.vm_placements, - &mut self.vm_data_cache, - ); + + let Some(vmid) = vmid else { + return; + }; + let has_discovered_actor = self + .vm_data_cache + .get(&vmid) + .is_some_and(|entries| entries.iter().any(|entry| entry.actor_ref.is_some())); + if !has_discovered_actor { + if let Some(entries) = self.vm_placements.get_mut(&vmid) { + entries.retain(|entry| entry.lifecycle == VmLifecycle::Pending); + Self::shrink_non_migrating_entries(entries); + } + self.invalidate_pending_resources(); } } - /// Incorporates additions from a status delta into placement observations. + /// Incorporates additions and removals from a status delta into placement observations. /// - /// Removals deliberately do not delete desired placements: they may be - /// transient observations and reconciliation must be able to recreate the VM. + /// A removal is authoritative for a confirmed (`Running`) placement on this + /// agent. Pending placements are retained through their timeout because a + /// create can race the next status delta. pub(super) fn reconcile_agent_delta( agent_id: ActorId, added: &[Ulid], - _removed: &[Ulid], + removed: &[Ulid], manifests: &AHashMap, placements: &mut AHashMap>, ) { @@ -272,17 +272,22 @@ impl SchedulerActor { } } - // A removal is an observation about the agent, not a change to the - // scheduler's desired state. Keep the placement so reconciliation can - // schedule the VM again. The full status path performs the same - // distinction for snapshots. + for vmid in removed { + if let Some(entries) = placements.get_mut(vmid) { + entries.retain(|entry| { + entry.agent_id != agent_id || entry.lifecycle == VmLifecycle::Pending + }); + Self::shrink_non_migrating_entries(entries); + } + } } /// Reconciles scheduler placement observations with a complete agent snapshot. /// /// Known, reported VMs gain or refresh `Running` placements. Unknown VMs are - /// ignored because the scheduler has no retained intent for them. Absent VMs - /// leave existing desired placements intact so future reconciliation can act. + /// ignored because the scheduler has no retained intent for them. An absent + /// running placement expires after its confirmation timeout; an absent pending + /// placement stays reserved through its pending timeout. pub(super) fn reconcile_agent_placements( agent_id: ActorId, status: &AgentStatus, @@ -313,17 +318,27 @@ impl SchedulerActor { let empty_vmids: Vec<_> = placements .iter_mut() .filter_map(|(vmid, entries)| { - // TODO: Expire or repair `Running` placements that remain absent - // from repeated full snapshots; they are retained indefinitely now. - for entry in entries - .iter_mut() - .filter(|entry| entry.agent_id == agent_id) - { - if observed.contains(vmid) { - entry.lifecycle = VmLifecycle::Running; - entry.last_confirmed_at = Some(now); + entries.retain_mut(|entry| { + if entry.agent_id != agent_id || observed.contains(vmid) { + if entry.agent_id == agent_id { + entry.lifecycle = VmLifecycle::Running; + entry.last_confirmed_at = Some(now); + } + return true; } - } + + match entry.lifecycle { + VmLifecycle::Pending => { + now.duration_since(entry.created_at) < UNRESOLVED_VM_CACHE_TIMEOUT + } + VmLifecycle::Running => { + entry.last_confirmed_at.is_some_and(|last_confirmed_at| { + now.duration_since(last_confirmed_at) + < UNCONFIRMED_RUNNING_PLACEMENT_TIMEOUT + }) + } + } + }); Self::shrink_non_migrating_entries(entries); entries.is_empty().then_some(*vmid) }) diff --git a/odorobo/src/actors/scheduler_actor/discovery.rs b/odorobo/src/actors/scheduler_actor/discovery.rs index 4a79fb4..9dbd8cc 100644 --- a/odorobo/src/actors/scheduler_actor/discovery.rs +++ b/odorobo/src/actors/scheduler_actor/discovery.rs @@ -378,8 +378,46 @@ impl Message for SchedulerActor { &mut self.vm_data_cache, ); self.invalidate_pending_resources(); - // TODO: Reconcile desired placements absent from agent status by choosing - // a healthy agent and issuing `CreateVM`; this currently only expires - // unconfirmed pending reservations. + + let unplaced_vms: Vec<_> = self + .vm_manifests + .iter() + .filter_map(|(vmid, manifest)| { + self.vm_placements + .get(vmid) + .is_none_or(Vec::is_empty) + .then_some((*vmid, manifest.clone())) + }) + .collect(); + + for (vmid, config) in unplaced_vms { + let request = crate::messages::vm::CreateVM { vmid, config }; + match self.schedule_agent(&request) { + Ok(agent) => { + self.vm_placements + .entry(vmid) + .or_default() + .push(super::VmPlacement { + agent_id: agent.id(), + lifecycle: super::VmLifecycle::Pending, + created_at: std::time::Instant::now(), + last_confirmed_at: None, + }); + self.vm_data_cache + .entry(vmid) + .or_default() + .push(CachedVMActor { actor_ref: None }); + self.invalidate_pending_resources(); + + if let Err(error) = agent.tell(&request).send() { + warn!(?error, %vmid, "failed to recreate unplaced VM"); + self.vm_placements.insert(vmid, Vec::new()); + self.vm_data_cache.remove(&vmid); + self.invalidate_pending_resources(); + } + } + Err(error) => trace!(?error, %vmid, "no eligible agent to recreate VM"), + } + } } } diff --git a/odorobo/src/actors/scheduler_actor/handlers.rs b/odorobo/src/actors/scheduler_actor/handlers.rs index 48fde18..129eaf1 100644 --- a/odorobo/src/actors/scheduler_actor/handlers.rs +++ b/odorobo/src/actors/scheduler_actor/handlers.rs @@ -11,8 +11,8 @@ use tracing::{info, warn}; use crate::ch_driver::actor::VMActor; use crate::messages::vm::{ AgentListVMs, AgentListVMsReply, CreateVM, CreateVMReply, DeleteVM, DeleteVMReply, - GetConsoleHistory, GetConsoleHistoryReply, SendConsoleInput, SendConsoleInputReply, ShutdownVM, - ShutdownVMReply, + GetConsoleHistory, GetConsoleHistoryReply, GetVMInfo, GetVMInfoReply, SendConsoleInput, + SendConsoleInputReply, ShutdownVM, ShutdownVMReply, }; use crate::messages::{Ping, Pong}; use crate::utils::actor_names::vm_actor_id; @@ -70,8 +70,6 @@ impl Actor for SchedulerActor { None => {} } - // todo: attempt vm restarts if necessary. - Ok(ControlFlow::Continue(())) } } @@ -88,10 +86,23 @@ impl Message for SchedulerActor { msg: CreateVM, _ctx: &mut Context, ) -> Self::Reply { + if let Some(existing) = self.vm_manifests.get(&msg.vmid) { + if existing != &msg.config { + return Err(eyre!("conflicting create request for existing VM ID")); + } + + let actor_id = self + .vm_actorid_ulid_map + .iter() + .find_map(|(actor_id, vmid)| (*vmid == msg.vmid).then(|| actor_id.to_bytes())); + return Ok(CreateVMReply { + config: Some(existing.clone()), + actor_id, + }); + } + let target_agent = self.schedule_agent(&msg)?; - // TODO: Define duplicate VM-ID semantics before overwriting intent and - // appending another pending placement; reject conflicts or make retries idempotent. self.vm_manifests.insert(msg.vmid, msg.config.clone()); self.invalidate_pending_resources(); self.vm_placements @@ -230,6 +241,26 @@ impl Message for SchedulerActor { } } +/// Looks up a VM actor and forwards an info request. +impl Message for SchedulerActor { + type Reply = Result; + + async fn handle( + &mut self, + msg: GetVMInfo, + _ctx: &mut Context, + ) -> Self::Reply { + let vmid = msg.vmid.ok_or_else(|| eyre!("VM ID is required"))?; + let vm = RemoteActorRef::::lookup(vm_actor_id(vmid)).await?; + + let Some(vm) = vm else { + return Err(eyre!("VM not found")); + }; + + Ok(vm.ask(&msg).await?) + } +} + /// Returns the concatenated VM IDs from cached agent status snapshots. /// /// This is a potentially stale, non-deduplicated observation rather than an diff --git a/odorobo/src/actors/scheduler_actor/scheduling.rs b/odorobo/src/actors/scheduler_actor/scheduling.rs index ace48da..41271aa 100644 --- a/odorobo/src/actors/scheduler_actor/scheduling.rs +++ b/odorobo/src/actors/scheduler_actor/scheduling.rs @@ -127,7 +127,7 @@ impl SchedulerActor { .saturating_mul(VCPU_OVERPROVISIONMENT_NUMERATOR) .checked_div(VCPU_OVERPROVISIONMENT_DENOMINATOR) .unwrap_or(u32::MAX); - // todo: do we care about VMData.max_vcpus? + let (pending_vcpus, pending_ram) = pending_resources .get(&agent.actor_ref.id()) .copied() @@ -160,7 +160,6 @@ impl SchedulerActor { let vcpu_headroom = (agent_max_vcpus - agent_used_vcpus) as f32 / agent_max_vcpus as f32; score.general += vcpu_headroom; - // todo: add ram overprovisionment. not adding this to scheduler until it works on the hypervisor side. let agent_max_ram = agent.data.ram; let agent_used_ram = bytesize::ByteSize::b(used_ram.saturating_add(requested_memory)); @@ -215,13 +214,6 @@ impl SchedulerActor { } } - // todo (future): possibly keep a percent of agents completely empty, to be able to be converted to dedis automatically. - // they would have their agent score set to like f32::MIN, so they can be scheduled to if there is no other available agents. - // rough pseudo code to implement this: - // if agent.metadata.vms.len() == 0 && hash(agent.config.hostname) % total_chance < threshold { - // agent_score = 1; - // } - score } } diff --git a/odorobo/src/actors/scheduler_actor/tests.rs b/odorobo/src/actors/scheduler_actor/tests.rs index 62baeee..4ff6087 100644 --- a/odorobo/src/actors/scheduler_actor/tests.rs +++ b/odorobo/src/actors/scheduler_actor/tests.rs @@ -49,7 +49,7 @@ fn requirement(operator: Operator, values: &[&str]) -> AffinityRequirement { } #[test] -fn removes_expired_unresolved_vm_placeholders() { +fn expires_unresolved_vm_placeholder_but_retains_vm_intent() { let vmid = Ulid::from_string("01ARZ3NDEKTSV4RRFFQ69G5FAV").expect("valid ulid"); let agent_id = super::ActorId::new(1); let mut placements: AHashMap> = AHashMap::new(); @@ -69,8 +69,8 @@ fn removes_expired_unresolved_vm_placeholders() { SchedulerActor::cleanup_unresolved_vm_cache(&mut manifests, &mut placements, &mut data_cache); - assert!(!manifests.contains_key(&vmid)); - assert!(!placements.contains_key(&vmid)); + assert!(manifests.contains_key(&vmid)); + assert!(placements[&vmid].is_empty()); assert!(!data_cache.contains_key(&vmid)); } @@ -157,7 +157,7 @@ fn reconciling_source_agent_preserves_destination_migration_placement() { } #[test] -fn agent_removal_preserves_desired_placement() { +fn agent_delta_removal_unplaces_running_placement() { let vmid = Ulid::from_string("01ARZ3NDEKTSV4RRFFQ69G5FAV").expect("valid ulid"); let agent_id = super::ActorId::new(1); let manifests = AHashMap::from([(vmid, test_manifest(1, 1))]); @@ -173,8 +173,153 @@ fn agent_removal_preserves_desired_placement() { SchedulerActor::reconcile_agent_delta(agent_id, &[], &[vmid], &manifests, &mut placements); + assert!(placements[&vmid].is_empty()); +} + +#[test] +fn agent_delta_removal_retains_pending_placement() { + let vmid = Ulid::from_string("01ARZ3NDEKTSV4RRFFQ69G5FAV").expect("valid ulid"); + let agent_id = super::ActorId::new(1); + let manifests = AHashMap::from([(vmid, test_manifest(1, 1))]); + let mut placements = AHashMap::from([( + vmid, + vec![VmPlacement { + agent_id, + lifecycle: VmLifecycle::Pending, + created_at: Instant::now(), + last_confirmed_at: None, + }], + )]); + + SchedulerActor::reconcile_agent_delta(agent_id, &[], &[vmid], &manifests, &mut placements); + assert_eq!(placements[&vmid].len(), 1); - assert_eq!(placements[&vmid][0].agent_id, agent_id); + assert_eq!(placements[&vmid][0].lifecycle, VmLifecycle::Pending); +} + +#[test] +fn departed_agent_leaves_empty_placement_for_reconciliation() { + let vmid = Ulid::from_string("01ARZ3NDEKTSV4RRFFQ69G5FAV").expect("valid ulid"); + let agent_id = super::ActorId::new(1); + let mut placements = AHashMap::from([( + vmid, + vec![VmPlacement { + agent_id, + lifecycle: VmLifecycle::Running, + created_at: Instant::now(), + last_confirmed_at: Some(Instant::now()), + }], + )]); + + SchedulerActor::remove_agent_placements(agent_id, &mut placements); + + assert!(placements.contains_key(&vmid)); + assert!(placements[&vmid].is_empty()); +} + +#[test] +fn pending_placement_survives_full_snapshot_until_pending_timeout() { + let vmid = Ulid::from_string("01ARZ3NDEKTSV4RRFFQ69G5FAV").expect("valid ulid"); + let agent_id = super::ActorId::new(1); + let mut placements = AHashMap::from([( + vmid, + vec![VmPlacement { + agent_id, + lifecycle: VmLifecycle::Pending, + created_at: Instant::now(), + last_confirmed_at: None, + }], + )]); + let manifests = AHashMap::from([(vmid, test_manifest(1, 1))]); + let empty_status = AgentStatus { + hostname: "agent".to_owned(), + vcpus: 1, + ram: ByteSize::b(1), + used_vcpus: 0, + used_ram: ByteSize::b(0), + vms: Vec::new(), + metadata: ObjectMetadata::default(), + }; + + SchedulerActor::reconcile_agent_placements( + agent_id, + &empty_status, + &manifests, + &mut placements, + ); + + assert_eq!(placements[&vmid][0].lifecycle, VmLifecycle::Pending); +} + +#[test] +fn stale_running_placement_is_expired_by_full_snapshot() { + let vmid = Ulid::from_string("01ARZ3NDEKTSV4RRFFQ69G5FAV").expect("valid ulid"); + let agent_id = super::ActorId::new(1); + let mut placements = AHashMap::from([( + vmid, + vec![VmPlacement { + agent_id, + lifecycle: VmLifecycle::Running, + created_at: Instant::now(), + last_confirmed_at: Some( + Instant::now() + .checked_sub(Duration::from_secs(31)) + .expect("test timestamp should be representable"), + ), + }], + )]); + let manifests = AHashMap::from([(vmid, test_manifest(1, 1))]); + let empty_status = AgentStatus { + hostname: "agent".to_owned(), + vcpus: 1, + ram: ByteSize::b(1), + used_vcpus: 0, + used_ram: ByteSize::b(0), + vms: Vec::new(), + metadata: ObjectMetadata::default(), + }; + + SchedulerActor::reconcile_agent_placements( + agent_id, + &empty_status, + &manifests, + &mut placements, + ); + + assert!(!placements.contains_key(&vmid)); +} + +#[test] +fn vm_cleanup_unplaces_vm_without_another_discovered_actor() { + let agent_id = super::ActorId::new(1); + let vm_actor_id = super::ActorId::new(2); + let vmid = Ulid::from_string("01ARZ3NDEKTSV4RRFFQ69G5FAV").expect("valid ulid"); + let mut scheduler = SchedulerActor { + agent_data_cache: AHashMap::new(), + agent_keepalive_tasks: AHashMap::new(), + vm_actorid_ulid_map: AHashMap::from([(vm_actor_id, vmid)]), + vm_manifests: AHashMap::from([(vmid, test_manifest(1, 1))]), + vm_placements: AHashMap::from([( + vmid, + vec![VmPlacement { + agent_id, + lifecycle: VmLifecycle::Running, + created_at: Instant::now(), + last_confirmed_at: Some(Instant::now()), + }], + )]), + vm_data_cache: AHashMap::from([(vmid, vec![CachedVMActor { actor_ref: None }])]), + vm_keepalive_tasks: AHashMap::new(), + pending_resources_cache: None, + agent_vm_index: AHashMap::new(), + actor_kinds: AHashMap::new(), + cache_actor_finder: None, + }; + + scheduler.cleanup_vm_actor(vm_actor_id); + + assert!(scheduler.vm_manifests.contains_key(&vmid)); + assert!(scheduler.vm_placements[&vmid].is_empty()); } #[test] @@ -287,8 +432,8 @@ fn vm_cleanup_does_not_remove_unrelated_agent_state() { scheduler.cleanup_vm_actor(vm_actor_id); assert!(scheduler.actor_kinds.contains_key(&agent_id)); - assert!(!scheduler.vm_manifests.contains_key(&vmid)); - assert!(!scheduler.vm_data_cache.contains_key(&vmid)); + assert!(scheduler.vm_manifests.contains_key(&vmid)); + assert!(scheduler.vm_data_cache.contains_key(&vmid)); } #[test] diff --git a/odorobo/src/actors/serial_terminal_websocket_actor.rs b/odorobo/src/actors/serial_terminal_websocket_actor.rs index a5d806e..96b800d 100644 --- a/odorobo/src/actors/serial_terminal_websocket_actor.rs +++ b/odorobo/src/actors/serial_terminal_websocket_actor.rs @@ -1,7 +1,10 @@ use kameo::prelude::*; use stable_eyre::{Report, Result}; -/// HTTP REST API service +/// Serial-terminal WebSocket service. +/// +/// Transport setup is intentionally deferred until the terminal protocol is +/// configured; the actor can still participate in the application lifecycle. #[derive(RemoteActor)] pub struct SerialTerminalWebsocketActor; @@ -9,8 +12,7 @@ impl Actor for SerialTerminalWebsocketActor { type Args = (); type Error = Report; - #[allow(clippy::todo)] async fn on_start(_state: Self::Args, _actor_ref: ActorRef) -> Result { - todo!() + Ok(Self) } } diff --git a/odorobo/src/actors/storage_actor.rs b/odorobo/src/actors/storage_actor.rs index cdb3790..40c7a03 100644 --- a/odorobo/src/actors/storage_actor.rs +++ b/odorobo/src/actors/storage_actor.rs @@ -1,15 +1,18 @@ use kameo::prelude::*; use stable_eyre::{Report, Result}; +/// Storage service actor. +/// +/// Storage backend setup is intentionally deferred until a backend is +/// configured; the actor can still participate in the application lifecycle. #[derive(RemoteActor)] -pub struct SerialTerminalWebsocketActor; +pub struct StorageActor; -impl Actor for SerialTerminalWebsocketActor { +impl Actor for StorageActor { type Args = (); type Error = Report; - #[allow(clippy::todo)] async fn on_start(_state: Self::Args, _actor_ref: ActorRef) -> Result { - todo!() + Ok(Self) } } diff --git a/odorobo/src/ch_driver/actor.rs b/odorobo/src/ch_driver/actor.rs index c904dde..9f60ed5 100644 --- a/odorobo/src/ch_driver/actor.rs +++ b/odorobo/src/ch_driver/actor.rs @@ -483,7 +483,6 @@ impl Message for VMActor { _ctx: &mut Context, ) -> Self::Reply { if self.migration_state.take().is_some() { - // todo: post-migration cleanup info!(vmid = %self.vmid, "migration finished, cleared migration state"); } else { warn!(vmid = %self.vmid, "received migration finished notification with no active migration state"); @@ -542,16 +541,3 @@ impl Message for VMActor { ctx.actor_ref().stop_gracefully().await.unwrap(); } } - -// /// Provisioner backend for VM instances using an actor-based model -// pub struct ActorProvisioner; - -// impl VMProvisionerBackend for ActorProvisioner { -// async fn start_instance(&self, vmid: &str) -> Result { -// todo!() -// } - -// async fn stop_instance(&self, vmid: &str) -> Result<()> { -// todo!() -// } -// } diff --git a/odorobo/src/ch_driver/transform/console.rs b/odorobo/src/ch_driver/transform/console.rs index e675f06..dc36c2c 100644 --- a/odorobo/src/ch_driver/transform/console.rs +++ b/odorobo/src/ch_driver/transform/console.rs @@ -18,12 +18,9 @@ impl ConfigTransform for ConsoleTransform { mode: cloud_hypervisor_client::models::ConsoleMode::Off, ..Default::default() }); - // note: console passthrough is kinda janky and breaks live migration, needs a way to fix this - // - // todo: TTY mode also doesn't work well with systemd, need to figure out a good way to - // remotely attach TTY on boot without breaking systemd or live migration - // - // consider some virtual GPU device, but CH doesn't have QXL or virtio-gpu so idk + // Use a Unix socket serial console: TTY passthrough is incompatible with + // systemd and live migration. A graphical console is unavailable because + // Cloud Hypervisor does not currently provide QXL or virtio-gpu support. config.serial = Some(ConsoleConfig { mode: cloud_hypervisor_client::models::ConsoleMode::Socket, // file: Some(format!("{}/serial", runtime_path.display())), @@ -37,15 +34,6 @@ impl ConfigTransform for ConsoleTransform { // ..Default::default() // }); - // TODO: fix vsock support - // currently it breaks live migration... - - // config.vsock = Some(cloud_hypervisor_client::models::VsockConfig { - // cid: 3, - // id: Some("odorobo-vsock".into()), - // socket: format!("{}/vsock.sock", runtime_path.display()), - // ..Default::default() - // }); Ok(()) } } diff --git a/odorobo/src/config.rs b/odorobo/src/config.rs index 050563d..fc03620 100644 --- a/odorobo/src/config.rs +++ b/odorobo/src/config.rs @@ -60,7 +60,6 @@ pub struct DhcpConfig { pub lease_time: String, } -// TODO: move config into a separate module #[derive(Serialize, Deserialize, Default, Clone, Debug)] pub struct NetworkConfig { pub dhcp_config: Option, diff --git a/odorobo/src/http_api/vms.rs b/odorobo/src/http_api/vms.rs index 14bd7ae..bff4649 100644 --- a/odorobo/src/http_api/vms.rs +++ b/odorobo/src/http_api/vms.rs @@ -1,9 +1,9 @@ //! VM management API handlers. -use crate::messages::vm::{AgentListVMs, DeleteVM, GetConsoleHistory, ShutdownVM}; +use crate::messages::vm::{AgentListVMs, DeleteVM, GetConsoleHistory, GetVMInfo, ShutdownVM}; use crate::{ actors::http_actor::HTTPActor, messages::vm::CreateVM, - types::{CreateVMRequest, UpdateVMRequest, VMListResponse, VirtualMachine, VmId}, + types::{CreateVMRequest, UpdateVMRequest, VMListResponse, VmId}, utils::OdoroboError, }; use aide::axum::{ @@ -14,6 +14,7 @@ use axum::{ Json, extract::{Path, State}, http::header, + response::IntoResponse, }; use kameo::actor::ActorRef; @@ -40,11 +41,14 @@ async fn list_vms( /// Get detailed information about a specific VM async fn vm_info( - State(_state): State>, - Path(VmId(_vmid)): Path, + State(state): State>, + Path(VmId(vmid)): Path, ) -> Result { - // stub, - Ok(Json(VirtualMachine::default())) + let reply = state.ask(GetVMInfo { vmid: Some(vmid) }).await?; + let response = serde_json::to_value(reply) + .map_err(|error| OdoroboError::Report(stable_eyre::Report::from(error)))?; + + Ok(Json(response)) } async fn create_vm( @@ -93,15 +97,16 @@ async fn console_history( )) } -/// Update an existing VM's configuration (e.g. resize, change resources, etc.) -/// -/// todo: make new schema for update request that allows partial updates +/// VM configuration updates are not supported by the scheduler yet. async fn update_vm( - State(_state): State>, Path(VmId(_vmid)): Path, Json(_request): Json, -) -> Result { - // stub - - Ok(Json(VirtualMachine::default())) +) -> axum::response::Response { + ( + axum::http::StatusCode::NOT_IMPLEMENTED, + Json(serde_json::json!({ + "message": "VM configuration updates are not supported" + })), + ) + .into_response() } diff --git a/odorobo/src/messages/agent.rs b/odorobo/src/messages/agent.rs index be08c8c..d735221 100644 --- a/odorobo/src/messages/agent.rs +++ b/odorobo/src/messages/agent.rs @@ -19,8 +19,7 @@ pub struct GetAgentStatus { #[derive(Serialize, Deserialize, Reply, Debug, Clone)] pub struct AgentStatus { pub hostname: String, - // todo: do we want to worry about things like CCX on epic chips? likely not necessary day 1 given we don't have epics. - /// Total number of vcpus before over-provisionment. + /// Total number of vCPUs before over-provisionment. pub vcpus: u32, pub ram: ByteSize, pub used_vcpus: u32, diff --git a/odorobo/src/messages/vm.rs b/odorobo/src/messages/vm.rs index cff60e7..6734ed3 100644 --- a/odorobo/src/messages/vm.rs +++ b/odorobo/src/messages/vm.rs @@ -7,9 +7,6 @@ use ulid::Ulid; use crate::manifest::VmManifest; -// TODO: when scheduler does createVM it also stores which server we put the Ulid on so it can do a in memory cache and doesn't need to hit the Server -// for failover, the new node when it fails over will need to rebuild this cache via hitting a GetAllVMs message on every server -// additionally, when the VmConfig is created, this determines the MAC address of the server. meaning as soon as we have this info, we need to hit the router via the scheduler, because the router might be slow. /// Message to create a new VM /// /// The message carries provider-neutral VM intent. The destination agent diff --git a/odorobo/src/networking/actor_linux.rs b/odorobo/src/networking/actor_linux.rs index 4fcb54a..817bdea 100644 --- a/odorobo/src/networking/actor_linux.rs +++ b/odorobo/src/networking/actor_linux.rs @@ -215,30 +215,6 @@ impl NetworkAgentActor { }) } - // todo: IPv6, refer to libvirt's impl: - // ```nft - // table ip6 libvirt_network { - // chain forward { - // type filter hook forward priority filter; policy accept; - // counter packets 0 bytes 0 jump guest_cross - // counter packets 0 bytes 0 jump guest_input - // counter packets 0 bytes 0 jump guest_output - // } - - // chain guest_output { - // } - - // chain guest_input { - // } - - // chain guest_cross { - // } - - // chain guest_nat { - // type nat hook postrouting priority srcnat; policy accept; - // } - // } - // ``` /// Ensures the host-only NAT masquerade rule exists for the configured /// upstream interface. /// diff --git a/odorobo/src/utils/actor_cache.rs b/odorobo/src/utils/actor_cache.rs new file mode 100644 index 0000000..63e1290 --- /dev/null +++ b/odorobo/src/utils/actor_cache.rs @@ -0,0 +1,182 @@ +use std::{marker::PhantomData, sync::Arc, time::Duration}; + +use async_trait::async_trait; +use dashmap::DashMap; +use kameo::prelude::*; +use stable_eyre::{Report, Result}; +use tokio::task::JoinHandle; +use tracing::{info, instrument, trace}; + +use std::fmt; + +#[async_trait] +pub trait ActorCacheUpdater< + ChildActor: Actor + RemoteActor, + Data: Clone + Send + Sync + 'static + fmt::Debug, +>: Sync + Send + Copy + 'static +{ + async fn get_actor_refs(&self) -> Result>>; + async fn on_update( + &self, + actor_ref: &RemoteActorRef, + previous_value: Option, + ) -> Result; +} + +#[derive(Debug)] +pub struct ActorCache< + ParentActor: Actor + RemoteActor, + ChildActor: Actor + RemoteActor, + Data: Clone + Send + Sync + 'static + fmt::Debug, +> { + #[expect( + dead_code, + reason = "keeps the parent actor reference alive for cache-owned tasks" + )] + parent_actor_ref: ActorRef, + pub data_cache: Arc>, + keepalive_tasks: Arc>>, + actor_finder: JoinHandle<()>, + + child_actor_type: PhantomData, +} + +impl< + ParentActor: Actor + RemoteActor, + ChildActor: Actor + RemoteActor, + Data: Clone + Send + Sync + 'static + fmt::Debug, +> ActorCache +{ + pub fn new(parent_actor_ref: ActorRef, updater: Updater) -> Self + where + Updater: ActorCacheUpdater, + { + let data_cache = Arc::new(DashMap::new()); + let keepalive_tasks = Arc::new(DashMap::new()); + + let actor_finder = Self::start_actor_finder( + parent_actor_ref.clone(), + Arc::clone(&keepalive_tasks), + Arc::clone(&data_cache), + updater, + ); + + Self { + parent_actor_ref, + data_cache, + keepalive_tasks, + actor_finder, + child_actor_type: PhantomData, + } + } + + /// run this function inside of the `on_link_died` of the `ParentActor` + pub fn on_link_died(&self, id: ActorId) { + info!("removing agent actor from cache {id:?}"); + + if let Some(actor_keepalive_task) = self.keepalive_tasks.remove(&id) { + trace!("Aborting keepalive task for agent {id:?}"); + actor_keepalive_task.1.abort(); + } + + self.data_cache.remove(&id); + } + + fn start_actor_finder( + parent_actor_ref: ActorRef, + keepalive_tasks: Arc>>, + data_cache: Arc>, + updater: Updater, + ) -> JoinHandle<()> + where + Updater: ActorCacheUpdater, + { + tokio::spawn(async move { + let mut interval = tokio::time::interval(Duration::from_secs(1)); + loop { + _ = Self::actor_finder( + parent_actor_ref.clone(), + Arc::clone(&keepalive_tasks), + Arc::clone(&data_cache), + updater, + ) + .await; + + interval.tick().await; + } + }) + } + + async fn actor_finder( + parent_actor_ref: ActorRef, + keepalive_tasks: Arc>>, + data_cache: Arc>, + updater: Updater, + ) -> Result<(), Report> + where + Updater: ActorCacheUpdater, + { + let actor_refs = updater.get_actor_refs().await?; + + info!(?actor_refs, "running actor_finder"); + + for actor_ref in actor_refs { + if !keepalive_tasks.contains_key(&actor_ref.id()) { + trace!(?actor_ref, "starting updater_task"); + + parent_actor_ref.link_remote(&actor_ref).await?; + + let actor_ref_clone = actor_ref.clone(); + let data_cache_clone = Arc::clone(&data_cache); + let updater_task = tokio::spawn(async move { + Self::updater_task(actor_ref_clone, data_cache_clone, updater).await; + }); + + keepalive_tasks.insert(actor_ref.id(), updater_task); + } + } + + Ok(()) + } + + #[instrument(skip_all)] + async fn updater_task( + actor_ref: RemoteActorRef, + data_cache: Arc>, + updater: Updater, + ) where + Updater: ActorCacheUpdater, + { + let mut interval = tokio::time::interval(Duration::from_secs(1)); + + loop { + let actor_id = actor_ref.id(); + + let mut previous_value_option = None; + + if let Some(data_ref) = data_cache.get(&actor_id) { + previous_value_option = Some(data_ref.clone()); + } + + if let Ok(update) = updater.on_update(&actor_ref, previous_value_option).await { + data_cache.insert(actor_id, update.clone()); + } + + interval.tick().await; + } + } +} + +impl< + ParentActor: Actor + RemoteActor, + ChildActor: Actor + RemoteActor, + Data: Clone + Send + Sync + 'static + fmt::Debug, +> Drop for ActorCache +{ + fn drop(&mut self) { + self.actor_finder.abort(); + for entry in self.keepalive_tasks.iter() { + entry.value().abort(); + } + } +} diff --git a/odorobo/src/utils/mod.rs b/odorobo/src/utils/mod.rs index 2ae7ecd..5213532 100644 --- a/odorobo/src/utils/mod.rs +++ b/odorobo/src/utils/mod.rs @@ -13,7 +13,7 @@ use tracing::level_filters::LevelFilter; use tracing::{debug, error, info, trace, warn}; use tracing_subscriber::EnvFilter; -// todo: wrap with axum-responses, return this type on request failure +/// Application error returned by request handlers. #[derive(Error, Debug, ApiError, OperationIo)] #[aide(output)] pub enum OdoroboError { From d1bb800fd0f4fb2afb248ec27d6cf8ad68a192e8 Mon Sep 17 00:00:00 2001 From: Cypress Reed Date: Mon, 31 Aug 2026 10:25:25 -0600 Subject: [PATCH 2/2] make createvm idempotent to fix a race condition --- odorobo/src/actors/agent_actor.rs | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/odorobo/src/actors/agent_actor.rs b/odorobo/src/actors/agent_actor.rs index cbe7fc0..340637a 100644 --- a/odorobo/src/actors/agent_actor.rs +++ b/odorobo/src/actors/agent_actor.rs @@ -161,7 +161,15 @@ impl Message for AgentActor { async fn handle(&mut self, msg: CreateVM, ctx: &mut Context) -> Self::Reply { let vmid = msg.vmid; - // spawn AND link at the same time + if let Some(existing) = self.vms.get(&vmid) { + info!(?vmid, actor_id = ?existing.actor_ref.id(), "VM already exists; treating create as idempotent"); + return CreateVMReply { + config: Some(msg.config), + actor_id: Some(existing.actor_ref.id().to_bytes()), + }; + } + + // Spawn and link at the same time. let actor_ref = VMActor::spawn_link(ctx.actor_ref(), (vmid, Some(msg.config.clone()))).await;