alien_core/resources/

compute_cluster.rs

//! ComputeCluster resource for long-running container workloads.
//!
//! A ComputeCluster represents the setup-owned compute boundary for containers.
//! Setup provisions:
//! - Auto Scaling Groups (AWS), Managed Instance Groups (GCP), or VM Scale Sets (Azure)
//! - IAM roles/service accounts for machine authentication
//! - Security groups/firewall rules
//! - Launch templates/instance configurations

use crate::error::{ErrorData, Result};
use crate::instance_catalog::{is_same_architecture_machine, Architecture};
use crate::resource::{ResourceDefinition, ResourceOutputsDefinition, ResourceRef};
use crate::{PermissionProfile, Platform, ResourceType};
use alien_error::AlienError;
use bon::Builder;
use serde::{Deserialize, Serialize};
use std::any::Any;
use std::collections::BTreeMap;
use std::fmt::Debug;

/// GPU specification for a capacity group.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct GpuSpec {
    /// GPU type identifier (e.g., "nvidia-a100", "nvidia-t4")
    #[serde(rename = "type")]
    pub gpu_type: String,
    /// Number of GPUs per machine
    pub count: u32,
}

/// Machine resource profile for a capacity group.
///
/// Represents the hardware specifications for machines in a capacity group.
/// These are hardware totals (what the instance type advertises), not allocatable
/// capacity. The managed container scheduler internally subtracts system reserves for planning.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct MachineProfile {
    /// CPU cores per machine (hardware total) - stored as string to preserve precision
    /// (e.g., "8.0", "4.5")
    pub cpu: String,
    /// Memory in bytes (hardware total)
    pub memory_bytes: u64,
    /// Ephemeral storage in bytes (hardware total)
    pub ephemeral_storage_bytes: u64,
    /// CPU architecture required or provided by this machine profile.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub architecture: Option<Architecture>,
    /// GPU specification (optional)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub gpu: Option<GpuSpec>,
    /// Additional empty block device each machine must provide, besides the
    /// disk holding the operating system. On AWS, GCP and Azure the runtime
    /// controller attaches one to every machine; on Machines, `join` refuses a
    /// host without one. Kubernetes rejects it.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub data_disk: Option<DataDiskRequirement>,
}

/// An additional block device a machine must provide with no partitions,
/// no filesystem signature, and no mounts, so the workload can format it.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct DataDiskRequirement {
    /// Minimum device size in bytes.
    pub size_bytes: u64,
}

/// Allowed range and default for a count selected by the installer.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct ComputeChoiceRange {
    /// Lowest allowed value.
    pub min: u32,
    /// Highest allowed value.
    pub max: u32,
    /// Default value recommended when no installer override is supplied.
    pub default: u32,
}

impl ComputeChoiceRange {
    /// Returns whether a selected value is inside the allowed range.
    pub fn contains(&self, value: u32) -> bool {
        self.min <= value && value <= self.max
    }
}

/// Source-declared scale policy for a capacity group.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase", tag = "type")]
pub enum CapacityGroupScalePolicy {
    /// A fixed-size pool where the installer can choose the fixed machine count.
    Fixed {
        /// Allowed fixed machine count range.
        machines: ComputeChoiceRange,
    },
    /// An autoscaling pool with separately bounded min and max counts.
    Autoscale {
        /// Allowed minimum machine count range.
        min: ComputeChoiceRange,
        /// Allowed maximum machine count range.
        max: ComputeChoiceRange,
    },
}

impl CapacityGroupScalePolicy {
    /// Derive the legacy policy represented by selected min/max values.
    pub fn from_selected_bounds(min_size: u32, max_size: u32) -> Self {
        if min_size == max_size {
            Self::Fixed {
                machines: ComputeChoiceRange {
                    min: min_size,
                    max: max_size,
                    default: min_size,
                },
            }
        } else {
            Self::Autoscale {
                min: ComputeChoiceRange {
                    min: min_size,
                    max: min_size,
                    default: min_size,
                },
                max: ComputeChoiceRange {
                    min: max_size,
                    max: max_size,
                    default: max_size,
                },
            }
        }
    }

