1use crate::error::{ErrorData, Result};
11use crate::instance_catalog::{is_same_architecture_aws_machine, Architecture};
12use crate::resource::{ResourceDefinition, ResourceOutputsDefinition, ResourceRef};
13use crate::ResourceType;
14use alien_error::AlienError;
15use bon::Builder;
16use serde::{Deserialize, Serialize};
17use std::any::Any;
18use std::collections::BTreeMap;
19use std::fmt::Debug;
20
21#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
23#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
24#[serde(rename_all = "camelCase")]
25pub struct GpuSpec {
26 #[serde(rename = "type")]
28 pub gpu_type: String,
29 pub count: u32,
31}
32
33#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
39#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
40#[serde(rename_all = "camelCase")]
41pub struct MachineProfile {
42 pub cpu: String,
45 pub memory_bytes: u64,
47 pub ephemeral_storage_bytes: u64,
49 #[serde(skip_serializing_if = "Option::is_none")]
51 pub architecture: Option<Architecture>,
52 #[serde(skip_serializing_if = "Option::is_none")]
54 pub gpu: Option<GpuSpec>,
55}
56
57#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
59#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
60#[serde(rename_all = "camelCase")]
61pub struct ComputeChoiceRange {
62 pub min: u32,
64 pub max: u32,
66 pub default: u32,
68}
69
70impl ComputeChoiceRange {
71 pub fn contains(&self, value: u32) -> bool {
73 self.min <= value && value <= self.max
74 }
75}
76
77#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
79#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
80#[serde(rename_all = "camelCase", tag = "type")]
81pub enum CapacityGroupScalePolicy {
82 Fixed {
84 machines: ComputeChoiceRange,
86 },
87 Autoscale {
89 min: ComputeChoiceRange,
91 max: ComputeChoiceRange,
93 },
94}
95
96impl CapacityGroupScalePolicy {
97 pub fn from_selected_bounds(min_size: u32, max_size: u32) -> Self {
99 if min_size == max_size {
100 Self::Fixed {
101 machines: ComputeChoiceRange {
102 min: min_size,
103 max: max_size,
104 default: min_size,
105 },
106 }
107 } else {
108 Self::Autoscale {
109 min: ComputeChoiceRange {
110 min: min_size,
111 max: min_size,
112 default: min_size,
113 },
114 max: ComputeChoiceRange {
115 min: max_size,
116 max: max_size,
117 default: max_size,
118 },
119 }
120 }
121 }
122
123 pub fn default_min_size(&self) -> u32 {
125 match self {
126 Self::Fixed { machines } => machines.default,
127 Self::Autoscale { min, .. } => min.default,
128 }
129 }
130
131 pub fn default_max_size(&self) -> u32 {
133 match self {
134 Self::Fixed { machines } => machines.default,
135 Self::Autoscale { max, .. } => max.default,
136 }
137 }
138}
139
140#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
146#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
147#[serde(rename_all = "camelCase")]
148pub struct CapacityGroup {
149 pub group_id: String,
151 #[serde(skip_serializing_if = "Option::is_none")]
155 pub instance_type: Option<String>,
156 #[serde(skip_serializing_if = "Option::is_none")]
158 pub profile: Option<MachineProfile>,
159 pub min_size: u32,
161 pub max_size: u32,
163 #[serde(skip_serializing_if = "Option::is_none")]
168 pub scale_policy: Option<CapacityGroupScalePolicy>,
169 #[serde(skip_serializing_if = "Option::is_none")]
175 pub nested_virtualization: Option<bool>,
176}
177
178#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Builder)]
215#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
216#[serde(rename_all = "camelCase", deny_unknown_fields)]
217#[builder(start_fn = new)]
218pub struct ComputeCluster {
219 #[builder(start_fn)]
222 pub id: String,
223
224 #[builder(field)]
227 pub capacity_groups: Vec<CapacityGroup>,
228
229 #[serde(default, skip_serializing_if = "Option::is_none")]
232 pub dynamic_container_pool: Option<String>,
233
234 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
237 #[builder(default)]
238 pub selected_failure_domains: BTreeMap<String, Vec<String>>,
239
240 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
243 #[builder(default)]
244 pub failure_domain_spread: BTreeMap<String, u8>,
245
246 #[serde(skip_serializing_if = "Option::is_none")]
250 pub container_cidr: Option<String>,
251}
252
253impl ComputeCluster {
254 pub const RESOURCE_TYPE: ResourceType = ResourceType::from_static("compute-cluster");
256
257 pub fn id(&self) -> &str {
259 &self.id
260 }
261
262 pub fn container_cidr(&self) -> &str {
264 self.container_cidr.as_deref().unwrap_or("10.244.0.0/16")
265 }
266}
267
268impl<S: compute_cluster_builder::State> ComputeClusterBuilder<S> {
269 pub fn capacity_group(mut self, group: CapacityGroup) -> Self {
271 self.capacity_groups.push(group);
272 self
273 }
274}
275
276#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
278#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
279#[serde(rename_all = "camelCase")]
280pub struct CapacityGroupStatus {
281 pub group_id: String,
283 pub current_machines: u32,
285 pub desired_machines: u32,
287 pub instance_type: String,
289}
290
291#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
293#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
294#[serde(rename_all = "camelCase")]
295pub struct ComputeClusterOutputs {
296 pub cluster_id: String,
298 pub horizon_ready: bool,
300 pub capacity_group_statuses: Vec<CapacityGroupStatus>,
302 pub total_machines: u32,
304}
305
306impl ResourceOutputsDefinition for ComputeClusterOutputs {
307 fn get_resource_type(&self) -> ResourceType {
308 ComputeCluster::RESOURCE_TYPE.clone()
309 }
310
311 fn as_any(&self) -> &dyn Any {
312 self
313 }
314
315 fn box_clone(&self) -> Box<dyn ResourceOutputsDefinition> {
316 Box::new(self.clone())
317 }
318
319 fn outputs_eq(&self, other: &dyn ResourceOutputsDefinition) -> bool {
320 other.as_any().downcast_ref::<ComputeClusterOutputs>() == Some(self)
321 }
322
323 fn to_json_value(&self) -> serde_json::Result<serde_json::Value> {
324 serde_json::to_value(self)
325 }
326}
327
328impl ResourceDefinition for ComputeCluster {
329 fn get_resource_type(&self) -> ResourceType {
330 Self::RESOURCE_TYPE
331 }
332
333 fn id(&self) -> &str {
334 &self.id
335 }
336
337 fn get_dependencies(&self) -> Vec<ResourceRef> {
338 Vec::new()
344 }
345
346 fn validate_update(&self, new_config: &dyn ResourceDefinition) -> Result<()> {
347 let new_cluster = new_config
348 .as_any()
349 .downcast_ref::<ComputeCluster>()
350 .ok_or_else(|| {
351 AlienError::new(ErrorData::UnexpectedResourceType {
352 resource_id: self.id.clone(),
353 expected: Self::RESOURCE_TYPE,
354 actual: new_config.get_resource_type(),
355 })
356 })?;
357
358 if self.id != new_cluster.id {
359 return Err(AlienError::new(ErrorData::InvalidResourceUpdate {
360 resource_id: self.id.clone(),
361 reason: "the 'id' field is immutable".to_string(),
362 }));
363 }
364
365 if self.container_cidr.is_some()
367 && new_cluster.container_cidr.is_some()
368 && self.container_cidr != new_cluster.container_cidr
369 {
370 return Err(AlienError::new(ErrorData::InvalidResourceUpdate {
371 resource_id: self.id.clone(),
372 reason: "the 'containerCidr' field is immutable once set".to_string(),
373 }));
374 }
375
376 for new_group in &new_cluster.capacity_groups {
378 if let Some(existing_group) = self
379 .capacity_groups
380 .iter()
381 .find(|g| g.group_id == new_group.group_id)
382 {
383 if let (Some(old), Some(new)) = (
387 existing_group.instance_type.as_deref(),
388 new_group.instance_type.as_deref(),
389 ) {
390 if old != new && !is_same_architecture_aws_machine(old, new) {
391 return Err(AlienError::new(ErrorData::InvalidResourceUpdate {
392 resource_id: self.id.clone(),
393 reason: format!(
394 "capacity group '{}' can't change machine from '{old}' to '{new}': only an AWS machine of the same CPU architecture can replace it",
395 new_group.group_id
396 ),
397 }));
398 }
399 }
400 }
401 }
402
403 Ok(())
404 }
405
406 fn as_any(&self) -> &dyn Any {
407 self
408 }
409
410 fn as_any_mut(&mut self) -> &mut dyn Any {
411 self
412 }
413
414 fn box_clone(&self) -> Box<dyn ResourceDefinition> {
415 Box::new(self.clone())
416 }
417
418 fn resource_eq(&self, other: &dyn ResourceDefinition) -> bool {
419 other.as_any().downcast_ref::<ComputeCluster>() == Some(self)
420 }
421
422 fn to_json_value(&self) -> serde_json::Result<serde_json::Value> {
423 serde_json::to_value(self)
424 }
425}
426
427#[cfg(test)]
428mod tests {
429 use super::*;
430
431 #[test]
432 fn test_compute_cluster_creation() {
433 let cluster = ComputeCluster::new("compute".to_string())
434 .capacity_group(CapacityGroup {
435 group_id: "general".to_string(),
436 instance_type: Some("m7g.xlarge".to_string()),
437 profile: None,
438 min_size: 1,
439 max_size: 5,
440 scale_policy: None,
441 nested_virtualization: None,
442 })
443 .build();
444
445 assert_eq!(cluster.id(), "compute");
446 assert_eq!(cluster.capacity_groups.len(), 1);
447 assert_eq!(cluster.capacity_groups[0].group_id, "general");
448 assert_eq!(cluster.container_cidr(), "10.244.0.0/16");
449 }
450
451 #[test]
452 fn test_compute_cluster_multiple_capacity_groups() {
453 let cluster = ComputeCluster::new("multi-pool".to_string())
454 .capacity_group(CapacityGroup {
455 group_id: "general".to_string(),
456 instance_type: Some("m7g.xlarge".to_string()),
457 profile: None,
458 min_size: 1,
459 max_size: 3,
460 scale_policy: None,
461 nested_virtualization: None,
462 })
463 .capacity_group(CapacityGroup {
464 group_id: "gpu".to_string(),
465 instance_type: Some("g5.xlarge".to_string()),
466 profile: Some(MachineProfile {
467 cpu: "4.0".to_string(),
468 memory_bytes: 17179869184, ephemeral_storage_bytes: 214748364800, architecture: None,
471 gpu: Some(GpuSpec {
472 gpu_type: "nvidia-a10g".to_string(),
473 count: 1,
474 }),
475 }),
476 min_size: 0,
477 max_size: 2,
478 scale_policy: None,
479 nested_virtualization: None,
480 })
481 .build();
482
483 assert_eq!(cluster.capacity_groups.len(), 2);
484 assert_eq!(cluster.capacity_groups[0].group_id, "general");
485 assert_eq!(cluster.capacity_groups[1].group_id, "gpu");
486 assert!(cluster.capacity_groups[1]
487 .profile
488 .as_ref()
489 .unwrap()
490 .gpu
491 .is_some());
492 }
493
494 #[test]
495 fn test_compute_cluster_custom_cidr() {
496 let cluster = ComputeCluster::new("custom-net".to_string())
497 .container_cidr("172.30.0.0/16".to_string())
498 .capacity_group(CapacityGroup {
499 group_id: "general".to_string(),
500 instance_type: None,
501 profile: None,
502 min_size: 1,
503 max_size: 5,
504 scale_policy: None,
505 nested_virtualization: None,
506 })
507 .build();
508
509 assert_eq!(cluster.container_cidr(), "172.30.0.0/16");
510 }
511
512 #[test]
513 fn test_compute_cluster_validate_update_immutable_id() {
514 let cluster1 = ComputeCluster::new("cluster-1".to_string())
515 .capacity_group(CapacityGroup {
516 group_id: "general".to_string(),
517 instance_type: None,
518 profile: None,
519 min_size: 1,
520 max_size: 5,
521 scale_policy: None,
522 nested_virtualization: None,
523 })
524 .build();
525
526 let cluster2 = ComputeCluster::new("cluster-2".to_string())
527 .capacity_group(CapacityGroup {
528 group_id: "general".to_string(),
529 instance_type: None,
530 profile: None,
531 min_size: 1,
532 max_size: 5,
533 scale_policy: None,
534 nested_virtualization: None,
535 })
536 .build();
537
538 let result = cluster1.validate_update(&cluster2);
539 assert!(result.is_err());
540 }
541
542 #[test]
543 fn test_compute_cluster_validate_update_scale_change() {
544 let cluster1 = ComputeCluster::new("compute".to_string())
545 .capacity_group(CapacityGroup {
546 group_id: "general".to_string(),
547 instance_type: Some("m7g.xlarge".to_string()),
548 profile: None,
549 min_size: 1,
550 max_size: 5,
551 scale_policy: None,
552 nested_virtualization: None,
553 })
554 .build();
555
556 let cluster2 = ComputeCluster::new("compute".to_string())
557 .capacity_group(CapacityGroup {
558 group_id: "general".to_string(),
559 instance_type: Some("m7g.xlarge".to_string()),
560 profile: None,
561 min_size: 2,
562 max_size: 10,
563 scale_policy: None,
564 nested_virtualization: None,
565 })
566 .build();
567
568 let result = cluster1.validate_update(&cluster2);
570 assert!(result.is_ok());
571 }
572
573 #[test]
574 fn machine_changes_stay_within_one_aws_architecture() {
575 let cluster = |machine: &str| {
576 ComputeCluster::new("compute".to_string())
577 .capacity_group(CapacityGroup {
578 group_id: "general".to_string(),
579 instance_type: Some(machine.to_string()),
580 profile: None,
581 min_size: 1,
582 max_size: 5,
583 scale_policy: None,
584 nested_virtualization: None,
585 })
586 .build()
587 };
588 cluster("t4g.small")
589 .validate_update(&cluster("t4g.medium"))
590 .expect("arm64 to arm64");
591 for (old, new) in [
592 ("t4g.small", "m7i.large"),
593 ("t4g.small", "t4g.unknown"),
594 ("n2-standard-2", "n2-standard-4"),
595 ] {
596 let error = cluster(old)
597 .validate_update(&cluster(new))
598 .expect_err("needs setup");
599 assert_eq!(error.code, "INVALID_RESOURCE_UPDATE", "{old} -> {new}");
600 }
601 }
602
603 #[test]
604 fn test_compute_cluster_serialization() {
605 let cluster = ComputeCluster::new("test-cluster".to_string())
606 .capacity_group(CapacityGroup {
607 group_id: "general".to_string(),
608 instance_type: Some("m7g.xlarge".to_string()),
609 profile: None,
610 min_size: 1,
611 max_size: 5,
612 scale_policy: None,
613 nested_virtualization: None,
614 })
615 .build();
616
617 let json = serde_json::to_string(&cluster).unwrap();
618 let deserialized: ComputeCluster = serde_json::from_str(&json).unwrap();
619 assert_eq!(cluster, deserialized);
620 }
621}