From 2350066a1458cb37a2467eef02666f279774d53b Mon Sep 17 00:00:00 2001 From: Willow C Reed Date: Thu, 20 Aug 2026 13:55:43 -0600 Subject: [PATCH 01/10] performance pass --- odorobo/src/actors/agent_actor.rs | 68 ++++++---- odorobo/src/actors/scheduler_actor.rs | 122 ++++++++++++------ odorobo/src/ch_driver/instance.rs | 40 +++--- .../src/ch_driver/transform/storage/iscsi.rs | 13 +- .../src/ch_driver/transform/storage/mod.rs | 4 +- .../src/ch_driver/transform/storage/rbd.rs | 51 ++++---- odorobo/src/networking/actor_linux.rs | 8 +- 7 files changed, 191 insertions(+), 115 deletions(-) diff --git a/odorobo/src/actors/agent_actor.rs b/odorobo/src/actors/agent_actor.rs index c3058c1..60102dc 100644 --- a/odorobo/src/actors/agent_actor.rs +++ b/odorobo/src/actors/agent_actor.rs @@ -1,7 +1,6 @@ use crate::{ ch_driver::actor::VMActor, config::Config, - manifest::VmManifest, messages::{ Ping, Pong, agent::{AgentStatus, GetAgentStatus}, @@ -29,13 +28,16 @@ use kameo::error::PanicError; pub struct VMCacheData { actor_ref: ActorRef, - config: VmManifest, + vcpus: u32, + memory_bytes: u64, } #[derive(RemoteActor)] pub struct AgentActor { pub vcpus: u32, pub memory: ByteSize, + used_vcpus: u32, + used_memory_bytes: u64, pub config: Config, pub vms: AHashMap, // pub network_actor: ActorRef, @@ -43,6 +45,24 @@ pub struct AgentActor { } impl AgentActor { + fn insert_vm(&mut self, vmid: Ulid, cache: VMCacheData) { + let vcpus = cache.vcpus; + let memory_bytes = cache.memory_bytes; + if let Some(previous) = self.vms.insert(vmid, cache) { + self.used_vcpus = self.used_vcpus.saturating_sub(previous.vcpus); + self.used_memory_bytes = self.used_memory_bytes.saturating_sub(previous.memory_bytes); + } + self.used_vcpus = self.used_vcpus.saturating_add(vcpus); + self.used_memory_bytes = self.used_memory_bytes.saturating_add(memory_bytes); + } + + fn remove_vm(&mut self, vmid: Ulid) -> Option { + let removed = self.vms.remove(&vmid)?; + self.used_vcpus = self.used_vcpus.saturating_sub(removed.vcpus); + self.used_memory_bytes = self.used_memory_bytes.saturating_sub(removed.memory_bytes); + Some(removed) + } + async fn lookup_vm_actor(vmid: Ulid) -> Option> { ActorRef::::lookup(format!("vm:{vmid}")) .await @@ -73,6 +93,8 @@ impl Actor for AgentActor { memory: ByteSize::b(sys.total_memory()), config: args, vms: AHashMap::new(), + used_vcpus: 0, + used_memory_bytes: 0, metadata: ObjectMetadata::default(), }) } @@ -101,7 +123,15 @@ impl Actor for AgentActor { ) -> Result> { warn!("Linked actor {id:?} died with reason {reason:?}"); - self.vms.retain(|_, vm| vm.actor_ref.id() != id); + let removed: Vec<_> = self + .vms + .iter() + .filter(|(_, vm)| vm.actor_ref.id() == id) + .map(|(vmid, _)| *vmid) + .collect(); + for vmid in removed { + self.remove_vm(vmid); + } Ok(ControlFlow::Continue(())) } @@ -119,11 +149,12 @@ impl Message for AgentActor { _ = actor_ref.register(vm_actor_id(vmid)).await; _ = actor_ref.register(VM).await; - self.vms.insert( + self.insert_vm( vmid, VMCacheData { actor_ref: actor_ref.clone(), - config: msg.config.clone(), + vcpus: msg.config.desired.compute.vcpus, + memory_bytes: msg.config.desired.compute.memory_bytes, }, ); @@ -149,11 +180,12 @@ impl Message for AgentActor { _ = actor_ref.register(vm_actor_id(vmid)).await; _ = actor_ref.register(VM).await; - self.vms.insert( + self.insert_vm( vmid, VMCacheData { actor_ref: actor_ref.clone(), - config: msg.config.clone(), + vcpus: msg.config.desired.compute.vcpus, + memory_bytes: msg.config.desired.compute.memory_bytes, }, ); @@ -174,7 +206,7 @@ impl Message for AgentActor { msg: DeleteVM, _ctx: &mut Context, ) -> Self::Reply { - match self.vms.remove(&msg.vmid) { + match self.remove_vm(msg.vmid) { Some(cache_data) => { let res = cache_data.actor_ref.tell(msg.clone()).await; if let Err(err) = res { @@ -302,27 +334,15 @@ impl Message for AgentActor { _msg: GetAgentStatus, _ctx: &mut Context, ) -> Self::Reply { - let vcpus_used_by_vms = self - .vms - .values() - .map(|vm| vm.config.desired.compute.vcpus) - .reduce(u32::saturating_add) - .unwrap_or(0); - - let ram_used_by_vms = self - .vms - .values() - .map(|vm| vm.config.desired.compute.memory_bytes) - .reduce(u64::saturating_add) - .unwrap_or(0); - AgentStatus { hostname: self.config.get_hostname().to_owned(), vcpus: self.vcpus, ram: self.memory, vms: self.vms.keys().copied().collect(), - used_vcpus: vcpus_used_by_vms.saturating_add(self.config.get_reserved_vcpus()), - used_ram: ByteSize::b(ram_used_by_vms), + used_vcpus: self + .used_vcpus + .saturating_add(self.config.get_reserved_vcpus()), + used_ram: ByteSize::b(self.used_memory_bytes), metadata: self.metadata.clone(), } } diff --git a/odorobo/src/actors/scheduler_actor.rs b/odorobo/src/actors/scheduler_actor.rs index 21c7602..8355359 100644 --- a/odorobo/src/actors/scheduler_actor.rs +++ b/odorobo/src/actors/scheduler_actor.rs @@ -16,7 +16,6 @@ use crate::messages::vm::{ SendConsoleInput, SendConsoleInputReply, ShutdownVM, ShutdownVMReply, }; use crate::messages::{Ping, Pong}; -use crate::types::ObjectMetadata; use crate::utils::actor_names::AGENT; use crate::utils::actor_names::VM; use crate::utils::actor_names::vm_actor_id; @@ -97,12 +96,17 @@ pub struct CachedVMActor { pub actor_ref: Option>, } +type MetadataTables<'a> = [( + &'a std::collections::BTreeMap, + &'a std::collections::BTreeMap, +)]; + // todo: we should improve the cache to not have agents and vms send the full data on every update. // I looked at kameo streams to make this better, but they aren't really intended for this kind of long term update use case. // They use rust futures::stream which seems to be more intended for you have an iterator for example that will create data, but not like full on sending messages. // This could likely be done pretty easily by having two get data messages. // Option 1: One that creates a session and sends the full data and then only sends diffs after that. -// Option 2: we can have one that sends full data, and then another that only sends data that changes. +// Option 2: One that sends full data, and then another that only sends data that changes. // Option 2 is easier to write and uses less compute, but uses more network bandwidth. #[derive(RemoteActor)] pub struct SchedulerActor { @@ -177,22 +181,29 @@ impl SchedulerActor { } } - /// takes a list of observed vmids and gives you a set of every vmid that could be on an agent. + /// Returns every VM that could be on an agent, without allocating a hash set. + /// + /// The observed list is normally the dominant source and is already unique; + /// migration placements are checked linearly to preserve deduplication while + /// avoiding a temporary hash table on every affinity evaluation. fn placement_vm_ids( placements: &AHashMap>, agent_id: ActorId, observed: &[Ulid], - ) -> AHashSet { - observed - .iter() - .copied() - .chain(placements.iter().filter_map(|(vmid, entries)| { - entries - .iter() - .any(|entry| entry.agent_id == agent_id) - .then_some(*vmid) - })) - .collect() + ) -> Vec { + let mut vmids = Vec::with_capacity(observed.len()); + vmids.extend_from_slice(observed); + for vmid in placements.iter().filter_map(|(vmid, entries)| { + entries + .iter() + .any(|entry| entry.agent_id == agent_id) + .then_some(vmid) + }) { + if !vmids.contains(vmid) { + vmids.push(*vmid); + } + } + vmids } fn remove_vm_state( @@ -248,11 +259,18 @@ impl SchedulerActor { fn rollback_failed_create( vmid: Ulid, actor_exists: bool, + actor_id: Option, + actor_map: &mut AHashMap, manifests: &mut AHashMap, placements: &mut AHashMap>, data_cache: &mut AHashMap>, ) { if !actor_exists { + if let Some(actor_id) = actor_id + && actor_map.get(&actor_id) == Some(&vmid) + { + actor_map.remove(&actor_id); + } Self::remove_vm_state(vmid, manifests, placements, data_cache); } } @@ -617,26 +635,31 @@ impl SchedulerActor { if !msg.config.desired.placement.affinity.is_empty() { let affinity_rules = &msg.config.desired.placement.affinity; for rule in affinity_rules { - let mut metadata_tables: Vec<&ObjectMetadata> = Vec::with_capacity(1); - - let vm_metadata: Vec = match rule.affinity_type { - AffinityType::VirtualMachine => Self::placement_vm_ids( - &self.vm_placements, - agent.actor_ref.id(), - &agent.data.vms, - ) - .into_iter() - .filter_map(|vmid| self.vm_manifests.get(&vmid)) - .map(|manifest| ObjectMetadata { - labels: manifest.desired.metadata.labels.clone(), - annotations: manifest.desired.metadata.annotations.clone(), - }) - .collect(), - AffinityType::Agent => Vec::new(), - }; - metadata_tables.extend(vm_metadata.iter()); - if matches!(rule.affinity_type, AffinityType::Agent) { - metadata_tables.push(&agent.data.metadata); + let mut metadata_tables = Vec::with_capacity(1); + match rule.affinity_type { + AffinityType::VirtualMachine => { + metadata_tables.extend( + Self::placement_vm_ids( + &self.vm_placements, + agent.actor_ref.id(), + &agent.data.vms, + ) + .into_iter() + .filter_map(|vmid| self.vm_manifests.get(&vmid)) + .map(|manifest| { + ( + &manifest.desired.metadata.labels, + &manifest.desired.metadata.annotations, + ) + }), + ); + } + AffinityType::Agent => { + metadata_tables.push(( + &agent.data.metadata.labels, + &agent.data.metadata.annotations, + )); + } } let follows_rule = evaluate_affinity_rule(&metadata_tables, rule); @@ -716,7 +739,7 @@ fn affinity_delta(strictness: &AffinityStrictness, follows_rule: bool) -> Option } fn evaluate_affinity_rule( - metadata_tables: &[&ObjectMetadata], + metadata_tables: &MetadataTables<'_>, rule: &crate::manifest::AffinityRule, ) -> bool { let mut follows_rule = false; @@ -726,8 +749,8 @@ fn evaluate_affinity_rule( for object_metadata in metadata_tables { let table = match requirement.table { - MetadataTable::Label => &object_metadata.labels, - MetadataTable::Annotation => &object_metadata.annotations, + MetadataTable::Label => object_metadata.0, + MetadataTable::Annotation => object_metadata.1, }; if !evaluate_table_value(table.get(&requirement.key), requirement) { @@ -935,6 +958,7 @@ mod tests { fn failed_create_rolls_back_state_without_an_actor() { let vmid = Ulid::from_string("01ARZ3NDEKTSV4RRFFQ69G5FAV").expect("valid ulid"); let mut manifests = AHashMap::from([(vmid, test_manifest(1, 1))]); + let mut actor_map = AHashMap::new(); let mut placements = AHashMap::from([( vmid, vec![VmPlacement { @@ -949,6 +973,8 @@ mod tests { SchedulerActor::rollback_failed_create( vmid, false, + None, + &mut actor_map, &mut manifests, &mut placements, &mut data_cache, @@ -963,12 +989,15 @@ mod tests { fn failed_create_keeps_state_if_actor_exists() { let vmid = Ulid::from_string("01ARZ3NDEKTSV4RRFFQ69G5FAV").expect("valid ulid"); let mut manifests = AHashMap::from([(vmid, test_manifest(1, 1))]); + let mut actor_map = AHashMap::new(); let mut placements = AHashMap::new(); let mut data_cache = AHashMap::new(); SchedulerActor::rollback_failed_create( vmid, true, + None, + &mut actor_map, &mut manifests, &mut placements, &mut data_cache, @@ -1093,7 +1122,10 @@ mod tests { inverse: true, requirements: vec![requirement(Operator::In, &["frontend"])], }; - assert!(!evaluate_affinity_rule(&[&metadata], &rule)); + assert!(!evaluate_affinity_rule( + &[(&metadata.labels, &metadata.annotations)], + &rule, + )); let non_matching_rule = AffinityRule { strictness: AffinityStrictness::Required, @@ -1407,6 +1439,13 @@ impl Message for SchedulerActor { Self::rollback_failed_create( msg.vmid, actor_exists, + reply.as_ref().ok().and_then(|reply| { + reply + .actor_id + .as_deref() + .and_then(|bytes| ActorId::from_bytes(bytes).ok()) + }), + &mut self.vm_actorid_ulid_map, &mut self.vm_manifests, &mut self.vm_placements, &mut self.vm_data_cache, @@ -1505,7 +1544,12 @@ impl Message for SchedulerActor { _msg: AgentListVMs, _ctx: &mut Context, ) -> Self::Reply { - let mut vms = Vec::new(); + let total_vms = self + .agent_data_cache + .values() + .map(|agent| agent.data.vms.len()) + .sum(); + let mut vms = Vec::with_capacity(total_vms); for agent in self.agent_data_cache.values() { vms.extend_from_slice(agent.data.vms.as_slice()); diff --git a/odorobo/src/ch_driver/instance.rs b/odorobo/src/ch_driver/instance.rs index 2faf8e1..f3eb1af 100644 --- a/odorobo/src/ch_driver/instance.rs +++ b/odorobo/src/ch_driver/instance.rs @@ -9,7 +9,9 @@ use stable_eyre::{ eyre::{Context, eyre}, }; use std::{ - env, fs, + env, + fs::{self, File}, + io::BufWriter, os::unix::fs::OpenOptionsExt, path::{Path, PathBuf}, }; @@ -333,6 +335,13 @@ impl VMInstance { &self.ch_socket_path } + async fn stop_child_after_failed_start(&mut self) { + if let Some(mut child) = self.child_process.take() { + _ = child.start_kill(); + _ = child.wait().await; + } + } + pub async fn info(&self) -> Result { self.conn() .vm_info_get() @@ -400,7 +409,10 @@ impl VMInstance { info!(vm_id = id, "CH socket available"); if let Some(vm_config) = vm_config { info!(boot, ?vm_config, "Creating VM config"); - instance.create_config(vm_config, boot).await?; + if let Err(error) = instance.create_config(vm_config, boot).await { + instance.stop_child_after_failed_start().await; + return Err(error); + } } return Ok(instance); } @@ -410,6 +422,7 @@ impl VMInstance { } } + instance.stop_child_after_failed_start().await; Err(eyre!( "CH socket not available after {} attempts for VM {}", MAX_ATTEMPTS, @@ -494,19 +507,18 @@ impl VMInstance { /// Load desired VM config from disk. pub fn load_config(&self) -> Result { - let config_data = fs::read_to_string(self.config_path()) - .wrap_err(eyre!("Failed to read config file for {}", self.vm_id()))?; - - serde_json::from_str(&config_data) - .wrap_err(eyre!("Failed to parse config JSON for {}", self.vm_id())) + serde_json::from_reader( + File::open(self.config_path()) + .wrap_err(eyre!("Failed to read config file for {}", self.vm_id()))?, + ) + .wrap_err(eyre!("Failed to parse config JSON for {}", self.vm_id())) } /// Save desired VM config to disk. pub fn save_config(&self, config: &models::VmConfig) -> Result<()> { - let config_data = - serde_json::to_string_pretty(config).wrap_err("Failed to serialize config to JSON")?; - - fs::write(self.config_path(), config_data) + let file = File::create(self.config_path()) + .wrap_err(eyre!("Failed to open config file for {}", self.vm_id()))?; + serde_json::to_writer_pretty(BufWriter::new(file), config) .wrap_err(eyre!("Failed to write config file for {}", self.vm_id())) } @@ -519,7 +531,7 @@ impl VMInstance { trace!(vm_id = self.vm_id(), "Creating VM with provided config"); trace!(vm_id = self.vm_id(), "Applying config transforms"); - let mut transformed_config = config.clone(); + let mut transformed_config = config; self.transformer .transform(self.vm_id(), &mut transformed_config) .wrap_err(eyre!( @@ -562,9 +574,7 @@ impl VMInstance { self.vm_id() ))?; - self.hook_manager - .before_boot(self.vm_id(), &config.clone()) - .await?; + self.hook_manager.before_boot(self.vm_id(), &config).await?; Ok(()) } diff --git a/odorobo/src/ch_driver/transform/storage/iscsi.rs b/odorobo/src/ch_driver/transform/storage/iscsi.rs index 28b847a..98f9bc1 100644 --- a/odorobo/src/ch_driver/transform/storage/iscsi.rs +++ b/odorobo/src/ch_driver/transform/storage/iscsi.rs @@ -35,7 +35,7 @@ impl ISCSITarget { Command::new("iscsiadm") .args(["-m", "node", "-T", &self.iqn, "-p", &self.host, "--login"]) - .output() + .status() .await .map_err(|e| eyre!("Failed to execute iscsiadm command: {e}"))?; Ok(self.to_device_path()) @@ -46,7 +46,7 @@ impl ISCSITarget { info!(?self, "Detaching iSCSI target"); Command::new("iscsiadm") .args(["-m", "node", "-T", &self.iqn, "-p", &self.host, "--logout"]) - .output() + .status() .await .map_err(|e| eyre!("Failed to execute iscsiadm command: {e}"))?; Ok(()) @@ -64,12 +64,9 @@ impl From<&Url> for ISCSITarget { let host_ip = uri.host_str().unwrap_or_default().to_owned(); let port = uri.port().unwrap_or(3260); let host = format!("{host_ip}:{port}"); - let path_segments: Vec<&str> = uri - .path_segments() - .map(std::iter::Iterator::collect) - .unwrap_or_default(); - let iqn = path_segments.first().unwrap_or(&"").to_string(); - let lun_str = path_segments.get(1).unwrap_or(&""); + let mut path_segments = uri.path_segments().into_iter().flatten(); + let iqn = path_segments.next().unwrap_or_default().to_owned(); + let lun_str = path_segments.next().unwrap_or_default(); let lun = lun_str .strip_prefix("lun") .unwrap_or(lun_str) diff --git a/odorobo/src/ch_driver/transform/storage/mod.rs b/odorobo/src/ch_driver/transform/storage/mod.rs index 4f06e8c..17bf41a 100644 --- a/odorobo/src/ch_driver/transform/storage/mod.rs +++ b/odorobo/src/ch_driver/transform/storage/mod.rs @@ -112,7 +112,7 @@ impl ConfigTransform for StorageDriverTransformer { }; for disk in disks { - let Some(ref path) = disk.path.clone() else { + let Some(path) = disk.path.as_deref() else { continue; }; @@ -132,7 +132,7 @@ impl ConfigTransform for StorageDriverTransformer { let new_disk_id = uri .clone() .query_pairs_mut() - .append_pair("id", &disk.id.clone().unwrap_or_else(|| "".into())) + .append_pair("id", disk.id.as_deref().unwrap_or("")) .finish() .to_string(); diff --git a/odorobo/src/ch_driver/transform/storage/rbd.rs b/odorobo/src/ch_driver/transform/storage/rbd.rs index c7f90b9..e20a08b 100644 --- a/odorobo/src/ch_driver/transform/storage/rbd.rs +++ b/odorobo/src/ch_driver/transform/storage/rbd.rs @@ -51,21 +51,28 @@ async fn rbd_map_list() -> Result> { fn rbd_lines_list(input: &str) -> Result> { let mut mappings = Vec::new(); for line in input.lines().skip(1) { - let parts: Vec<&str> = line.split_whitespace().collect(); + let mut parts = line.split_whitespace(); + let _id = parts.next(); + let Some(pool) = parts.next() else { + continue; + }; + let Some(field) = parts.next() else { + continue; + }; + let Some(next) = parts.next() else { + continue; + }; + let Some(fifth) = parts.next() else { + continue; + }; + // id pool namespace image snap device // 0 pool foo testimg - /dev/rbd0 - if parts.len() == 6 { - let rbd_path = format!("{}/{}", parts[1], parts[3]); - let device_path = parts[5].to_owned(); - mappings.push((rbd_path, device_path)); - } - if parts.len() == 5 { - // if namespace is empty, it might be omitted from the output, so we need to handle that case as well - // id pool image snap device - // 0 pool testimg - /dev/rbd0 - let rbd_path = format!("{}/{}", parts[1], parts[2]); - let device_path = parts[4].to_owned(); - mappings.push((rbd_path, device_path)); + if let Some(device) = parts.next() { + mappings.push((format!("{pool}/{next}"), device.to_owned())); + } else if fifth.starts_with("/dev/") { + // If namespace is empty, it may be omitted from the output. + mappings.push((format!("{pool}/{field}"), fifth.to_owned())); } } @@ -101,17 +108,16 @@ impl RbdImage { } info!(?rbd_path, "Mapping RBD image to device"); - let output = Command::new("rbd") + let status = Command::new("rbd") .args(rbd_extra_args()) .arg("device") .arg("map") .arg(&rbd_path) - .output() + .status() .await .map_err(|e| eyre!("Failed to execute rbd command: {e}"))?; - if !output.status.success() { - let stderr = String::from_utf8_lossy(&output.stderr); - return Err(eyre!("rbd map failed: {stderr}")); + if !status.success() { + return Err(eyre!("rbd map failed with status {status}")); } Ok(()) } @@ -121,17 +127,16 @@ impl RbdImage { pub async fn unmap(&self) -> Result<()> { let rbd_path = self.rbd_path(); info!(?rbd_path, "Unmapping RBD image"); - let output = Command::new("rbd") + let status = Command::new("rbd") .args(rbd_extra_args()) .arg("device") .arg("unmap") .arg(&rbd_path) - .output() + .status() .await .map_err(|e| eyre!("Failed to execute rbd command: {e}"))?; - if !output.status.success() { - let stderr = String::from_utf8_lossy(&output.stderr); - return Err(eyre!("rbd unmap failed: {stderr}")); + if !status.success() { + return Err(eyre!("rbd unmap failed with status {status}")); } Ok(()) } diff --git a/odorobo/src/networking/actor_linux.rs b/odorobo/src/networking/actor_linux.rs index a4be038..4fcb54a 100644 --- a/odorobo/src/networking/actor_linux.rs +++ b/odorobo/src/networking/actor_linux.rs @@ -363,19 +363,19 @@ impl Actor for NetworkAgentActor { let (connection, handle, _) = rtnetlink::new_connection()?; let netlink_thread = tokio::spawn(connection); - let common = match args.network_mode.clone() { + let common = match &args.network_mode { NetworkMode::HostonlyNat { bridge, subnet, upstream_iface, .. } => NetworkConfigCommon { - bridge, + bridge: bridge.clone(), subnet: subnet.to_string(), - upstream_iface: Some(upstream_iface), + upstream_iface: Some(upstream_iface.clone()), }, NetworkMode::Bridged { bridge, subnet, .. } => NetworkConfigCommon { - bridge, + bridge: bridge.clone(), subnet: subnet.to_string(), upstream_iface: None, }, From f21ccf6e9d504e4716655f2c40486b63494f4d2b Mon Sep 17 00:00:00 2001 From: Willow C Reed Date: Thu, 20 Aug 2026 14:04:07 -0600 Subject: [PATCH 02/10] more perf stuff and benchmarking --- odorobo/Cargo.toml | 4 ++ odorobo/benches/agent_status.rs | 69 +++++++++++++++++++++++++ odorobo/src/actors/agent_actor.rs | 74 ++++++++++++++++++++++++--- odorobo/src/actors/scheduler_actor.rs | 57 ++++++++++++++++----- odorobo/src/lib.rs | 1 + odorobo/src/messages/agent.rs | 61 +++++++++++++++++++++- 6 files changed, 244 insertions(+), 22 deletions(-) create mode 100644 odorobo/benches/agent_status.rs diff --git a/odorobo/Cargo.toml b/odorobo/Cargo.toml index 5dd55fe..17d5a99 100644 --- a/odorobo/Cargo.toml +++ b/odorobo/Cargo.toml @@ -68,5 +68,9 @@ signal-hook = "0.4.4" rtnetlink = "0.23" nftables = { version = "0.6", features = ["tokio"] } +[[bench]] +name = "agent_status" +harness = false + [lints] workspace = true diff --git a/odorobo/benches/agent_status.rs b/odorobo/benches/agent_status.rs new file mode 100644 index 0000000..d0308ca --- /dev/null +++ b/odorobo/benches/agent_status.rs @@ -0,0 +1,69 @@ +use std::hint::black_box; +use std::time::Instant; + +use bytesize::ByteSize; +use odorobo::messages::agent::{AgentStatus, AgentStatusUpdate, apply_status_update}; +use odorobo::types::ObjectMetadata; +use ulid::Ulid; + +fn status(vm_count: usize) -> AgentStatus { + AgentStatus { + hostname: "benchmark-agent".to_owned(), + vcpus: 64, + ram: ByteSize::gb(256), + used_vcpus: u32::try_from(vm_count).expect("benchmark VM count fits in u32"), + used_ram: ByteSize::gb(vm_count as u64), + vms: (0..vm_count).map(|_| Ulid::generate()).collect(), + metadata: ObjectMetadata::default(), + } +} + +fn main() { + let iterations = std::env::var("ITERATIONS") + .ok() + .and_then(|value| value.parse().ok()) + .unwrap_or(10_000usize); + + println!("AgentStatus benchmark; iterations={iterations}"); + println!("Set ITERATIONS to change the sample count."); + + for vm_count in [10, 100, 1_000, 10_000] { + let base = status(vm_count); + let added = vec![Ulid::generate()]; + let removed = vec![base.vms[vm_count / 2]]; + + let full_update = AgentStatusUpdate::Full { + revision: 1, + status: base.clone(), + }; + let delta_update = AgentStatusUpdate::Delta { + revision: 1, + added, + removed, + used_vcpus: base.used_vcpus, + used_ram: base.used_ram, + }; + + let full_payload = serde_json::to_vec(&full_update).expect("full update serializes"); + let delta_payload = serde_json::to_vec(&delta_update).expect("delta update serializes"); + + let start = Instant::now(); + for _ in 0..iterations { + black_box(full_update.clone()); + } + let full_elapsed = start.elapsed(); + + let start = Instant::now(); + for _ in 0..iterations { + let mut applied = base.clone(); + black_box(apply_status_update(&mut applied, delta_update.clone())); + } + let delta_elapsed = start.elapsed(); + + println!( + "vms={vm_count:>5} full={full_elapsed:?} delta={delta_elapsed:?} full_payload={:>8}B delta_payload={:>5}B", + full_payload.len(), + delta_payload.len(), + ); + } +} diff --git a/odorobo/src/actors/agent_actor.rs b/odorobo/src/actors/agent_actor.rs index 60102dc..9999471 100644 --- a/odorobo/src/actors/agent_actor.rs +++ b/odorobo/src/actors/agent_actor.rs @@ -3,7 +3,10 @@ use crate::{ config::Config, messages::{ Ping, Pong, - agent::{AgentStatus, GetAgentStatus}, + agent::{ + AgentStatus, AgentStatusUpdate, GetAgentStatus, MembershipChange, + STATUS_CHANGE_HISTORY_LIMIT, StatusChangeHistory, + }, debug::PanicAgent, vm::{ AgentListVMs, AgentListVMsReply, CreateVM, CreateVMReply, DeleteVM, DeleteVMReply, @@ -38,6 +41,8 @@ pub struct AgentActor { pub memory: ByteSize, used_vcpus: u32, used_memory_bytes: u64, + membership_revision: u64, + status_history: StatusChangeHistory, pub config: Config, pub vms: AHashMap, // pub network_actor: ActorRef, @@ -45,12 +50,26 @@ pub struct AgentActor { } impl AgentActor { + fn record_membership_change(&mut self, vmid: Ulid, added: bool) { + self.membership_revision = self.membership_revision.saturating_add(1); + self.status_history.push_back(MembershipChange { + revision: self.membership_revision, + vmid, + added, + }); + while self.status_history.len() > STATUS_CHANGE_HISTORY_LIMIT { + self.status_history.pop_front(); + } + } + fn insert_vm(&mut self, vmid: Ulid, cache: VMCacheData) { let vcpus = cache.vcpus; let memory_bytes = cache.memory_bytes; if let Some(previous) = self.vms.insert(vmid, cache) { self.used_vcpus = self.used_vcpus.saturating_sub(previous.vcpus); self.used_memory_bytes = self.used_memory_bytes.saturating_sub(previous.memory_bytes); + } else { + self.record_membership_change(vmid, true); } self.used_vcpus = self.used_vcpus.saturating_add(vcpus); self.used_memory_bytes = self.used_memory_bytes.saturating_add(memory_bytes); @@ -58,6 +77,7 @@ impl AgentActor { fn remove_vm(&mut self, vmid: Ulid) -> Option { let removed = self.vms.remove(&vmid)?; + self.record_membership_change(vmid, false); self.used_vcpus = self.used_vcpus.saturating_sub(removed.vcpus); self.used_memory_bytes = self.used_memory_bytes.saturating_sub(removed.memory_bytes); Some(removed) @@ -95,6 +115,8 @@ impl Actor for AgentActor { vms: AHashMap::new(), used_vcpus: 0, used_memory_bytes: 0, + membership_revision: 0, + status_history: StatusChangeHistory::new(), metadata: ObjectMetadata::default(), }) } @@ -327,23 +349,59 @@ impl Message for AgentActor { #[remote_message] #[allow(clippy::unused_async_trait_impl)] impl Message for AgentActor { - type Reply = AgentStatus; + type Reply = AgentStatusUpdate; async fn handle( &mut self, - _msg: GetAgentStatus, + msg: GetAgentStatus, _ctx: &mut Context, ) -> Self::Reply { - AgentStatus { + let used_vcpus = self + .used_vcpus + .saturating_add(self.config.get_reserved_vcpus()); + let used_ram = ByteSize::b(self.used_memory_bytes); + let full_status = || AgentStatus { hostname: self.config.get_hostname().to_owned(), vcpus: self.vcpus, ram: self.memory, vms: self.vms.keys().copied().collect(), - used_vcpus: self - .used_vcpus - .saturating_add(self.config.get_reserved_vcpus()), - used_ram: ByteSize::b(self.used_memory_bytes), + used_vcpus, + used_ram, metadata: self.metadata.clone(), + }; + + if msg.since_revision == 0 + || msg.since_revision > self.membership_revision + || self + .status_history + .front() + .is_some_and(|change| msg.since_revision.saturating_add(1) < change.revision) + { + return AgentStatusUpdate::Full { + revision: self.membership_revision, + status: full_status(), + }; + } + + let mut added = Vec::new(); + let mut removed = Vec::new(); + for change in self + .status_history + .iter() + .filter(|change| change.revision > msg.since_revision) + { + if change.added { + added.push(change.vmid); + } else { + removed.push(change.vmid); + } + } + AgentStatusUpdate::Delta { + revision: self.membership_revision, + added, + removed, + used_vcpus, + used_ram, } } } diff --git a/odorobo/src/actors/scheduler_actor.rs b/odorobo/src/actors/scheduler_actor.rs index 8355359..f3fc68d 100644 --- a/odorobo/src/actors/scheduler_actor.rs +++ b/odorobo/src/actors/scheduler_actor.rs @@ -9,7 +9,7 @@ use crate::ch_driver::actor::VMActor; use crate::manifest::{ AffinityRequirement, AffinityStrictness, AffinityType, MetadataTable, Operator, VmManifest, }; -use crate::messages::agent::{AgentStatus, GetAgentStatus}; +use crate::messages::agent::{AgentStatus, AgentStatusUpdate, GetAgentStatus, apply_status_update}; use crate::messages::vm::{ AgentListVMs, AgentListVMsReply, CreateVM, CreateVMReply, DeleteVM, DeleteVMReply, GetConsoleHistory, GetConsoleHistoryReply, GetVMHeartbeat, GetVMInfo, GetVMInfoReply, @@ -60,7 +60,7 @@ struct VmUpdaterStopped { struct AgentUpdated { actor_id: ActorId, actor_ref: RemoteActorRef, - data: AgentStatus, + update: AgentStatusUpdate, } #[derive(Debug)] @@ -75,6 +75,7 @@ struct ReconcileVmPlacements; pub struct CachedAgentActor { pub actor_ref: RemoteActorRef, pub data: AgentStatus, + pub status_revision: u64, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -469,14 +470,24 @@ impl SchedulerActor { async fn agent_updater_task(scheduler: ActorRef, actor_ref: RemoteActorRef) { let mut interval = tokio::time::interval(Duration::from_secs(1)); + let mut status_revision = 0; let mut fails: u8 = 0; loop { - if let Ok(data) = actor_ref.ask(&GetAgentStatus).await { + if let Ok(update) = actor_ref + .ask(&GetAgentStatus { + since_revision: status_revision, + }) + .await + { + status_revision = match &update { + AgentStatusUpdate::Full { revision, .. } + | AgentStatusUpdate::Delta { revision, .. } => *revision, + }; let send_result = scheduler .tell(AgentUpdated { actor_id: actor_ref.id(), actor_ref: actor_ref.clone(), - data, + update, }) .send() .await @@ -1352,19 +1363,41 @@ impl Message for SchedulerActor { type Reply = (); async fn handle(&mut self, msg: AgentUpdated, _ctx: &mut Context) { + let Some(cached) = self.agent_data_cache.get_mut(&msg.actor_id) else { + if let AgentStatusUpdate::Full { revision, status } = msg.update { + Self::reconcile_agent_placements( + msg.actor_id, + &status, + &self.vm_manifests, + &mut self.vm_placements, + ); + self.agent_data_cache.insert( + msg.actor_id, + CachedAgentActor { + actor_ref: msg.actor_ref, + data: status, + status_revision: revision, + }, + ); + } + return; + }; + + let revision = match &msg.update { + AgentStatusUpdate::Full { revision, .. } + | AgentStatusUpdate::Delta { revision, .. } => *revision, + }; + if revision <= cached.status_revision { + return; + } + cached.status_revision = apply_status_update(&mut cached.data, msg.update); + cached.actor_ref = msg.actor_ref; Self::reconcile_agent_placements( msg.actor_id, - &msg.data, + &cached.data, &self.vm_manifests, &mut self.vm_placements, ); - self.agent_data_cache.insert( - msg.actor_id, - CachedAgentActor { - actor_ref: msg.actor_ref, - data: msg.data, - }, - ); } } diff --git a/odorobo/src/lib.rs b/odorobo/src/lib.rs index 5a950bf..58628df 100644 --- a/odorobo/src/lib.rs +++ b/odorobo/src/lib.rs @@ -1,2 +1,3 @@ pub mod manifest; +pub mod messages; pub mod types; diff --git a/odorobo/src/messages/agent.rs b/odorobo/src/messages/agent.rs index 44f4193..dcd410e 100644 --- a/odorobo/src/messages/agent.rs +++ b/odorobo/src/messages/agent.rs @@ -1,12 +1,18 @@ use bytesize::ByteSize; use kameo::Reply; use serde::{Deserialize, Serialize}; +use std::collections::VecDeque; + use ulid::Ulid; use crate::types::ObjectMetadata; -#[derive(Serialize, Deserialize)] -pub struct GetAgentStatus; +#[derive(Serialize, Deserialize, Debug, Clone, Copy)] +pub struct GetAgentStatus { + /// Membership revision already applied by the caller. Revision zero requests + /// a full snapshot; stale revisions are also answered with a full snapshot. + pub since_revision: u64, +} #[derive(Serialize, Deserialize, Reply, Debug, Clone)] pub struct AgentStatus { @@ -20,3 +26,54 @@ pub struct AgentStatus { pub vms: Vec, pub metadata: ObjectMetadata, } + +#[derive(Serialize, Deserialize, Reply, Debug, Clone)] +pub enum AgentStatusUpdate { + Full { + revision: u64, + status: AgentStatus, + }, + Delta { + revision: u64, + added: Vec, + removed: Vec, + used_vcpus: u32, + used_ram: ByteSize, + }, +} + +#[derive(Debug, Clone)] +pub struct MembershipChange { + pub revision: u64, + pub vmid: Ulid, + pub added: bool, +} + +pub const STATUS_CHANGE_HISTORY_LIMIT: usize = 256; + +pub type StatusChangeHistory = VecDeque; + +pub fn apply_status_update(status: &mut AgentStatus, update: AgentStatusUpdate) -> u64 { + match update { + AgentStatusUpdate::Full { + revision, + status: next, + } => { + *status = next; + revision + } + AgentStatusUpdate::Delta { + revision, + added, + removed, + used_vcpus, + used_ram, + } => { + status.vms.retain(|vmid| !removed.contains(vmid)); + status.vms.extend(added); + status.used_vcpus = used_vcpus; + status.used_ram = used_ram; + revision + } + } +} From 69a0b6825fc7d588c3815e88a8613d6b6eb2192b Mon Sep 17 00:00:00 2001 From: Willow C Reed Date: Thu, 20 Aug 2026 14:10:01 -0600 Subject: [PATCH 03/10] reduce load from heartbeats --- odorobo/src/actors/scheduler_actor.rs | 70 ++++++++++++++++++++++++--- 1 file changed, 62 insertions(+), 8 deletions(-) diff --git a/odorobo/src/actors/scheduler_actor.rs b/odorobo/src/actors/scheduler_actor.rs index f3fc68d..c11c1f6 100644 --- a/odorobo/src/actors/scheduler_actor.rs +++ b/odorobo/src/actors/scheduler_actor.rs @@ -312,6 +312,46 @@ impl SchedulerActor { } } + fn reconcile_agent_delta( + agent_id: ActorId, + added: &[Ulid], + removed: &[Ulid], + manifests: &AHashMap, + placements: &mut AHashMap>, + ) { + let now = Instant::now(); + for vmid in added { + if !manifests.contains_key(vmid) { + continue; + } + let entries = placements.entry(*vmid).or_default(); + if let Some(entry) = entries.iter_mut().find(|entry| entry.agent_id == agent_id) { + entry.lifecycle = VmLifecycle::Running; + entry.last_confirmed_at = Some(now); + } else { + entries.push(VmPlacement { + agent_id, + lifecycle: VmLifecycle::Running, + created_at: now, + last_confirmed_at: Some(now), + }); + } + } + + for vmid in removed { + let Some(entries) = placements.get_mut(vmid) else { + continue; + }; + entries.retain(|entry| { + entry.agent_id != agent_id || entry.lifecycle == VmLifecycle::Pending + }); + Self::shrink_non_migrating_entries(entries); + if entries.is_empty() { + placements.remove(vmid); + } + } + } + fn reconcile_agent_placements( agent_id: ActorId, status: &AgentStatus, @@ -1390,14 +1430,28 @@ impl Message for SchedulerActor { if revision <= cached.status_revision { return; } - cached.status_revision = apply_status_update(&mut cached.data, msg.update); - cached.actor_ref = msg.actor_ref; - Self::reconcile_agent_placements( - msg.actor_id, - &cached.data, - &self.vm_manifests, - &mut self.vm_placements, - ); + if let AgentStatusUpdate::Delta { added, removed, .. } = &msg.update { + let added = added.clone(); + let removed = removed.clone(); + cached.status_revision = apply_status_update(&mut cached.data, msg.update); + cached.actor_ref = msg.actor_ref; + Self::reconcile_agent_delta( + msg.actor_id, + &added, + &removed, + &self.vm_manifests, + &mut self.vm_placements, + ); + } else { + cached.status_revision = apply_status_update(&mut cached.data, msg.update); + cached.actor_ref = msg.actor_ref; + Self::reconcile_agent_placements( + msg.actor_id, + &cached.data, + &self.vm_manifests, + &mut self.vm_placements, + ); + } } } From b8b7696e2e9c29dcf8a616326de7afe8c2509993 Mon Sep 17 00:00:00 2001 From: Willow C Reed Date: Thu, 20 Aug 2026 14:18:11 -0600 Subject: [PATCH 04/10] more perf stuff again --- odorobo/benches/agent_status.rs | 6 +- odorobo/src/actors/agent_actor.rs | 10 ++- odorobo/src/actors/scheduler_actor.rs | 94 +++++++++++++++++++++++++-- odorobo/src/messages/agent.rs | 13 +++- 4 files changed, 113 insertions(+), 10 deletions(-) diff --git a/odorobo/benches/agent_status.rs b/odorobo/benches/agent_status.rs index d0308ca..56c133f 100644 --- a/odorobo/benches/agent_status.rs +++ b/odorobo/benches/agent_status.rs @@ -13,7 +13,11 @@ fn status(vm_count: usize) -> AgentStatus { ram: ByteSize::gb(256), used_vcpus: u32::try_from(vm_count).expect("benchmark VM count fits in u32"), used_ram: ByteSize::gb(vm_count as u64), - vms: (0..vm_count).map(|_| Ulid::generate()).collect(), + vms: { + let mut vms: Vec<_> = (0..vm_count).map(|_| Ulid::generate()).collect(); + vms.sort_unstable(); + vms + }, metadata: ObjectMetadata::default(), } } diff --git a/odorobo/src/actors/agent_actor.rs b/odorobo/src/actors/agent_actor.rs index 9999471..7f7a9d0 100644 --- a/odorobo/src/actors/agent_actor.rs +++ b/odorobo/src/actors/agent_actor.rs @@ -364,7 +364,11 @@ impl Message for AgentActor { hostname: self.config.get_hostname().to_owned(), vcpus: self.vcpus, ram: self.memory, - vms: self.vms.keys().copied().collect(), + vms: { + let mut vms: Vec<_> = self.vms.keys().copied().collect(); + vms.sort_unstable(); + vms + }, used_vcpus, used_ram, metadata: self.metadata.clone(), @@ -383,8 +387,8 @@ impl Message for AgentActor { }; } - let mut added = Vec::new(); - let mut removed = Vec::new(); + let mut added = Vec::with_capacity(self.status_history.len()); + let mut removed = Vec::with_capacity(self.status_history.len()); for change in self .status_history .iter() diff --git a/odorobo/src/actors/scheduler_actor.rs b/odorobo/src/actors/scheduler_actor.rs index c11c1f6..f1de5d1 100644 --- a/odorobo/src/actors/scheduler_actor.rs +++ b/odorobo/src/actors/scheduler_actor.rs @@ -122,6 +122,8 @@ pub struct SchedulerActor { /// this is a vec because a vmid/ulid can be scheduled on multiple boxes simultaneously during migration pub vm_data_cache: AHashMap>, pub vm_keepalive_tasks: AHashMap>, + pending_resources_cache: Option>, + agent_vm_index: AHashMap>, actor_kinds: AHashMap, pub cache_actor_finder: Option>, @@ -189,11 +191,20 @@ impl SchedulerActor { /// avoiding a temporary hash table on every affinity evaluation. fn placement_vm_ids( placements: &AHashMap>, + indexed: Option<&AHashSet>, agent_id: ActorId, observed: &[Ulid], ) -> Vec { - let mut vmids = Vec::with_capacity(observed.len()); + let indexed_len = indexed.map_or(0, |index| index.len()); + let mut vmids = Vec::with_capacity(observed.len().max(indexed_len)); vmids.extend_from_slice(observed); + if let Some(indexed) = indexed { + for vmid in indexed { + if !vmids.contains(vmid) { + vmids.push(*vmid); + } + } + } for vmid in placements.iter().filter_map(|(vmid, entries)| { entries .iter() @@ -282,6 +293,8 @@ impl SchedulerActor { keepalive_task.abort(); } 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, @@ -296,6 +309,7 @@ impl SchedulerActor { keepalive_task.abort(); } 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 @@ -599,13 +613,33 @@ impl SchedulerActor { /// - the cache likely needs to be updated automatically when a new vm is scheduled for info like used resources, because otherwise we have to deal with latency on that data we are using /// and then if someone tries to schedule lets say 10 VMs in a batch, we could end up scheduling them all to the same agent because the metadata hasn't updated. /// - there are a few solutions for this but they all kinda suck, mostly due to also making sure we deal with latency properly. I am ignoring the issue for now. - fn schedule_agent(&self, msg: &CreateVM) -> Result, Report> { - let pending_resources = pending_resources_by_agent(&self.vm_manifests, &self.vm_placements); + fn pending_resources(&mut self) -> &AHashMap { + if self.pending_resources_cache.is_none() { + self.pending_resources_cache = Some(pending_resources_by_agent( + &self.vm_manifests, + &self.vm_placements, + )); + } + self.pending_resources_cache + .as_ref() + .expect("pending resources cache was just initialized") + } + + fn invalidate_pending_resources(&mut self) { + self.pending_resources_cache = None; + } + + fn schedule_agent(&mut self, msg: &CreateVM) -> Result, Report> { + self.pending_resources(); + let pending_resources = self + .pending_resources_cache + .as_ref() + .expect("pending resources cache was just initialized"); let mut best_agent = None; let mut best_score = AgentScore::REJECTED; for agent in self.agent_data_cache.values() { - let score = self.score_agent(msg, agent, &pending_resources); + let score = self.score_agent(msg, agent, pending_resources); if score > best_score { best_agent = Some(agent.actor_ref.clone()); @@ -618,6 +652,32 @@ impl SchedulerActor { best_agent.ok_or_eyre("No valid agents found.") } + #[expect(dead_code, reason = "reserved for a future batch create message")] + fn schedule_agents( + &mut self, + msgs: &[CreateVM], + ) -> Vec, Report>> { + self.pending_resources(); + let pending_resources = self + .pending_resources_cache + .as_ref() + .expect("pending resources cache was just initialized"); + msgs.iter() + .map(|msg| { + let mut best_agent = None; + let mut best_score = AgentScore::REJECTED; + for agent in self.agent_data_cache.values() { + let score = self.score_agent(msg, agent, pending_resources); + if score > best_score { + best_agent = Some(agent.actor_ref.clone()); + best_score = score; + } + } + best_agent.ok_or_eyre("No valid agents found.") + }) + .collect() + } + // this function intentionally only checks against the cache. this has some positives and negatives: // positive: it will never trigger any network requests so its very fast, and having to do network requests for scoring whenever we want to schedule a vm is likely a bad idea // negative: it technically has a delayed view of the cluster, meaning that some things that happened in the future, may not exist yet. so we need to be careful about how this is done so affinity rules are not accidentally broken. mostly this means, if we do anything that could affect the outcome of an affinity rule (ex: network request to an agent), we need to update the cache, before we do the action. @@ -692,6 +752,7 @@ impl SchedulerActor { metadata_tables.extend( Self::placement_vm_ids( &self.vm_placements, + self.agent_vm_index.get(&agent.actor_ref.id()), agent.actor_ref.id(), &agent.data.vms, ) @@ -1079,6 +1140,8 @@ mod tests { )]), 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::from([(agent_id, CachedActorKind::Agent)]), cache_actor_finder: None, }; @@ -1104,6 +1167,8 @@ mod tests { vm_placements: AHashMap::new(), 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::from([(agent_id, CachedActorKind::Agent)]), cache_actor_finder: None, }; @@ -1267,6 +1332,8 @@ impl Actor for SchedulerActor { vm_placements: AHashMap::new(), vm_data_cache: AHashMap::new(), vm_keepalive_tasks: AHashMap::new(), + pending_resources_cache: None, + agent_vm_index: AHashMap::new(), actor_kinds: AHashMap::new(), cache_actor_finder: None, }; @@ -1405,12 +1472,16 @@ impl Message for SchedulerActor { async fn handle(&mut self, msg: AgentUpdated, _ctx: &mut Context) { let Some(cached) = self.agent_data_cache.get_mut(&msg.actor_id) else { if let AgentStatusUpdate::Full { revision, status } = msg.update { + self.agent_vm_index + .insert(msg.actor_id, status.vms.iter().copied().collect()); + self.invalidate_pending_resources(); Self::reconcile_agent_placements( msg.actor_id, &status, &self.vm_manifests, &mut self.vm_placements, ); + self.invalidate_pending_resources(); self.agent_data_cache.insert( msg.actor_id, CachedAgentActor { @@ -1435,6 +1506,16 @@ impl Message for SchedulerActor { let removed = removed.clone(); cached.status_revision = apply_status_update(&mut cached.data, msg.update); cached.actor_ref = msg.actor_ref; + self.agent_vm_index + .entry(msg.actor_id) + .or_default() + .extend(added.iter().copied()); + if let Some(index) = self.agent_vm_index.get_mut(&msg.actor_id) { + for vmid in &removed { + index.remove(vmid); + } + } + self.invalidate_pending_resources(); Self::reconcile_agent_delta( msg.actor_id, &added, @@ -1445,12 +1526,15 @@ impl Message for SchedulerActor { } else { cached.status_revision = apply_status_update(&mut cached.data, msg.update); cached.actor_ref = msg.actor_ref; + self.agent_vm_index + .insert(msg.actor_id, cached.data.vms.iter().copied().collect()); Self::reconcile_agent_placements( msg.actor_id, &cached.data, &self.vm_manifests, &mut self.vm_placements, ); + self.invalidate_pending_resources(); } } } @@ -1480,6 +1564,7 @@ impl Message for SchedulerActor { &mut self.vm_placements, &mut self.vm_data_cache, ); + self.invalidate_pending_resources(); } } @@ -1494,6 +1579,7 @@ impl Message for SchedulerActor { let target_agent = self.schedule_agent(&msg)?; self.vm_manifests.insert(msg.vmid, msg.config.clone()); + self.invalidate_pending_resources(); self.vm_placements .entry(msg.vmid) .or_default() diff --git a/odorobo/src/messages/agent.rs b/odorobo/src/messages/agent.rs index dcd410e..b981b23 100644 --- a/odorobo/src/messages/agent.rs +++ b/odorobo/src/messages/agent.rs @@ -69,8 +69,17 @@ pub fn apply_status_update(status: &mut AgentStatus, update: AgentStatusUpdate) used_vcpus, used_ram, } => { - status.vms.retain(|vmid| !removed.contains(vmid)); - status.vms.extend(added); + for vmid in removed { + if let Ok(index) = status.vms.binary_search(&vmid) { + status.vms.remove(index); + } + } + for vmid in added { + match status.vms.binary_search(&vmid) { + Ok(_) => {} + Err(index) => status.vms.insert(index, vmid), + } + } status.used_vcpus = used_vcpus; status.used_ram = used_ram; revision From fcbbbf705ee27ab939164fa7e371682ba0c7b4d6 Mon Sep 17 00:00:00 2001 From: Willow C Reed Date: Thu, 27 Aug 2026 16:48:49 -0600 Subject: [PATCH 05/10] replace dedup with AHashSet, remove todo comments, preserve VM ordering and placement fallback --- odorobo/src/actors/scheduler_actor.rs | 26 +++++++++++--------------- 1 file changed, 11 insertions(+), 15 deletions(-) diff --git a/odorobo/src/actors/scheduler_actor.rs b/odorobo/src/actors/scheduler_actor.rs index f1de5d1..d81325f 100644 --- a/odorobo/src/actors/scheduler_actor.rs +++ b/odorobo/src/actors/scheduler_actor.rs @@ -102,13 +102,6 @@ type MetadataTables<'a> = [( &'a std::collections::BTreeMap, )]; -// todo: we should improve the cache to not have agents and vms send the full data on every update. -// I looked at kameo streams to make this better, but they aren't really intended for this kind of long term update use case. -// They use rust futures::stream which seems to be more intended for you have an iterator for example that will create data, but not like full on sending messages. -// This could likely be done pretty easily by having two get data messages. -// Option 1: One that creates a session and sends the full data and then only sends diffs after that. -// Option 2: One that sends full data, and then another that only sends data that changes. -// Option 2 is easier to write and uses less compute, but uses more network bandwidth. #[derive(RemoteActor)] pub struct SchedulerActor { pub agent_data_cache: AHashMap, @@ -184,11 +177,8 @@ impl SchedulerActor { } } - /// Returns every VM that could be on an agent, without allocating a hash set. - /// - /// The observed list is normally the dominant source and is already unique; - /// migration placements are checked linearly to preserve deduplication while - /// avoiding a temporary hash table on every affinity evaluation. + /// Returns every VM that could be on an agent, deduplicating each source in + /// constant expected time. fn placement_vm_ids( placements: &AHashMap>, indexed: Option<&AHashSet>, @@ -197,10 +187,16 @@ impl SchedulerActor { ) -> Vec { let indexed_len = indexed.map_or(0, |index| index.len()); let mut vmids = Vec::with_capacity(observed.len().max(indexed_len)); - vmids.extend_from_slice(observed); + let mut seen = AHashSet::with_capacity(observed.len().saturating_add(indexed_len)); + + for vmid in observed { + if seen.insert(*vmid) { + vmids.push(*vmid); + } + } if let Some(indexed) = indexed { for vmid in indexed { - if !vmids.contains(vmid) { + if seen.insert(*vmid) { vmids.push(*vmid); } } @@ -211,7 +207,7 @@ impl SchedulerActor { .any(|entry| entry.agent_id == agent_id) .then_some(vmid) }) { - if !vmids.contains(vmid) { + if seen.insert(*vmid) { vmids.push(*vmid); } } From 8d66b7b2747a4b93a33700062b6fb5733a6aebd0 Mon Sep 17 00:00:00 2001 From: Willow C Reed Date: Thu, 27 Aug 2026 16:52:24 -0600 Subject: [PATCH 06/10] add more comments, fix divergence from changes with manifest --- odorobo/src/actors/scheduler_actor.rs | 95 +++++++++++++++------------ 1 file changed, 54 insertions(+), 41 deletions(-) diff --git a/odorobo/src/actors/scheduler_actor.rs b/odorobo/src/actors/scheduler_actor.rs index d81325f..dc7c578 100644 --- a/odorobo/src/actors/scheduler_actor.rs +++ b/odorobo/src/actors/scheduler_actor.rs @@ -104,21 +104,33 @@ type MetadataTables<'a> = [( #[derive(RemoteActor)] pub struct SchedulerActor { + /// Latest status snapshot for each known agent, used for scheduling decisions. pub agent_data_cache: AHashMap, + /// Heartbeat tasks that refresh the corresponding entry in `agent_data_cache`. pub agent_keepalive_tasks: AHashMap>, + /// Maps discovered VM actor IDs to their canonical VM IDs. pub vm_actorid_ulid_map: AHashMap, - /// Canonical manifest storage; placements and actor entries refer to the VM ID. + /// Canonical VM intent. A manifest remains available while the VM is being + /// reconciled or migrated. pub vm_manifests: AHashMap, - /// A VM may be placed on multiple agents while it is migrating. + /// Desired VM placements. Entries are retained when an agent observation says + /// a VM is absent so reconciliation can restore the desired placement. pub vm_placements: AHashMap>, - /// this is a vec because a vmid/ulid can be scheduled on multiple boxes simultaneously during migration + /// Cached VM actor references, with one entry per concurrent placement during + /// migration. `None` represents a placement whose actor is not discovered yet. pub vm_data_cache: AHashMap>, + /// Heartbeat tasks that refresh the corresponding VM actor cache entry. pub vm_keepalive_tasks: AHashMap>, + /// Cached resource totals for pending placements; invalidated on state changes. pending_resources_cache: Option>, + /// VM membership observed from each agent, used to discover placements without + /// scanning every placement during scheduling. agent_vm_index: AHashMap>, + /// Whether a discovered actor is an agent or VM actor. actor_kinds: AHashMap, + /// Background task that discovers actors and periodically triggers cleanup. pub cache_actor_finder: Option>, } @@ -325,7 +337,7 @@ impl SchedulerActor { fn reconcile_agent_delta( agent_id: ActorId, added: &[Ulid], - removed: &[Ulid], + _removed: &[Ulid], manifests: &AHashMap, placements: &mut AHashMap>, ) { @@ -348,18 +360,10 @@ impl SchedulerActor { } } - for vmid in removed { - let Some(entries) = placements.get_mut(vmid) else { - continue; - }; - entries.retain(|entry| { - entry.agent_id != agent_id || entry.lifecycle == VmLifecycle::Pending - }); - Self::shrink_non_migrating_entries(entries); - if entries.is_empty() { - placements.remove(vmid); - } - } + // 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. } fn reconcile_agent_placements( @@ -392,11 +396,6 @@ impl SchedulerActor { let empty_vmids: Vec<_> = placements .iter_mut() .filter_map(|(vmid, entries)| { - entries.retain(|entry| { - entry.agent_id != agent_id - || entry.lifecycle == VmLifecycle::Pending - || observed.contains(vmid) - }); for entry in entries .iter_mut() .filter(|entry| entry.agent_id == agent_id) @@ -591,24 +590,6 @@ impl SchedulerActor { })); } - /// Determine the best agent to schedule a specific VM creation request to. - /// - /// Rough explanation of the algorithm: - /// Loop through every known agent. - /// Go through a set of rules to determine if the VM can be scheduled on this agent at all, and an affinity score and a general score. - /// - /// Based on these scores, pick the best agent. - /// First the affinity score is used, because these are things the customer specifically wanted. - /// If the affinity score is tied, we use the general score as a tie breaker. - /// The general score uses things like resource utilization to not over load any specific agent. - /// - /// - /// Affinity rules are roughly based on . - /// - /// todo: - /// - the cache likely needs to be updated automatically when a new vm is scheduled for info like used resources, because otherwise we have to deal with latency on that data we are using - /// and then if someone tries to schedule lets say 10 VMs in a batch, we could end up scheduling them all to the same agent because the metadata hasn't updated. - /// - there are a few solutions for this but they all kinda suck, mostly due to also making sure we deal with latency properly. I am ignoring the issue for now. fn pending_resources(&mut self) -> &AHashMap { if self.pending_resources_cache.is_none() { self.pending_resources_cache = Some(pending_resources_by_agent( @@ -625,6 +606,12 @@ impl SchedulerActor { self.pending_resources_cache = None; } + /// Determine the best agent to schedule a specific VM creation request to. + /// + /// The scheduler first filters agents by capacity and affinity requirements, + /// then uses affinity and general resource scores to select the best match. + /// Affinity rules are roughly based on + /// . fn schedule_agent(&mut self, msg: &CreateVM) -> Result, Report> { self.pending_resources(); let pending_resources = self @@ -1058,8 +1045,34 @@ mod tests { let remaining = placements .get(&vmid) .expect("destination placement remains"); - assert_eq!(remaining.len(), 1); - assert_eq!(remaining[0].agent_id, destination_agent); + assert_eq!(remaining.len(), 2); + assert!(remaining.iter().any(|entry| entry.agent_id == source_agent)); + assert!( + remaining + .iter() + .any(|entry| entry.agent_id == destination_agent) + ); + } + + #[test] + fn agent_removal_preserves_desired_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::Running, + created_at: Instant::now(), + last_confirmed_at: Some(Instant::now()), + }], + )]); + + 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); } #[test] From 15467c79ccba402a1204481375ca2660e7cd458e Mon Sep 17 00:00:00 2001 From: Willow C Reed Date: Thu, 27 Aug 2026 18:23:38 -0600 Subject: [PATCH 07/10] allow synchronous scheduler status updates --- odorobo/src/actors/scheduler_actor.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/odorobo/src/actors/scheduler_actor.rs b/odorobo/src/actors/scheduler_actor.rs index dc7c578..8dca596 100644 --- a/odorobo/src/actors/scheduler_actor.rs +++ b/odorobo/src/actors/scheduler_actor.rs @@ -1475,6 +1475,7 @@ impl Message for SchedulerActor { } } +#[allow(clippy::unused_async_trait_impl)] impl Message for SchedulerActor { type Reply = (); From 8b2ccc27de45a53938867b9a3054e643ad00b3af Mon Sep 17 00:00:00 2001 From: Cypress Reed Date: Fri, 28 Aug 2026 11:00:34 -0600 Subject: [PATCH 08/10] fix test failure --- odorobo/src/actors/scheduler_actor.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/odorobo/src/actors/scheduler_actor.rs b/odorobo/src/actors/scheduler_actor.rs index 8dca596..843e00d 100644 --- a/odorobo/src/actors/scheduler_actor.rs +++ b/odorobo/src/actors/scheduler_actor.rs @@ -1258,7 +1258,10 @@ mod tests { inverse: true, requirements: vec![requirement(Operator::In, &["backend"])], }; - assert!(evaluate_affinity_rule(&[&metadata], &non_matching_rule)); + assert!(evaluate_affinity_rule( + &[(&metadata.labels, &metadata.annotations)], + &non_matching_rule, + )); let empty_rule = AffinityRule { strictness: AffinityStrictness::Required, From 893efe028fd43a6dfee51f8e1252d5b19c49a01b Mon Sep 17 00:00:00 2001 From: Cypress Reed Date: Mon, 31 Aug 2026 10:06:50 -0600 Subject: [PATCH 09/10] fix stale scheduler state, avoid full agent snapshots, restore subprocess error diagnostics --- odorobo/src/actors/agent_actor.rs | 2 +- odorobo/src/actors/scheduler_actor.rs | 5 +++++ .../src/ch_driver/transform/storage/iscsi.rs | 16 ++++++++++++---- odorobo/src/ch_driver/transform/storage/rbd.rs | 18 ++++++++++-------- odorobo/src/messages/agent.rs | 6 ++++-- 5 files changed, 32 insertions(+), 15 deletions(-) diff --git a/odorobo/src/actors/agent_actor.rs b/odorobo/src/actors/agent_actor.rs index 7f7a9d0..bf81bf0 100644 --- a/odorobo/src/actors/agent_actor.rs +++ b/odorobo/src/actors/agent_actor.rs @@ -374,7 +374,7 @@ impl Message for AgentActor { metadata: self.metadata.clone(), }; - if msg.since_revision == 0 + if msg.initial || msg.since_revision > self.membership_revision || self .status_history diff --git a/odorobo/src/actors/scheduler_actor.rs b/odorobo/src/actors/scheduler_actor.rs index 843e00d..1de4b01 100644 --- a/odorobo/src/actors/scheduler_actor.rs +++ b/odorobo/src/actors/scheduler_actor.rs @@ -520,11 +520,13 @@ impl SchedulerActor { async fn agent_updater_task(scheduler: ActorRef, actor_ref: RemoteActorRef) { let mut interval = tokio::time::interval(Duration::from_secs(1)); let mut status_revision = 0; + let mut initial_status = true; let mut fails: u8 = 0; loop { if let Ok(update) = actor_ref .ask(&GetAgentStatus { since_revision: status_revision, + initial: initial_status, }) .await { @@ -532,6 +534,7 @@ impl SchedulerActor { AgentStatusUpdate::Full { revision, .. } | AgentStatusUpdate::Delta { revision, .. } => *revision, }; + initial_status = false; let send_result = scheduler .tell(AgentUpdated { actor_id: actor_ref.id(), @@ -1559,12 +1562,14 @@ impl Message for SchedulerActor { self.agent_keepalive_tasks.remove(&msg.actor_id); self.actor_kinds.remove(&msg.actor_id); self.agent_data_cache.remove(&msg.actor_id); + self.agent_vm_index.remove(&msg.actor_id); Self::remove_agent_placements( msg.actor_id, &mut self.vm_manifests, &mut self.vm_placements, &mut self.vm_data_cache, ); + self.invalidate_pending_resources(); } } diff --git a/odorobo/src/ch_driver/transform/storage/iscsi.rs b/odorobo/src/ch_driver/transform/storage/iscsi.rs index 98f9bc1..c62ef7d 100644 --- a/odorobo/src/ch_driver/transform/storage/iscsi.rs +++ b/odorobo/src/ch_driver/transform/storage/iscsi.rs @@ -33,22 +33,30 @@ impl ISCSITarget { // do iscsiadm login to the target, then find the corresponding device path in /dev/disk/by-path info!(?self, "Attaching iSCSI target"); - Command::new("iscsiadm") + let output = Command::new("iscsiadm") .args(["-m", "node", "-T", &self.iqn, "-p", &self.host, "--login"]) - .status() + .output() .await .map_err(|e| eyre!("Failed to execute iscsiadm command: {e}"))?; + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + return Err(eyre!("iscsiadm login failed: {stderr}")); + } Ok(self.to_device_path()) } #[tracing::instrument(skip(self))] pub async fn detach(&self) -> Result<()> { info!(?self, "Detaching iSCSI target"); - Command::new("iscsiadm") + let output = Command::new("iscsiadm") .args(["-m", "node", "-T", &self.iqn, "-p", &self.host, "--logout"]) - .status() + .output() .await .map_err(|e| eyre!("Failed to execute iscsiadm command: {e}"))?; + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + return Err(eyre!("iscsiadm logout failed: {stderr}")); + } Ok(()) } } diff --git a/odorobo/src/ch_driver/transform/storage/rbd.rs b/odorobo/src/ch_driver/transform/storage/rbd.rs index e20a08b..00fd499 100644 --- a/odorobo/src/ch_driver/transform/storage/rbd.rs +++ b/odorobo/src/ch_driver/transform/storage/rbd.rs @@ -108,16 +108,17 @@ impl RbdImage { } info!(?rbd_path, "Mapping RBD image to device"); - let status = Command::new("rbd") + let output = Command::new("rbd") .args(rbd_extra_args()) .arg("device") .arg("map") .arg(&rbd_path) - .status() + .output() .await .map_err(|e| eyre!("Failed to execute rbd command: {e}"))?; - if !status.success() { - return Err(eyre!("rbd map failed with status {status}")); + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + return Err(eyre!("rbd map failed: {stderr}")); } Ok(()) } @@ -127,16 +128,17 @@ impl RbdImage { pub async fn unmap(&self) -> Result<()> { let rbd_path = self.rbd_path(); info!(?rbd_path, "Unmapping RBD image"); - let status = Command::new("rbd") + let output = Command::new("rbd") .args(rbd_extra_args()) .arg("device") .arg("unmap") .arg(&rbd_path) - .status() + .output() .await .map_err(|e| eyre!("Failed to execute rbd command: {e}"))?; - if !status.success() { - return Err(eyre!("rbd unmap failed with status {status}")); + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + return Err(eyre!("rbd unmap failed: {stderr}")); } Ok(()) } diff --git a/odorobo/src/messages/agent.rs b/odorobo/src/messages/agent.rs index b981b23..be08c8c 100644 --- a/odorobo/src/messages/agent.rs +++ b/odorobo/src/messages/agent.rs @@ -9,9 +9,11 @@ use crate::types::ObjectMetadata; #[derive(Serialize, Deserialize, Debug, Clone, Copy)] pub struct GetAgentStatus { - /// Membership revision already applied by the caller. Revision zero requests - /// a full snapshot; stale revisions are also answered with a full snapshot. + /// Membership revision already applied by the caller. pub since_revision: u64, + /// Requests the initial full snapshot. Later requests can use revision zero + /// without forcing a full snapshot when the agent has not changed. + pub initial: bool, } #[derive(Serialize, Deserialize, Reply, Debug, Clone)] From c7e2c943dbef93f4fb91d4389f5f1d30824d3f1e Mon Sep 17 00:00:00 2001 From: Cypress Reed Date: Mon, 31 Aug 2026 10:18:13 -0600 Subject: [PATCH 10/10] fix a delta issue --- odorobo/src/actors/agent_actor.rs | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/odorobo/src/actors/agent_actor.rs b/odorobo/src/actors/agent_actor.rs index bf81bf0..6229b83 100644 --- a/odorobo/src/actors/agent_actor.rs +++ b/odorobo/src/actors/agent_actor.rs @@ -387,19 +387,25 @@ impl Message for AgentActor { }; } - let mut added = Vec::with_capacity(self.status_history.len()); - let mut removed = Vec::with_capacity(self.status_history.len()); + let mut latest_changes = AHashMap::new(); for change in self .status_history .iter() .filter(|change| change.revision > msg.since_revision) { - if change.added { - added.push(change.vmid); + latest_changes.insert(change.vmid, change.added); + } + let mut added = Vec::with_capacity(latest_changes.len()); + let mut removed = Vec::with_capacity(latest_changes.len()); + for (vmid, added_change) in latest_changes { + if added_change { + added.push(vmid); } else { - removed.push(change.vmid); + removed.push(vmid); } } + added.sort_unstable(); + removed.sort_unstable(); AgentStatusUpdate::Delta { revision: self.membership_revision, added,