    /// Default selected min bound.
    pub fn default_min_size(&self) -> u32 {
        match self {
            Self::Fixed { machines } => machines.default,
            Self::Autoscale { min, .. } => min.default,
        }
    }

    /// Default selected max bound.
    pub fn default_max_size(&self) -> u32 {
        match self {
            Self::Fixed { machines } => machines.default,
            Self::Autoscale { max, .. } => max.default,
        }
    }
}

/// Capacity group definition.
///
/// A capacity group represents machines with identical hardware profiles.
/// Each group becomes a separate Auto Scaling Group (AWS), Managed Instance Group (GCP),
/// or VM Scale Set (Azure).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct CapacityGroup {
    /// Unique identifier for this capacity group (must be lowercase alphanumeric with hyphens)
    pub group_id: String,
    /// Provider machine selected at deployment time.
    /// `alien.ts` should declare portable requirements; preflight materialization
    /// fills this field from `StackSettings.compute`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub instance_type: Option<String>,
    /// Machine resource profile (auto-derived from instance_type if not specified)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub profile: Option<MachineProfile>,
    /// Minimum number of machines (can be 0 for scale-to-zero)
    pub min_size: u32,
    /// Maximum number of machines (must be ≤ 10)
    pub max_size: u32,
    /// Source-declared scale policy and installer-editable bounds.
    ///
    /// Older stacks only have `minSize` and `maxSize`; planners derive an exact
    /// fixed/autoscale policy from those selected bounds when this field is absent.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub scale_policy: Option<CapacityGroupScalePolicy>,
    /// Require instance types that expose nested virtualization (VT-x/EPT)
    /// to guest VMs. This is needed by workloads that boot nested VMs inside
    /// containers.
    /// When true, the controller's instance-type selector is constrained
    /// to a vetted nested-virt-capable allowlist.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub nested_virtualization: Option<bool>,
    /// Data disk sizes, in GiB, a deployment may choose from. Present only when
    /// the source declares a range; `profile.data_disk` then holds the default
    /// (absent when the default is 0, which means no data disk). Preflight
    /// writes the deployment's choice into `profile.data_disk`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub data_disk_gib_range: Option<ComputeChoiceRange>,
}

impl CapacityGroup {
    /// AWS root volume size, including filesystem headroom and runtime overhead.
    pub fn aws_root_volume_size_gib(&self) -> i32 {
        self.aws_filesystem_size_gib()
            .saturating_add(12)
            .max(20)
            .min(i32::MAX as u64) as i32
    }

    /// AWS second EBS volume: the declared data disk, or for a nested
    /// virtualization pool without one, the volume such pools have always
    /// received, sized from ephemeral storage.
    pub fn aws_runtime_data_volume_size_gib(&self) -> Option<i32> {
        if let Some(size_gib) = self.data_disk_size_gib() {
            return Some(size_gib.min(i32::MAX as u64) as i32);
        }
        (self.nested_virtualization == Some(true))
            .then(|| self.aws_filesystem_size_gib().max(20).min(i32::MAX as u64) as i32)
    }

    /// Whether `other` asks for different disks than this group: another root
    /// storage size or data disk. Runtime controllers replace machines for it.
    pub fn disks_differ(&self, other: &CapacityGroup) -> bool {
        let disks = |group: &CapacityGroup| {
            group
                .profile
                .as_ref()
                .map(|profile| (profile.ephemeral_storage_bytes, profile.data_disk.clone()))
        };
        disks(self) != disks(other)
    }

