diff --git a/src/controller.rs b/src/controller.rs index 98b4fe4..962f84f 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -568,18 +568,46 @@ impl Controller { let fleet: Vec<_> = self.instances.values().cloned().collect(); self.persist_new_classifications(&fleet).await?; - let plan = self.engine.evaluate_fleet(&fleet, &context).await; + let decisions = self.engine.evaluate_fleet(&fleet, &context).await; + let mut planned_ids = HashSet::new(); + let mut plan = Vec::new(); + for (sequence, item) in decisions.into_iter().enumerate() { + if !planned_ids.insert(item.instance_id.clone()) { + tracing::warn!( + target: "controller", + vm = %item.instance_id, + "engine returned duplicate decisions; keeping the first" + ); + continue; + } + let Some(instance) = self.instances.get(&item.instance_id) else { + tracing::warn!( + target: "controller", + vm = %item.instance_id, + "engine returned a decision for an unknown VM" + ); + continue; + }; + let action = item.decision; + let Some(target) = action.target() else { + continue; + }; + let status = instance.status.read().await; + let current = status.thread_count; + if target == current { + continue; + } + plan.push((target > current, sequence, item.instance_id, action)); + } + plan.sort_by_key(|(is_up, sequence, _, _)| (*is_up, *sequence)); + let cap = self.cfg.max_instances_adjusted_per_poll as usize; let mut adjusted = 0usize; - for item in plan { - let action = item.decision; + for (_, _, instance_id, action) in plan { if cap > 0 && adjusted >= cap { continue; } - if self - .apply_engine_decision(&item.instance_id, action) - .await? - { + if self.apply_engine_decision(&instance_id, action).await? { adjusted += 1; } } diff --git a/src/engines/mod.rs b/src/engines/mod.rs index 08f8976..0c31ec0 100644 --- a/src/engines/mod.rs +++ b/src/engines/mod.rs @@ -203,6 +203,10 @@ pub trait ScalingEngine: Send + Sync { async fn evaluate(&self, instance: &Arc, context: &EngineTickContext) -> ScaleAction; /// Evaluate eligible VMs concurrently and collect one coherent plan. + /// + /// Sticky or backend-disabled VMs remain visible to fleet-aware engines + /// that override this method, but the default per-VM implementation does + /// not evaluate them. async fn evaluate_fleet( &self, instances: &[Arc], @@ -212,7 +216,9 @@ pub trait ScalingEngine: Send + Sync { // to first look at all the instances first and then make decisions join_all(instances.iter().map(|instance| async move { let status = instance.status.read().await; - if status.manual_scaling_sticky || !status.scaling_allowed { + let scaling_allowed = status.scaling_allowed; + let sticky = status.manual_scaling_sticky; + if sticky || !scaling_allowed { return None; } Some(InstanceDecision::new( @@ -298,4 +304,86 @@ mod tests { Err(other) => panic!("expected NoSuchEngine, got {other}"), } } + + use std::sync::atomic::{AtomicUsize, Ordering}; + + use crate::{ + backends::BackendClientError, + instance::{InstanceClient, ThreadPoolSnapshot}, + }; + + struct SnapshotClient; + + #[async_trait] + impl InstanceClient for SnapshotClient { + async fn set_thread_count(&self, _count: u32) -> Result<(), BackendClientError> { + Ok(()) + } + + async fn get_thread_pool_snapshot(&self) -> Result { + Ok(ThreadPoolSnapshot { + thread_count: 1, + vcpu_count: 2, + perf: None, + per_thread_util: Some(0.0), + }) + } + + async fn close(&self) {} + } + + struct PerInstanceEngine { + calls: AtomicUsize, + } + + #[async_trait] + impl ScalingEngine for PerInstanceEngine { + fn name(&self) -> &'static str { + "test" + } + + fn dump_config(&self) -> serde_json::Value { + serde_json::Value::Null + } + + async fn evaluate( + &self, + _instance: &Arc, + _context: &EngineTickContext, + ) -> ScaleAction { + self.calls.fetch_add(1, Ordering::Relaxed); + ScaleAction::None + } + } + + fn test_instance(id: &str) -> Arc { + Arc::new(Instance::new( + id.to_string(), + Path::new(""), + 0, + SnapshotClient, + )) + } + + /// Test that default fleet `evaluate` skips ineligible VMs and only + /// plans eligible ones. + #[tokio::test] + async fn default_fleet_evaluation_skips_ineligible_instances() { + let engine = PerInstanceEngine { + calls: AtomicUsize::new(0), + }; + let first = test_instance("first"); + let sticky = test_instance("sticky"); + sticky.status.write().await.manual_scaling_sticky = true; + + let plan = engine + .evaluate_fleet(&[first.clone(), sticky], &EngineTickContext::default()) + .await; + + assert_eq!(engine.calls.load(Ordering::Relaxed), 1); + assert_eq!( + plan, + vec![InstanceDecision::new(&first.id, ScaleAction::None)] + ); + } }