diff --git a/.zed/debug.json b/.zed/debug.json new file mode 100644 index 0000000..00271f1 --- /dev/null +++ b/.zed/debug.json @@ -0,0 +1,14 @@ +[ + { + "label": "Run manager agent", + "build": { + "command": "cargo", + "args": ["build"] + }, + "sourceLanguages": ["rust"], + "program": "$ZED_WORKTREE_ROOT/target/debug/odorobo", + "args": ["--manager-enabled"], + "request": "launch", + "adapter": "CodeLLDB" + } +] diff --git a/odorobo/src/actors/agent_actor.rs b/odorobo/src/actors/agent_actor.rs index 7a14a9b..5beade8 100644 --- a/odorobo/src/actors/agent_actor.rs +++ b/odorobo/src/actors/agent_actor.rs @@ -317,6 +317,7 @@ impl Message for AgentActor { 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), + metadata: self.metadata.clone(), } } } diff --git a/odorobo/src/actors/scheduler_actor.rs b/odorobo/src/actors/scheduler_actor.rs index ee4ef59..74012dd 100644 --- a/odorobo/src/actors/scheduler_actor.rs +++ b/odorobo/src/actors/scheduler_actor.rs @@ -1,4 +1,7 @@ +use std::cmp::Ordering; use std::ops::ControlFlow; +use std::sync::Arc; +use std::time::Duration; use crate::actors::agent_actor::AgentActor; use crate::ch_driver::actor::VMActor; @@ -8,23 +11,87 @@ use crate::messages::vm::{ GetVMInfoReply, ShutdownVM, ShutdownVMReply, }; use crate::messages::{Ping, Pong}; -use crate::utils::actor_cache::ActorCache; -use crate::utils::actor_cache::ActorCacheUpdater; +use crate::types::AffinityRequirement; +use crate::types::AffinityStrictness; +use crate::types::AffinityType; +use crate::types::MetadataTable; +use crate::types::ObjectMetadata; +use crate::types::Operator; use crate::utils::actor_names::AGENT; use crate::utils::actor_names::VM; use crate::utils::actor_names::vm_actor_id; -use async_trait::async_trait; +use ahash::AHashSet; +use dashmap::DashMap; +use dashmap::mapref::multiple::RefMulti; use kameo::prelude::*; use libp2p::futures::TryStreamExt; use stable_eyre::eyre::OptionExt; use stable_eyre::{Report, Result, eyre::eyre}; -use tracing::info_span; +use tokio::task::JoinHandle; +use tracing::trace; use tracing::{info, warn}; +use ulid::Ulid; +#[derive(Debug, Clone)] +pub struct CachedAgentActor { + pub actor_ref: RemoteActorRef, + pub data: AgentStatus, + /// this is a set of all VMs that may be on an agent. it is used for rules such as affinity to make sure we don't schedule things in ways that arent allowed + /// We don't know for a fact these VMs are scheduled due to latency and boot up delay, but they may be scheduled. + pub extended_vm_set: AHashSet, +} + +#[derive(Debug, Clone)] +pub struct CachedVMActor { + pub actor_ref: Option>, + /// The agent this VM was scheduled on. Used by the updater task to clean up + /// the agent's `extended_vm_set` if the VM actor dies before it's linked. + pub scheduled_agent_id: ActorId, + pub data: GetVMInfoReply, +} + +// todo: i dont like the way this cache is setup. I think we may need to change it later, but it is hard to figure out what the optimal solution is without doing it at least once. +// especially when we haven't fully made decisions about some other things. +// +// 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 is easier to write and uses less compute, but uses more network bandwidth. #[derive(RemoteActor)] pub struct SchedulerActor { - pub agent_actor_cache: ActorCache, - pub vm_actor_cache: ActorCache, + pub agent_data_cache: Arc>, + pub agent_keepalive_tasks: Arc>>, + + // todo: we might need a better way to store this. + // we 100% need a way to store vms even if we don't know their actorid (ex: actor hasn't been started or is shutdown) + // we also might want to be able to store them without a ulid, possibly + // so we might need a vec of vms and then to just store maps/indexes of actorid and ulid to vector index + // and then like a freelist or something. + // i dont really love that option either though cause it feels overkill. + // maybe we sure just be using a proper database entirely? + // idk. will figure it out later. + // + // new related problem: i just realized vmid, actorid pairs dont have to be unique. + // if a vm is migrating from one actor to another, there might be two actors with the same vmid. + // + // additional context (05/05/2026): we almost may need a way to store them without a ulid, due to how CH migration works. + // the question becomes if we want to abstract CH migration away entirely from the scheduler. + // we could also possibly ignore it for the non-HA scheduler. + // I (caleb) want to ask cappy (and possibly Lea) about these problems. + // + // the best solution for at least some of this is almost certainly having an external reliable DB (such as etcd) to store some of these things permanently. + // we will need that specifically for what VMs are supposed to be running, because if a large percentage of the cluster goes down, including the manager, we need a way to recover. + // and i dont think leaving that on dashboard which could have high latency is a good idea. + // alternatively we could have the other manager nodes try to keep track of that data, but i think we are going to run into issues with keeping the state consistent between all nodes. + // we may need to make some architecture designs about db consistency vs uptime vs speed in that situation, and im not doing that on my own. + pub vm_actorid_ulid_map: Arc>, + pub vm_data_cache: Arc>, + pub vm_keepalive_tasks: Arc>>, + + pub cache_actor_finder: Option>, } // todo: this might need to be a runtime thing but this makes it easy to write for now and could easily be switched out later. @@ -32,182 +99,479 @@ static VCPU_OVERPROVISIONMENT_NUMERATOR: u32 = 2; static VCPU_OVERPROVISIONMENT_DENOMINATOR: u32 = 1; impl SchedulerActor { - #[expect( - dead_code, - reason = "scheduler lookup helper reserved for explicit placement by actor id" - )] - fn lookup_by_actor_id(&self, actor_id: &ActorId) -> Option> { - self.agent_actor_cache - .data_cache + #[expect(dead_code, reason = "reserved for explicit placement by actor id")] + fn lookup_agent_by_actor_id(&self, actor_id: &ActorId) -> Option> { + self.agent_data_cache .get(actor_id) .map(|data| data.actor_ref.clone()) } - #[expect( - dead_code, - reason = "scheduler lookup helper reserved for explicit placement by hostname" - )] - fn lookup_by_hostname(&self, hostname: &str) -> Option> { - self.agent_actor_cache - .data_cache + #[expect(dead_code, reason = "reserved for explicit placement by hostname")] + fn lookup_agent_by_hostname(&self, hostname: &str) -> Option> { + self.agent_data_cache .iter() - .find(|data| data.metadata.hostname == hostname) + .find(|data| data.data.hostname == hostname) .map(|data| data.actor_ref.clone()) } - /// current scheduling algo info: - /// this is vaguely based on / - /// when a vm is attempted to be scheduled, we loop through every agent and score it based on some rules - /// there are hard rules that will simply throw out an agent entirely. - /// otherwise, we take whatever the best agent we can find is. - /// - /// additionally, because caleb is way too performance brained, he used integer math for the entire scoring algorithm just so we didnt have to convert to floats. - fn schedule_agent(&self, _msg: &CreateVM) -> Result, Report> { - let mut best_agent = None; - let mut best_agent_score = 0u32; - - // todo: this arguably could be done as map-reduce. is that better? - let span = info_span!("schedule_agent"); - span.in_scope(|| { - for agent in self.agent_actor_cache.data_cache.iter() { - let mut agent_score = 0u32; - - let agent_max_vcpus = agent - .metadata - .vcpus - .checked_mul(VCPU_OVERPROVISIONMENT_NUMERATOR) - .and_then(|value| value.checked_div(VCPU_OVERPROVISIONMENT_DENOMINATOR)) - .unwrap_or(u32::MAX); - - if agent.metadata.used_vcpus >= agent_max_vcpus { - continue; - } + // someone should likely give caleb a firm talking to about code duplication due to this section, but things are just different enough that trying to make them one function requires usage of a lot of generics which feels even worse. so i dont know what to do. cappy please fix. i hate this. + async fn vm_actor_finder( + parent_actor_ref: RemoteActorRef, + vm_actorid_ulid_map: Arc>, + data_cache: Arc>, + keepalive_tasks: Arc>>, + agent_data_cache: Arc>, + ) -> Result<(), Report> { + trace!("running vm_actor_finder"); + + let mut vm_actor_stream = RemoteActorRef::::lookup_all(VM); + + while let Some(vm_actor) = vm_actor_stream.try_next().await? { + if !keepalive_tasks.contains_key(&vm_actor.id()) { + trace!(?vm_actor, "starting vm_updater_task"); + + parent_actor_ref.link_remote(&vm_actor).await?; + + let vm_actor_id = vm_actor.id(); + + let vm_actorid_ulid_map_clone = Arc::clone(&vm_actorid_ulid_map); + let data_cache_clone = Arc::clone(&data_cache); + let agent_data_cache_clone = Arc::clone(&agent_data_cache); + let updater_task = tokio::spawn(async move { + Self::vm_updater_task( + vm_actor, + vm_actorid_ulid_map_clone, + data_cache_clone, + agent_data_cache_clone, + ) + .await; + }); + + keepalive_tasks.insert(vm_actor_id, updater_task); + } + } + + Ok(()) + } - agent_score = agent_score.saturating_add( - agent_max_vcpus - .saturating_sub(agent.metadata.used_vcpus) - .checked_mul(1024) - .and_then(|value| value.checked_div(agent_max_vcpus)) - .unwrap_or(u32::MAX), + async fn vm_updater_task( + actor_ref: RemoteActorRef, + vm_actorid_ulid_map: Arc>, + data_cache: Arc>, + agent_data_cache: Arc>, + ) { + let mut interval = tokio::time::interval(Duration::from_secs(1)); + let mut fails: u8 = 0; + loop { + if let Ok(data) = actor_ref.ask(&GetVMInfo { vmid: None }).await { + let vmid = data.vmid; + + vm_actorid_ulid_map.insert(actor_ref.id(), vmid); // should we be doing this on every loop? idk. but we at least need to do it on the first iteration given we don't know the mapping before that + + data_cache.insert( + vmid, + CachedVMActor { + actor_ref: Some(actor_ref.clone()), + // preserve the original scheduled_agent_id if an entry already exists + // (e.g. from CreateVM::handle inserting it before the actor was reachable), + // otherwise default to the vm actor's own id (won't match any agent entry, + // so no accidental cleanup of wrong agent). + scheduled_agent_id: data_cache + .get(&vmid) + .map_or_else(|| actor_ref.id(), |e| e.scheduled_agent_id), + data, + }, ); - // todo: add ram overprovisionment. not adding this to scheduler until it works on the hypervisor side. - let agent_max_ram = agent.metadata.ram; + fails = 0; + } else { + fails = fails.saturating_add(1); + } + + if fails > 5 { + warn!( + ?actor_ref, + "can no longer reach vm actor, cleaning up cache entries" + ); - if agent.metadata.used_ram >= agent_max_ram { - continue; + let vmid = vm_actorid_ulid_map + .remove(&actor_ref.id()) + .map(|(_, vmid)| vmid); + + // if the actor was never reachable, there's no vm_actorid_ulid_map entry. + // try to recover the vmid from the data_cache by scanning for a stale entry + // that matches this actor_ref. + let vmid = vmid.or_else(|| { + data_cache.iter().find_map(|entry| { + (entry.actor_ref.as_ref().map(RemoteActorRef::id) == Some(actor_ref.id())) + .then(|| *entry.key()) + }) + }); + + if let Some(vmid) = vmid { + if let Some(cached_vm) = data_cache.get(&vmid) { + agent_data_cache.alter(&cached_vm.scheduled_agent_id, |_, mut v| { + v.extended_vm_set.remove(&vmid); + v + }); + } + + data_cache.remove(&vmid); } - agent_score = agent_score.saturating_add( - u32::try_from( - agent_max_ram - .as_u64() - .saturating_sub(agent.metadata.used_ram.as_u64()) - .checked_mul(1024) - .and_then(|value| value.checked_div(agent_max_ram.as_u64())) - .unwrap_or(u64::MAX), - ) - .unwrap_or(u32::MAX), - ); + return; + } + + interval.tick().await; + } + } + + async fn agent_actor_finder( + parent_actor_ref: RemoteActorRef, + data_cache: Arc>, + keepalive_tasks: Arc>>, + ) -> Result<(), Report> { + trace!("running agent_actor_finder"); + + let mut agent_actor_stream = RemoteActorRef::::lookup_all(AGENT); - // todo: https://kubernetes.io/docs/concepts/scheduling-eviction/assign-pod-node/ + while let Some(agent_actor) = agent_actor_stream.try_next().await? { + if !keepalive_tasks.contains_key(&agent_actor.id()) { + trace!(?agent_actor, "starting agent_updater_task"); - // 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 1, so they can be scheduled to if there is no other avaliable agents. - // rough pseudo code to implement this: - // if agent.metadata.vms.len() == 0 && hash(agent.config.hostname) % total_chance < threshold { - // agent_score = 1; - // } + parent_actor_ref.link_remote(&agent_actor).await?; - info!(agent=?agent.value(), score=agent_score); + let agent_actor_id = agent_actor.id(); - if agent_score > best_agent_score { - best_agent = Some(agent.actor_ref.clone()); - best_agent_score = agent_score; + let data_cache_clone = Arc::clone(&data_cache); + let updater_task = tokio::spawn(async move { + Self::agent_updater_task(agent_actor, data_cache_clone).await; + }); + + keepalive_tasks.insert(agent_actor_id, updater_task); + } + } + + Ok(()) + } + + async fn agent_updater_task( + actor_ref: RemoteActorRef, + data_cache: Arc>, + ) { + let mut interval = tokio::time::interval(Duration::from_secs(1)); + let mut fails: u8 = 0; + loop { + if let Ok(data) = actor_ref.ask(&GetAgentStatus).await { + if data_cache.contains_key(&actor_ref.id()) { + data_cache.alter(&actor_ref.id(), |_, mut v| { + v.data = data; + + v.extended_vm_set.extend(v.data.vms.iter()); + + v + }); + } else { + data_cache.insert( + actor_ref.id(), + CachedAgentActor { + actor_ref: actor_ref.clone(), + data: data.clone(), + extended_vm_set: data.vms.iter().copied().collect(), + }, + ); } + + fails = 0; + } else { + fails = fails.saturating_add(1); } - }); - best_agent.ok_or_eyre("No valid agents found.") + if fails > 5 { + warn!( + ?actor_ref, + "can no longer reach agent actor, stopping updater" + ); + data_cache.remove(&actor_ref.id()); + return; + } + + interval.tick().await; + } } -} -#[derive(Copy, Clone)] -struct AgentActorCacheUpdater; + fn start_actor_finder(&mut self, actor_ref: RemoteActorRef) { + let agent_data_cache_arc_clone = Arc::clone(&self.agent_data_cache); + let agent_keepalive_tasks_arc_clone = Arc::clone(&self.agent_keepalive_tasks); + + let vm_actorid_ulid_map_arc_clone = Arc::clone(&self.vm_actorid_ulid_map); + let vm_data_cache_arc_clone = Arc::clone(&self.vm_data_cache); + let vm_keepalive_tasks_arc_clone = Arc::clone(&self.vm_keepalive_tasks); + + self.cache_actor_finder = Some(tokio::spawn(async move { + let mut interval = tokio::time::interval(Duration::from_secs(1)); + loop { + let vm_join_handle = Self::vm_actor_finder( + actor_ref.clone(), + Arc::clone(&vm_actorid_ulid_map_arc_clone), + Arc::clone(&vm_data_cache_arc_clone), + Arc::clone(&vm_keepalive_tasks_arc_clone), + Arc::clone(&agent_data_cache_arc_clone), + ); + + let agent_join_handle = Self::agent_actor_finder( + actor_ref.clone(), + Arc::clone(&agent_data_cache_arc_clone), + Arc::clone(&agent_keepalive_tasks_arc_clone), + ); -#[derive(Debug, Clone)] -pub struct CachedAgentActor { - pub actor_ref: RemoteActorRef, - pub metadata: AgentStatus, -} + // intentionally ignoring results because we want to keep finding actors even if an attempt fails + let (vm_result, agent_result) = tokio::join!(vm_join_handle, agent_join_handle); + if let Err(error) = vm_result { + warn!(?error, "VM actor discovery failed"); + } + if let Err(error) = agent_result { + warn!(?error, "agent actor discovery failed"); + } + + //info!(?vm_data_cache_arc_clone); + //info!(?agent_data_cache_arc_clone); + + interval.tick().await; + } + })); + } + + /// 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 schedule_agent(&self, msg: &CreateVM) -> Result, Report> { + let mut best_agent = None; + let mut best_score = AgentScore::REJECTED; -#[async_trait] -impl ActorCacheUpdater for AgentActorCacheUpdater { - async fn get_actor_refs(&self) -> Result>> { - let mut agent_actors_lookup = RemoteActorRef::::lookup_all(AGENT); - let mut actor_ref_vec = Vec::new(); + for agent in self.agent_data_cache.iter() { + let score = self.score_agent(msg, &agent); - while let Some(agent_actor) = agent_actors_lookup.try_next().await? { - actor_ref_vec.push(agent_actor); + if score > best_score { + best_agent = Some(agent.actor_ref.clone()); + best_score = score; + } } - Ok(actor_ref_vec) + info!(?best_score, ?best_agent, "best agent"); + + best_agent.ok_or_eyre("No valid agents found.") } - async fn on_update( + // 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. + fn score_agent( &self, - actor_ref: &RemoteActorRef, - previous_value: Option, - ) -> Result { - let output_actor_ref = match previous_value { - Some(value) => value.actor_ref, - _ => actor_ref.clone(), - }; + msg: &CreateVM, + agent: &RefMulti<'_, ActorId, CachedAgentActor>, + ) -> AgentScore { + let mut score = AgentScore::default(); + + let agent_max_vcpus = agent + .data + .vcpus + .saturating_mul(VCPU_OVERPROVISIONMENT_NUMERATOR) + .checked_div(VCPU_OVERPROVISIONMENT_DENOMINATOR) + .unwrap_or(u32::MAX); + // todo: do we care about VMData.max_vcpus? + let agent_used_vcpus = agent.data.used_vcpus.saturating_add(msg.config.data.vcpus); + + if agent_used_vcpus >= agent_max_vcpus { + return AgentScore::REJECTED; + } + + #[expect( + clippy::cast_precision_loss, + reason = "the scheduler score intentionally uses f32 ratios" + )] + #[expect( + clippy::arithmetic_side_effects, + reason = "the preceding capacity check guarantees non-negative subtraction" + )] + 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( + agent + .data + .used_ram + .as_u64() + .saturating_add(msg.config.data.memory.as_u64()), + ); + + if agent_used_ram >= agent_max_ram { + return AgentScore::REJECTED; + } + + #[expect( + clippy::cast_precision_loss, + reason = "the scheduler score intentionally uses f32 ratios" + )] + let ram_headroom = agent_max_ram + .as_u64() + .saturating_sub(agent_used_ram.as_u64()) as f32 + / agent_max_ram.as_u64() as f32; + score.general += ram_headroom; + + // Roughly based on . + if let Some(affinity_rules) = &msg.config.affinity { + for rule in affinity_rules { + let mut metadata_tables: Vec = Vec::with_capacity(1); + + match rule.affinity_type { + AffinityType::VirtualMachine => { + for vmid in &agent.extended_vm_set { + let Some(vm_data_cache_ref) = &self.vm_data_cache.get(vmid) else { + continue; + }; + + let Some(vm_manifest) = &vm_data_cache_ref.data.config else { + continue; + }; + + if let Some(metadata) = &vm_manifest.metadata { + metadata_tables.push(metadata.clone()); + } + } + } + AffinityType::Agent => metadata_tables.push(agent.data.metadata.clone()), + } + + let mut follows_rule = false; + + for requirement in &rule.requirements { + let mut requirement_outcome = true; - Ok(CachedAgentActor { - actor_ref: output_actor_ref, - metadata: actor_ref.ask(&GetAgentStatus).await?, - }) + for object_metadata in &metadata_tables { + let table = match requirement.table { + MetadataTable::Label => &object_metadata.labels, + MetadataTable::Annotation => &object_metadata.annotations, + }; + + let value_option = table.get(&requirement.key); + + if !evaluate_table_value(value_option, requirement) { + requirement_outcome = false; + break; + } + } + + if requirement_outcome { + follows_rule = true; + break; + } + } + + follows_rule ^= rule.inverse; + + match (rule.strictness, follows_rule) { + (AffinityStrictness::Required, false) => return AgentScore::REJECTED, + (AffinityStrictness::Required, true) => {} // specifically do nothing + (AffinityStrictness::Preferred { weight }, follows_rule) => { + let follows_rule = i64::from(follows_rule); + score.affinity = score + .affinity + .saturating_add(follows_rule.saturating_mul(weight)); + } + } + } + } + + // 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 } } -// todo: this code is really bad, and we should not have effectively two copies of ths same thing. -#[derive(Copy, Clone)] -struct VMActorCacheUpdater; +fn evaluate_table_value(value_option: Option<&String>, requirement: &AffinityRequirement) -> bool { + let Some(value) = value_option else { + return false; + }; + + match requirement.operator { + Operator::In => requirement.values.contains(value), + Operator::NotIn => !requirement.values.contains(value), + Operator::Lt | Operator::Gt => { + if requirement.values.len() != 1 { + return false; + } -#[derive(Debug, Clone)] -pub struct CachedVMActor { - pub actor_ref: RemoteActorRef, - pub metadata: GetVMInfoReply, -} + let Ok(value_number): Result = value.parse() else { + return false; + }; -#[async_trait] -impl ActorCacheUpdater for VMActorCacheUpdater { - async fn get_actor_refs(&self) -> Result>> { - let mut agent_actors_lookup = RemoteActorRef::::lookup_all(VM); - let mut actor_ref_vec = Vec::new(); + let Ok(requirement_value_number): Result = requirement.values[0].parse() else { + return false; + }; - while let Some(agent_actor) = agent_actors_lookup.try_next().await? { - actor_ref_vec.push(agent_actor); + if requirement.operator == Operator::Lt { + value_number < requirement_value_number + } else { + value_number > requirement_value_number + } } + } +} + +#[derive(Debug, Clone, Copy, PartialEq)] +struct AgentScore { + general: f32, + affinity: i64, +} - Ok(actor_ref_vec) +impl AgentScore { + pub const REJECTED: Self = Self { + general: f32::NEG_INFINITY, + affinity: i64::MIN, + }; +} + +impl Default for AgentScore { + fn default() -> Self { + Self { + general: 0.0, + affinity: 0, + } } +} - async fn on_update( - &self, - actor_ref: &RemoteActorRef, - previous_value: Option, - ) -> Result { - let output_actor_ref = match previous_value { - Some(value) => value.actor_ref, - _ => actor_ref.clone(), - }; +impl PartialOrd for AgentScore { + fn partial_cmp(&self, other: &Self) -> Option { + let affinity_cmp = self.affinity.cmp(&other.affinity); + + if affinity_cmp != Ordering::Equal { + return Some(affinity_cmp); + } - Ok(CachedVMActor { - actor_ref: output_actor_ref, - metadata: actor_ref.ask(&GetVMInfo { vmid: None }).await?, - }) + self.general.partial_cmp(&other.general) } } @@ -220,10 +584,18 @@ impl Actor for SchedulerActor { info!(?peer_id, "Scheduler Actor started!"); - Ok(Self { - agent_actor_cache: ActorCache::new(actor_ref.clone(), AgentActorCacheUpdater), - vm_actor_cache: ActorCache::new(actor_ref, VMActorCacheUpdater), - }) + let mut scheduler_actor = Self { + agent_data_cache: Arc::new(DashMap::new()), + agent_keepalive_tasks: Arc::new(DashMap::new()), + vm_actorid_ulid_map: Arc::new(DashMap::new()), + vm_data_cache: Arc::new(DashMap::new()), + vm_keepalive_tasks: Arc::new(DashMap::new()), + cache_actor_finder: None, + }; + + scheduler_actor.start_actor_finder(actor_ref.into_remote_ref().await); + + Ok(scheduler_actor) } async fn on_link_died( @@ -232,16 +604,42 @@ impl Actor for SchedulerActor { id: ActorId, reason: ActorStopReason, ) -> Result, Self::Error> { - warn!("Linked actor {id:?} died with reason {reason:?}"); + warn!(?id, ?reason, "Linked actor died"); + // check that scheduler actor is still alive. let Some(_) = actor_ref.upgrade() else { return Ok(ControlFlow::Break(ActorStopReason::Killed)); }; - self.agent_actor_cache.on_link_died(id); - self.vm_actor_cache.on_link_died(id); + if let Some((_, keepalive_task)) = self.agent_keepalive_tasks.remove(&id) { + trace!(?id, "Aborting agent keepalive task"); + keepalive_task.abort(); + } + + self.agent_data_cache.remove(&id); + + if let Some((_, keepalive_task)) = self.vm_keepalive_tasks.remove(&id) { + trace!(?id, "Aborting vm keepalive task"); + keepalive_task.abort(); + } + + if let Some((_, vmid)) = self.vm_actorid_ulid_map.remove(&id) { + if let Some(cached_vm) = self.vm_data_cache.get(&vmid) + && let Some(mut agent) = + self.agent_data_cache.get_mut(&cached_vm.scheduled_agent_id) + { + agent.extended_vm_set.remove(&vmid); + } + + // todo: we likely should keep a copy of the VirtualMachine manifest in the cache. + // instead of removing the vm entirely, we should just modify the status to shutdown or crashed or something. + // + // another potential issue is that the link dying doesn't guarantee that the vm is dead. if there is a networking partition, things get iffy. + trace!(?id, ?vmid, "Removing vm from vm_data_cache"); + self.vm_data_cache.remove(&vmid); + } - info!(vm_actor_cache=?self.agent_actor_cache.data_cache, agent_actor_cache=?self.vm_actor_cache.data_cache, "data caches post actor removal"); + // todo: attempt vm restarts if necessary. Ok(ControlFlow::Continue(())) } @@ -255,16 +653,40 @@ impl Message for SchedulerActor { msg: CreateVM, _ctx: &mut Context, ) -> Self::Reply { - loop { - let target_agent = self.schedule_agent(&msg)?; + let target_agent = self.schedule_agent(&msg)?; - match target_agent.ask(&msg).await { - Ok(reply) => return Ok(reply), - Err(err) => { - warn!("CreateVM forwarding failed, trying again: {err}"); - } - } + // we add to cache first, because we want to make sure future requests assume this vm exists. if the message fails, we clean it up afterward. + if let Some(mut cached_data) = self.agent_data_cache.get_mut(&target_agent.id()) { + cached_data.extended_vm_set.insert(msg.vmid); + } else { + return Err(eyre!("target agent is not in data cache")); } + + self.vm_data_cache.insert( + msg.vmid, + CachedVMActor { + actor_ref: None, + scheduled_agent_id: target_agent.id(), + data: GetVMInfoReply { + vmid: msg.vmid, + config: Some(msg.config.clone()), + }, + }, + ); + + let reply = target_agent.ask(&msg).await; + + // remove from caches if the agent rejected the request + if reply.is_err() { + self.agent_data_cache.alter(&target_agent.id(), |_, mut v| { + v.extended_vm_set.remove(&msg.vmid); + + v + }); + self.vm_data_cache.remove(&msg.vmid); + } + + Ok(reply?) } } @@ -279,6 +701,7 @@ impl Message for SchedulerActor { let vm = RemoteActorRef::::lookup(vm_actor_id(msg.vmid)).await?; tracing::trace!(?vm, "DeleteVM"); if let Some(vm) = vm { + // don't update cache, because we rely on link dying and updater task to remove from cache once the VM is fully down. vm.tell(&msg).send()?; Ok(DeleteVMReply) } else { @@ -298,6 +721,7 @@ impl Message for SchedulerActor { let vm = RemoteActorRef::::lookup(vm_actor_id(msg.vmid)).await?; tracing::trace!(?vm, "ShutdownVM"); if let Some(vm) = vm { + // don't update cache, because we rely on link dying and updater task to remove from cache once the VM is fully down. vm.tell(&msg).send()?; Ok(ShutdownVMReply) } else { @@ -319,8 +743,8 @@ impl Message for SchedulerActor { ) -> Self::Reply { let mut vms = Vec::new(); - for agent in self.agent_actor_cache.data_cache.iter() { - vms.extend_from_slice(agent.metadata.vms.as_slice()); + for agent in self.agent_data_cache.iter() { + vms.extend_from_slice(agent.data.vms.as_slice()); } Ok(AgentListVMsReply { vms }) diff --git a/odorobo/src/ch_driver/actor.rs b/odorobo/src/ch_driver/actor.rs index da985a5..122eed3 100644 --- a/odorobo/src/ch_driver/actor.rs +++ b/odorobo/src/ch_driver/actor.rs @@ -30,17 +30,22 @@ pub struct VMActor { /// path to the Cloud Hypervisor socket, in /run/odorobo/vms//ch.sock pub vm_instance: VMInstance, pub migration_state: Option, + pub manifest: Option, } impl Actor for VMActor { - // tuple of VM ID and optional config + // tuple of VM ID and manifest type Args = (ulid::Ulid, Option); type Error = Report; #[tracing::instrument(skip_all)] async fn on_start((vmid, vm_config): Self::Args, actor_ref: ActorRef) -> Result { - let mut vminstance = - VMInstance::spawn(&vmid.to_string(), vm_config.map(VmConfig::from), None).await?; + let mut vminstance = VMInstance::spawn( + &vmid.to_string(), + vm_config.clone().map(VmConfig::from), + None, + ) + .await?; // Take the child process out so we can watch for unexpected death. // destroy() handles a missing child_process gracefully. @@ -72,6 +77,7 @@ impl Actor for VMActor { vmid, vm_instance: vminstance, migration_state: None, + manifest: vm_config, }) } @@ -159,7 +165,7 @@ impl Message for VMActor { ) -> Self::Reply { GetVMInfoReply { vmid: self.vmid, - config: self.vm_instance.vm_config.clone(), + config: self.manifest.clone(), // we likely dont want to send the entire manifest on every update, but some of this data is required and this is easier for now. } } } diff --git a/odorobo/src/messages/agent.rs b/odorobo/src/messages/agent.rs index eb1fd4d..44f4193 100644 --- a/odorobo/src/messages/agent.rs +++ b/odorobo/src/messages/agent.rs @@ -3,6 +3,8 @@ use kameo::Reply; use serde::{Deserialize, Serialize}; use ulid::Ulid; +use crate::types::ObjectMetadata; + #[derive(Serialize, Deserialize)] pub struct GetAgentStatus; @@ -16,4 +18,5 @@ pub struct AgentStatus { pub used_vcpus: u32, pub used_ram: ByteSize, pub vms: Vec, + pub metadata: ObjectMetadata, } diff --git a/odorobo/src/messages/vm.rs b/odorobo/src/messages/vm.rs index c46d9b2..c48ce5c 100644 --- a/odorobo/src/messages/vm.rs +++ b/odorobo/src/messages/vm.rs @@ -99,5 +99,5 @@ pub struct GetVMInfo { #[derive(Serialize, Deserialize, Reply, Debug, Clone)] pub struct GetVMInfoReply { pub vmid: Ulid, - pub config: Option, + pub config: Option, } diff --git a/odorobo/src/types.rs b/odorobo/src/types.rs index 4fadcc1..be0cf26 100644 --- a/odorobo/src/types.rs +++ b/odorobo/src/types.rs @@ -175,12 +175,14 @@ pub struct VirtualMachine { pub struct AffinityRule { pub strictness: AffinityStrictness, pub affinity_type: AffinityType, - pub direction: AffinityDirection, + /// If true, the outcome of the requirements is inverted. + #[serde(default)] + pub inverse: bool, /// `ORed` together pub requirements: Vec, } -#[derive(Serialize, Deserialize, Debug, JsonSchema, Clone)] +#[derive(Serialize, Deserialize, Debug, JsonSchema, Clone, Copy)] pub enum AffinityStrictness { Required, Preferred { weight: i64 }, @@ -192,12 +194,8 @@ pub enum AffinityType { Agent, } -#[derive(Serialize, Deserialize, Debug, JsonSchema, Clone)] -pub enum AffinityDirection { - Normal, - Anti, -} - +/// If there are several metadata tables, their results will be `ANDed` together +/// EX: if a requirement is checked against several VMs, it must pass all VMs for the requirement to be fulfilled. #[derive(Serialize, Deserialize, Debug, JsonSchema, Clone)] pub struct AffinityRequirement { pub key: String, @@ -212,8 +210,7 @@ pub enum MetadataTable { Annotation, } -// todo: possibly replace with std::ops -#[derive(Serialize, Deserialize, Debug, JsonSchema, Clone)] +#[derive(Serialize, Deserialize, Debug, JsonSchema, Clone, PartialEq, Eq)] pub enum Operator { In, NotIn, diff --git a/odorobo/src/utils/mod.rs b/odorobo/src/utils/mod.rs index a8b1569..2ae7ecd 100644 --- a/odorobo/src/utils/mod.rs +++ b/odorobo/src/utils/mod.rs @@ -1,4 +1,3 @@ -pub mod actor_cache; pub mod actor_names; pub mod lockfile; diff --git a/odoroboctl/src/cli.rs b/odoroboctl/src/cli.rs index b59842b..8450d37 100644 --- a/odoroboctl/src/cli.rs +++ b/odoroboctl/src/cli.rs @@ -1,6 +1,11 @@ +use std::collections::BTreeMap; + use bytesize::ByteSize; use clap::{Parser, Subcommand}; -use odorobo::types::{CreateVMRequest, VMData, VirtualMachine}; +use odorobo::types::{ + AffinityRequirement, AffinityRule, AffinityStrictness, AffinityType, CreateVMRequest, + MetadataTable, ObjectMetadata, Operator, VMData, VirtualMachine, +}; use reqwest::{Client, Response}; use serde::Deserialize; use stable_eyre::Result; @@ -105,9 +110,27 @@ pub async fn run_command(cli: Cli) -> Result<()> { vcpus: 4, max_vcpus: None, memory: ByteSize::gib(4), - image: "/var/lib/odorobo/f43.raw".to_owned(), + image: "/var/lib/odorobo/f43-1.raw".to_owned(), ..Default::default() }, + metadata: Some(ObjectMetadata { + labels: BTreeMap::new(), + annotations: BTreeMap::from([( + String::from("distribution"), + String::from("different"), + )]), + }), + affinity: Some(vec![AffinityRule { + strictness: AffinityStrictness::Required, + affinity_type: AffinityType::VirtualMachine, + inverse: false, + requirements: vec![AffinityRequirement { + key: String::from("distribution"), + table: MetadataTable::Annotation, + operator: Operator::NotIn, + values: vec![String::from("different")], + }], + }]), ..Default::default() };