    /// Data disk sizes a deployment may choose, in GiB: the declared range, or
    /// just the declared size. `None` when the pool declares no data disk.
    pub fn data_disk_choices(&self) -> Option<ComputeChoiceRange> {
        if let Some(range) = &self.data_disk_gib_range {
            return Some(range.clone());
        }
        let size_gib = u32::try_from(self.data_disk_size_gib()?).unwrap_or(u32::MAX);
        Some(ComputeChoiceRange {
            min: size_gib,
            max: size_gib,
            default: size_gib,
        })
    }

    /// Data disk size, in GiB, this pool runs with for a deployment that chose
    /// `selected` (`None` uses the default). 0 means no data disk.
    pub fn effective_data_disk_gib(
        &self,
        selected: Option<u32>,
    ) -> std::result::Result<Option<u32>, String> {
        let Some(choices) = self.data_disk_choices() else {
            return match selected {
                None | Some(0) => Ok(None),
                Some(size) => Err(format!(
                    "pool '{}' declares no data disk, so dataDiskGib {size} cannot be chosen",
                    self.group_id
                )),
            };
        };
        let size = selected.unwrap_or(choices.default);
        if !choices.contains(size) {
            return Err(if choices.min == choices.max {
                format!(
                    "pool '{}' has a fixed {} GiB data disk; dataDiskGib {size} cannot be chosen",
                    self.group_id, choices.min
                )
            } else {
                format!(
                    "pool '{}' data disk must be between {} and {} GiB; dataDiskGib {size} is outside that range",
                    self.group_id, choices.min, choices.max
                )
            });
        }
        Ok(Some(size))
    }

    /// The data disk this pool's machines get for a deployment that chose
    /// `selected` GiB (`None` uses the default; 0 means none).
    pub fn selected_data_disk(
        &self,
        selected: Option<u32>,
    ) -> std::result::Result<Option<DataDiskRequirement>, String> {
        let size_gib = self.effective_data_disk_gib(selected)?;
        if self.data_disk_gib_range.is_none() {
            // A fixed size stands as declared, to the byte: "500G" must keep
            // accepting a 500 GB disk rather than the 466 GiB it rounds up to.
            return Ok(self
                .profile
                .as_ref()
                .and_then(|profile| profile.data_disk.clone()));
        }
        Ok(size_gib
            .filter(|size_gib| *size_gib > 0)
            .map(|size_gib| DataDiskRequirement {
                size_bytes: u64::from(size_gib) * 1024 * 1024 * 1024,
            }))
    }

    /// Declared data disk size, rounded up to a whole GiB.
    pub fn data_disk_size_gib(&self) -> Option<u64> {
        let data_disk = self.profile.as_ref()?.data_disk.as_ref()?;
        Some(data_disk.size_bytes.div_ceil(1024 * 1024 * 1024))
    }

    fn aws_filesystem_size_gib(&self) -> u64 {
        self.profile
            .as_ref()
            .map_or(0, |profile| profile.ephemeral_storage_bytes)
            .div_ceil(1024 * 1024 * 1024)
            .saturating_mul(5)
            .div_ceil(4)
    }
}

/// ComputeCluster resource for running long-running container workloads.
///
/// A ComputeCluster provides the setup-owned machine boundary for containers.
/// Alien may manage the worker fleet inside that boundary when setup grants
/// `compute-cluster/management`.
///
/// ## Architecture
///
/// - **Setup** creates cloud resources: ASGs/MIGs/VMSSs, IAM roles, security groups
/// - **Alien** manages allowed fleet operations: machine count, runtime
///   machine image rollout and, on AWS, the machine type within one CPU architecture
/// - A node agent runs on each machine from the selected runtime image channel
///
/// ## Example
///
/// ```rust
/// use alien_core::{CapacityGroup, ComputeCluster, MachineProfile};
///
/// let cluster = ComputeCluster::new("compute".to_string())
///     .capacity_group(CapacityGroup {
///         group_id: "general".to_string(),
///         instance_type: None,
///         profile: Some(MachineProfile {
///             cpu: "4.0".to_string(),
///             memory_bytes: 16 * 1024 * 1024 * 1024,
///             ephemeral_storage_bytes: 20 * 1024 * 1024 * 1024,
///             architecture: None,
///             gpu: None,
///             data_disk: None,
///         }),
///         min_size: 1,
///         max_size: 5,
///         scale_policy: None,
///         nested_virtualization: None,
///         data_disk_gib_range: None,
///     })
///     .build();
/// ```
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Builder)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[builder(start_fn = new)]
pub struct ComputeCluster {
    /// Unique identifier for the container cluster.
    /// Must contain only alphanumeric characters, hyphens, and underscores.
    #[builder(start_fn)]
    pub id: String,

