1use crate::{
8 instance_catalog::{self, Architecture, WorkloadRequirements},
9 CapacityGroup, CapacityGroupScalePolicy, ComputeChoiceRange, ComputePoolSelection, Container,
10 Daemon, ErrorData, FailureDomainSelection, GpuSpec, MachineProfile, Platform, ResourceSpec,
11 Stack, StackState,
12};
13use alien_error::{AlienError, Result};
14use serde::{Deserialize, Serialize};
15use std::collections::HashMap;
16
17#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
19#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
20#[serde(rename_all = "camelCase")]
21pub struct ComputePlan {
22 pub pools: Vec<ComputePoolPlan>,
24 #[serde(default, skip_serializing_if = "Vec::is_empty")]
26 pub containers: Vec<ContainerResourcePlan>,
27}
28
29#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
31#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
32#[serde(rename_all = "camelCase")]
33pub struct ContainerResourcePlan {
34 pub container_id: String,
36 pub choices: crate::ContainerResourceChoices,
38 pub cpu: ResourceSpec,
40 pub memory: ResourceSpec,
42}
43
44#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
46#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
47#[serde(rename_all = "camelCase")]
48pub struct ComputePoolPlan {
49 pub pool_id: String,
51 pub workloads: Vec<String>,
53 pub requirements: MachineProfile,
55 pub scale: CapacityGroupScalePolicy,
57 pub selected: ComputePoolSelection,
59 pub recommended: ComputePoolSelection,
61 pub machines: Vec<ComputeMachineOption>,
63 #[serde(default, skip_serializing_if = "Vec::is_empty")]
65 pub errors: Vec<String>,
66}
67
68#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
70#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
71#[serde(rename_all = "camelCase")]
72pub struct ComputeMachineOption {
73 pub machine: String,
75 pub profile: MachineProfile,
77 pub recommended: bool,
79}
80
81pub fn default_persistent_failure_domains() -> FailureDomainSelection {
85 FailureDomainSelection {
86 spread: 1,
87 selected_failure_domains: Vec::new(),
88 }
89}
90
91pub fn plan_compute(
93 stack: &Stack,
94 platform: Platform,
95 selected_settings: Option<&crate::ComputeSettings>,
96) -> Result<ComputePlan, ErrorData> {
97 plan_compute_with_state(stack, platform, selected_settings, None)
98}
99
100pub fn plan_compute_with_state(
103 stack: &Stack,
104 platform: Platform,
105 selected_settings: Option<&crate::ComputeSettings>,
106 previous_state: Option<&StackState>,
107) -> Result<ComputePlan, ErrorData> {
108 let mut resolved_stack = stack.clone();
109 crate::container_resources::resolve_container_resources(
110 &mut resolved_stack,
111 selected_settings,
112 )?;
113 let stack = &resolved_stack;
114 let mut groups = collect_workload_groups(stack)?;
115 merge_explicit_compute_groups(stack, &mut groups)?;
116
117 let mut pool_ids: Vec<String> = groups.keys().cloned().collect();
118 pool_ids.sort();
119
120 let mut pools = Vec::new();
121 for pool_id in pool_ids {
122 let mut group = groups.remove(&pool_id).expect("pool id came from map keys");
123 let mut requirements = group.requirements;
124 let selected = selected_settings.and_then(|settings| settings.pools.get(&pool_id));
125 let declared_machine = stack
126 .resources
127 .values()
128 .filter_map(|entry| entry.config.downcast_ref::<crate::ComputeCluster>())
129 .flat_map(|cluster| &cluster.capacity_groups)
130 .find(|group| group.group_id == pool_id)
131 .and_then(|group| group.instance_type.as_deref());
132 let machine_override = match selected {
133 Some(selection) => selection.machine(),
134 None => declared_machine,
135 };
136 if requirements.architecture.is_none() {
137 requirements.architecture = machine_override
138 .and_then(|machine| instance_catalog::find_instance_type(platform, machine))
139 .map(|spec| spec.architecture)
140 .or_else(|| {
141 previous_state.and_then(|state| {
142 state
143 .resources
144 .values()
145 .filter_map(|resource| {
146 resource.config.downcast_ref::<crate::ComputeCluster>()
147 })
148 .flat_map(|cluster| &cluster.capacity_groups)
149 .find(|group| group.group_id == pool_id)
150 .and_then(|group| group.instance_type.as_deref())
151 .and_then(|machine| {
152 instance_catalog::find_instance_type(platform, machine)
153 })
154 .map(|spec| spec.architecture)
155 })
156 });
157 }
158 if group.generated {
159 group.scale = generated_pool_scale_policy(
160 group.scale.default_min_size(),
161 group.scale.default_max_size(),
162 selected.map(ComputePoolSelection::max_size),
163 );
164 }
165 let machines = selected
166 .map(ComputePoolSelection::max_size)
167 .unwrap_or(group.scale.default_max_size())
168 .max(1);
169 let mut sizing = requirements.clone();
170 sizing.max_cpu_per_container = sizing
172 .max_cpu_per_container
173 .max(sizing.total_cpu_at_desired / f64::from(machines));
174 sizing.max_memory_per_container = sizing.max_memory_per_container.max(
175 sizing
176 .total_memory_bytes_at_desired
177 .div_ceil(u64::from(machines)),
178 );
179 let recommended = recommended_selection(
180 platform,
181 &sizing,
182 &group.scale,
183 group.requires_failure_domain,
184 group.generated,
185 machine_override,
186 )?;
187 let mut selected_choice = selected.cloned().unwrap_or_else(|| recommended.clone());
188 if selected.is_none() {
189 if let Some(declared_machine) = declared_machine {
190 match &mut selected_choice {
191 ComputePoolSelection::Fixed { machine, .. }
192 | ComputePoolSelection::Autoscale { machine, .. } => {
193 *machine = Some(declared_machine.to_string());
194 }
195 }
196 }
197 }
198 match &mut selected_choice {
202 ComputePoolSelection::Fixed { machine, .. }
203 | ComputePoolSelection::Autoscale { machine, .. } => {
204 if machine.is_none() {
205 *machine = recommended.machine().map(ToString::to_string);
206 }
207 }
208 }
209 if selected_choice.failure_domains().is_none() {
210 if let Some(default_failure_domains) = recommended.failure_domains().cloned() {
211 match &mut selected_choice {
212 ComputePoolSelection::Fixed {
213 failure_domains, ..
214 }
215 | ComputePoolSelection::Autoscale {
216 failure_domains, ..
217 } => *failure_domains = Some(default_failure_domains),
218 }
219 }
220 }
221 let mut errors = validate_compute_pool_selection(
222 platform,
223 &pool_id,
224 &selected_choice,
225 &requirements,
226 &group.scale,
227 );
228 if let Some(workload_requirements) = &group.workload_requirements {
231 if let Err(message) =
232 check_pool_capacity(platform, &pool_id, &selected_choice, workload_requirements)
233 {
234 errors.push(message);
235 }
236 }
237 let machines = machine_options(
238 platform,
239 &requirements,
240 selected_choice.machine(),
241 recommended.machine(),
242 )?;
243
244 pools.push(ComputePoolPlan {
245 pool_id,
246 workloads: group.workloads,
247 requirements: requirements_to_profile(&requirements),
248 scale: group.scale,
249 selected: selected_choice,
250 recommended,
251 machines,
252 errors,
253 });
254 }
255
256 let mut containers: Vec<_> = stack
257 .resources
258 .values()
259 .filter_map(|entry| {
260 let container = entry.config.downcast_ref::<Container>()?;
261 let choices = container.resource_choices.clone()?;
262 Some(ContainerResourcePlan {
263 container_id: container.id.clone(),
264 choices,
265 cpu: container.cpu.clone(),
266 memory: container.memory.clone(),
267 })
268 })
269 .collect();
270 containers.sort_by(|a, b| a.container_id.cmp(&b.container_id));
271 Ok(ComputePlan { pools, containers })
272}
273
274#[derive(Debug, Clone)]
275struct PlannedGroup {
276 workloads: Vec<String>,
277 workload_requirements: Option<WorkloadRequirements>,
279 requirements: WorkloadRequirements,
280 scale: CapacityGroupScalePolicy,
281 requires_failure_domain: bool,
282 generated: bool,
285}
286
287pub const GENERATED_POOL_MAX_MACHINES: u32 = 100;
292
293pub fn generated_pool_scale_policy(
305 default_min: u32,
306 default_max: u32,
307 selected_max: Option<u32>,
308) -> CapacityGroupScalePolicy {
309 let offered = GENERATED_POOL_MAX_MACHINES;
310 let ceiling = offered.max(selected_max.unwrap_or(0));
311 let default_min = default_min.min(offered);
312 CapacityGroupScalePolicy::Autoscale {
313 min: ComputeChoiceRange {
314 min: default_min,
315 max: ceiling,
316 default: default_min,
317 },
318 max: ComputeChoiceRange {
319 min: default_min.max(1),
320 max: ceiling,
321 default: default_max.clamp(default_min.max(1), offered),
322 },
323 }
324}
325
326pub fn check_pool_capacity(
330 platform: Platform,
331 pool_id: &str,
332 selection: &ComputePoolSelection,
333 requirements: &WorkloadRequirements,
334) -> std::result::Result<(), String> {
335 let Some(spec) = selection
336 .machine()
337 .and_then(|machine| instance_catalog::find_instance_type(platform, machine))
338 else {
339 return Ok(());
341 };
342 let machines = selection.max_size();
343 let per_machine_cpu = instance_catalog::allocatable_cpu(spec);
344 let per_machine_memory = instance_catalog::allocatable_memory_bytes(spec);
345 if per_machine_cpu + f64::EPSILON < requirements.max_cpu_per_container
346 || per_machine_memory < requirements.max_memory_per_container
347 {
348 return Err(format!(
349 "Pool '{pool_id}' machine '{}' cannot fit one replica after reserving host resources",
350 spec.name,
351 ));
352 }
353 let cpu = f64::from(machines) * per_machine_cpu;
354 let memory = u64::from(machines) * per_machine_memory;
355 if cpu + f64::EPSILON >= requirements.total_cpu_at_desired
356 && memory >= requirements.total_memory_bytes_at_desired
357 {
358 return Ok(());
359 }
360 const GIB: f64 = 1024.0 * 1024.0 * 1024.0;
361 Err(format!(
362 "Pool '{pool_id}' is too small for its workloads: {machines} x {} has {cpu:.2} vCPU and {:.1} GiB, and the workloads request {:.2} vCPU and {:.1} GiB at their desired replica counts",
363 spec.name,
364 memory as f64 / GIB,
365 requirements.total_cpu_at_desired,
366 requirements.total_memory_bytes_at_desired as f64 / GIB,
367 ))
368}
369
370fn collect_workload_groups(stack: &Stack) -> Result<HashMap<String, PlannedGroup>, ErrorData> {
371 let mut groups: HashMap<String, Vec<Workload>> = HashMap::new();
372
373 for entry in stack.resources.values() {
374 if let Some(container) = entry.config.downcast_ref::<Container>() {
375 groups
376 .entry(
377 container
378 .pool
379 .clone()
380 .unwrap_or_else(|| needed_container_pool(container).to_string()),
381 )
382 .or_default()
383 .push(Workload::from_container(container)?);
384 }
385 if let Some(daemon) = entry.config.downcast_ref::<Daemon>() {
386 if daemon.cluster.is_some() {
387 groups
388 .entry(daemon.pool.clone().unwrap_or_else(|| "general".to_string()))
389 .or_default()
390 .push(Workload::from_daemon(daemon)?);
391 }
392 }
393 }
394
395 let mut planned = HashMap::new();
396 for (pool_id, workloads) in groups {
397 let requirements = aggregate_workloads(&workloads);
398 let requires_failure_domain = workloads
399 .iter()
400 .any(|workload| workload.requires_failure_domain);
401 let min_size = default_min_machines(&requirements);
402 let max_size = default_max_machines(&requirements);
403 planned.insert(
404 pool_id,
405 PlannedGroup {
406 workloads: workloads.into_iter().map(|w| w.id).collect(),
407 workload_requirements: Some(requirements.clone()),
408 scale: CapacityGroupScalePolicy::from_selected_bounds(min_size, max_size),
410 requirements,
411 requires_failure_domain,
412 generated: true,
413 },
414 );
415 }
416 Ok(planned)
417}
418
419fn merge_explicit_compute_groups(
420 stack: &Stack,
421 groups: &mut HashMap<String, PlannedGroup>,
422) -> Result<(), ErrorData> {
423 for entry in stack.resources.values() {
424 let Some(cluster) = entry.config.downcast_ref::<crate::ComputeCluster>() else {
425 continue;
426 };
427 for group in &cluster.capacity_groups {
428 let explicit_requirements = profile_to_requirements(
429 group.profile.as_ref(),
430 group.nested_virtualization.unwrap_or(false),
431 );
432 let scale = group.scale_policy.clone().unwrap_or_else(|| {
433 CapacityGroupScalePolicy::from_selected_bounds(group.min_size, group.max_size)
434 });
435 groups
436 .entry(group.group_id.clone())
437 .and_modify(|planned| {
438 if group.instance_type.is_none() {
439 merge_requirements(&mut planned.requirements, &explicit_requirements);
440 } else {
441 planned.requirements.architecture = planned
448 .requirements
449 .architecture
450 .or(explicit_requirements.architecture);
451 planned.requirements.nested_virt |= explicit_requirements.nested_virt;
452 }
453 planned.scale = merge_scale_policy(&planned.scale, &scale);
454 planned.generated = false;
455 })
456 .or_insert_with(|| PlannedGroup {
457 workloads: Vec::new(),
458 workload_requirements: None,
459 scale,
460 requirements: explicit_requirements,
461 requires_failure_domain: false,
462 generated: false,
463 });
464 }
465 }
466 Ok(())
467}
468
469fn recommended_selection(
470 platform: Platform,
471 requirements: &WorkloadRequirements,
472 scale: &CapacityGroupScalePolicy,
473 requires_failure_domain: bool,
474 generated: bool,
475 machine_override: Option<&str>,
476) -> Result<ComputePoolSelection, ErrorData> {
477 let machine = match platform {
478 Platform::Aws | Platform::Gcp | Platform::Azure => Some(
479 match instance_catalog::select_instance_type(platform, requirements) {
480 Ok(selection) => selection.instance_type.to_string(),
481 Err(message) => {
482 let fallback = machine_override
486 .and_then(|machine| instance_catalog::find_instance_type(platform, machine))
487 .filter(|spec| {
488 instance_satisfies(
489 spec,
490 requirements,
491 requirements.architecture.unwrap_or(spec.architecture),
492 )
493 });
494 fallback.map(|spec| spec.name.to_string()).ok_or_else(|| {
495 AlienError::new(ErrorData::GenericError {
496 message: format!("Failed to select {platform} machine: {message}"),
497 })
498 })?
499 }
500 },
501 ),
502 Platform::Local | Platform::Kubernetes | Platform::Machines | Platform::Test => None,
503 };
504
505 let failure_domains = (requires_failure_domain
506 && matches!(platform, Platform::Aws | Platform::Gcp | Platform::Azure))
507 .then(default_persistent_failure_domains);
508
509 match scale {
510 CapacityGroupScalePolicy::Fixed { machines } => Ok(ComputePoolSelection::Fixed {
511 machines: machines.default.max(1),
512 machine,
513 failure_domains,
514 }),
515 CapacityGroupScalePolicy::Autoscale { min, max }
517 if generated && min.default == max.default =>
518 {
519 Ok(ComputePoolSelection::Fixed {
520 machines: min.default.max(1),
521 machine,
522 failure_domains,
523 })
524 }
525 CapacityGroupScalePolicy::Autoscale { min, max } => Ok(ComputePoolSelection::Autoscale {
526 min: min.default,
527 max: max.default.max(min.default),
528 machine,
529 failure_domains,
530 }),
531 }
532}
533
534pub fn validate_compute_pool_selection(
537 platform: Platform,
538 pool_id: &str,
539 selection: &ComputePoolSelection,
540 requirements: &WorkloadRequirements,
541 scale: &CapacityGroupScalePolicy,
542) -> Vec<String> {
543 let mut errors = Vec::new();
544 if let Err(message) = selection.validate() {
545 errors.push(message);
546 }
547 if let Err(message) = validate_selection_against_scale(selection, scale) {
548 errors.push(format!("Pool '{pool_id}' {message}"));
549 }
550 if matches!(platform, Platform::Aws | Platform::Gcp | Platform::Azure) {
551 match selection.machine() {
552 Some(machine) => match instance_catalog::find_instance_type(platform, machine) {
553 Some(spec) => {
554 let architecture = requirements.architecture.unwrap_or(spec.architecture);
555 if !instance_satisfies(spec, requirements, architecture) {
556 errors.push(format!(
557 "{} machine '{}' does not satisfy pool '{}' requirements",
558 platform, machine, pool_id
559 ));
560 }
561 }
562 None => errors.push(format!(
563 "Unknown {} machine '{}' for pool '{}'",
564 platform, machine, pool_id
565 )),
566 },
567 None => errors.push(format!(
568 "Pool '{}' requires a provider machine on {}",
569 pool_id, platform
570 )),
571 }
572 }
573 errors
574}
575
576pub fn capacity_group_requirements(group: &CapacityGroup) -> WorkloadRequirements {
578 profile_to_requirements(
579 group.profile.as_ref(),
580 group.nested_virtualization.unwrap_or(false),
581 )
582}
583
584fn machine_options(
585 platform: Platform,
586 requirements: &WorkloadRequirements,
587 selected_machine: Option<&str>,
588 recommended_machine: Option<&str>,
589) -> Result<Vec<ComputeMachineOption>, ErrorData> {
590 if !matches!(platform, Platform::Aws | Platform::Gcp | Platform::Azure) {
591 return Ok(Vec::new());
592 }
593 let resolved_architecture = requirements
594 .architecture
595 .or_else(|| {
596 selected_machine.and_then(|machine| {
597 instance_catalog::find_instance_type(platform, machine)
598 .map(|spec| spec.architecture)
599 })
600 })
601 .or_else(|| {
602 recommended_machine
603 .and_then(|machine| instance_catalog::find_instance_type(platform, machine))
604 .map(|spec| spec.architecture)
605 })
606 .ok_or_else(|| {
607 AlienError::new(ErrorData::GenericError {
608 message: format!("Selected {platform} machine has no CPU architecture"),
609 })
610 })?;
611
612 let mut options: Vec<ComputeMachineOption> = instance_catalog::catalog_for_platform(platform)
613 .into_iter()
614 .filter(|spec| instance_satisfies(spec, requirements, resolved_architecture))
615 .map(|spec| ComputeMachineOption {
616 machine: spec.name.to_string(),
617 profile: spec.to_machine_profile_for_storage(requirements.max_ephemeral_storage_bytes),
618 recommended: Some(spec.name) == recommended_machine
619 || Some(spec.name) == selected_machine,
620 })
621 .collect();
622 options.sort_by(|a, b| a.machine.cmp(&b.machine));
623 Ok(options)
624}
625
626fn instance_satisfies(
627 spec: &instance_catalog::InstanceTypeSpec,
628 requirements: &WorkloadRequirements,
629 resolved_architecture: Architecture,
630) -> bool {
631 if spec.architecture != resolved_architecture {
632 return false;
633 }
634 if requirements.nested_virt && !spec.is_nested_virt_capable() {
635 return false;
636 }
637 if spec.vcpu < requirements.max_cpu_per_container.ceil() as u32 {
638 return false;
639 }
640 if spec.memory_bytes < requirements.max_memory_per_container {
641 return false;
642 }
643 if spec.has_configurable_ephemeral_storage() {
644 if crate::instance_catalog::max_configurable_ephemeral_storage_bytes(spec.platform)
645 .is_none_or(|max| requirements.max_ephemeral_storage_bytes > max)
646 {
647 return false;
648 }
649 } else if spec.ephemeral_storage_bytes < requirements.max_ephemeral_storage_bytes {
650 return false;
651 }
652 match (&requirements.gpu, spec.gpu) {
653 (Some(required), Some(actual)) => {
654 (required.gpu_type == "any" || required.gpu_type == actual.gpu_type)
655 && actual.count >= required.count
656 }
657 (Some(_), None) => false,
658 (None, _) => true,
659 }
660}
661
662#[derive(Debug, Clone)]
663struct Workload {
664 id: String,
665 cpu: f64,
666 memory_bytes: u64,
667 desired_replicas: f64,
668 max_replicas: f64,
669 ephemeral_storage_bytes: u64,
670 gpu: Option<GpuSpec>,
671 requires_failure_domain: bool,
672}
673
674impl Workload {
675 fn from_container(container: &Container) -> Result<Self, ErrorData> {
676 let cpu = parse_cpu(&container.id, &container.cpu)?;
677 let memory_bytes = parse_memory(&container.id, &container.memory)?;
678 let desired_replicas = container
679 .autoscaling
680 .as_ref()
681 .map(|a| a.desired)
682 .or(container.replicas)
683 .unwrap_or(1) as f64;
684 let max_replicas = container
685 .autoscaling
686 .as_ref()
687 .map(|a| a.max)
688 .or(container.replicas)
689 .unwrap_or(1) as f64;
690 let ephemeral_storage_bytes = container
691 .ephemeral_storage
692 .as_deref()
693 .map(instance_catalog::parse_memory_bytes)
694 .transpose()
695 .map_err(|message| {
696 AlienError::new(ErrorData::GenericError {
697 message: format!(
698 "Failed to parse ephemeral storage for '{}': {message}",
699 container.id
700 ),
701 })
702 })?
703 .unwrap_or(0);
704
705 Ok(Self {
706 id: container.id.clone(),
707 cpu,
708 memory_bytes,
709 desired_replicas,
710 max_replicas,
711 ephemeral_storage_bytes,
712 gpu: container.gpu.as_ref().map(|gpu| GpuSpec {
713 gpu_type: gpu.gpu_type.clone(),
714 count: gpu.count,
715 }),
716 requires_failure_domain: container.stateful && container.persistent_storage.is_some(),
717 })
718 }
719
720 fn from_daemon(daemon: &Daemon) -> Result<Self, ErrorData> {
721 Ok(Self {
722 id: daemon.id.clone(),
723 cpu: parse_cpu(&daemon.id, &daemon.cpu)?,
724 memory_bytes: parse_memory(&daemon.id, &daemon.memory)?,
725 desired_replicas: 1.0,
726 max_replicas: 1.0,
727 ephemeral_storage_bytes: 0,
728 gpu: None,
729 requires_failure_domain: false,
730 })
731 }
732}
733
734fn parse_cpu(resource_id: &str, spec: &ResourceSpec) -> Result<f64, ErrorData> {
735 instance_catalog::parse_cpu(&spec.desired).map_err(|message| {
736 AlienError::new(ErrorData::GenericError {
737 message: format!(
738 "Failed to parse CPU requirement '{}' for '{}': {message}",
739 spec.desired, resource_id
740 ),
741 })
742 })
743}
744
745fn parse_memory(resource_id: &str, spec: &ResourceSpec) -> Result<u64, ErrorData> {
746 instance_catalog::parse_memory_bytes(&spec.desired).map_err(|message| {
747 AlienError::new(ErrorData::GenericError {
748 message: format!(
749 "Failed to parse memory requirement '{}' for '{}': {message}",
750 spec.desired, resource_id
751 ),
752 })
753 })
754}
755
756fn aggregate_workloads(workloads: &[Workload]) -> WorkloadRequirements {
757 let mut requirements = default_requirements();
758 requirements.total_cpu_at_desired = 0.0;
759 requirements.total_memory_bytes_at_desired = 0;
760 requirements.total_cpu_at_max = 0.0;
761 requirements.total_memory_bytes_at_max = 0;
762 requirements.max_cpu_per_container = 0.0;
763 requirements.max_memory_per_container = 0;
764 requirements.max_ephemeral_storage_bytes = 0;
765 requirements.gpu = None;
766
767 for workload in workloads {
768 requirements.total_cpu_at_desired += workload.cpu * workload.desired_replicas;
769 requirements.total_cpu_at_max += workload.cpu * workload.max_replicas;
770 requirements.total_memory_bytes_at_desired +=
771 (workload.memory_bytes as f64 * workload.desired_replicas) as u64;
772 requirements.total_memory_bytes_at_max +=
773 (workload.memory_bytes as f64 * workload.max_replicas) as u64;
774 requirements.max_cpu_per_container = requirements.max_cpu_per_container.max(workload.cpu);
775 requirements.max_memory_per_container = requirements
776 .max_memory_per_container
777 .max(workload.memory_bytes);
778 requirements.max_ephemeral_storage_bytes = requirements
779 .max_ephemeral_storage_bytes
780 .max(workload.ephemeral_storage_bytes);
781 if requirements.gpu.is_none() {
782 requirements.gpu = workload.gpu.clone();
783 }
784 }
785 requirements
786}
787
788fn default_requirements() -> WorkloadRequirements {
789 WorkloadRequirements {
790 total_cpu_at_desired: 1.0,
791 total_memory_bytes_at_desired: 2 * 1024 * 1024 * 1024,
792 total_cpu_at_max: 1.0,
793 total_memory_bytes_at_max: 2 * 1024 * 1024 * 1024,
794 max_cpu_per_container: 1.0,
795 max_memory_per_container: 2 * 1024 * 1024 * 1024,
796 max_ephemeral_storage_bytes: 0,
797 gpu: None,
798 architecture: None,
799 nested_virt: false,
800 }
801}
802
803fn profile_to_requirements(
804 profile: Option<&MachineProfile>,
805 nested_virt: bool,
806) -> WorkloadRequirements {
807 let Some(profile) = profile else {
808 return WorkloadRequirements {
809 nested_virt,
810 ..default_requirements()
811 };
812 };
813 let cpu = instance_catalog::parse_cpu(&profile.cpu).unwrap_or(1.0);
814 WorkloadRequirements {
815 total_cpu_at_desired: cpu,
816 total_memory_bytes_at_desired: profile.memory_bytes,
817 total_cpu_at_max: cpu,
818 total_memory_bytes_at_max: profile.memory_bytes,
819 max_cpu_per_container: cpu,
820 max_memory_per_container: profile.memory_bytes,
821 max_ephemeral_storage_bytes: profile.ephemeral_storage_bytes,
822 gpu: profile.gpu.clone(),
823 architecture: profile.architecture,
824 nested_virt,
825 }
826}
827
828fn merge_requirements(existing: &mut WorkloadRequirements, declared: &WorkloadRequirements) {
829 existing.total_cpu_at_desired = existing
830 .total_cpu_at_desired
831 .max(declared.total_cpu_at_desired);
832 existing.total_memory_bytes_at_desired = existing
833 .total_memory_bytes_at_desired
834 .max(declared.total_memory_bytes_at_desired);
835 existing.total_cpu_at_max = existing.total_cpu_at_max.max(declared.total_cpu_at_max);
836 existing.total_memory_bytes_at_max = existing
837 .total_memory_bytes_at_max
838 .max(declared.total_memory_bytes_at_max);
839 existing.max_cpu_per_container = existing
840 .max_cpu_per_container
841 .max(declared.max_cpu_per_container);
842 existing.max_memory_per_container = existing
843 .max_memory_per_container
844 .max(declared.max_memory_per_container);
845 existing.max_ephemeral_storage_bytes = existing
846 .max_ephemeral_storage_bytes
847 .max(declared.max_ephemeral_storage_bytes);
848 if existing.gpu.is_none() {
849 existing.gpu = declared.gpu.clone();
850 }
851 existing.nested_virt |= declared.nested_virt;
852 if existing.architecture.is_none() {
853 existing.architecture = declared.architecture;
854 }
855}
856
857fn requirements_to_profile(requirements: &WorkloadRequirements) -> MachineProfile {
858 MachineProfile {
859 cpu: requirements.max_cpu_per_container.to_string(),
860 memory_bytes: requirements.max_memory_per_container,
861 ephemeral_storage_bytes: requirements.max_ephemeral_storage_bytes,
862 architecture: requirements.architecture,
863 gpu: requirements.gpu.clone(),
864 }
865}
866
867fn merge_scale_policy(
868 existing: &CapacityGroupScalePolicy,
869 declared: &CapacityGroupScalePolicy,
870) -> CapacityGroupScalePolicy {
871 match (existing, declared) {
872 (
873 CapacityGroupScalePolicy::Fixed {
874 machines: existing_machines,
875 },
876 CapacityGroupScalePolicy::Fixed {
877 machines: declared_machines,
878 },
879 ) => CapacityGroupScalePolicy::Fixed {
880 machines: merge_choice_range(existing_machines, declared_machines),
881 },
882 (_, declared) => declared.clone(),
883 }
884}
885
886fn merge_choice_range(
887 existing: &ComputeChoiceRange,
888 declared: &ComputeChoiceRange,
889) -> ComputeChoiceRange {
890 ComputeChoiceRange {
891 min: existing.min.max(declared.min),
892 max: existing.max.max(declared.max),
893 default: declared.default,
894 }
895}
896
897fn validate_selection_against_scale(
898 selection: &ComputePoolSelection,
899 scale: &CapacityGroupScalePolicy,
900) -> std::result::Result<(), String> {
901 match (selection, scale) {
904 (
905 ComputePoolSelection::Fixed { machines, .. },
906 CapacityGroupScalePolicy::Fixed { machines: allowed },
907 ) => check_machine_count("fixed machine count", *machines, allowed.min, allowed.max),
908 (
909 ComputePoolSelection::Autoscale { min, max, .. },
910 CapacityGroupScalePolicy::Autoscale {
911 min: allowed_min,
912 max: allowed_max,
913 },
914 ) => {
915 check_machine_count("autoscale minimum", *min, allowed_min.min, allowed_min.max)?;
916 check_machine_count("autoscale maximum", *max, allowed_max.min, allowed_max.max)
917 }
918 (
919 ComputePoolSelection::Fixed { machines, .. },
920 CapacityGroupScalePolicy::Autoscale {
921 min: allowed_min,
922 max: allowed_max,
923 },
924 ) => {
925 let lowest = allowed_min.min.max(allowed_max.min);
926 let highest = allowed_min.max.min(allowed_max.max);
927 if lowest > highest {
928 return Err(format!(
929 "must use autoscale mode (minimum {}-{}, maximum {}-{})",
930 allowed_min.min, allowed_min.max, allowed_max.min, allowed_max.max
931 ));
932 }
933 check_machine_count("fixed machine count", *machines, lowest, highest)
934 }
935 (
936 ComputePoolSelection::Autoscale { min, max, .. },
937 CapacityGroupScalePolicy::Fixed { machines: allowed },
938 ) => {
939 if min != max {
940 return Err(format!(
941 "must use fixed mode with {}-{} machines; autoscale {min}-{max} is not allowed",
942 allowed.min, allowed.max
943 ));
944 }
945 check_machine_count("machine count", *min, allowed.min, allowed.max)
946 }
947 }
948}
949
950fn check_machine_count(
951 label: &str,
952 value: u32,
953 lowest: u32,
954 highest: u32,
955) -> std::result::Result<(), String> {
956 if (lowest..=highest).contains(&value) {
957 Ok(())
958 } else {
959 Err(format!(
960 "{label} {value} is outside the allowed range {lowest}-{highest}"
961 ))
962 }
963}
964
965fn needed_container_pool(container: &Container) -> &'static str {
966 if container.stateful && container.persistent_storage.is_some() {
967 return "stateful";
968 }
969 if container.gpu.is_some() {
970 return "gpu";
971 }
972 if let Some(storage) = &container.ephemeral_storage {
973 if instance_catalog::parse_memory_bytes(storage).unwrap_or(0) > 200 * 1024 * 1024 * 1024 {
974 return "storage";
975 }
976 }
977 "general"
978}
979
980fn default_min_machines(requirements: &WorkloadRequirements) -> u32 {
981 if requirements.total_cpu_at_desired > 0.0 || requirements.total_memory_bytes_at_desired > 0 {
982 1
983 } else {
984 0
985 }
986}
987
988fn default_max_machines(requirements: &WorkloadRequirements) -> u32 {
989 let min = default_min_machines(requirements);
990 let by_cpu =
991 (requirements.total_cpu_at_max / requirements.max_cpu_per_container.max(1.0)).ceil() as u32;
992 let by_mem = requirements
993 .total_memory_bytes_at_max
994 .div_ceil(requirements.max_memory_per_container.max(1)) as u32;
995 min.max(by_cpu).max(by_mem).max(1)
996}
997
998#[cfg(test)]
999mod tests {
1000 use super::*;
1001 use crate::{
1002 instance_catalog::Architecture, CapacityGroup, CapacityGroupScalePolicy,
1003 ComputeChoiceRange, ComputeCluster, ComputeSettings, ContainerCode, DaemonCode, Resource,
1004 ResourceEntry, ResourceLifecycle, Stack, Storage, Worker, WorkerCode,
1005 };
1006
1007 #[test]
1008 fn deployment_resources_drive_machine_planning_on_every_cloud() {
1009 let mut stack = stack_with_container();
1010 let container = stack
1011 .resources
1012 .get_mut("api")
1013 .unwrap()
1014 .config
1015 .downcast_mut::<Container>()
1016 .unwrap();
1017 container.resource_choices = Some(crate::ContainerResourceChoices {
1018 cpu: Some(crate::ResourceChoiceRange {
1019 min: "0.5".into(),
1020 max: "8".into(),
1021 default: "2".into(),
1022 }),
1023 memory: Some(crate::ResourceChoiceRange {
1024 min: "512Mi".into(),
1025 max: "16Gi".into(),
1026 default: "4Gi".into(),
1027 }),
1028 });
1029 let settings: ComputeSettings = serde_json::from_value(serde_json::json!({
1030 "pools": { "general": { "mode": "fixed", "machines": 1 } },
1031 "containers": { "api": { "cpu": 8, "memory": "16Gi" } }
1032 }))
1033 .unwrap();
1034 for platform in [Platform::Aws, Platform::Gcp, Platform::Azure] {
1035 let default_plan = plan_compute(&stack, platform, None).unwrap();
1036 let selected_plan = plan_compute(&stack, platform, Some(&settings)).unwrap();
1037 let pool = &selected_plan.pools[0];
1038 assert!(pool.errors.is_empty(), "{platform}: {:?}", pool.errors);
1039 assert_eq!(pool.requirements.cpu, "8");
1040 assert_eq!(pool.requirements.memory_bytes, 16 * 1024 * 1024 * 1024);
1041 assert_ne!(
1042 pool.selected.machine(),
1043 default_plan.pools[0].selected.machine()
1044 );
1045 assert_eq!(selected_plan.containers[0].cpu.min, "8");
1046 assert_eq!(selected_plan.containers[0].memory.desired, "16Gi");
1047 assert_eq!(pool.selected.min_size(), 1);
1048 assert_eq!(pool.selected.max_size(), 1);
1049 }
1050 assert_eq!(
1051 settings.pools["general"].machine(),
1052 None,
1053 "resolving a plan must preserve automatic machine intent"
1054 );
1055 let original = stack.resources["api"]
1056 .config
1057 .downcast_ref::<Container>()
1058 .unwrap();
1059 assert_eq!(
1060 original.cpu.desired, "2",
1061 "planning must not mutate release defaults"
1062 );
1063 }
1064
1065 #[test]
1066 fn explicit_machine_can_fit_a_low_cpu_high_memory_workload() {
1067 let mut stack = stack_with_container();
1068 let container = stack
1069 .resources
1070 .get_mut("api")
1071 .unwrap()
1072 .config
1073 .downcast_mut::<Container>()
1074 .unwrap();
1075 container.cpu = ResourceSpec {
1076 min: "0.5".into(),
1077 desired: "0.5".into(),
1078 };
1079 container.memory = ResourceSpec {
1080 min: "8Gi".into(),
1081 desired: "8Gi".into(),
1082 };
1083 let settings: ComputeSettings = serde_json::from_value(serde_json::json!({
1084 "pools": {"general":{"mode":"fixed","machines":1,"machine":"n2-standard-4"}}
1085 }))
1086 .unwrap();
1087 let plan = plan_compute(&stack, Platform::Gcp, Some(&settings)).unwrap();
1088 assert!(
1089 plan.pools[0].errors.is_empty(),
1090 "{:?}",
1091 plan.pools[0].errors
1092 );
1093 assert_eq!(plan.pools[0].selected.machine(), Some("n2-standard-4"));
1094 }
1095
1096 #[test]
1097 fn pin_must_leave_host_capacity_for_one_replica() {
1098 let mut stack = stack_with_container();
1099 stack
1100 .resources
1101 .get_mut("api")
1102 .unwrap()
1103 .config
1104 .downcast_mut::<Container>()
1105 .unwrap()
1106 .cpu = ResourceSpec {
1107 min: "4".into(),
1108 desired: "4".into(),
1109 };
1110 let settings: ComputeSettings = serde_json::from_value(serde_json::json!({
1111 "pools":{"general":{"mode":"fixed","machines":1,"machine":"t3.xlarge"}}
1112 }))
1113 .unwrap();
1114 let plan = plan_compute(&stack, Platform::Aws, Some(&settings)).unwrap();
1115 assert!(
1116 plan.pools[0]
1117 .errors
1118 .iter()
1119 .any(|error| error.contains("host resources")),
1120 "{:?}",
1121 plan.pools[0].errors
1122 );
1123 }
1124
1125 #[test]
1126 fn valid_pin_does_not_require_automatic_recommendation_headroom() {
1127 let machine = instance_catalog::catalog_for_platform(Platform::Gcp)
1128 .into_iter()
1129 .max_by_key(|machine| machine.memory_bytes)
1130 .unwrap();
1131 let mut stack = stack_with_container();
1132 let container = stack
1133 .resources
1134 .get_mut("api")
1135 .unwrap()
1136 .config
1137 .downcast_mut::<Container>()
1138 .unwrap();
1139 container.cpu = ResourceSpec {
1140 min: "0.5".into(),
1141 desired: "0.5".into(),
1142 };
1143 let memory = instance_catalog::allocatable_memory_bytes(machine).to_string();
1144 container.memory = ResourceSpec {
1145 min: memory.clone(),
1146 desired: memory,
1147 };
1148 assert!(plan_compute(&stack, Platform::Gcp, None).is_err());
1149 let settings: ComputeSettings = serde_json::from_value(serde_json::json!({
1150 "pools":{"general":{"mode":"fixed","machines":1,"machine":machine.name}}
1151 }))
1152 .unwrap();
1153 let plan = plan_compute(&stack, Platform::Gcp, Some(&settings)).unwrap();
1154 assert!(
1155 plan.pools[0].errors.is_empty(),
1156 "{:?}",
1157 plan.pools[0].errors
1158 );
1159 assert_eq!(plan.pools[0].selected.machine(), Some(machine.name));
1160 }
1161
1162 #[test]
1163 fn invalid_deployment_resources_fail_before_capacity_planning() {
1164 let mut stack = stack_with_container();
1165 stack
1166 .resources
1167 .get_mut("api")
1168 .unwrap()
1169 .config
1170 .downcast_mut::<Container>()
1171 .unwrap()
1172 .resource_choices = Some(crate::ContainerResourceChoices {
1173 cpu: Some(crate::ResourceChoiceRange {
1174 min: "0.5".into(),
1175 max: "4".into(),
1176 default: "2".into(),
1177 }),
1178 memory: Some(crate::ResourceChoiceRange {
1179 min: "512Mi".into(),
1180 max: "8Gi".into(),
1181 default: "4Gi".into(),
1182 }),
1183 });
1184 for selection in [
1185 serde_json::json!({"api":{"cpu": 0}}),
1186 serde_json::json!({"api":{"cpu": 5}}),
1187 serde_json::json!({"api":{"memory": "16Gi"}}),
1188 serde_json::json!({"api":{"memory": "NaNGi"}}),
1189 serde_json::json!({"api":{"memory": "-1Gi"}}),
1190 serde_json::json!({"missing":{"cpu": 1}}),
1191 ] {
1192 let settings: ComputeSettings =
1193 serde_json::from_value(serde_json::json!({"containers": selection})).unwrap();
1194 let error = plan_compute(&stack, Platform::Aws, Some(&settings)).unwrap_err();
1195 assert_eq!(error.code, "CONTAINER_RESOURCE_SELECTION_INVALID");
1196 }
1197 }
1198
1199 fn stack_with_container() -> Stack {
1200 let container = Container::new("api".to_string())
1201 .code(ContainerCode::Image {
1202 image: "api:latest".to_string(),
1203 })
1204 .cpu(ResourceSpec {
1205 min: "1".to_string(),
1206 desired: "2".to_string(),
1207 })
1208 .memory(ResourceSpec {
1209 min: "2Gi".to_string(),
1210 desired: "4Gi".to_string(),
1211 })
1212 .permissions("api".to_string())
1213 .build();
1214 Stack {
1215 id: "test".to_string(),
1216 resources: [(
1217 "api".to_string(),
1218 ResourceEntry {
1219 config: Resource::new(container),
1220 lifecycle: ResourceLifecycle::Live,
1221 dependencies: Vec::new(),
1222 remote_access: false,
1223 enabled_when: None,
1224 },
1225 )]
1226 .into_iter()
1227 .collect(),
1228 permissions: crate::permissions::PermissionsConfig::default(),
1229 supported_platforms: None,
1230 inputs: vec![],
1231 dynamic_container_repositories: Vec::new(),
1232 dynamic_container_image_resources: Vec::new(),
1233 operations: None,
1234 }
1235 }
1236
1237 #[test]
1238 fn serverless_workers_and_storage_do_not_allocate_machine_pools() {
1239 let mut stack = stack_with_container();
1240 stack.resources.clear();
1241 for (id, resource) in [
1242 (
1243 "api",
1244 Resource::new(
1245 Worker::new("api".to_string())
1246 .code(WorkerCode::Image {
1247 image: "example/worker:v1".to_string(),
1248 })
1249 .permissions("execution".to_string())
1250 .build(),
1251 ),
1252 ),
1253 (
1254 "data",
1255 Resource::new(Storage::new("data".to_string()).build()),
1256 ),
1257 ] {
1258 stack.resources.insert(
1259 id.to_string(),
1260 ResourceEntry {
1261 config: resource,
1262 lifecycle: ResourceLifecycle::Frozen,
1263 dependencies: Vec::new(),
1264 remote_access: false,
1265 enabled_when: None,
1266 },
1267 );
1268 }
1269
1270 for platform in [Platform::Aws, Platform::Gcp, Platform::Azure] {
1271 let plan = plan_compute(&stack, platform, None).expect("plan should build");
1272 assert!(plan.pools.is_empty(), "serverless plan allocated {plan:?}");
1273 }
1274 }
1275
1276 #[test]
1277 fn empty_stack_does_not_allocate_machine_pools() {
1278 let mut stack = stack_with_container();
1279 stack.resources.clear();
1280 let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");
1281 assert!(plan.pools.is_empty());
1282 }
1283
1284 #[test]
1285 fn explicit_cluster_without_workloads_only_plans_declared_pools() {
1286 let mut stack = stack_with_container();
1287 stack.resources.clear();
1288 let cluster = ComputeCluster::new("compute".to_string())
1289 .capacity_group(CapacityGroup {
1290 group_id: "batch".to_string(),
1291 instance_type: Some("t4g.medium".to_string()),
1292 profile: None,
1293 min_size: 1,
1294 max_size: 1,
1295 scale_policy: None,
1296 nested_virtualization: None,
1297 })
1298 .build();
1299 stack.resources.insert(
1300 "compute".to_string(),
1301 ResourceEntry {
1302 config: Resource::new(cluster),
1303 lifecycle: ResourceLifecycle::Frozen,
1304 dependencies: Vec::new(),
1305 remote_access: false,
1306 enabled_when: None,
1307 },
1308 );
1309
1310 let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");
1311 assert_eq!(plan.pools.len(), 1);
1312 assert_eq!(plan.pools[0].pool_id, "batch");
1313 assert!(plan.pools[0].errors.is_empty());
1314 }
1315
1316 #[test]
1317 fn cloud_plan_recommends_provider_machine_without_mutating_selection() {
1318 let stack = stack_with_container();
1319
1320 let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");
1321
1322 let pool = plan.pools.first().expect("general pool should exist");
1323 assert_eq!(pool.pool_id, "general");
1324 assert_eq!(pool.workloads, vec!["api"]);
1325 assert!(pool.selected.machine().is_some());
1326 assert!(pool.machines.iter().any(|machine| machine.recommended));
1327 }
1328
1329 #[test]
1330 fn cloud_plan_only_offers_machines_matching_the_default_image_target() {
1331 let stack = stack_with_container();
1332
1333 for platform in [Platform::Aws, Platform::Gcp, Platform::Azure] {
1334 let plan = plan_compute(&stack, platform, None).expect("plan should build");
1335 let pool = plan.pools.first().expect("general pool should exist");
1336 let expected = instance_catalog::default_architecture(platform)
1337 .expect("managed cloud should have a default architecture");
1338
1339 assert!(pool.machines.iter().all(|machine| {
1340 instance_catalog::find_instance_type(platform, &machine.machine)
1341 .is_some_and(|spec| spec.architecture == expected)
1342 }));
1343 }
1344 }
1345
1346 #[test]
1347 fn selected_machine_is_preserved_as_static_deployment_choice() {
1348 let stack = stack_with_container();
1349 let settings = ComputeSettings {
1350 containers: Default::default(),
1351 pools: [(
1352 "general".to_string(),
1353 ComputePoolSelection::Fixed {
1354 machines: 1,
1355 machine: Some("m7g.xlarge".to_string()),
1356 failure_domains: None,
1357 },
1358 )]
1359 .into_iter()
1360 .collect(),
1361 };
1362
1363 let plan = plan_compute(&stack, Platform::Aws, Some(&settings)).expect("plan should build");
1364
1365 let pool = plan.pools.first().expect("general pool should exist");
1366 assert_eq!(pool.selected.machine(), Some("m7g.xlarge"));
1367 assert!(pool.errors.is_empty());
1368 }
1369
1370 #[test]
1371 fn configurable_cloud_disk_satisfies_storage_above_fixed_local_catalog() {
1372 let mut stack = stack_with_container();
1373 let container = stack
1374 .resources
1375 .get_mut("api")
1376 .and_then(|entry| entry.config.downcast_mut::<Container>())
1377 .expect("test stack should contain a container");
1378 container.ephemeral_storage = Some("8000Gi".to_string());
1379 let settings = ComputeSettings {
1380 containers: Default::default(),
1381 pools: [(
1382 "storage".to_string(),
1383 ComputePoolSelection::Fixed {
1384 machines: 1,
1385 machine: Some("m8i.2xlarge".to_string()),
1386 failure_domains: None,
1387 },
1388 )]
1389 .into_iter()
1390 .collect(),
1391 };
1392
1393 let plan = plan_compute(&stack, Platform::Aws, Some(&settings)).expect("plan should build");
1394 let pool = plan.pools.first().expect("general pool should exist");
1395 assert!(pool.errors.is_empty());
1396 let selected = pool
1397 .machines
1398 .iter()
1399 .find(|option| option.machine == "m8i.2xlarge")
1400 .expect("selected machine should remain an option");
1401 assert_eq!(
1402 selected.profile.ephemeral_storage_bytes,
1403 8_000 * 1024 * 1024 * 1024
1404 );
1405 }
1406
1407 #[test]
1408 fn selected_machine_defines_architecture_when_workloads_do_not() {
1409 let stack = stack_with_container();
1410 let settings = ComputeSettings {
1411 containers: Default::default(),
1412 pools: [(
1413 "general".to_string(),
1414 ComputePoolSelection::Fixed {
1415 machines: 1,
1416 machine: Some("m7i.xlarge".to_string()),
1417 failure_domains: None,
1418 },
1419 )]
1420 .into_iter()
1421 .collect(),
1422 };
1423
1424 let plan = plan_compute(&stack, Platform::Aws, Some(&settings)).expect("plan should build");
1425
1426 let pool = plan.pools.first().expect("general pool should exist");
1427 assert!(pool.errors.is_empty());
1428 assert!(pool.machines.iter().all(|machine| {
1429 instance_catalog::find_instance_type(Platform::Aws, &machine.machine)
1430 .is_some_and(|spec| spec.architecture == Architecture::X86_64)
1431 }));
1432 }
1433
1434 #[test]
1435 fn graviton4_compute_machine_can_be_selected() {
1436 let stack = stack_with_container();
1437 let settings = ComputeSettings {
1438 containers: Default::default(),
1439 pools: [(
1440 "general".to_string(),
1441 ComputePoolSelection::Fixed {
1442 machines: 1,
1443 machine: Some("c8g.2xlarge".to_string()),
1444 failure_domains: None,
1445 },
1446 )]
1447 .into_iter()
1448 .collect(),
1449 };
1450
1451 let plan = plan_compute(&stack, Platform::Aws, Some(&settings)).expect("plan should build");
1452
1453 let pool = plan.pools.first().expect("general pool should exist");
1454 assert!(pool.errors.is_empty(), "{:?}", pool.errors);
1455 let selected = pool
1456 .machines
1457 .iter()
1458 .find(|option| option.machine == "c8g.2xlarge")
1459 .expect("selected machine should be offered");
1460 assert!(selected.recommended);
1461 assert_eq!(selected.profile.architecture, Some(Architecture::Arm64));
1462 }
1463
1464 #[test]
1465 fn explicit_capacity_group_requirements_are_merged_with_workloads() {
1466 let mut stack = stack_with_container();
1467 let cluster = ComputeCluster::new("compute".to_string())
1468 .capacity_group(CapacityGroup {
1469 group_id: "general".to_string(),
1470 instance_type: None,
1471 profile: Some(MachineProfile {
1472 cpu: "4".to_string(),
1473 memory_bytes: 16 * 1024 * 1024 * 1024,
1474 ephemeral_storage_bytes: 20 * 1024 * 1024 * 1024,
1475 architecture: Some(Architecture::X86_64),
1476 gpu: None,
1477 }),
1478 min_size: 2,
1479 max_size: 5,
1480 scale_policy: None,
1481 nested_virtualization: Some(true),
1482 })
1483 .build();
1484 stack.resources.insert(
1485 "compute".to_string(),
1486 ResourceEntry {
1487 config: Resource::new(cluster),
1488 lifecycle: ResourceLifecycle::Frozen,
1489 dependencies: Vec::new(),
1490 remote_access: false,
1491 enabled_when: None,
1492 },
1493 );
1494
1495 let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");
1496
1497 let pool = plan.pools.first().expect("general pool should exist");
1498 let machine = pool
1499 .selected
1500 .machine()
1501 .expect("AWS selection should include a machine");
1502 let spec = instance_catalog::find_instance_type(Platform::Aws, machine)
1503 .expect("selected machine should exist in the catalog");
1504 assert!(spec.is_nested_virt_capable());
1505 assert_eq!(pool.selected.min_size(), 2);
1506 assert_eq!(pool.selected.max_size(), 5);
1507 assert!(pool.errors.is_empty());
1508 }
1509
1510 #[test]
1511 fn materialized_machine_profile_is_not_counted_as_workload_demand() {
1512 let mut stack = stack_with_container();
1513 let selected = instance_catalog::find_instance_type(Platform::Aws, "m7g.xlarge")
1514 .expect("machine should exist in catalog");
1515 let cluster = ComputeCluster::new("compute".to_string())
1516 .capacity_group(CapacityGroup {
1517 group_id: "general".to_string(),
1518 instance_type: Some(selected.name.to_string()),
1519 profile: Some(selected.to_machine_profile()),
1520 min_size: 1,
1521 max_size: 1,
1522 scale_policy: None,
1523 nested_virtualization: None,
1524 })
1525 .build();
1526 stack.resources.insert(
1527 "compute".to_string(),
1528 ResourceEntry {
1529 config: Resource::new(cluster),
1530 lifecycle: ResourceLifecycle::Frozen,
1531 dependencies: Vec::new(),
1532 remote_access: false,
1533 enabled_when: None,
1534 },
1535 );
1536
1537 let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");
1538 let pool = plan.pools.first().expect("general pool should exist");
1539
1540 assert_eq!(pool.requirements.cpu, "2");
1541 assert_eq!(pool.requirements.memory_bytes, 4 * 1024 * 1024 * 1024);
1542 assert_eq!(pool.recommended.machine(), Some("m7g.xlarge"));
1543 assert!(pool
1544 .machines
1545 .iter()
1546 .any(|option| option.machine == "m7g.xlarge"));
1547 }
1548
1549 #[test]
1550 fn nested_x86_fixed_range_pool_preserves_bounds_and_rejects_graviton() {
1551 let daemon = Daemon::new("vm-runtime-loader".to_string())
1552 .code(DaemonCode::Image {
1553 image: "example.com/vm-runtime:latest".to_string(),
1554 })
1555 .cluster("vm-runtime".to_string())
1556 .cpu(ResourceSpec {
1557 min: "2".to_string(),
1558 desired: "2".to_string(),
1559 })
1560 .memory(ResourceSpec {
1561 min: "4Gi".to_string(),
1562 desired: "4Gi".to_string(),
1563 })
1564 .permissions("loader".to_string())
1565 .build();
1566 let cluster = ComputeCluster::new("vm-runtime".to_string())
1567 .capacity_group(CapacityGroup {
1568 group_id: "general".to_string(),
1569 instance_type: None,
1570 profile: Some(MachineProfile {
1571 cpu: "4".to_string(),
1572 memory_bytes: 16 * 1024 * 1024 * 1024,
1573 ephemeral_storage_bytes: 20 * 1024 * 1024 * 1024,
1574 architecture: Some(Architecture::X86_64),
1575 gpu: None,
1576 }),
1577 min_size: 2,
1578 max_size: 2,
1579 scale_policy: Some(CapacityGroupScalePolicy::Fixed {
1580 machines: ComputeChoiceRange {
1581 min: 1,
1582 max: 5,
1583 default: 2,
1584 },
1585 }),
1586 nested_virtualization: Some(true),
1587 })
1588 .build();
1589 let stack = Stack {
1590 id: "vm-runtime".to_string(),
1591 resources: [
1592 (
1593 "vm-runtime-loader".to_string(),
1594 ResourceEntry {
1595 config: Resource::new(daemon),
1596 lifecycle: ResourceLifecycle::Live,
1597 dependencies: Vec::new(),
1598 remote_access: false,
1599 enabled_when: None,
1600 },
1601 ),
1602 (
1603 "vm-runtime".to_string(),
1604 ResourceEntry {
1605 config: Resource::new(cluster),
1606 lifecycle: ResourceLifecycle::Frozen,
1607 dependencies: Vec::new(),
1608 remote_access: false,
1609 enabled_when: None,
1610 },
1611 ),
1612 ]
1613 .into_iter()
1614 .collect(),
1615 permissions: crate::permissions::PermissionsConfig::default(),
1616 supported_platforms: None,
1617 inputs: vec![],
1618 dynamic_container_repositories: Vec::new(),
1619 dynamic_container_image_resources: Vec::new(),
1620 operations: None,
1621 };
1622
1623 let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");
1624 let pool = plan.pools.first().expect("general pool should exist");
1625 assert_eq!(pool.recommended.machine(), Some("m8i.2xlarge"));
1626 assert_eq!(pool.recommended.min_size(), 2);
1627 assert_eq!(pool.recommended.max_size(), 2);
1628 assert_eq!(
1629 pool.scale,
1630 CapacityGroupScalePolicy::Fixed {
1631 machines: ComputeChoiceRange {
1632 min: 1,
1633 max: 5,
1634 default: 2,
1635 },
1636 }
1637 );
1638 assert!(!pool
1639 .machines
1640 .iter()
1641 .any(|option| option.machine == "m7g.2xlarge"));
1642
1643 let invalid_settings = ComputeSettings {
1644 containers: Default::default(),
1645 pools: [(
1646 "general".to_string(),
1647 ComputePoolSelection::Fixed {
1648 machines: 2,
1649 machine: Some("m7g.2xlarge".to_string()),
1650 failure_domains: None,
1651 },
1652 )]
1653 .into_iter()
1654 .collect(),
1655 };
1656 let invalid_plan = plan_compute(&stack, Platform::Aws, Some(&invalid_settings))
1657 .expect("plan should build");
1658 assert!(!invalid_plan.pools[0].errors.is_empty());
1659 }
1660
1661 fn declared_pool_errors(
1662 scale: CapacityGroupScalePolicy,
1663 selection: ComputePoolSelection,
1664 ) -> Vec<String> {
1665 let mut stack = stack_with_container();
1666 let cluster = ComputeCluster::new("compute".to_string())
1667 .capacity_group(CapacityGroup {
1668 group_id: "general".to_string(),
1669 instance_type: None,
1670 profile: None,
1671 min_size: scale.default_min_size(),
1672 max_size: scale.default_max_size(),
1673 scale_policy: Some(scale),
1674 nested_virtualization: None,
1675 })
1676 .build();
1677 stack.resources.insert(
1678 "compute".to_string(),
1679 ResourceEntry {
1680 config: Resource::new(cluster),
1681 lifecycle: ResourceLifecycle::Frozen,
1682 dependencies: Vec::new(),
1683 remote_access: false,
1684 enabled_when: None,
1685 },
1686 );
1687 let settings = ComputeSettings {
1688 containers: Default::default(),
1689 pools: [("general".to_string(), selection)].into_iter().collect(),
1690 };
1691 plan_compute(&stack, Platform::Aws, Some(&settings))
1692 .expect("plan should build")
1693 .pools
1694 .remove(0)
1695 .errors
1696 }
1697
1698 fn range(min: u32, max: u32, default: u32) -> ComputeChoiceRange {
1699 ComputeChoiceRange { min, max, default }
1700 }
1701
1702 #[test]
1703 fn generated_pool_offers_the_recommendation_as_a_default_not_a_bound() {
1704 let mut stack = stack_with_container();
1705 let api = stack
1706 .resources
1707 .get_mut("api")
1708 .and_then(|entry| entry.config.downcast_mut::<Container>())
1709 .expect("test stack should contain a container");
1710 api.autoscaling = Some(crate::ContainerAutoscaling {
1711 min: 1,
1712 desired: 1,
1713 max: 3,
1714 target_cpu_percent: Some(70.0),
1715 target_memory_percent: None,
1716 target_http_in_flight_per_replica: None,
1717 max_http_p95_latency_ms: None,
1718 });
1719
1720 let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");
1721 let pool = &plan.pools[0];
1722 assert_eq!(
1723 pool.scale,
1724 CapacityGroupScalePolicy::Autoscale {
1725 min: range(1, 100, 1),
1726 max: range(1, 100, 3),
1727 }
1728 );
1729 assert_eq!(
1730 (pool.recommended.min_size(), pool.recommended.max_size()),
1731 (1, 3)
1732 );
1733
1734 let plan =
1737 plan_compute(&stack_with_container(), Platform::Aws, None).expect("plan should build");
1738 assert_eq!(
1739 plan.pools[0].scale,
1740 CapacityGroupScalePolicy::Autoscale {
1741 min: range(1, 100, 1),
1742 max: range(1, 100, 1),
1743 }
1744 );
1745 assert!(matches!(
1746 plan.pools[0].recommended,
1747 ComputePoolSelection::Fixed { machines: 1, .. }
1748 ));
1749 assert!(plan.pools[0].errors.is_empty());
1750 }
1751
1752 #[test]
1753 fn generated_pool_rejects_a_fleet_too_small_for_desired_replicas() {
1754 let mut stack = stack_with_container();
1755 let api = stack
1756 .resources
1757 .get_mut("api")
1758 .and_then(|entry| entry.config.downcast_mut::<Container>())
1759 .expect("test stack should contain a container");
1760 api.replicas = Some(3);
1762 let selection = |machines| ComputeSettings {
1763 containers: Default::default(),
1764 pools: [(
1765 "general".to_string(),
1766 ComputePoolSelection::Fixed {
1767 machines,
1768 machine: Some("m7g.xlarge".to_string()),
1769 failure_domains: None,
1770 },
1771 )]
1772 .into_iter()
1773 .collect(),
1774 };
1775
1776 let errors = plan_compute(&stack, Platform::Aws, Some(&selection(1)))
1778 .expect("plan should build")
1779 .pools
1780 .remove(0)
1781 .errors;
1782 assert_eq!(errors.len(), 1, "{errors:?}");
1783 assert!(
1784 errors[0].starts_with(
1785 "Pool 'general' is too small for its workloads: 1 x m7g.xlarge has 3.50 vCPU"
1786 ),
1787 "{errors:?}"
1788 );
1789 assert!(errors[0].contains("request 6.00 vCPU"), "{errors:?}");
1790
1791 let errors = plan_compute(&stack, Platform::Aws, Some(&selection(2)))
1792 .expect("plan should build")
1793 .pools
1794 .remove(0)
1795 .errors;
1796 assert!(errors.is_empty(), "{errors:?}");
1797 }
1798
1799 #[test]
1800 fn fixed_and_equal_autoscale_bounds_are_interchangeable_for_declared_pools() {
1801 let fixed = |machines| ComputePoolSelection::Fixed {
1802 machines,
1803 machine: Some("m7g.xlarge".to_string()),
1804 failure_domains: None,
1805 };
1806 let autoscale = |min, max| ComputePoolSelection::Autoscale {
1807 min,
1808 max,
1809 machine: Some("m7g.xlarge".to_string()),
1810 failure_domains: None,
1811 };
1812 let declared_fixed = CapacityGroupScalePolicy::Fixed {
1813 machines: range(2, 4, 2),
1814 };
1815 let declared_autoscale = CapacityGroupScalePolicy::Autoscale {
1816 min: range(1, 3, 1),
1817 max: range(2, 6, 4),
1818 };
1819
1820 assert!(declared_pool_errors(declared_fixed.clone(), autoscale(3, 3)).is_empty());
1821 assert_eq!(
1822 declared_pool_errors(declared_fixed.clone(), autoscale(2, 4)),
1823 vec!["Pool 'general' must use fixed mode with 2-4 machines; autoscale 2-4 is not allowed"]
1824 );
1825 assert_eq!(
1826 declared_pool_errors(declared_fixed, autoscale(5, 5)),
1827 vec!["Pool 'general' machine count 5 is outside the allowed range 2-4"]
1828 );
1829
1830 assert!(declared_pool_errors(declared_autoscale.clone(), fixed(3)).is_empty());
1832 assert_eq!(
1833 declared_pool_errors(declared_autoscale.clone(), fixed(1)),
1834 vec!["Pool 'general' fixed machine count 1 is outside the allowed range 2-3"]
1835 );
1836 assert_eq!(
1837 declared_pool_errors(
1838 CapacityGroupScalePolicy::Autoscale {
1839 min: range(1, 1, 1),
1840 max: range(3, 5, 3),
1841 },
1842 fixed(1),
1843 ),
1844 vec!["Pool 'general' must use autoscale mode (minimum 1-1, maximum 3-5)"]
1845 );
1846 assert_eq!(
1847 declared_pool_errors(declared_autoscale, autoscale(1, 7)),
1848 vec!["Pool 'general' autoscale maximum 7 is outside the allowed range 2-6"]
1849 );
1850 }
1851
1852 #[test]
1853 fn local_plan_has_no_provider_machine_choices() {
1854 let stack = stack_with_container();
1855
1856 let plan = plan_compute(&stack, Platform::Local, None).expect("plan should build");
1857
1858 let pool = plan.pools.first().expect("general pool should exist");
1859 assert_eq!(pool.selected.machine(), None);
1860 assert!(pool.machines.is_empty());
1861 assert!(pool.errors.is_empty());
1862 }
1863}