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 87c8518..751b03a 100644 --- a/odorobo/src/actors/agent_actor.rs +++ b/odorobo/src/actors/agent_actor.rs @@ -316,6 +316,7 @@ impl Message for AgentActor { vms: self.vms.keys().copied().collect(), used_vcpus: vcpus_used_by_vms + self.config.get_reserved_vcpus(), used_ram: ByteSize::b(ram_used_by_vms), + metadata: self.metadata.clone(), } } } diff --git a/odorobo/src/actors/ip_management_actor.rs b/odorobo/src/actors/ip_management_actor.rs index 9796b38..2f1fdd9 100644 --- a/odorobo/src/actors/ip_management_actor.rs +++ b/odorobo/src/actors/ip_management_actor.rs @@ -1,6 +1,3 @@ -use kameo::prelude::*; -use stable_eyre::{Report, Result}; - // idk if we ever agreed upon an OUI for fyra, but im reserving `FYR` for this // -cappy pub const FYRA_OUI: [u8; 3] = [0x46, 0x59, 0x52]; @@ -21,18 +18,3 @@ fn test_calculate_mac_address() { let mac = calculate_mac_address(ip); assert_eq!(mac, [0x46, 0x59, 0x52, 168, 0x01, 0x01]); } - -/// HTTP REST API service -#[derive(RemoteActor)] -pub struct IPManagementActor; - -impl Actor for IPManagementActor { - type Args = (); - type Error = Report; - - async fn on_start(_state: Self::Args, _actor_ref: ActorRef) -> Result { - // if we need to like prep the router stuff - - Ok(Self) - } -} diff --git a/odorobo/src/actors/scheduler_actor.rs b/odorobo/src/actors/scheduler_actor.rs index dda999e..151c0e8 100644 --- a/odorobo/src/actors/scheduler_actor.rs +++ b/odorobo/src/actors/scheduler_actor.rs @@ -1,27 +1,94 @@ +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; use crate::messages::agent::*; use crate::messages::vm::*; 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. @@ -29,170 +96,452 @@ 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" - )] - async fn lookup_by_actor_id( + async fn lookup_agent_by_actor_id( &mut self, actor_id: &ActorId, ) -> Option> { - self.agent_actor_cache - .data_cache + 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" - )] - async fn lookup_by_hostname(&mut self, hostname: &str) -> Option> { - self.agent_actor_cache - .data_cache + async fn lookup_agent_by_hostname( + &mut 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 https://kubernetes.io/docs/concepts/scheduling-eviction/assign-pod-node/ - /// 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. - async fn schedule_agent( - &mut self, - _msg: &CreateVM, - ) -> Result, Report> { - let mut best_agent = None; - let mut best_agent_score = 0u32; + // 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); + } + } - // 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; + Ok(()) + } - let agent_max_vcpus = agent.metadata.vcpus * VCPU_OVERPROVISIONMENT_NUMERATOR - / VCPU_OVERPROVISIONMENT_DENOMINATOR; + 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 = 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(|e| e.scheduled_agent_id) + .unwrap_or(actor_ref.id()), + data, + }, + ); + + fails = 0; + } else { + fails += 1; + } - if agent.metadata.used_vcpus >= agent_max_vcpus { - continue; + if fails > 5 { + warn!( + ?actor_ref, + "can no longer reach vm actor, cleaning up cache entries" + ); + + 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| { + if entry.actor_ref.as_ref().map(|r| r.id()) == Some(actor_ref.id()) { + Some(*entry.key()) + } else { + None + } + }) + }); + + 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_max_vcpus - agent.metadata.used_vcpus) * 1024 / agent_max_vcpus; + return; + } + + interval.tick().await; + } + } - // todo: add ram overprovisionment. not adding this to scheduler until it works on the hypervisor side. - let agent_max_ram = agent.metadata.ram; + async fn agent_actor_finder( + parent_actor_ref: RemoteActorRef, + data_cache: Arc>, + keepalive_tasks: Arc>>, + ) -> Result<(), Report> { + trace!("running agent_actor_finder"); - if agent.metadata.used_ram >= agent_max_ram { - continue; - } + let mut agent_actor_stream = RemoteActorRef::::lookup_all(AGENT); + + 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"); - agent_score += ((agent_max_ram.as_u64() - agent.metadata.used_ram.as_u64()) * 1024 - / agent_max_ram.as_u64()) as u32; + parent_actor_ref.link_remote(&agent_actor).await?; - // todo: https://kubernetes.io/docs/concepts/scheduling-eviction/assign-pod-node/ + let agent_actor_id = agent_actor.id(); - // 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; - // } + 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; + }); - info!(agent=?agent.value(), score=agent_score); + keepalive_tasks.insert(agent_actor_id, updater_task); + } + } + + Ok(()) + } - if agent_score > best_agent_score { - best_agent = Some(agent.actor_ref.clone()); - best_agent_score = agent_score; + async fn agent_updater_task( + actor_ref: RemoteActorRef, + data_cache: Arc>, + ) { + let mut interval = tokio::time::interval(Duration::from_secs(1)); + let mut fails = 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: AHashSet::from_iter(data.vms.iter().copied()), + }, + ); } + + fails = 0; + } else { + fails += 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), + ); + + // intentionally ignoring results because we want to keep finding actors even if an attempt fails + let _ = tokio::join!(vm_join_handle, agent_join_handle); + + //info!(?vm_data_cache_arc_clone); + //info!(?agent_data_cache_arc_clone); + + interval.tick().await; + } + })); + } -#[derive(Debug, Clone)] -pub struct CachedAgentActor { - pub actor_ref: RemoteActorRef, - pub metadata: AgentStatus, -} + /// 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 https://kubernetes.io/docs/concepts/scheduling-eviction/assign-pod-node/ + /// + /// 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(&mut 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 * VCPU_OVERPROVISIONMENT_NUMERATOR + / VCPU_OVERPROVISIONMENT_DENOMINATOR; + // todo: do we care about VMData.max_vcpus? + let agent_used_vcpus = agent.data.used_vcpus + msg.config.data.vcpus; + + if agent_used_vcpus >= agent_max_vcpus { + return AgentScore::REJECTED; + } + + score.general += (agent_max_vcpus - agent_used_vcpus) as f32 / agent_max_vcpus as f32; + + // 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 = agent.data.used_ram + msg.config.data.memory; + + if agent_used_ram >= agent_max_ram { + return AgentScore::REJECTED; + } - Ok(CachedAgentActor { - actor_ref: output_actor_ref, - metadata: actor_ref.ask(&GetAgentStatus).await?, - }) + score.general += (agent_max_ram.as_u64() - agent_used_ram.as_u64()) as f32 + / agent_max_ram.as_u64() as f32; + + // roughly based on: https://kubernetes.io/docs/concepts/scheduling-eviction/assign-pod-node/ + 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; + + 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) => { + score.affinity += follows_rule as i64 * 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, +} + +impl AgentScore { + pub const REJECTED: Self = Self { + general: f32::NEG_INFINITY, + affinity: i64::MIN, + }; +} - Ok(actor_ref_vec) +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) } } @@ -205,28 +554,62 @@ 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 = SchedulerActor { + 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( &mut self, actor_ref: WeakActorRef, - id: ActorId, + actor_id: ActorId, reason: ActorStopReason, ) -> Result, Self::Error> { - warn!("Linked actor {id:?} died with reason {reason:?}"); + warn!(?actor_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).await; - self.vm_actor_cache.on_link_died(id).await; + if let Some((_, keepalive_task)) = self.agent_keepalive_tasks.remove(&actor_id) { + trace!(?actor_id, "Aborting agent keepalive task"); + keepalive_task.abort(); + }; + + self.agent_data_cache.remove(&actor_id); - info!(vm_actor_cache=?self.agent_actor_cache.data_cache, agent_actor_cache=?self.vm_actor_cache.data_cache, "data caches post actor removal"); + if let Some((_, keepalive_task)) = self.vm_keepalive_tasks.remove(&actor_id) { + trace!(?actor_id, "Aborting vm keepalive task"); + keepalive_task.abort(); + }; + + if let Some((_, vmid)) = self.vm_actorid_ulid_map.remove(&actor_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!(?actor_id, ?vmid, "Removing vm from vm_data_cache"); + self.vm_data_cache.remove(&vmid); + } + + // todo: attempt vm restarts if necessary. Ok(ControlFlow::Continue(())) } @@ -240,16 +623,40 @@ impl Message for SchedulerActor { msg: CreateVM, _ctx: &mut Context, ) -> Self::Reply { - loop { - let target_agent = self.schedule_agent(&msg).await?; + 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?) } } @@ -264,6 +671,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 { @@ -283,6 +691,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 { @@ -304,8 +713,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 c5f3654..5b78cdf 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 512a793..8318a5a 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 1c177c0..e425bf1 100644 --- a/odorobo/src/types.rs +++ b/odorobo/src/types.rs @@ -165,7 +165,8 @@ pub struct VirtualMachine { /// Metadata pub metadata: Option, - /// List of Affinity rules for scheduling. These are ANDed / summed together depending on the strictness. + /// List of Affinity rules for scheduling. + /// Affinity rules are ANDed or summed based on strictness pub affinity: Option>, } @@ -173,12 +174,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 }, @@ -190,12 +193,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, @@ -210,8 +209,7 @@ pub enum MetadataTable { Annotation, } -// todo: possibly replace with std::ops -#[derive(Serialize, Deserialize, Debug, JsonSchema, Clone)] +#[derive(Serialize, Deserialize, Debug, JsonSchema, Clone, PartialEq)] pub enum Operator { In, NotIn, diff --git a/odorobo/src/utils/actor_cache.rs b/odorobo/src/utils/actor_cache.rs deleted file mode 100644 index 50415ab..0000000 --- a/odorobo/src/utils/actor_cache.rs +++ /dev/null @@ -1,181 +0,0 @@ -use std::{marker::PhantomData, sync::Arc, time::Duration}; - -use async_trait::async_trait; -use dashmap::DashMap; -use kameo::prelude::*; -use stable_eyre::{Report, Result}; -use tokio::task::JoinHandle; -use tracing::{info, instrument, trace}; - -use std::fmt; - -// TODO: refactor to use derive macro, but I (caleb) don't know how to write a derive macro. -// -// The best way to write this would be using a derive similar to kameo -// so you would create a struct with #[derive(ActorCache)]. -// Then you would set the types and then write get_actor_refs and on_update. -// This would combine everything into one struct and make it a lot easier to work with. -// -// similar to: https://github.com/tqwewe/kameo/blob/1d498c0566b613b9afe6d54965c4b191c84432e0/src/actor.rs#L122 -// -// -// other things we might want: -// - a default get_actor_refs that just finds all actor_refs with a specific actor string. -// - change get_actor_refs to use an iterator - -#[async_trait] -pub trait ActorCacheUpdater< - ChildActor: Actor + RemoteActor, - Data: Clone + Send + Sync + 'static + fmt::Debug, ->: Sync + Send + Copy + 'static -{ - // todo: this could probably be better if it was an iterator, but I am lazy and don't want to right now. - async fn get_actor_refs(&self) -> Result>>; - async fn on_update( - &self, - actor_ref: &RemoteActorRef, - previous_value: Option, - ) -> Result; -} - -#[derive(Debug)] -pub struct ActorCache< - ParentActor: Actor + RemoteActor, - ChildActor: Actor + RemoteActor, - Data: Clone + Send + Sync + 'static + fmt::Debug, -> { - #[expect( - dead_code, - reason = "keeps the parent actor reference alive for cache-owned tasks" - )] - parent_actor_ref: ActorRef, - pub data_cache: Arc>, - keepalive_tasks: Arc>>, - #[expect( - dead_code, - reason = "retains the background finder task handle for future shutdown/drop handling" - )] - actor_finder: Option>, - - child_actor_type: PhantomData, -} - -// todo: impl Drop to automatically kill all the keepalive_tasks and the actor_finder task. - -impl< - ParentActor: Actor + RemoteActor, - ChildActor: Actor + RemoteActor, - Data: Clone + Send + Sync + 'static + fmt::Debug, -> ActorCache -{ - pub fn new( - parent_actor_ref: ActorRef, - updater: impl ActorCacheUpdater, - ) -> Result { - let data_cache = Arc::new(DashMap::new()); - let keepalive_tasks = Arc::new(DashMap::new()); - - let actor_cache = ActorCache { - parent_actor_ref: parent_actor_ref.clone(), - data_cache, - keepalive_tasks, - actor_finder: None, - - child_actor_type: PhantomData, - }; - - actor_cache.start_actor_finder(parent_actor_ref, updater); - - Ok(actor_cache) - } - - /// run this function inside of the on_link_died of the ParentActor - pub async fn on_link_died(&self, id: ActorId) { - info!("removing agent actor from cache {id:?}"); - - if let Some(actor_keepalive_task) = self.keepalive_tasks.remove(&id) { - trace!("Aborting keepalive task for agent {id:?}"); - actor_keepalive_task.1.abort(); - }; - - self.data_cache.remove(&id); - } - - fn start_actor_finder( - &self, - parent_actor_ref: ActorRef, - updater: impl ActorCacheUpdater, - ) { - let keepalive_tasks_clone = Arc::clone(&self.keepalive_tasks); - let data_cache_clone = Arc::clone(&self.data_cache); - - tokio::spawn(async move { - let mut interval = tokio::time::interval(Duration::from_secs(1)); - loop { - let _ = Self::actor_finder( - parent_actor_ref.clone(), - Arc::clone(&keepalive_tasks_clone), - Arc::clone(&data_cache_clone), - updater, - ) - .await; - - interval.tick().await; - } - }); - } - - async fn actor_finder( - parent_actor_ref: ActorRef, - keepalive_tasks: Arc>>, - data_cache: Arc>, - updater: impl ActorCacheUpdater, - ) -> Result<(), Report> { - let actor_refs = updater.get_actor_refs().await?; - - info!(?actor_refs, "running actor_finder"); - - for actor_ref in actor_refs { - if !keepalive_tasks.contains_key(&actor_ref.id()) { - trace!(?actor_ref, "starting updater_task"); - - parent_actor_ref.link_remote(&actor_ref).await?; - - let actor_ref_clone = actor_ref.clone(); - let data_cache_clone = Arc::clone(&data_cache); - let updater_task = tokio::spawn(async move { - Self::updater_task(actor_ref_clone, data_cache_clone, updater).await; - }); - - keepalive_tasks.insert(actor_ref.id(), updater_task); - } - } - - Ok(()) - } - - #[instrument(skip_all)] - async fn updater_task( - actor_ref: RemoteActorRef, - data_cache: Arc>, - updater: impl ActorCacheUpdater, - ) { - let mut interval = tokio::time::interval(Duration::from_secs(1)); - - loop { - let actor_id = actor_ref.id(); - - let mut previous_value_option = None; - - if let Some(data_ref) = data_cache.get(&actor_id) { - previous_value_option = Some(data_ref.clone()); - } - - if let Ok(update) = updater.on_update(&actor_ref, previous_value_option).await { - data_cache.insert(actor_id, update.clone()); - } - - interval.tick().await; - } - } -} diff --git a/odorobo/src/utils/mod.rs b/odorobo/src/utils/mod.rs index a874761..3bbae60 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; use aide::OperationIo; diff --git a/odoroboctl/src/cli.rs b/odoroboctl/src/cli.rs index ea7e031..4c6503a 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_string(), + image: "/var/lib/odorobo/f43-1.raw".to_string(), ..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() };