    /// Capacity groups defining the machine pools for this cluster.
    /// Each group becomes a separate ASG/MIG/VMSS.
    #[builder(field)]
    pub capacity_groups: Vec<CapacityGroup>,

    /// Explicit grants for the node identity, keyed by concrete resource ID.
    /// Independent of workload permission profiles; absent grants no data access.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub node_permissions: Option<PermissionProfile>,

    /// Cloud platforms on which the explicit node grants apply; absent applies everywhere.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub node_permissions_platforms: Option<Vec<Platform>>,

    /// Pool reserved for containers created after a deployment is installed.
    /// If absent, the runtime uses the `general` pool when it exists.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub dynamic_container_pool: Option<String>,

    /// Concrete provider failure domains selected during setup, keyed by capacity group.
    /// Empty preserves the existing aggregate layout when no spread policy is configured.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    #[builder(default)]
    pub selected_failure_domains: BTreeMap<String, Vec<String>>,

    /// Requested failure-domain spread keyed by capacity group.
    /// Empty preserves the existing aggregate layout.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    #[builder(default)]
    pub failure_domain_spread: BTreeMap<String, u8>,

    /// Container CIDR block for internal container networking.
    /// Auto-generated as "10.244.0.0/16" if not specified.
    /// Each machine gets a /24 subnet from this range.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub container_cidr: Option<String>,
}

impl ComputeCluster {
    /// The resource type identifier for ComputeCluster
    pub const RESOURCE_TYPE: ResourceType = ResourceType::from_static("compute-cluster");

    /// Returns the cluster's unique identifier.
    pub fn id(&self) -> &str {
        &self.id
    }

    /// Returns the container CIDR, defaulting to "10.244.0.0/16" if not specified.
    pub fn container_cidr(&self) -> &str {
        self.container_cidr.as_deref().unwrap_or("10.244.0.0/16")
    }
}

impl<S: compute_cluster_builder::State> ComputeClusterBuilder<S> {
    /// Adds a capacity group to the cluster.
    pub fn capacity_group(mut self, group: CapacityGroup) -> Self {
        self.capacity_groups.push(group);
        self
    }
}

/// Status of a single capacity group within a ComputeCluster.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct CapacityGroupStatus {
    /// Capacity group ID
    pub group_id: String,
    /// Current number of machines
    pub current_machines: u32,
    /// Desired number of machines (from the managed container capacity plan)
    pub desired_machines: u32,
    /// Instance type being used
    pub instance_type: String,
}

/// Outputs generated by a successfully provisioned ComputeCluster.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct ComputeClusterOutputs {
    /// Managed container cluster ID (workspace/project/deployment/resourceid format)
    pub cluster_id: String,
    /// Whether the managed container cluster is ready
    pub horizon_ready: bool,
    /// Status of each capacity group
    pub capacity_group_statuses: Vec<CapacityGroupStatus>,
    /// Total number of machines across all capacity groups
    pub total_machines: u32,
}

impl ResourceOutputsDefinition for ComputeClusterOutputs {
    fn get_resource_type(&self) -> ResourceType {
        ComputeCluster::RESOURCE_TYPE.clone()
    }

    fn as_any(&self) -> &dyn Any {
        self
    }

