use crate::{
instance_catalog::{self, Architecture, WorkloadRequirements},
CapacityGroup, CapacityGroupScalePolicy, ComputeChoiceRange, ComputePoolSelection, Container,
Daemon, ErrorData, FailureDomainSelection, GpuSpec, MachineProfile, Platform, ResourceSpec,
Stack, StackState,
};
use alien_error::{AlienError, Result};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct ComputePlan {
pub pools: Vec<ComputePoolPlan>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub containers: Vec<ContainerResourcePlan>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct ContainerResourcePlan {
pub container_id: String,
pub choices: crate::ContainerResourceChoices,
pub cpu: ResourceSpec,
pub memory: ResourceSpec,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub volume_size_gib: Option<u32>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub minimum_volume_size_gib: Option<u32>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub maximum_volume_size_gib: Option<u32>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct ComputePoolPlan {
pub pool_id: String,
pub workloads: Vec<String>,
pub requirements: MachineProfile,
pub scale: CapacityGroupScalePolicy,
pub selected: ComputePoolSelection,
pub recommended: ComputePoolSelection,
pub machines: Vec<ComputeMachineOption>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub errors: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct ComputeMachineOption {
pub machine: String,
pub profile: MachineProfile,
pub recommended: bool,
}
pub fn default_persistent_failure_domains() -> FailureDomainSelection {
FailureDomainSelection {
spread: 1,
selected_failure_domains: Vec::new(),
}
}
pub fn plan_compute(
stack: &Stack,
platform: Platform,
selected_settings: Option<&crate::ComputeSettings>,
) -> Result<ComputePlan, ErrorData> {
plan_compute_with_state(stack, platform, selected_settings, None)
}
pub fn plan_compute_with_state(
stack: &Stack,
platform: Platform,
selected_settings: Option<&crate::ComputeSettings>,
previous_state: Option<&StackState>,
) -> Result<ComputePlan, ErrorData> {
let mut resolved_stack = stack.clone();
crate::container_resources::resolve_container_resources(
&mut resolved_stack,
selected_settings,
)?;
if let Some(previous_state) = previous_state {
crate::container_resources::retain_installed_volume_sizes(
&mut resolved_stack,
selected_settings,
previous_state,
);
}
let (minimum, maximum) = match platform {
Platform::Gcp => (10, 65536),
Platform::Azure => (1, 32767),
_ => (1, 65536),
};
for (id, entry) in &resolved_stack.resources {
let Some(storage) = entry
.config
.downcast_ref::<Container>()
.and_then(|container| container.persistent_storage.as_ref())
else {
continue;
};
let size = crate::container_resources::parse_volume_size_gib(&storage.size)
.map_err(|reason| crate::container_resources::invalid(id, "volumeSizeGib", &reason))?;
if size < minimum || size > maximum {
return Err(AlienError::new(
ErrorData::ContainerResourceSelectionInvalid {
resource_id: id.clone(),
dimension: "volumeSizeGib".into(),
reason: format!(
"capacity for {platform} must be between {minimum} and {maximum} GiB"
),
},
));
}
}
if let Some(previous_state) = previous_state {
for (id, entry) in &resolved_stack.resources {
let desired = entry
.config
.downcast_ref::<Container>()
.and_then(|container| container.persistent_storage.as_ref());
let previous = previous_state
.resources
.get(id)
.and_then(|resource| resource.config.downcast_ref::<Container>())
.and_then(|container| container.persistent_storage.as_ref());
if previous.is_some() && desired.is_none() {
return Err(crate::container_resources::invalid(
id,
"volumeSizeGib",
"removing installed persistent storage is unsupported",
));
}
if let (Some(desired), Some(previous)) = (desired, previous) {
if desired.mount_path != previous.mount_path {
return Err(crate::container_resources::invalid(
id,
"volumeSizeGib",
"changing an installed persistent storage mount path is unsupported",
));
}
let target = crate::container_resources::parse_volume_size_gib(&desired.size);
let installed = crate::container_resources::parse_volume_size_gib(&previous.size);
if let (Ok(target), Ok(installed)) = (target, installed) {
if target < installed {
return Err(AlienError::new(ErrorData::ContainerResourceSelectionInvalid {
resource_id: id.clone(), dimension: "volumeSizeGib".into(),
reason: format!("existing volumes cannot shrink from {installed} to {target} GiB"),
}));
}
}
}
}
}
let stack = &resolved_stack;
let mut groups = collect_workload_groups(stack)?;
merge_explicit_compute_groups(stack, &mut groups)?;
let mut pool_ids: Vec<String> = groups.keys().cloned().collect();
pool_ids.sort();
let mut pools = Vec::new();
for pool_id in pool_ids {
let mut group = groups.remove(&pool_id).expect("pool id came from map keys");
let mut requirements = group.requirements;
let selected = selected_settings.and_then(|settings| settings.pools.get(&pool_id));
let declared_machine = stack
.resources
.values()
.filter_map(|entry| entry.config.downcast_ref::<crate::ComputeCluster>())
.flat_map(|cluster| &cluster.capacity_groups)
.find(|group| group.group_id == pool_id)
.and_then(|group| group.instance_type.as_deref());
let machine_override = match selected {
Some(selection) => selection.machine(),
None => declared_machine,
};
if requirements.architecture.is_none() {
requirements.architecture = machine_override
.and_then(|machine| instance_catalog::find_instance_type(platform, machine))
.map(|spec| spec.architecture)
.or_else(|| {
previous_state.and_then(|state| {
state
.resources
.values()
.filter_map(|resource| {
resource.config.downcast_ref::<crate::ComputeCluster>()
})
.flat_map(|cluster| &cluster.capacity_groups)
.find(|group| group.group_id == pool_id)
.and_then(|group| group.instance_type.as_deref())
.and_then(|machine| {
instance_catalog::find_instance_type(platform, machine)
})
.map(|spec| spec.architecture)
})
});
}
if group.generated {
group.scale = generated_pool_scale_policy(
group.scale.default_min_size(),
group.scale.default_max_size(),
selected.map(ComputePoolSelection::max_size),
);
}
let machines = selected
.map(ComputePoolSelection::max_size)
.unwrap_or(group.scale.default_max_size())
.max(1);
let mut sizing = requirements.clone();
sizing.max_cpu_per_container = sizing
.max_cpu_per_container
.max(sizing.total_cpu_at_desired / f64::from(machines));
sizing.max_memory_per_container = sizing.max_memory_per_container.max(
sizing
.total_memory_bytes_at_desired
.div_ceil(u64::from(machines)),
);
let recommended = recommended_selection(
platform,
&sizing,
&group.scale,
group.requires_failure_domain,
group.generated,
machine_override,
)?;
let mut selected_choice = selected.cloned().unwrap_or_else(|| recommended.clone());
if selected.is_none() {
if let Some(declared_machine) = declared_machine {
match &mut selected_choice {
ComputePoolSelection::Fixed { machine, .. }
| ComputePoolSelection::Autoscale { machine, .. } => {
*machine = Some(declared_machine.to_string());
}
}
}
}
match &mut selected_choice {
ComputePoolSelection::Fixed { machine, .. }
| ComputePoolSelection::Autoscale { machine, .. } => {
if machine.is_none() {
*machine = recommended.machine().map(ToString::to_string);
}
}
}
if selected_choice.failure_domains().is_none() {
if let Some(default_failure_domains) = recommended.failure_domains().cloned() {
match &mut selected_choice {
ComputePoolSelection::Fixed {
failure_domains, ..
}
| ComputePoolSelection::Autoscale {
failure_domains, ..
} => *failure_domains = Some(default_failure_domains),
}
}
}
let mut errors = validate_compute_pool_selection(
platform,
&pool_id,
&selected_choice,
&requirements,
&group.scale,
);
if let Some(workload_requirements) = &group.workload_requirements {
if let Err(message) =
check_pool_capacity(platform, &pool_id, &selected_choice, workload_requirements)
{
errors.push(message);
}
}
let machines = machine_options(
platform,
&requirements,
selected_choice.machine(),
recommended.machine(),
)?;
pools.push(ComputePoolPlan {
pool_id,
workloads: group.workloads,
requirements: requirements_to_profile(&requirements),
scale: group.scale,
selected: selected_choice,
recommended,
machines,
errors,
});
}
let mut containers: Vec<_> = stack
.resources
.values()
.filter_map(|entry| entry.config.downcast_ref::<Container>())
.filter(|container| {
container.resource_choices.is_some() || container.persistent_storage.is_some()
})
.map(|container| {
let capacity = |storage: &crate::PersistentStorage| {
crate::container_resources::parse_volume_size_gib(&storage.size).map_err(|reason| {
crate::container_resources::invalid(&container.id, "volumeSizeGib", &reason)
})
};
let volume_size_gib = container
.persistent_storage
.as_ref()
.map(capacity)
.transpose()?;
let installed = previous_state
.and_then(|state| state.resources.get(&container.id))
.and_then(|resource| resource.config.downcast_ref::<Container>())
.and_then(|container| container.persistent_storage.as_ref())
.map(capacity)
.transpose()?;
Ok(ContainerResourcePlan {
container_id: container.id.clone(),
choices: container.resource_choices.clone().unwrap_or_default(),
cpu: container.cpu.clone(),
memory: container.memory.clone(),
volume_size_gib,
minimum_volume_size_gib: volume_size_gib
.map(|_| installed.unwrap_or(if platform == Platform::Gcp { 10 } else { 1 })),
maximum_volume_size_gib: volume_size_gib.map(|_| {
if platform == Platform::Azure {
32767
} else {
65536
}
}),
})
})
.collect::<Result<Vec<_>, ErrorData>>()?;
containers.sort_by(|a, b| a.container_id.cmp(&b.container_id));
Ok(ComputePlan { pools, containers })
}
#[derive(Debug, Clone)]
struct PlannedGroup {
workloads: Vec<String>,
workload_requirements: Option<WorkloadRequirements>,
requirements: WorkloadRequirements,
scale: CapacityGroupScalePolicy,
requires_failure_domain: bool,
generated: bool,
}
pub const GENERATED_POOL_MAX_MACHINES: u32 = 100;
pub fn generated_pool_scale_policy(
default_min: u32,
default_max: u32,
selected_max: Option<u32>,
) -> CapacityGroupScalePolicy {
let offered = GENERATED_POOL_MAX_MACHINES;
let ceiling = offered.max(selected_max.unwrap_or(0));
let default_min = default_min.min(offered);
CapacityGroupScalePolicy::Autoscale {
min: ComputeChoiceRange {
min: default_min,
max: ceiling,
default: default_min,
},
max: ComputeChoiceRange {
min: default_min.max(1),
max: ceiling,
default: default_max.clamp(default_min.max(1), offered),
},
}
}
pub fn check_pool_capacity(
platform: Platform,
pool_id: &str,
selection: &ComputePoolSelection,
requirements: &WorkloadRequirements,
) -> std::result::Result<(), String> {
let Some(spec) = selection
.machine()
.and_then(|machine| instance_catalog::find_instance_type(platform, machine))
else {
return Ok(());
};
let machines = selection.max_size();
let per_machine_cpu = instance_catalog::allocatable_cpu(spec);
let per_machine_memory = instance_catalog::allocatable_memory_bytes(spec);
if per_machine_cpu + f64::EPSILON < requirements.max_cpu_per_container
|| per_machine_memory < requirements.max_memory_per_container
{
return Err(format!(
"Pool '{pool_id}' machine '{}' cannot fit one replica after reserving host resources",
spec.name,
));
}
let cpu = f64::from(machines) * per_machine_cpu;
let memory = u64::from(machines) * per_machine_memory;
if cpu + f64::EPSILON >= requirements.total_cpu_at_desired
&& memory >= requirements.total_memory_bytes_at_desired
{
return Ok(());
}
const GIB: f64 = 1024.0 * 1024.0 * 1024.0;
Err(format!(
"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",
spec.name,
memory as f64 / GIB,
requirements.total_cpu_at_desired,
requirements.total_memory_bytes_at_desired as f64 / GIB,
))
}
fn collect_workload_groups(stack: &Stack) -> Result<HashMap<String, PlannedGroup>, ErrorData> {
let mut groups: HashMap<String, Vec<Workload>> = HashMap::new();
for entry in stack.resources.values() {
if let Some(container) = entry.config.downcast_ref::<Container>() {
groups
.entry(
container
.pool
.clone()
.unwrap_or_else(|| needed_container_pool(container).to_string()),
)
.or_default()
.push(Workload::from_container(container)?);
}
if let Some(daemon) = entry.config.downcast_ref::<Daemon>() {
if daemon.cluster.is_some() {
groups
.entry(daemon.pool.clone().unwrap_or_else(|| "general".to_string()))
.or_default()
.push(Workload::from_daemon(daemon)?);
}
}
}
let mut planned = HashMap::new();
for (pool_id, workloads) in groups {
let requirements = aggregate_workloads(&workloads);
let requires_failure_domain = workloads
.iter()
.any(|workload| workload.requires_failure_domain);
let min_size = default_min_machines(&requirements);
let max_size = default_max_machines(&requirements);
planned.insert(
pool_id,
PlannedGroup {
workloads: workloads.into_iter().map(|w| w.id).collect(),
workload_requirements: Some(requirements.clone()),
scale: CapacityGroupScalePolicy::from_selected_bounds(min_size, max_size),
requirements,
requires_failure_domain,
generated: true,
},
);
}
Ok(planned)
}
fn merge_explicit_compute_groups(
stack: &Stack,
groups: &mut HashMap<String, PlannedGroup>,
) -> Result<(), ErrorData> {
for entry in stack.resources.values() {
let Some(cluster) = entry.config.downcast_ref::<crate::ComputeCluster>() else {
continue;
};
for group in &cluster.capacity_groups {
let explicit_requirements = profile_to_requirements(
group.profile.as_ref(),
group.nested_virtualization.unwrap_or(false),
);
let scale = group.scale_policy.clone().unwrap_or_else(|| {
CapacityGroupScalePolicy::from_selected_bounds(group.min_size, group.max_size)
});
groups
.entry(group.group_id.clone())
.and_modify(|planned| {
if group.instance_type.is_none() {
merge_requirements(&mut planned.requirements, &explicit_requirements);
} else {
planned.requirements.architecture = planned
.requirements
.architecture
.or(explicit_requirements.architecture);
planned.requirements.nested_virt |= explicit_requirements.nested_virt;
}
planned.scale = merge_scale_policy(&planned.scale, &scale);
planned.generated = false;
})
.or_insert_with(|| PlannedGroup {
workloads: Vec::new(),
workload_requirements: None,
scale,
requirements: explicit_requirements,
requires_failure_domain: false,
generated: false,
});
}
}
Ok(())
}
fn recommended_selection(
platform: Platform,
requirements: &WorkloadRequirements,
scale: &CapacityGroupScalePolicy,
requires_failure_domain: bool,
generated: bool,
machine_override: Option<&str>,
) -> Result<ComputePoolSelection, ErrorData> {
let machine = match platform {
Platform::Aws | Platform::Gcp | Platform::Azure => Some(
match instance_catalog::select_instance_type(platform, requirements) {
Ok(selection) => selection.instance_type.to_string(),
Err(message) => {
let fallback = machine_override
.and_then(|machine| instance_catalog::find_instance_type(platform, machine))
.filter(|spec| {
instance_satisfies(
spec,
requirements,
requirements.architecture.unwrap_or(spec.architecture),
)
});
fallback.map(|spec| spec.name.to_string()).ok_or_else(|| {
AlienError::new(ErrorData::GenericError {
message: format!("Failed to select {platform} machine: {message}"),
})
})?
}
},
),
Platform::Local | Platform::Kubernetes | Platform::Machines | Platform::Test => None,
};
let failure_domains = (requires_failure_domain
&& matches!(platform, Platform::Aws | Platform::Gcp | Platform::Azure))
.then(default_persistent_failure_domains);
match scale {
CapacityGroupScalePolicy::Fixed { machines } => Ok(ComputePoolSelection::Fixed {
machines: machines.default.max(1),
machine,
failure_domains,
}),
CapacityGroupScalePolicy::Autoscale { min, max }
if generated && min.default == max.default =>
{
Ok(ComputePoolSelection::Fixed {
machines: min.default.max(1),
machine,
failure_domains,
})
}
CapacityGroupScalePolicy::Autoscale { min, max } => Ok(ComputePoolSelection::Autoscale {
min: min.default,
max: max.default.max(min.default),
machine,
failure_domains,
}),
}
}
pub fn validate_compute_pool_selection(
platform: Platform,
pool_id: &str,
selection: &ComputePoolSelection,
requirements: &WorkloadRequirements,
scale: &CapacityGroupScalePolicy,
) -> Vec<String> {
let mut errors = Vec::new();
if let Err(message) = selection.validate() {
errors.push(message);
}
if let Err(message) = validate_selection_against_scale(selection, scale) {
errors.push(format!("Pool '{pool_id}' {message}"));
}
if matches!(platform, Platform::Aws | Platform::Gcp | Platform::Azure) {
match selection.machine() {
Some(machine) => match instance_catalog::find_instance_type(platform, machine) {
Some(spec) => {
let architecture = requirements.architecture.unwrap_or(spec.architecture);
if !instance_satisfies(spec, requirements, architecture) {
errors.push(format!(
"{} machine '{}' does not satisfy pool '{}' requirements",
platform, machine, pool_id
));
}
}
None => errors.push(format!(
"Unknown {} machine '{}' for pool '{}'",
platform, machine, pool_id
)),
},
None => errors.push(format!(
"Pool '{}' requires a provider machine on {}",
pool_id, platform
)),
}
}
errors
}
pub fn capacity_group_requirements(group: &CapacityGroup) -> WorkloadRequirements {
profile_to_requirements(
group.profile.as_ref(),
group.nested_virtualization.unwrap_or(false),
)
}
fn machine_options(
platform: Platform,
requirements: &WorkloadRequirements,
selected_machine: Option<&str>,
recommended_machine: Option<&str>,
) -> Result<Vec<ComputeMachineOption>, ErrorData> {
if !matches!(platform, Platform::Aws | Platform::Gcp | Platform::Azure) {
return Ok(Vec::new());
}
let resolved_architecture = requirements
.architecture
.or_else(|| {
selected_machine.and_then(|machine| {
instance_catalog::find_instance_type(platform, machine)
.map(|spec| spec.architecture)
})
})
.or_else(|| {
recommended_machine
.and_then(|machine| instance_catalog::find_instance_type(platform, machine))
.map(|spec| spec.architecture)
})
.ok_or_else(|| {
AlienError::new(ErrorData::GenericError {
message: format!("Selected {platform} machine has no CPU architecture"),
})
})?;
let mut options: Vec<ComputeMachineOption> = instance_catalog::catalog_for_platform(platform)
.into_iter()
.filter(|spec| instance_satisfies(spec, requirements, resolved_architecture))
.map(|spec| ComputeMachineOption {
machine: spec.name.to_string(),
profile: spec.to_machine_profile_for_storage(requirements.max_ephemeral_storage_bytes),
recommended: Some(spec.name) == recommended_machine
|| Some(spec.name) == selected_machine,
})
.collect();
options.sort_by(|a, b| a.machine.cmp(&b.machine));
Ok(options)
}
fn instance_satisfies(
spec: &instance_catalog::InstanceTypeSpec,
requirements: &WorkloadRequirements,
resolved_architecture: Architecture,
) -> bool {
if spec.architecture != resolved_architecture {
return false;
}
if requirements.nested_virt && !spec.is_nested_virt_capable() {
return false;
}
if spec.vcpu < requirements.max_cpu_per_container.ceil() as u32 {
return false;
}
if spec.memory_bytes < requirements.max_memory_per_container {
return false;
}
if spec.has_configurable_ephemeral_storage() {
if crate::instance_catalog::max_configurable_ephemeral_storage_bytes(spec.platform)
.is_none_or(|max| requirements.max_ephemeral_storage_bytes > max)
{
return false;
}
} else if spec.ephemeral_storage_bytes < requirements.max_ephemeral_storage_bytes {
return false;
}
match (&requirements.gpu, spec.gpu) {
(Some(required), Some(actual)) => {
(required.gpu_type == "any" || required.gpu_type == actual.gpu_type)
&& actual.count >= required.count
}
(Some(_), None) => false,
(None, _) => true,
}
}
#[derive(Debug, Clone)]
struct Workload {
id: String,
cpu: f64,
memory_bytes: u64,
desired_replicas: f64,
max_replicas: f64,
ephemeral_storage_bytes: u64,
gpu: Option<GpuSpec>,
requires_failure_domain: bool,
}
impl Workload {
fn from_container(container: &Container) -> Result<Self, ErrorData> {
let cpu = parse_cpu(&container.id, &container.cpu)?;
let memory_bytes = parse_memory(&container.id, &container.memory)?;
let desired_replicas = container
.autoscaling
.as_ref()
.map(|a| a.desired)
.or(container.replicas)
.unwrap_or(1) as f64;
let max_replicas = container
.autoscaling
.as_ref()
.map(|a| a.max)
.or(container.replicas)
.unwrap_or(1) as f64;
let ephemeral_storage_bytes = container
.ephemeral_storage
.as_deref()
.map(instance_catalog::parse_memory_bytes)
.transpose()
.map_err(|message| {
AlienError::new(ErrorData::GenericError {
message: format!(
"Failed to parse ephemeral storage for '{}': {message}",
container.id
),
})
})?
.unwrap_or(0);
Ok(Self {
id: container.id.clone(),
cpu,
memory_bytes,
desired_replicas,
max_replicas,
ephemeral_storage_bytes,
gpu: container.gpu.as_ref().map(|gpu| GpuSpec {
gpu_type: gpu.gpu_type.clone(),
count: gpu.count,
}),
requires_failure_domain: container.stateful && container.persistent_storage.is_some(),
})
}
fn from_daemon(daemon: &Daemon) -> Result<Self, ErrorData> {
Ok(Self {
id: daemon.id.clone(),
cpu: parse_cpu(&daemon.id, &daemon.cpu)?,
memory_bytes: parse_memory(&daemon.id, &daemon.memory)?,
desired_replicas: 1.0,
max_replicas: 1.0,
ephemeral_storage_bytes: 0,
gpu: None,
requires_failure_domain: false,
})
}
}
fn parse_cpu(resource_id: &str, spec: &ResourceSpec) -> Result<f64, ErrorData> {
instance_catalog::parse_cpu(&spec.desired).map_err(|message| {
AlienError::new(ErrorData::GenericError {
message: format!(
"Failed to parse CPU requirement '{}' for '{}': {message}",
spec.desired, resource_id
),
})
})
}
fn parse_memory(resource_id: &str, spec: &ResourceSpec) -> Result<u64, ErrorData> {
instance_catalog::parse_memory_bytes(&spec.desired).map_err(|message| {
AlienError::new(ErrorData::GenericError {
message: format!(
"Failed to parse memory requirement '{}' for '{}': {message}",
spec.desired, resource_id
),
})
})
}
fn aggregate_workloads(workloads: &[Workload]) -> WorkloadRequirements {
let mut requirements = default_requirements();
requirements.total_cpu_at_desired = 0.0;
requirements.total_memory_bytes_at_desired = 0;
requirements.total_cpu_at_max = 0.0;
requirements.total_memory_bytes_at_max = 0;
requirements.max_cpu_per_container = 0.0;
requirements.max_memory_per_container = 0;
requirements.max_ephemeral_storage_bytes = 0;
requirements.gpu = None;
for workload in workloads {
requirements.total_cpu_at_desired += workload.cpu * workload.desired_replicas;
requirements.total_cpu_at_max += workload.cpu * workload.max_replicas;
requirements.total_memory_bytes_at_desired +=
(workload.memory_bytes as f64 * workload.desired_replicas) as u64;
requirements.total_memory_bytes_at_max +=
(workload.memory_bytes as f64 * workload.max_replicas) as u64;
requirements.max_cpu_per_container = requirements.max_cpu_per_container.max(workload.cpu);
requirements.max_memory_per_container = requirements
.max_memory_per_container
.max(workload.memory_bytes);
requirements.max_ephemeral_storage_bytes = requirements
.max_ephemeral_storage_bytes
.max(workload.ephemeral_storage_bytes);
if requirements.gpu.is_none() {
requirements.gpu = workload.gpu.clone();
}
}
requirements
}
fn default_requirements() -> WorkloadRequirements {
WorkloadRequirements {
total_cpu_at_desired: 1.0,
total_memory_bytes_at_desired: 2 * 1024 * 1024 * 1024,
total_cpu_at_max: 1.0,
total_memory_bytes_at_max: 2 * 1024 * 1024 * 1024,
max_cpu_per_container: 1.0,
max_memory_per_container: 2 * 1024 * 1024 * 1024,
max_ephemeral_storage_bytes: 0,
gpu: None,
architecture: None,
nested_virt: false,
}
}
fn profile_to_requirements(
profile: Option<&MachineProfile>,
nested_virt: bool,
) -> WorkloadRequirements {
let Some(profile) = profile else {
return WorkloadRequirements {
nested_virt,
..default_requirements()
};
};
let cpu = instance_catalog::parse_cpu(&profile.cpu).unwrap_or(1.0);
WorkloadRequirements {
total_cpu_at_desired: cpu,
total_memory_bytes_at_desired: profile.memory_bytes,
total_cpu_at_max: cpu,
total_memory_bytes_at_max: profile.memory_bytes,
max_cpu_per_container: cpu,
max_memory_per_container: profile.memory_bytes,
max_ephemeral_storage_bytes: profile.ephemeral_storage_bytes,
gpu: profile.gpu.clone(),
architecture: profile.architecture,
nested_virt,
}
}
fn merge_requirements(existing: &mut WorkloadRequirements, declared: &WorkloadRequirements) {
existing.total_cpu_at_desired = existing
.total_cpu_at_desired
.max(declared.total_cpu_at_desired);
existing.total_memory_bytes_at_desired = existing
.total_memory_bytes_at_desired
.max(declared.total_memory_bytes_at_desired);
existing.total_cpu_at_max = existing.total_cpu_at_max.max(declared.total_cpu_at_max);
existing.total_memory_bytes_at_max = existing
.total_memory_bytes_at_max
.max(declared.total_memory_bytes_at_max);
existing.max_cpu_per_container = existing
.max_cpu_per_container
.max(declared.max_cpu_per_container);
existing.max_memory_per_container = existing
.max_memory_per_container
.max(declared.max_memory_per_container);
existing.max_ephemeral_storage_bytes = existing
.max_ephemeral_storage_bytes
.max(declared.max_ephemeral_storage_bytes);
if existing.gpu.is_none() {
existing.gpu = declared.gpu.clone();
}
existing.nested_virt |= declared.nested_virt;
if existing.architecture.is_none() {
existing.architecture = declared.architecture;
}
}
fn requirements_to_profile(requirements: &WorkloadRequirements) -> MachineProfile {
MachineProfile {
cpu: requirements.max_cpu_per_container.to_string(),
memory_bytes: requirements.max_memory_per_container,
ephemeral_storage_bytes: requirements.max_ephemeral_storage_bytes,
architecture: requirements.architecture,
gpu: requirements.gpu.clone(),
data_disk: None,
}
}
fn merge_scale_policy(
existing: &CapacityGroupScalePolicy,
declared: &CapacityGroupScalePolicy,
) -> CapacityGroupScalePolicy {
match (existing, declared) {
(
CapacityGroupScalePolicy::Fixed {
machines: existing_machines,
},
CapacityGroupScalePolicy::Fixed {
machines: declared_machines,
},
) => CapacityGroupScalePolicy::Fixed {
machines: merge_choice_range(existing_machines, declared_machines),
},
(_, declared) => declared.clone(),
}
}
fn merge_choice_range(
existing: &ComputeChoiceRange,
declared: &ComputeChoiceRange,
) -> ComputeChoiceRange {
ComputeChoiceRange {
min: existing.min.max(declared.min),
max: existing.max.max(declared.max),
default: declared.default,
}
}
fn validate_selection_against_scale(
selection: &ComputePoolSelection,
scale: &CapacityGroupScalePolicy,
) -> std::result::Result<(), String> {
match (selection, scale) {
(
ComputePoolSelection::Fixed { machines, .. },
CapacityGroupScalePolicy::Fixed { machines: allowed },
) => check_machine_count("fixed machine count", *machines, allowed.min, allowed.max),
(
ComputePoolSelection::Autoscale { min, max, .. },
CapacityGroupScalePolicy::Autoscale {
min: allowed_min,
max: allowed_max,
},
) => {
check_machine_count("autoscale minimum", *min, allowed_min.min, allowed_min.max)?;
check_machine_count("autoscale maximum", *max, allowed_max.min, allowed_max.max)
}
(
ComputePoolSelection::Fixed { machines, .. },
CapacityGroupScalePolicy::Autoscale {
min: allowed_min,
max: allowed_max,
},
) => {
let lowest = allowed_min.min.max(allowed_max.min);
let highest = allowed_min.max.min(allowed_max.max);
if lowest > highest {
return Err(format!(
"must use autoscale mode (minimum {}-{}, maximum {}-{})",
allowed_min.min, allowed_min.max, allowed_max.min, allowed_max.max
));
}
check_machine_count("fixed machine count", *machines, lowest, highest)
}
(
ComputePoolSelection::Autoscale { min, max, .. },
CapacityGroupScalePolicy::Fixed { machines: allowed },
) => {
if min != max {
return Err(format!(
"must use fixed mode with {}-{} machines; autoscale {min}-{max} is not allowed",
allowed.min, allowed.max
));
}
check_machine_count("machine count", *min, allowed.min, allowed.max)
}
}
}
fn check_machine_count(
label: &str,
value: u32,
lowest: u32,
highest: u32,
) -> std::result::Result<(), String> {
if (lowest..=highest).contains(&value) {
Ok(())
} else {
Err(format!(
"{label} {value} is outside the allowed range {lowest}-{highest}"
))
}
}
fn needed_container_pool(container: &Container) -> &'static str {
if container.stateful && container.persistent_storage.is_some() {
return "stateful";
}
if container.gpu.is_some() {
return "gpu";
}
if let Some(storage) = &container.ephemeral_storage {
if instance_catalog::parse_memory_bytes(storage).unwrap_or(0) > 200 * 1024 * 1024 * 1024 {
return "storage";
}
}
"general"
}
fn default_min_machines(requirements: &WorkloadRequirements) -> u32 {
if requirements.total_cpu_at_desired > 0.0 || requirements.total_memory_bytes_at_desired > 0 {
1
} else {
0
}
}
fn default_max_machines(requirements: &WorkloadRequirements) -> u32 {
let min = default_min_machines(requirements);
let by_cpu =
(requirements.total_cpu_at_max / requirements.max_cpu_per_container.max(1.0)).ceil() as u32;
let by_mem = requirements
.total_memory_bytes_at_max
.div_ceil(requirements.max_memory_per_container.max(1)) as u32;
min.max(by_cpu).max(by_mem).max(1)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{
instance_catalog::Architecture, CapacityGroup, CapacityGroupScalePolicy,
ComputeChoiceRange, ComputeCluster, ComputeSettings, ContainerCode, DaemonCode, Resource,
ResourceEntry, ResourceLifecycle, Stack, Storage, Worker, WorkerCode,
};
#[test]
fn deployment_resources_drive_machine_planning_on_every_cloud() {
let mut stack = stack_with_container();
let container = stack
.resources
.get_mut("api")
.unwrap()
.config
.downcast_mut::<Container>()
.unwrap();
container.resource_choices = Some(crate::ContainerResourceChoices {
cpu: Some(crate::ResourceChoiceRange {
min: "0.5".into(),
max: "8".into(),
default: "2".into(),
}),
memory: Some(crate::ResourceChoiceRange {
min: "512Mi".into(),
max: "16Gi".into(),
default: "4Gi".into(),
}),
});
let settings: ComputeSettings = serde_json::from_value(serde_json::json!({
"pools": { "general": { "mode": "fixed", "machines": 1 } },
"containers": { "api": { "cpu": 8, "memory": "16Gi" } }
}))
.unwrap();
for platform in [Platform::Aws, Platform::Gcp, Platform::Azure] {
let default_plan = plan_compute(&stack, platform, None).unwrap();
let selected_plan = plan_compute(&stack, platform, Some(&settings)).unwrap();
let pool = &selected_plan.pools[0];
assert!(pool.errors.is_empty(), "{platform}: {:?}", pool.errors);
assert_eq!(pool.requirements.cpu, "8");
assert_eq!(pool.requirements.memory_bytes, 16 * 1024 * 1024 * 1024);
assert_ne!(
pool.selected.machine(),
default_plan.pools[0].selected.machine()
);
assert_eq!(selected_plan.containers[0].cpu.min, "8");
assert_eq!(selected_plan.containers[0].memory.desired, "16Gi");
assert_eq!(pool.selected.min_size(), 1);
assert_eq!(pool.selected.max_size(), 1);
}
assert_eq!(
settings.pools["general"].machine(),
None,
"resolving a plan must preserve automatic machine intent"
);
let original = stack.resources["api"]
.config
.downcast_ref::<Container>()
.unwrap();
assert_eq!(
original.cpu.desired, "2",
"planning must not mutate release defaults"
);
}
#[test]
fn explicit_machine_can_fit_a_low_cpu_high_memory_workload() {
let mut stack = stack_with_container();
let container = stack
.resources
.get_mut("api")
.unwrap()
.config
.downcast_mut::<Container>()
.unwrap();
container.cpu = ResourceSpec {
min: "0.5".into(),
desired: "0.5".into(),
};
container.memory = ResourceSpec {
min: "8Gi".into(),
desired: "8Gi".into(),
};
let settings: ComputeSettings = serde_json::from_value(serde_json::json!({
"pools": {"general":{"mode":"fixed","machines":1,"machine":"n2-standard-4"}}
}))
.unwrap();
let plan = plan_compute(&stack, Platform::Gcp, Some(&settings)).unwrap();
assert!(
plan.pools[0].errors.is_empty(),
"{:?}",
plan.pools[0].errors
);
assert_eq!(plan.pools[0].selected.machine(), Some("n2-standard-4"));
}
#[test]
fn pin_must_leave_host_capacity_for_one_replica() {
let mut stack = stack_with_container();
stack
.resources
.get_mut("api")
.unwrap()
.config
.downcast_mut::<Container>()
.unwrap()
.cpu = ResourceSpec {
min: "4".into(),
desired: "4".into(),
};
let settings: ComputeSettings = serde_json::from_value(serde_json::json!({
"pools":{"general":{"mode":"fixed","machines":1,"machine":"t3.xlarge"}}
}))
.unwrap();
let plan = plan_compute(&stack, Platform::Aws, Some(&settings)).unwrap();
assert!(
plan.pools[0]
.errors
.iter()
.any(|error| error.contains("host resources")),
"{:?}",
plan.pools[0].errors
);
}
#[test]
fn valid_pin_does_not_require_automatic_recommendation_headroom() {
let machine = instance_catalog::catalog_for_platform(Platform::Gcp)
.into_iter()
.max_by_key(|machine| machine.memory_bytes)
.unwrap();
let mut stack = stack_with_container();
let container = stack
.resources
.get_mut("api")
.unwrap()
.config
.downcast_mut::<Container>()
.unwrap();
container.cpu = ResourceSpec {
min: "0.5".into(),
desired: "0.5".into(),
};
let memory = instance_catalog::allocatable_memory_bytes(machine).to_string();
container.memory = ResourceSpec {
min: memory.clone(),
desired: memory,
};
assert!(plan_compute(&stack, Platform::Gcp, None).is_err());
let settings: ComputeSettings = serde_json::from_value(serde_json::json!({
"pools":{"general":{"mode":"fixed","machines":1,"machine":machine.name}}
}))
.unwrap();
let plan = plan_compute(&stack, Platform::Gcp, Some(&settings)).unwrap();
assert!(
plan.pools[0].errors.is_empty(),
"{:?}",
plan.pools[0].errors
);
assert_eq!(plan.pools[0].selected.machine(), Some(machine.name));
}
#[test]
fn invalid_deployment_resources_fail_before_capacity_planning() {
let mut stack = stack_with_container();
stack
.resources
.get_mut("api")
.unwrap()
.config
.downcast_mut::<Container>()
.unwrap()
.resource_choices = Some(crate::ContainerResourceChoices {
cpu: Some(crate::ResourceChoiceRange {
min: "0.5".into(),
max: "4".into(),
default: "2".into(),
}),
memory: Some(crate::ResourceChoiceRange {
min: "512Mi".into(),
max: "8Gi".into(),
default: "4Gi".into(),
}),
});
for selection in [
serde_json::json!({"api":{"cpu": 0}}),
serde_json::json!({"api":{"cpu": 5}}),
serde_json::json!({"api":{"memory": "16Gi"}}),
serde_json::json!({"api":{"memory": "NaNGi"}}),
serde_json::json!({"api":{"memory": "-1Gi"}}),
serde_json::json!({"missing":{"cpu": 1}}),
] {
let settings: ComputeSettings =
serde_json::from_value(serde_json::json!({"containers": selection})).unwrap();
let error = plan_compute(&stack, Platform::Aws, Some(&settings)).unwrap_err();
assert_eq!(error.code, "CONTAINER_RESOURCE_SELECTION_INVALID");
}
}
#[test]
fn deployment_volume_selection_is_applied_without_changing_release_defaults() {
let mut release = stack_with_container();
release
.resources
.get_mut("api")
.unwrap()
.config
.downcast_mut::<Container>()
.unwrap()
.persistent_storage = Some(crate::PersistentStorage {
size: "20Gi".into(),
mount_path: "/data".into(),
backups: crate::VolumeBackups::default(),
encryption_key: None,
});
let settings: ComputeSettings = serde_json::from_value(serde_json::json!({
"containers": {"api": {"volumeSizeGib": 40}}
}))
.unwrap();
let mut deployment = release.clone();
crate::container_resources::resolve_container_resources(&mut deployment, Some(&settings))
.unwrap();
let volume = |stack: &Stack| {
stack.resources["api"]
.config
.downcast_ref::<Container>()
.unwrap()
.persistent_storage
.as_ref()
.unwrap()
.size
.clone()
};
assert_eq!(volume(&deployment), "40Gi");
assert_eq!(volume(&release), "20Gi");
let mut equivalent = release.clone();
equivalent
.resources
.get_mut("api")
.unwrap()
.config
.downcast_mut::<Container>()
.unwrap()
.persistent_storage
.as_mut()
.unwrap()
.size = "1Ti".into();
let same_capacity: ComputeSettings = serde_json::from_value(serde_json::json!({
"containers": {"api": {"volumeSizeGib": 1024}}
}))
.unwrap();
crate::container_resources::resolve_container_resources(
&mut equivalent,
Some(&same_capacity),
)
.unwrap();
assert_eq!(
volume(&equivalent),
"1Ti",
"equivalent selections preserve PVC quantities"
);
crate::container_resources::resolve_container_resources(&mut deployment, Some(&settings))
.unwrap();
assert_eq!(volume(&deployment), "40Gi");
let smaller_initial: ComputeSettings = serde_json::from_value(serde_json::json!({
"containers": {"api": {"volumeSizeGib": 10}}
}))
.unwrap();
crate::container_resources::resolve_container_resources(
&mut release.clone(),
Some(&smaller_initial),
)
.unwrap();
let mut installed = StackState::new(Platform::Aws);
installed.resources.insert(
"api".into(),
crate::StackResourceState::new_pending(
"Container".into(),
deployment.resources["api"].config.clone(),
None,
vec![],
),
);
let shrinking: ComputeSettings = serde_json::from_value(serde_json::json!({
"containers": {"api": {"volumeSizeGib": 30}}
}))
.unwrap();
let error =
plan_compute_with_state(&release, Platform::Aws, Some(&shrinking), Some(&installed))
.unwrap_err();
assert_eq!(error.code, "CONTAINER_RESOURCE_SELECTION_INVALID");
assert!(error.to_string().contains("cannot shrink from 40 to 30"));
let imported: ComputeSettings = serde_json::from_value(serde_json::json!({
"containers": {"api": {"volumeSizeGib": "40"}}
}))
.unwrap();
assert_eq!(imported.containers["api"].volume_size_gib, Some(40));
for value in [
serde_json::json!(-1),
serde_json::json!(1.5),
serde_json::json!("4294967296"),
] {
assert!(
serde_json::from_value::<ComputeSettings>(serde_json::json!({
"containers": {"api": {"volumeSizeGib": value}}
}))
.is_err()
);
}
for (platform, size) in [(Platform::Gcp, 9), (Platform::Azure, 32768)] {
let invalid: ComputeSettings = serde_json::from_value(serde_json::json!({
"containers": {"api": {"volumeSizeGib": size}}
}))
.unwrap();
let error = plan_compute(&release, platform, Some(&invalid)).unwrap_err();
assert_eq!(error.code, "CONTAINER_RESOURCE_SELECTION_INVALID");
assert!(error.to_string().contains("must be between"));
}
for size in [0, 65537] {
let invalid: ComputeSettings = serde_json::from_value(serde_json::json!({
"containers": {"api": {"volumeSizeGib": size}}
}))
.unwrap();
assert_eq!(
crate::container_resources::resolve_container_resources(
&mut release.clone(),
Some(&invalid)
)
.unwrap_err()
.code,
"CONTAINER_RESOURCE_SELECTION_INVALID"
);
}
assert_eq!(
crate::container_resources::resolve_container_resources(
&mut stack_with_container(),
Some(&settings)
)
.unwrap_err()
.code,
"CONTAINER_RESOURCE_SELECTION_INVALID"
);
}
fn stack_with_volume(size: &str) -> Stack {
let mut stack = stack_with_container();
stack
.resources
.get_mut("api")
.unwrap()
.config
.downcast_mut::<Container>()
.unwrap()
.persistent_storage = Some(crate::PersistentStorage {
size: size.into(),
mount_path: "/data".into(),
backups: crate::VolumeBackups::default(),
encryption_key: None,
});
stack
}
fn installed_with_volume(platform: Platform, size: &str) -> StackState {
let mut installed = StackState::new(platform);
installed.resources.insert(
"api".into(),
crate::StackResourceState::new_pending(
"Container".into(),
stack_with_volume(size).resources["api"].config.clone(),
None,
vec![],
),
);
installed
}
fn api_volume(plan: &ComputePlan) -> (Option<u32>, Option<u32>) {
let api = plan
.containers
.iter()
.find(|container| container.container_id == "api")
.expect("api is planned");
(api.volume_size_gib, api.minimum_volume_size_gib)
}
#[test]
fn a_lower_release_default_keeps_installed_volumes_without_a_selection() {
for platform in [Platform::Aws, Platform::Gcp, Platform::Azure] {
let installed = installed_with_volume(platform, "30Gi");
let plan = plan_compute_with_state(
&stack_with_volume("20Gi"),
platform,
None,
Some(&installed),
)
.expect("a lower release default keeps the installed volume");
assert_eq!(api_volume(&plan), (Some(30), Some(30)), "{platform}");
let plan = plan_compute_with_state(
&stack_with_volume("40Gi"),
platform,
None,
Some(&installed),
)
.expect("a larger release default grows the volume");
assert_eq!(api_volume(&plan), (Some(40), Some(30)), "{platform}");
let shrinking: ComputeSettings = serde_json::from_value(serde_json::json!({
"containers": {"api": {"volumeSizeGib": 25}}
}))
.unwrap();
let error = plan_compute_with_state(
&stack_with_volume("20Gi"),
platform,
Some(&shrinking),
Some(&installed),
)
.unwrap_err();
assert!(
error.to_string().contains("cannot shrink from 30 to 25"),
"{error}"
);
}
}
#[test]
fn setup_source_volume_sizes_below_installed_keep_the_installed_size() {
let mut installed = installed_with_volume(Platform::Aws, "30Gi");
let mut worker = stack_with_volume("10Gi").resources["api"].config.clone();
worker
.downcast_mut::<Container>()
.unwrap()
.persistent_storage = None;
installed.resources.insert(
"worker".into(),
crate::StackResourceState::new_pending("Container".into(), worker, None, vec![]),
);
let mut settings: ComputeSettings = serde_json::from_value(serde_json::json!({
"containers": {
"api": {"volumeSizeGib": 25, "cpu": 2},
"worker": {"volumeSizeGib": 5},
"new": {"volumeSizeGib": 8}
}
}))
.unwrap();
let kept = crate::container_resources::keep_volume_sizes_over_setup_source(
&mut settings,
Some(&installed),
None,
);
assert_eq!(
kept,
vec![crate::container_resources::KeptVolumeSize {
container_id: "api".into(),
requested_gib: 25,
kept_gib: 30,
}]
);
assert_eq!(settings.containers["api"].volume_size_gib, Some(30));
assert_eq!(
settings.containers["api"].cpu,
Some(serde_json::Number::from(2))
);
assert_eq!(settings.containers["worker"].volume_size_gib, Some(5));
assert_eq!(settings.containers["new"].volume_size_gib, Some(8));
let plan = plan_compute_with_state(
&stack_with_volume("20Gi"),
Platform::Aws,
Some(&ComputeSettings {
containers: [("api".to_string(), settings.containers["api"].clone())].into(),
..Default::default()
}),
Some(&installed_with_volume(Platform::Aws, "30Gi")),
)
.expect("the kept size plans");
assert_eq!(api_volume(&plan), (Some(30), Some(30)));
let mut growing: ComputeSettings = serde_json::from_value(serde_json::json!({
"containers": {"api": {"volumeSizeGib": 35}}
}))
.unwrap();
assert!(
crate::container_resources::keep_volume_sizes_over_setup_source(
&mut growing,
Some(&installed),
None,
)
.is_empty()
);
assert_eq!(growing.containers["api"].volume_size_gib, Some(35));
let chosen: ComputeSettings = serde_json::from_value(serde_json::json!({
"containers": {"api": {"volumeSizeGib": 40}}
}))
.unwrap();
let mut stale: ComputeSettings = serde_json::from_value(serde_json::json!({
"containers": {"api": {"volumeSizeGib": 25}}
}))
.unwrap();
let kept = crate::container_resources::keep_volume_sizes_over_setup_source(
&mut stale,
Some(&installed),
Some(&chosen),
);
assert_eq!(kept[0].kept_gib, 40);
assert_eq!(stale.containers["api"].volume_size_gib, Some(40));
let mut correction: ComputeSettings = serde_json::from_value(serde_json::json!({
"containers": {"api": {"volumeSizeGib": 20}}
}))
.unwrap();
for nothing_installed in [None, Some(&StackState::new(Platform::Aws))] {
assert!(
crate::container_resources::keep_volume_sizes_over_setup_source(
&mut correction,
nothing_installed,
Some(&chosen),
)
.is_empty()
);
assert_eq!(correction.containers["api"].volume_size_gib, Some(20));
}
}
#[test]
fn installed_storage_cannot_be_removed_or_remounted_on_any_provider() {
for platform in [
Platform::Aws,
Platform::Gcp,
Platform::Azure,
Platform::Kubernetes,
] {
let mut release = stack_with_container();
release
.resources
.get_mut("api")
.unwrap()
.config
.downcast_mut::<Container>()
.unwrap()
.persistent_storage = Some(crate::PersistentStorage {
size: "20Gi".into(),
mount_path: "/data".into(),
backups: crate::VolumeBackups::default(),
encryption_key: None,
});
let mut installed = StackState::new(platform);
installed.resources.insert(
"api".into(),
crate::StackResourceState::new_pending(
"Container".into(),
release.resources["api"].config.clone(),
None,
vec![],
),
);
for remove in [false, true] {
let mut desired = release.clone();
let storage = &mut desired
.resources
.get_mut("api")
.unwrap()
.config
.downcast_mut::<Container>()
.unwrap()
.persistent_storage;
if remove {
*storage = None;
} else {
storage.as_mut().unwrap().mount_path = "/other".into();
}
let error = plan_compute_with_state(&desired, platform, None, Some(&installed))
.unwrap_err();
assert_eq!(error.code, "CONTAINER_RESOURCE_SELECTION_INVALID");
assert!(
error.to_string().contains(if remove {
"removing installed"
} else {
"mount path"
}),
"{error}"
);
}
}
}
#[test]
fn release_volume_defaults_obey_provider_bounds_without_a_selection() {
for (platform, size) in [(Platform::Gcp, "5Gi"), (Platform::Azure, "40000Gi")] {
let mut stack = stack_with_container();
stack
.resources
.get_mut("api")
.unwrap()
.config
.downcast_mut::<Container>()
.unwrap()
.persistent_storage = Some(crate::PersistentStorage {
size: size.into(),
mount_path: "/data".into(),
backups: crate::VolumeBackups::default(),
encryption_key: None,
});
let error = plan_compute(&stack, platform, None).unwrap_err();
assert_eq!(error.code, "CONTAINER_RESOURCE_SELECTION_INVALID");
assert!(error.to_string().contains("must be between"));
}
}
#[test]
fn volume_capacity_parsing_rejects_overflow_and_preserves_legacy_units() {
use crate::container_resources::parse_volume_size_gib;
for (value, size) in [
("20Gi", 20),
("20GiB", 20),
("20GB", 20),
("1Ti", 1024),
("1TB", 1000),
("1.5Ti", 1536),
("1536Mi", 2),
("0.5Gi", 1),
("1073741824", 1),
] {
assert_eq!(parse_volume_size_gib(value).unwrap(), size);
}
for value in ["0Gi", "-1Gi", "NaNGi", "4294967295Ti", "65537Gi"] {
assert!(parse_volume_size_gib(value).is_err(), "{value}");
}
}
fn stack_with_container() -> Stack {
let container = Container::new("api".to_string())
.code(ContainerCode::Image {
image: "api:latest".to_string(),
})
.cpu(ResourceSpec {
min: "1".to_string(),
desired: "2".to_string(),
})
.memory(ResourceSpec {
min: "2Gi".to_string(),
desired: "4Gi".to_string(),
})
.permissions("api".to_string())
.build();
Stack {
id: "test".to_string(),
resources: [(
"api".to_string(),
ResourceEntry {
config: Resource::new(container),
lifecycle: ResourceLifecycle::Live,
dependencies: Vec::new(),
display_name: None,
description: None,
remote_access: false,
enabled_when: None,
},
)]
.into_iter()
.collect(),
permissions: crate::permissions::PermissionsConfig::default(),
supported_platforms: None,
inputs: vec![],
dynamic_container_repositories: Vec::new(),
dynamic_container_image_resources: Vec::new(),
operations: None,
}
}
#[test]
fn serverless_workers_and_storage_do_not_allocate_machine_pools() {
let mut stack = stack_with_container();
stack.resources.clear();
for (id, resource) in [
(
"api",
Resource::new(
Worker::new("api".to_string())
.code(WorkerCode::Image {
image: "example/worker:v1".to_string(),
})
.permissions("execution".to_string())
.build(),
),
),
(
"data",
Resource::new(Storage::new("data".to_string()).build()),
),
] {
stack.resources.insert(
id.to_string(),
ResourceEntry {
config: resource,
lifecycle: ResourceLifecycle::Frozen,
dependencies: Vec::new(),
display_name: None,
description: None,
remote_access: false,
enabled_when: None,
},
);
}
for platform in [Platform::Aws, Platform::Gcp, Platform::Azure] {
let plan = plan_compute(&stack, platform, None).expect("plan should build");
assert!(plan.pools.is_empty(), "serverless plan allocated {plan:?}");
}
}
#[test]
fn empty_stack_does_not_allocate_machine_pools() {
let mut stack = stack_with_container();
stack.resources.clear();
let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");
assert!(plan.pools.is_empty());
}
#[test]
fn explicit_cluster_without_workloads_only_plans_declared_pools() {
let mut stack = stack_with_container();
stack.resources.clear();
let cluster = ComputeCluster::new("compute".to_string())
.capacity_group(CapacityGroup {
group_id: "batch".to_string(),
instance_type: Some("t4g.medium".to_string()),
profile: None,
min_size: 1,
max_size: 1,
scale_policy: None,
nested_virtualization: None,
})
.build();
stack.resources.insert(
"compute".to_string(),
ResourceEntry {
config: Resource::new(cluster),
lifecycle: ResourceLifecycle::Frozen,
dependencies: Vec::new(),
display_name: None,
description: None,
remote_access: false,
enabled_when: None,
},
);
let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");
assert_eq!(plan.pools.len(), 1);
assert_eq!(plan.pools[0].pool_id, "batch");
assert!(plan.pools[0].errors.is_empty());
}
#[test]
fn cloud_plan_recommends_provider_machine_without_mutating_selection() {
let stack = stack_with_container();
let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");
let pool = plan.pools.first().expect("general pool should exist");
assert_eq!(pool.pool_id, "general");
assert_eq!(pool.workloads, vec!["api"]);
assert!(pool.selected.machine().is_some());
assert!(pool.machines.iter().any(|machine| machine.recommended));
}
#[test]
fn cloud_plan_only_offers_machines_matching_the_default_image_target() {
let stack = stack_with_container();
for platform in [Platform::Aws, Platform::Gcp, Platform::Azure] {
let plan = plan_compute(&stack, platform, None).expect("plan should build");
let pool = plan.pools.first().expect("general pool should exist");
let expected = instance_catalog::default_architecture(platform)
.expect("managed cloud should have a default architecture");
assert!(pool.machines.iter().all(|machine| {
instance_catalog::find_instance_type(platform, &machine.machine)
.is_some_and(|spec| spec.architecture == expected)
}));
}
}
#[test]
fn selected_machine_is_preserved_as_static_deployment_choice() {
let stack = stack_with_container();
let settings = ComputeSettings {
containers: Default::default(),
pools: [(
"general".to_string(),
ComputePoolSelection::Fixed {
machines: 1,
machine: Some("m7g.xlarge".to_string()),
failure_domains: None,
},
)]
.into_iter()
.collect(),
};
let plan = plan_compute(&stack, Platform::Aws, Some(&settings)).expect("plan should build");
let pool = plan.pools.first().expect("general pool should exist");
assert_eq!(pool.selected.machine(), Some("m7g.xlarge"));
assert!(pool.errors.is_empty());
}
#[test]
fn configurable_cloud_disk_satisfies_storage_above_fixed_local_catalog() {
let mut stack = stack_with_container();
let container = stack
.resources
.get_mut("api")
.and_then(|entry| entry.config.downcast_mut::<Container>())
.expect("test stack should contain a container");
container.ephemeral_storage = Some("8000Gi".to_string());
let settings = ComputeSettings {
containers: Default::default(),
pools: [(
"storage".to_string(),
ComputePoolSelection::Fixed {
machines: 1,
machine: Some("m8i.2xlarge".to_string()),
failure_domains: None,
},
)]
.into_iter()
.collect(),
};
let plan = plan_compute(&stack, Platform::Aws, Some(&settings)).expect("plan should build");
let pool = plan.pools.first().expect("general pool should exist");
assert!(pool.errors.is_empty());
let selected = pool
.machines
.iter()
.find(|option| option.machine == "m8i.2xlarge")
.expect("selected machine should remain an option");
assert_eq!(
selected.profile.ephemeral_storage_bytes,
8_000 * 1024 * 1024 * 1024
);
}
#[test]
fn selected_machine_defines_architecture_when_workloads_do_not() {
let stack = stack_with_container();
let settings = ComputeSettings {
containers: Default::default(),
pools: [(
"general".to_string(),
ComputePoolSelection::Fixed {
machines: 1,
machine: Some("m7i.xlarge".to_string()),
failure_domains: None,
},
)]
.into_iter()
.collect(),
};
let plan = plan_compute(&stack, Platform::Aws, Some(&settings)).expect("plan should build");
let pool = plan.pools.first().expect("general pool should exist");
assert!(pool.errors.is_empty());
assert!(pool.machines.iter().all(|machine| {
instance_catalog::find_instance_type(Platform::Aws, &machine.machine)
.is_some_and(|spec| spec.architecture == Architecture::X86_64)
}));
}
#[test]
fn graviton4_compute_machine_can_be_selected() {
let stack = stack_with_container();
let settings = ComputeSettings {
containers: Default::default(),
pools: [(
"general".to_string(),
ComputePoolSelection::Fixed {
machines: 1,
machine: Some("c8g.2xlarge".to_string()),
failure_domains: None,
},
)]
.into_iter()
.collect(),
};
let plan = plan_compute(&stack, Platform::Aws, Some(&settings)).expect("plan should build");
let pool = plan.pools.first().expect("general pool should exist");
assert!(pool.errors.is_empty(), "{:?}", pool.errors);
let selected = pool
.machines
.iter()
.find(|option| option.machine == "c8g.2xlarge")
.expect("selected machine should be offered");
assert!(selected.recommended);
assert_eq!(selected.profile.architecture, Some(Architecture::Arm64));
}
#[test]
fn explicit_capacity_group_requirements_are_merged_with_workloads() {
let mut stack = stack_with_container();
let cluster = ComputeCluster::new("compute".to_string())
.capacity_group(CapacityGroup {
group_id: "general".to_string(),
instance_type: None,
profile: Some(MachineProfile {
cpu: "4".to_string(),
memory_bytes: 16 * 1024 * 1024 * 1024,
ephemeral_storage_bytes: 20 * 1024 * 1024 * 1024,
architecture: Some(Architecture::X86_64),
gpu: None,
data_disk: None,
}),
min_size: 2,
max_size: 5,
scale_policy: None,
nested_virtualization: Some(true),
})
.build();
stack.resources.insert(
"compute".to_string(),
ResourceEntry {
config: Resource::new(cluster),
lifecycle: ResourceLifecycle::Frozen,
dependencies: Vec::new(),
display_name: None,
description: None,
remote_access: false,
enabled_when: None,
},
);
let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");
let pool = plan.pools.first().expect("general pool should exist");
let machine = pool
.selected
.machine()
.expect("AWS selection should include a machine");
let spec = instance_catalog::find_instance_type(Platform::Aws, machine)
.expect("selected machine should exist in the catalog");
assert!(spec.is_nested_virt_capable());
assert_eq!(pool.selected.min_size(), 2);
assert_eq!(pool.selected.max_size(), 5);
assert!(pool.errors.is_empty());
}
#[test]
fn materialized_machine_profile_is_not_counted_as_workload_demand() {
let mut stack = stack_with_container();
let selected = instance_catalog::find_instance_type(Platform::Aws, "m7g.xlarge")
.expect("machine should exist in catalog");
let cluster = ComputeCluster::new("compute".to_string())
.capacity_group(CapacityGroup {
group_id: "general".to_string(),
instance_type: Some(selected.name.to_string()),
profile: Some(selected.to_machine_profile()),
min_size: 1,
max_size: 1,
scale_policy: None,
nested_virtualization: None,
})
.build();
stack.resources.insert(
"compute".to_string(),
ResourceEntry {
config: Resource::new(cluster),
lifecycle: ResourceLifecycle::Frozen,
dependencies: Vec::new(),
display_name: None,
description: None,
remote_access: false,
enabled_when: None,
},
);
let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");
let pool = plan.pools.first().expect("general pool should exist");
assert_eq!(pool.requirements.cpu, "2");
assert_eq!(pool.requirements.memory_bytes, 4 * 1024 * 1024 * 1024);
assert_eq!(pool.recommended.machine(), Some("m7g.xlarge"));
assert!(pool
.machines
.iter()
.any(|option| option.machine == "m7g.xlarge"));
}
#[test]
fn nested_x86_fixed_range_pool_preserves_bounds_and_rejects_graviton() {
let daemon = Daemon::new("vm-runtime-loader".to_string())
.code(DaemonCode::Image {
image: "example.com/vm-runtime:latest".to_string(),
})
.cluster("vm-runtime".to_string())
.cpu(ResourceSpec {
min: "2".to_string(),
desired: "2".to_string(),
})
.memory(ResourceSpec {
min: "4Gi".to_string(),
desired: "4Gi".to_string(),
})
.permissions("loader".to_string())
.build();
let cluster = ComputeCluster::new("vm-runtime".to_string())
.capacity_group(CapacityGroup {
group_id: "general".to_string(),
instance_type: None,
profile: Some(MachineProfile {
cpu: "4".to_string(),
memory_bytes: 16 * 1024 * 1024 * 1024,
ephemeral_storage_bytes: 20 * 1024 * 1024 * 1024,
architecture: Some(Architecture::X86_64),
gpu: None,
data_disk: None,
}),
min_size: 2,
max_size: 2,
scale_policy: Some(CapacityGroupScalePolicy::Fixed {
machines: ComputeChoiceRange {
min: 1,
max: 5,
default: 2,
},
}),
nested_virtualization: Some(true),
})
.build();
let stack = Stack {
id: "vm-runtime".to_string(),
resources: [
(
"vm-runtime-loader".to_string(),
ResourceEntry {
config: Resource::new(daemon),
lifecycle: ResourceLifecycle::Live,
dependencies: Vec::new(),
display_name: None,
description: None,
remote_access: false,
enabled_when: None,
},
),
(
"vm-runtime".to_string(),
ResourceEntry {
config: Resource::new(cluster),
lifecycle: ResourceLifecycle::Frozen,
dependencies: Vec::new(),
display_name: None,
description: None,
remote_access: false,
enabled_when: None,
},
),
]
.into_iter()
.collect(),
permissions: crate::permissions::PermissionsConfig::default(),
supported_platforms: None,
inputs: vec![],
dynamic_container_repositories: Vec::new(),
dynamic_container_image_resources: Vec::new(),
operations: None,
};
let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");
let pool = plan.pools.first().expect("general pool should exist");
assert_eq!(pool.recommended.machine(), Some("m8i.2xlarge"));
assert_eq!(pool.recommended.min_size(), 2);
assert_eq!(pool.recommended.max_size(), 2);
assert_eq!(
pool.scale,
CapacityGroupScalePolicy::Fixed {
machines: ComputeChoiceRange {
min: 1,
max: 5,
default: 2,
},
}
);
assert!(!pool
.machines
.iter()
.any(|option| option.machine == "m7g.2xlarge"));
let invalid_settings = ComputeSettings {
containers: Default::default(),
pools: [(
"general".to_string(),
ComputePoolSelection::Fixed {
machines: 2,
machine: Some("m7g.2xlarge".to_string()),
failure_domains: None,
},
)]
.into_iter()
.collect(),
};
let invalid_plan = plan_compute(&stack, Platform::Aws, Some(&invalid_settings))
.expect("plan should build");
assert!(!invalid_plan.pools[0].errors.is_empty());
}
fn declared_pool_errors(
scale: CapacityGroupScalePolicy,
selection: ComputePoolSelection,
) -> Vec<String> {
let mut stack = stack_with_container();
let cluster = ComputeCluster::new("compute".to_string())
.capacity_group(CapacityGroup {
group_id: "general".to_string(),
instance_type: None,
profile: None,
min_size: scale.default_min_size(),
max_size: scale.default_max_size(),
scale_policy: Some(scale),
nested_virtualization: None,
})
.build();
stack.resources.insert(
"compute".to_string(),
ResourceEntry {
config: Resource::new(cluster),
lifecycle: ResourceLifecycle::Frozen,
dependencies: Vec::new(),
display_name: None,
description: None,
remote_access: false,
enabled_when: None,
},
);
let settings = ComputeSettings {
containers: Default::default(),
pools: [("general".to_string(), selection)].into_iter().collect(),
};
plan_compute(&stack, Platform::Aws, Some(&settings))
.expect("plan should build")
.pools
.remove(0)
.errors
}
fn range(min: u32, max: u32, default: u32) -> ComputeChoiceRange {
ComputeChoiceRange { min, max, default }
}
#[test]
fn generated_pool_offers_the_recommendation_as_a_default_not_a_bound() {
let mut stack = stack_with_container();
let api = stack
.resources
.get_mut("api")
.and_then(|entry| entry.config.downcast_mut::<Container>())
.expect("test stack should contain a container");
api.autoscaling = Some(crate::ContainerAutoscaling {
min: 1,
desired: 1,
max: 3,
target_cpu_percent: Some(70.0),
target_memory_percent: None,
target_http_in_flight_per_replica: None,
max_http_p95_latency_ms: None,
});
let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");
let pool = &plan.pools[0];
assert_eq!(
pool.scale,
CapacityGroupScalePolicy::Autoscale {
min: range(1, 100, 1),
max: range(1, 100, 3),
}
);
assert_eq!(
(pool.recommended.min_size(), pool.recommended.max_size()),
(1, 3)
);
let plan =
plan_compute(&stack_with_container(), Platform::Aws, None).expect("plan should build");
assert_eq!(
plan.pools[0].scale,
CapacityGroupScalePolicy::Autoscale {
min: range(1, 100, 1),
max: range(1, 100, 1),
}
);
assert!(matches!(
plan.pools[0].recommended,
ComputePoolSelection::Fixed { machines: 1, .. }
));
assert!(plan.pools[0].errors.is_empty());
}
#[test]
fn generated_pool_rejects_a_fleet_too_small_for_desired_replicas() {
let mut stack = stack_with_container();
let api = stack
.resources
.get_mut("api")
.and_then(|entry| entry.config.downcast_mut::<Container>())
.expect("test stack should contain a container");
api.replicas = Some(3);
let selection = |machines| ComputeSettings {
containers: Default::default(),
pools: [(
"general".to_string(),
ComputePoolSelection::Fixed {
machines,
machine: Some("m7g.xlarge".to_string()),
failure_domains: None,
},
)]
.into_iter()
.collect(),
};
let errors = plan_compute(&stack, Platform::Aws, Some(&selection(1)))
.expect("plan should build")
.pools
.remove(0)
.errors;
assert_eq!(errors.len(), 1, "{errors:?}");
assert!(
errors[0].starts_with(
"Pool 'general' is too small for its workloads: 1 x m7g.xlarge has 3.50 vCPU"
),
"{errors:?}"
);
assert!(errors[0].contains("request 6.00 vCPU"), "{errors:?}");
let errors = plan_compute(&stack, Platform::Aws, Some(&selection(2)))
.expect("plan should build")
.pools
.remove(0)
.errors;
assert!(errors.is_empty(), "{errors:?}");
}
#[test]
fn fixed_and_equal_autoscale_bounds_are_interchangeable_for_declared_pools() {
let fixed = |machines| ComputePoolSelection::Fixed {
machines,
machine: Some("m7g.xlarge".to_string()),
failure_domains: None,
};
let autoscale = |min, max| ComputePoolSelection::Autoscale {
min,
max,
machine: Some("m7g.xlarge".to_string()),
failure_domains: None,
};
let declared_fixed = CapacityGroupScalePolicy::Fixed {
machines: range(2, 4, 2),
};
let declared_autoscale = CapacityGroupScalePolicy::Autoscale {
min: range(1, 3, 1),
max: range(2, 6, 4),
};
assert!(declared_pool_errors(declared_fixed.clone(), autoscale(3, 3)).is_empty());
assert_eq!(
declared_pool_errors(declared_fixed.clone(), autoscale(2, 4)),
vec!["Pool 'general' must use fixed mode with 2-4 machines; autoscale 2-4 is not allowed"]
);
assert_eq!(
declared_pool_errors(declared_fixed, autoscale(5, 5)),
vec!["Pool 'general' machine count 5 is outside the allowed range 2-4"]
);
assert!(declared_pool_errors(declared_autoscale.clone(), fixed(3)).is_empty());
assert_eq!(
declared_pool_errors(declared_autoscale.clone(), fixed(1)),
vec!["Pool 'general' fixed machine count 1 is outside the allowed range 2-3"]
);
assert_eq!(
declared_pool_errors(
CapacityGroupScalePolicy::Autoscale {
min: range(1, 1, 1),
max: range(3, 5, 3),
},
fixed(1),
),
vec!["Pool 'general' must use autoscale mode (minimum 1-1, maximum 3-5)"]
);
assert_eq!(
declared_pool_errors(declared_autoscale, autoscale(1, 7)),
vec!["Pool 'general' autoscale maximum 7 is outside the allowed range 2-6"]
);
}
#[test]
fn local_plan_has_no_provider_machine_choices() {
let stack = stack_with_container();
let plan = plan_compute(&stack, Platform::Local, None).expect("plan should build");
let pool = plan.pools.first().expect("general pool should exist");
assert_eq!(pool.selected.machine(), None);
assert!(pool.machines.is_empty());
assert!(pool.errors.is_empty());
}
}