    fn box_clone(&self) -> Box<dyn ResourceOutputsDefinition> {
        Box::new(self.clone())
    }

    fn outputs_eq(&self, other: &dyn ResourceOutputsDefinition) -> bool {
        other.as_any().downcast_ref::<ComputeClusterOutputs>() == Some(self)
    }

    fn to_json_value(&self) -> serde_json::Result<serde_json::Value> {
        serde_json::to_value(self)
    }
}

impl ResourceDefinition for ComputeCluster {
    fn get_resource_type(&self) -> ResourceType {
        Self::RESOURCE_TYPE
    }

    fn id(&self) -> &str {
        &self.id
    }

    fn replace_after_failed_create_is_safe(&self) -> bool {
        // A create that failed waiting for instances may already run some; deleting it terminates
        // them, and in a capacity-starved zone the capacity may not come back.
        false
    }

    fn get_dependencies(&self) -> Vec<ResourceRef> {
        // ComputeCluster has no static dependencies.
        // Network dependency is platform-specific:
        // - AWS/GCP/Azure: Added by ComputeClusterMutation
        // - Local/Kubernetes: Not needed (Docker/K8s handles networking)
        // Platform controllers use require_dependency() at runtime to access Network state.
        Vec::new()
    }

    fn validate_update(&self, new_config: &dyn ResourceDefinition) -> Result<()> {
        let new_cluster = new_config
            .as_any()
            .downcast_ref::<ComputeCluster>()
            .ok_or_else(|| {
                AlienError::new(ErrorData::UnexpectedResourceType {
                    resource_id: self.id.clone(),
                    expected: Self::RESOURCE_TYPE,
                    actual: new_config.get_resource_type(),
                })
            })?;

        if self.id != new_cluster.id {
            return Err(AlienError::new(ErrorData::InvalidResourceUpdate {
                resource_id: self.id.clone(),
                reason: "the 'id' field is immutable".to_string(),
            }));
        }

        // Container CIDR is immutable once set
        if self.container_cidr.is_some()
            && new_cluster.container_cidr.is_some()
            && self.container_cidr != new_cluster.container_cidr
        {
            return Err(AlienError::new(ErrorData::InvalidResourceUpdate {
                resource_id: self.id.clone(),
                reason: "the 'containerCidr' field is immutable once set".to_string(),
            }));
        }

        // Validate capacity groups
        for new_group in &new_cluster.capacity_groups {
            if let Some(existing_group) = self
                .capacity_groups
                .iter()
                .find(|g| g.group_id == new_group.group_id)
            {
                // The controller rolls the fleet onto another machine of the same architecture;
                // the stack's images may not run on any other machine. Preflights also limit
                // this to the installed cloud, which this check can't see.
                if let (Some(old), Some(new)) = (
                    existing_group.instance_type.as_deref(),
                    new_group.instance_type.as_deref(),
                ) {
                    if old != new
                        && ![Platform::Aws, Platform::Gcp, Platform::Azure]
                            .into_iter()
                            .any(|platform| is_same_architecture_machine(platform, old, new))
                    {
                        return Err(AlienError::new(ErrorData::InvalidResourceUpdate {
                            resource_id: self.id.clone(),
                            reason: format!(
                                "capacity group '{}' can't change machine from '{old}' to '{new}': only a catalog machine in the same cloud and of the same CPU architecture can replace it",
                                new_group.group_id
                            ),
                        }));
                    }
                }
            }
        }

        Ok(())
    }

    fn as_any(&self) -> &dyn Any {
        self
    }

    fn as_any_mut(&mut self) -> &mut dyn Any {
        self
    }

    fn box_clone(&self) -> Box<dyn ResourceDefinition> {
        Box::new(self.clone())
    }

    fn resource_eq(&self, other: &dyn ResourceDefinition) -> bool {
        other.as_any().downcast_ref::<ComputeCluster>() == Some(self)
    }

    fn to_json_value(&self) -> serde_json::Result<serde_json::Value> {
        serde_json::to_value(self)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn aws_disk_sizing_preserves_rounding_minimums_and_saturation() {
        let gib = 1024 * 1024 * 1024;
        for (bytes, root, data) in [
            (0, 20, 20),
            (gib, 20, 20),
            (20 * gib, 37, 25),
            (20 * gib + 1, 39, 27),
            (u64::MAX, i32::MAX, i32::MAX),
        ] {
            let mut group = CapacityGroup {
                group_id: "general".into(),
                instance_type: None,
                profile: Some(MachineProfile {
                    cpu: "4.0".into(),
                    memory_bytes: 8 * gib,
                    ephemeral_storage_bytes: bytes,
                    architecture: None,
                    gpu: None,
                    data_disk: None,
                }),
                min_size: 1,
                max_size: 1,
                scale_policy: None,
                nested_virtualization: None,
                data_disk_gib_range: None,
            };
            assert_eq!(group.aws_root_volume_size_gib(), root);
            assert_eq!(group.aws_runtime_data_volume_size_gib(), None);
            group.nested_virtualization = Some(true);
            assert_eq!(group.aws_root_volume_size_gib(), root);
            assert_eq!(group.aws_runtime_data_volume_size_gib(), Some(data));
        }
    }

    /// A fixed size keeps its declared bytes, so a host disk that met it still
    /// does; a size picked from a range is whole GiB.
    #[test]
    fn a_fixed_data_disk_keeps_its_declared_bytes() {
        let gib: u64 = 1024 * 1024 * 1024;
        let declared = DataDiskRequirement {
            size_bytes: 500_000_000_000,
        };
        let mut group = CapacityGroup {
            group_id: "general".into(),
            instance_type: None,
            profile: Some(MachineProfile {
                cpu: "4.0".into(),
                memory_bytes: 8 * gib,
                ephemeral_storage_bytes: 20 * gib,
                architecture: None,
                gpu: None,
                data_disk: Some(declared.clone()),
            }),
            min_size: 1,
            max_size: 1,
            scale_policy: None,
            nested_virtualization: None,
            data_disk_gib_range: None,
        };
        // 500 GB rounds up to 466 GiB, which is the one size a deployment may name.
        assert_eq!(group.selected_data_disk(None), Ok(Some(declared.clone())));
        assert_eq!(group.selected_data_disk(Some(466)), Ok(Some(declared)));
        assert!(group.selected_data_disk(Some(500)).is_err());

        group.data_disk_gib_range = Some(ComputeChoiceRange {
            min: 0,
            max: 1024,
            default: 466,
        });
        assert_eq!(
            group.selected_data_disk(None),
            Ok(Some(DataDiskRequirement {
                size_bytes: 466 * gib
            }))
        );
        assert_eq!(group.selected_data_disk(Some(0)), Ok(None));
    }

    #[test]
    fn declared_data_disk_replaces_the_nested_virtualization_volume() {
        let gib = 1024 * 1024 * 1024;
        let group = |nested: Option<bool>, data_disk_bytes: u64| CapacityGroup {
            group_id: "general".into(),
            instance_type: None,
            profile: Some(MachineProfile {
                cpu: "4.0".into(),
                memory_bytes: 8 * gib,
                ephemeral_storage_bytes: 20 * gib,
                architecture: None,
                gpu: None,
                data_disk: Some(DataDiskRequirement {
                    size_bytes: data_disk_bytes,
                }),
            }),
            min_size: 1,
            max_size: 1,
            scale_policy: None,
            nested_virtualization: nested,
            data_disk_gib_range: None,
        };
        for nested in [None, Some(false), Some(true)] {
            // Root keeps the ephemeral storage formula; the second volume is
            // exactly the declared size, not the 25 GiB implicit volume.
            assert_eq!(group(nested, 30 * gib).aws_root_volume_size_gib(), 37);
            assert_eq!(
                group(nested, 30 * gib).aws_runtime_data_volume_size_gib(),
                Some(30)
            );
            assert_eq!(
                group(nested, 30 * gib + 1).aws_runtime_data_volume_size_gib(),
                Some(31)
            );
        }
        assert_eq!(group(None, 512 * 1024 * 1024).data_disk_size_gib(), Some(1));
    }

    #[test]
    fn test_compute_cluster_creation() {
        let cluster = ComputeCluster::new("compute".to_string())
            .capacity_group(CapacityGroup {
                group_id: "general".to_string(),
                instance_type: Some("m7g.xlarge".to_string()),
                profile: None,
                min_size: 1,
                max_size: 5,
                scale_policy: None,
                nested_virtualization: None,
                data_disk_gib_range: None,
            })
            .build();

        assert_eq!(cluster.id(), "compute");
        assert_eq!(cluster.capacity_groups.len(), 1);
        assert_eq!(cluster.capacity_groups[0].group_id, "general");
        assert_eq!(cluster.container_cidr(), "10.244.0.0/16");
    }

    #[test]
    fn test_compute_cluster_multiple_capacity_groups() {
        let cluster = ComputeCluster::new("multi-pool".to_string())
            .capacity_group(CapacityGroup {
                group_id: "general".to_string(),
                instance_type: Some("m7g.xlarge".to_string()),
                profile: None,
                min_size: 1,
                max_size: 3,
                scale_policy: None,
                nested_virtualization: None,
                data_disk_gib_range: None,
            })
            .capacity_group(CapacityGroup {
                group_id: "gpu".to_string(),
                instance_type: Some("g5.xlarge".to_string()),
                profile: Some(MachineProfile {
                    cpu: "4.0".to_string(),
                    memory_bytes: 17179869184,             // 16 GiB
                    ephemeral_storage_bytes: 214748364800, // 200 GiB
                    architecture: None,
                    gpu: Some(GpuSpec {
                        gpu_type: "nvidia-a10g".to_string(),
                        count: 1,
                    }),
                    data_disk: None,
                }),
                min_size: 0,
                max_size: 2,
                scale_policy: None,
                nested_virtualization: None,
                data_disk_gib_range: None,
            })
            .build();

        assert_eq!(cluster.capacity_groups.len(), 2);
        assert_eq!(cluster.capacity_groups[0].group_id, "general");
        assert_eq!(cluster.capacity_groups[1].group_id, "gpu");
        assert!(cluster.capacity_groups[1]
            .profile
            .as_ref()
            .unwrap()
            .gpu
            .is_some());
    }

    #[test]
    fn test_compute_cluster_custom_cidr() {
        let cluster = ComputeCluster::new("custom-net".to_string())
            .container_cidr("172.30.0.0/16".to_string())
            .capacity_group(CapacityGroup {
                group_id: "general".to_string(),
                instance_type: None,
                profile: None,
                min_size: 1,
                max_size: 5,
                scale_policy: None,
                nested_virtualization: None,
                data_disk_gib_range: None,
            })
            .build();

        assert_eq!(cluster.container_cidr(), "172.30.0.0/16");
    }

    #[test]
    fn test_compute_cluster_validate_update_immutable_id() {
        let cluster1 = ComputeCluster::new("cluster-1".to_string())
            .capacity_group(CapacityGroup {
                group_id: "general".to_string(),
                instance_type: None,
                profile: None,
                min_size: 1,
                max_size: 5,
                scale_policy: None,
                nested_virtualization: None,
                data_disk_gib_range: None,
            })
            .build();

        let cluster2 = ComputeCluster::new("cluster-2".to_string())
            .capacity_group(CapacityGroup {
                group_id: "general".to_string(),
                instance_type: None,
                profile: None,
                min_size: 1,
                max_size: 5,
                scale_policy: None,
                nested_virtualization: None,
                data_disk_gib_range: None,
            })
            .build();

        let result = cluster1.validate_update(&cluster2);
        assert!(result.is_err());
    }

    #[test]
    fn test_compute_cluster_validate_update_scale_change() {
        let cluster1 = ComputeCluster::new("compute".to_string())
            .capacity_group(CapacityGroup {
                group_id: "general".to_string(),
                instance_type: Some("m7g.xlarge".to_string()),
                profile: None,
                min_size: 1,
                max_size: 5,
                scale_policy: None,
                nested_virtualization: None,
                data_disk_gib_range: None,
            })
            .build();

        let cluster2 = ComputeCluster::new("compute".to_string())
            .capacity_group(CapacityGroup {
                group_id: "general".to_string(),
                instance_type: Some("m7g.xlarge".to_string()),
                profile: None,
                min_size: 2,
                max_size: 10,
                scale_policy: None,
                nested_virtualization: None,
                data_disk_gib_range: None,
            })
            .build();

        // Scale changes should be allowed
        let result = cluster1.validate_update(&cluster2);
        assert!(result.is_ok());
    }

    #[test]
    fn machine_changes_stay_within_one_cloud_and_architecture() {
        let cluster = |machine: &str| {
            ComputeCluster::new("compute".to_string())
                .capacity_group(CapacityGroup {
                    group_id: "general".to_string(),
                    instance_type: Some(machine.to_string()),
                    profile: None,
                    min_size: 1,
                    max_size: 5,
                    scale_policy: None,
                    nested_virtualization: None,
                    data_disk_gib_range: None,
                })
                .build()
        };
        for (old, new) in [
            ("t4g.small", "t4g.medium"),
            ("e2-medium", "n2-standard-4"),
            ("Standard_D2s_v5", "Standard_D4s_v5"),
        ] {
            cluster(old)
                .validate_update(&cluster(new))
                .expect("same cloud and architecture");
        }
        for (old, new) in [
            ("t4g.small", "m7i.large"),
            ("t4g.small", "t4g.unknown"),
            ("n2-standard-2", "m7i.large"),
            ("Standard_D2s_v5", "n2-standard-4"),
        ] {
            let error = cluster(old)
                .validate_update(&cluster(new))
                .expect_err("needs setup");
            assert_eq!(error.code, "INVALID_RESOURCE_UPDATE", "{old} -> {new}");
        }
    }

    #[test]
    fn node_permissions_are_inline_optional_and_roundtrip_exactly() {
        let mut cluster = ComputeCluster::new("compute".to_string()).build();
        let absent = serde_json::to_value(&cluster).unwrap();
        assert!(absent.get("nodePermissions").is_none());
        assert_eq!(
            serde_json::from_value::<ComputeCluster>(absent).unwrap(),
            cluster
        );
        cluster.node_permissions =
            Some(PermissionProfile::new().resource("objects", ["storage/data-read"]));
        let json = serde_json::to_value(&cluster).unwrap();
        assert_eq!(
            json["nodePermissions"],
            serde_json::json!({"objects": ["storage/data-read"]})
        );
        assert_eq!(
            serde_json::from_value::<ComputeCluster>(json).unwrap(),
            cluster
        );
        assert_eq!(ResourceDefinition::get_permissions(&cluster), None);
    }

    #[test]
    fn test_compute_cluster_serialization() {
        let cluster = ComputeCluster::new("test-cluster".to_string())
            .capacity_group(CapacityGroup {
                group_id: "general".to_string(),
                instance_type: Some("m7g.xlarge".to_string()),
                profile: None,
                min_size: 1,
                max_size: 5,
                scale_policy: None,
                nested_virtualization: None,
                data_disk_gib_range: None,
            })
            .build();

        let json = serde_json::to_string(&cluster).unwrap();
        let deserialized: ComputeCluster = serde_json::from_str(&json).unwrap();
        assert_eq!(cluster, deserialized);
    }
}