Coverage for scripts / accelerator_catalog.py: 100.00%
566 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1#!/usr/bin/env python3
2"""Validate and maintain GCO's EC2 accelerator catalog.
4Normal CI is deliberately offline: ``validate`` compares the checked-in catalog
5with Karpenter NodePools, the capacity-history watch lists in ``cdk.json`` and
6``ConfigLoader``, and the Spot Placement Score instance pools declared below. The monthly dependency workflow runs ``check-online`` to
7compare that catalog with the union of NVIDIA GPU and AWS Neuron instance types
8returned by EC2 in every enabled commercial Region.
10Online reads are sequential and paginated. Botocore adaptive retries protect the
11monthly scan from transient EC2 throttling without making the deterministic test
12suite depend on credentials or a mutable cloud catalog.
13"""
15from __future__ import annotations
17import argparse
18import ast
19import json
20import sys
21from dataclasses import dataclass
22from datetime import UTC, datetime
23from pathlib import Path
24from typing import Any, Literal, cast
26import yaml
28ROOT = Path(__file__).resolve().parents[1]
29DEFAULT_CATALOG_PATH = ROOT / "gco" / "config" / "accelerator_catalog.json"
30DEFAULT_CDK_PATH = ROOT / "cdk.json"
31DEFAULT_CONFIG_LOADER_PATH = ROOT / "gco" / "config" / "config_loader.py"
32DEFAULT_MANIFESTS_PATH = ROOT / "lambda" / "kubectl-applier-simple" / "manifests"
34Accelerator = Literal["nvidia", "neuron"]
35Lifecycle = Literal["active", "announced", "deprecated", "end-of-life"]
37_ALLOWED_ACCELERATORS = {"nvidia", "neuron"}
38_ALLOWED_LIFECYCLES = {"active", "announced", "deprecated", "end-of-life"}
39_DEPRECATED_LIFECYCLES = {"deprecated", "end-of-life"}
40_ENABLED_REGION_STATUSES = {"opt-in-not-required", "opted-in"}
41_EC2_TO_KUBERNETES_ARCH = {"x86_64": "amd64", "arm64": "arm64"}
44class CatalogError(ValueError):
45 """Raised when checked-in catalog or repository input is malformed."""
48def _mapping(value: object, label: str) -> dict[str, object]:
49 if not isinstance(value, dict) or not all(isinstance(key, str) for key in value):
50 raise CatalogError(f"{label} must be a JSON/YAML object with string keys")
51 return cast(dict[str, object], value)
54def _string(value: object, label: str) -> str:
55 if not isinstance(value, str) or not value:
56 raise CatalogError(f"{label} must be a non-empty string")
57 return value
60def _utc_timestamp(value: object, label: str) -> str:
61 timestamp = _string(value, label)
62 if "T" not in timestamp or not timestamp.endswith("Z"):
63 raise CatalogError(f"{label} must be an ISO 8601 UTC timestamp ending in Z")
64 try:
65 datetime.fromisoformat(f"{timestamp[:-1]}+00:00")
66 except ValueError as exc:
67 raise CatalogError(f"{label} must be an ISO 8601 UTC timestamp ending in Z") from exc
68 return timestamp
71def _current_utc_timestamp() -> str:
72 return datetime.now(tz=UTC).isoformat(timespec="seconds").replace("+00:00", "Z")
75def _string_list(value: object, label: str, *, allow_empty: bool = False) -> tuple[str, ...]:
76 if not isinstance(value, list) or not all(isinstance(item, str) and item for item in value):
77 raise CatalogError(f"{label} must be a list of non-empty strings")
78 result = tuple(cast(list[str], value))
79 if not allow_empty and not result:
80 raise CatalogError(f"{label} must not be empty")
81 return result
84def _family_for_instance_type(instance_type: str) -> str:
85 family, separator, size = instance_type.partition(".")
86 if not separator or not family or not size:
87 raise CatalogError(f"invalid EC2 instance type in catalog: {instance_type!r}")
88 return family
91@dataclass(frozen=True)
92class FamilyPolicy:
93 """Reviewed lifecycle and generation metadata for one EC2 family."""
95 name: str
96 accelerator: Accelerator
97 architectures: tuple[str, ...]
98 track: str
99 generation: int
100 lifecycle: Lifecycle
101 manifest_allowed: bool
102 reason: str | None
103 replacements: tuple[str, ...]
105 @classmethod
106 def from_mapping(cls, name: str, value: object) -> FamilyPolicy:
107 raw = _mapping(value, f"families.{name}")
108 accelerator_value = _string(raw.get("accelerator"), f"families.{name}.accelerator")
109 if accelerator_value not in _ALLOWED_ACCELERATORS:
110 raise CatalogError(
111 f"families.{name}.accelerator must be one of {sorted(_ALLOWED_ACCELERATORS)}"
112 )
113 lifecycle_value = _string(raw.get("lifecycle"), f"families.{name}.lifecycle")
114 if lifecycle_value not in _ALLOWED_LIFECYCLES:
115 raise CatalogError(
116 f"families.{name}.lifecycle must be one of {sorted(_ALLOWED_LIFECYCLES)}"
117 )
118 generation = raw.get("generation")
119 if isinstance(generation, bool) or not isinstance(generation, int) or generation < 0:
120 raise CatalogError(f"families.{name}.generation must be a non-negative integer")
121 if lifecycle_value == "announced" and "manifest_allowed" not in raw:
122 raise CatalogError(
123 f"families.{name}.manifest_allowed is required for announced families"
124 )
125 manifest_allowed_value = raw.get("manifest_allowed", lifecycle_value == "active")
126 if not isinstance(manifest_allowed_value, bool):
127 raise CatalogError(f"families.{name}.manifest_allowed must be a boolean")
128 reason_value = raw.get("reason")
129 if reason_value is not None and not isinstance(reason_value, str):
130 raise CatalogError(f"families.{name}.reason must be a string when present")
131 replacements_value = raw.get("replacements", [])
132 replacements = _string_list(
133 replacements_value,
134 f"families.{name}.replacements",
135 allow_empty=True,
136 )
137 return cls(
138 name=name,
139 accelerator=cast(Accelerator, accelerator_value),
140 architectures=_string_list(raw.get("architectures"), f"families.{name}.architectures"),
141 track=_string(raw.get("track"), f"families.{name}.track"),
142 generation=generation,
143 lifecycle=cast(Lifecycle, lifecycle_value),
144 manifest_allowed=manifest_allowed_value,
145 reason=reason_value,
146 replacements=replacements,
147 )
149 def to_mapping(self) -> dict[str, object]:
150 result: dict[str, object] = {
151 "accelerator": self.accelerator,
152 "architectures": list(self.architectures),
153 "track": self.track,
154 "generation": self.generation,
155 "lifecycle": self.lifecycle,
156 }
157 default_allowed = self.lifecycle == "active"
158 if self.manifest_allowed != default_allowed or self.lifecycle == "announced":
159 result["manifest_allowed"] = self.manifest_allowed
160 if self.reason is not None:
161 result["reason"] = self.reason
162 if self.replacements:
163 result["replacements"] = list(self.replacements)
164 return result
167@dataclass(frozen=True)
168class Catalog:
169 """Normalized checked-in accelerator catalog."""
171 schema_version: int
172 last_refreshed_at: str
173 source: dict[str, object]
174 families: dict[str, FamilyPolicy]
175 instance_types: tuple[str, ...]
177 @classmethod
178 def load(cls, path: Path = DEFAULT_CATALOG_PATH) -> Catalog:
179 try:
180 parsed: object = json.loads(path.read_text())
181 except (OSError, json.JSONDecodeError) as exc:
182 raise CatalogError(f"cannot read accelerator catalog {path}: {exc}") from exc
183 raw = _mapping(parsed, str(path))
184 schema_version = raw.get("schema_version")
185 if schema_version != 1:
186 raise CatalogError(f"{path}: schema_version must be 1")
187 last_refreshed_at = _utc_timestamp(
188 raw.get("last_refreshed_at"), f"{path}: last_refreshed_at"
189 )
190 source = _mapping(raw.get("source"), f"{path}: source")
191 family_values = _mapping(raw.get("families"), f"{path}: families")
192 families = {
193 name: FamilyPolicy.from_mapping(name, value)
194 for name, value in sorted(family_values.items())
195 }
196 instance_types = _string_list(raw.get("instance_types"), f"{path}: instance_types")
197 if instance_types != tuple(sorted(instance_types)):
198 raise CatalogError(f"{path}: instance_types must be sorted lexicographically")
199 if len(instance_types) != len(set(instance_types)):
200 raise CatalogError(f"{path}: instance_types contains duplicates")
201 for instance_type in instance_types:
202 family = _family_for_instance_type(instance_type)
203 if family not in families:
204 raise CatalogError(
205 f"{path}: {instance_type} has no reviewed families.{family} policy"
206 )
207 return cls(
208 schema_version=1,
209 last_refreshed_at=last_refreshed_at,
210 source=source,
211 families=families,
212 instance_types=instance_types,
213 )
215 @property
216 def live_families(self) -> frozenset[str]:
217 return frozenset(_family_for_instance_type(item) for item in self.instance_types)
219 def to_mapping(self) -> dict[str, object]:
220 return {
221 "schema_version": self.schema_version,
222 "last_refreshed_at": self.last_refreshed_at,
223 "source": self.source,
224 "families": {
225 name: policy.to_mapping() for name, policy in sorted(self.families.items())
226 },
227 "instance_types": list(self.instance_types),
228 }
231@dataclass(frozen=True)
232class NodePoolReference:
233 """Accelerator families and architectures selected by one NodePool."""
235 path: Path
236 name: str
237 families: tuple[str, ...]
238 architectures: tuple[str, ...]
240 @property
241 def location(self) -> str:
242 try:
243 display_path = self.path.relative_to(ROOT)
244 except ValueError:
245 display_path = self.path
246 return f"{display_path} (NodePool {self.name})"
249@dataclass(frozen=True)
250class Finding:
251 """One deterministic, actionable offline validation failure."""
253 code: str
254 title: str
255 detail: str
256 recommendation: str
257 locations: tuple[str, ...] = ()
259 def sort_key(self) -> tuple[str, str, tuple[str, ...]]:
260 return (self.code, self.title, self.locations)
263@dataclass(frozen=True)
264class ValidationReport:
265 findings: tuple[Finding, ...]
267 @property
268 def ok(self) -> bool:
269 return not self.findings
271 def to_text(self) -> str:
272 if self.ok:
273 return (
274 "Accelerator catalog validation passed: NodePools, both watch lists, "
275 "and the instance pools are current.\n"
276 )
277 lines = [f"Accelerator catalog validation failed with {len(self.findings)} finding(s):"]
278 for finding in self.findings:
279 lines.append(f"\nERROR [{finding.code}] {finding.title}")
280 if finding.locations:
281 lines.append(f" Location: {', '.join(finding.locations)}")
282 lines.append(f" Why: {finding.detail}")
283 lines.append(f" Recommended change: {finding.recommendation}")
284 return "\n".join(lines) + "\n"
286 def to_markdown(self) -> str:
287 status = "PASS" if self.ok else "ACTION REQUIRED"
288 lines = [
289 "## Accelerator catalog and NodePool policy",
290 "",
291 f"**Status: {status}.**",
292 ]
293 if self.ok:
294 lines.extend(
295 [
296 "",
297 "The checked-in EC2 catalog, Karpenter families, capacity-history "
298 "watch lists in `cdk.json` and `ConfigLoader`, and the Spot "
299 "Placement Score instance pools are synchronized.",
300 ]
301 )
302 return "\n".join(lines) + "\n"
303 lines.extend(["", f"{len(self.findings)} actionable finding(s):"])
304 for finding in self.findings:
305 lines.extend(["", f"### {finding.title}", ""])
306 if finding.locations:
307 lines.append(f"- **Location:** {', '.join(finding.locations)}")
308 lines.append(f"- **Why:** {finding.detail}")
309 lines.append(f"- **Recommended change:** {finding.recommendation}")
310 return "\n".join(lines) + "\n"
313def load_nodepools(manifests_path: Path = DEFAULT_MANIFESTS_PATH) -> tuple[NodePoolReference, ...]:
314 """Load every NodePool manifest that declares explicit instance families."""
315 pools: list[NodePoolReference] = []
316 for path in sorted(manifests_path.glob("*nodepool*.yaml")):
317 try:
318 parsed_documents: list[object] = list(yaml.safe_load_all(path.read_text()))
319 except (OSError, yaml.YAMLError) as exc:
320 raise CatalogError(f"cannot read NodePool manifest {path}: {exc}") from exc
321 root: dict[str, object] | None = None
322 for document_index, parsed in enumerate(parsed_documents):
323 if parsed is None:
324 continue
325 candidate = _mapping(parsed, f"{path}: document {document_index + 1}")
326 if candidate.get("kind") == "NodePool":
327 root = candidate
328 break
329 if root is None:
330 continue
331 metadata = _mapping(root.get("metadata"), f"{path}: metadata")
332 spec = _mapping(root.get("spec"), f"{path}: spec")
333 template = _mapping(spec.get("template"), f"{path}: spec.template")
334 template_spec = _mapping(template.get("spec"), f"{path}: spec.template.spec")
335 requirements_value = template_spec.get("requirements", [])
336 if not isinstance(requirements_value, list):
337 raise CatalogError(f"{path}: spec.template.spec.requirements must be a list")
338 families: tuple[str, ...] = ()
339 architectures: tuple[str, ...] = ()
340 for index, requirement_value in enumerate(requirements_value):
341 requirement = _mapping(requirement_value, f"{path}: requirements[{index}]")
342 key = requirement.get("key")
343 if key == "eks.amazonaws.com/instance-family":
344 families = _string_list(
345 requirement.get("values"), f"{path}: instance-family values"
346 )
347 elif key == "kubernetes.io/arch":
348 architectures = _string_list(
349 requirement.get("values"), f"{path}: architecture values"
350 )
351 if families:
352 pools.append(
353 NodePoolReference(
354 path=path,
355 name=_string(metadata.get("name"), f"{path}: metadata.name"),
356 families=families,
357 architectures=architectures,
358 )
359 )
360 return tuple(pools)
363def validate_nodepools(
364 catalog: Catalog, nodepools: tuple[NodePoolReference, ...]
365) -> tuple[Finding, ...]:
366 """Validate lifecycle, architecture, and newest-generation NodePool policy."""
367 findings: list[Finding] = []
368 referenced_families = {family for pool in nodepools for family in pool.families}
370 for pool in nodepools:
371 for family in pool.families:
372 policy = catalog.families.get(family)
373 if policy is None:
374 findings.append(
375 Finding(
376 code="unknown-family",
377 title=f"{pool.name} references unreviewed family {family}",
378 locations=(pool.location,),
379 detail=(
380 "The family has no lifecycle, architecture, or generation policy in "
381 "gco/config/accelerator_catalog.json."
382 ),
383 recommendation=(
384 f"Review {family} against EC2, add explicit family metadata to the "
385 "catalog, then rerun this validator; do not silently allow unknown "
386 "families."
387 ),
388 )
389 )
390 continue
391 if not policy.manifest_allowed or policy.lifecycle in _DEPRECATED_LIFECYCLES:
392 replacements = ", ".join(policy.replacements) or "a reviewed active family"
393 findings.append(
394 Finding(
395 code="deprecated-family",
396 title=(f"{pool.name} references {policy.lifecycle} family {policy.name}"),
397 locations=(pool.location,),
398 detail=policy.reason
399 or f"Family {policy.name} is marked {policy.lifecycle} by project policy.",
400 recommendation=(
401 f"Remove {policy.name} from this NodePool and use {replacements} "
402 "instead. Keep capacity-history observation separate from scheduling "
403 "eligibility."
404 ),
405 )
406 )
407 if pool.architectures and not set(pool.architectures).intersection(
408 policy.architectures
409 ):
410 findings.append(
411 Finding(
412 code="architecture-mismatch",
413 title=f"{pool.name} cannot launch {policy.name}",
414 locations=(pool.location,),
415 detail=(
416 f"The NodePool requires {list(pool.architectures)}, but {policy.name} "
417 f"is cataloged for {list(policy.architectures)}."
418 ),
419 recommendation=(
420 f"Move {policy.name} to an architecture-compatible NodePool or correct "
421 "the NodePool's kubernetes.io/arch requirement."
422 ),
423 )
424 )
426 active_by_track: dict[str, list[FamilyPolicy]] = {}
427 for family in catalog.live_families:
428 policy = catalog.families[family]
429 if policy.lifecycle == "active":
430 active_by_track.setdefault(policy.track, []).append(policy)
432 for track, policies in sorted(active_by_track.items()):
433 latest_generation = max(policy.generation for policy in policies)
434 latest = sorted(
435 policy.name for policy in policies if policy.generation == latest_generation
436 )
437 if referenced_families.intersection(latest):
438 continue
439 latest_architectures = {
440 architecture
441 for policy in policies
442 if policy.generation == latest_generation
443 for architecture in policy.architectures
444 }
445 candidates: list[NodePoolReference] = []
446 for pool in nodepools:
447 pool_tracks = {
448 catalog.families[family].track
449 for family in pool.families
450 if family in catalog.families
451 }
452 architecture_matches = not pool.architectures or bool(
453 latest_architectures.intersection(pool.architectures)
454 )
455 if track in pool_tracks and architecture_matches:
456 candidates.append(pool)
457 locations = tuple(pool.location for pool in candidates)
458 latest_display = ", ".join(latest)
459 if locations:
460 target = ", ".join(locations)
461 recommendation = (
462 f"Update {target}: add a reviewed family from [{latest_display}] after confirming "
463 "EKS Auto Mode labels and workload compatibility."
464 )
465 else:
466 recommendation = (
467 f"Create an architecture-compatible NodePool for [{latest_display}], or document "
468 f"why the {track} track is intentionally unsupported."
469 )
470 findings.append(
471 Finding(
472 code="newer-generation-unreferenced",
473 title=f"New {track} generation is absent from all NodePools",
474 locations=locations,
475 detail=(
476 f"The EC2 catalog contains generation {latest_generation} family/families "
477 f"[{latest_display}], but no NodePool references any of them."
478 ),
479 recommendation=recommendation,
480 )
481 )
483 return tuple(findings)
486def _watch_list_findings(
487 catalog: Catalog,
488 watched: tuple[str, ...],
489 *,
490 location: Path,
491 code_prefix: str,
492 subject: str,
493 target: str,
494 peer: str,
495) -> tuple[Finding, ...]:
496 """Compare one capacity-history watch list with the normalized catalog."""
497 findings: list[Finding] = []
498 if len(watched) != len(set(watched)):
499 duplicates = sorted(item for item in set(watched) if watched.count(item) > 1)
500 findings.append(
501 Finding(
502 code=f"{code_prefix}-duplicates",
503 title=f"{subject} contains duplicate instance types",
504 locations=(str(location),),
505 detail=f"Duplicate values: {', '.join(duplicates)}.",
506 recommendation=f"Remove duplicate entries from {target}.",
507 )
508 )
510 expected = set(catalog.instance_types)
511 actual = set(watched)
512 missing = sorted(expected - actual)
513 unexpected = sorted(actual - expected)
514 if missing:
515 findings.append(
516 Finding(
517 code=f"{code_prefix}-missing",
518 title=f"{subject} omits accelerator instance types",
519 locations=(str(location),),
520 detail=f"Missing {len(missing)} catalog type(s): {', '.join(missing)}.",
521 recommendation=(
522 f"Add every listed type to {target} and mirror the same default in {peer}."
523 ),
524 )
525 )
526 if unexpected:
527 findings.append(
528 Finding(
529 code=f"{code_prefix}-unexpected",
530 title=f"{subject} contains types outside the catalog",
531 locations=(str(location),),
532 detail=f"Unexpected {len(unexpected)} type(s): {', '.join(unexpected)}.",
533 recommendation=(
534 "Refresh and review the catalog before retaining these entries, or remove "
535 f"them from {target}."
536 ),
537 )
538 )
539 if not missing and not unexpected and watched != catalog.instance_types:
540 findings.append(
541 Finding(
542 code=f"{code_prefix}-order",
543 title=f"{subject} is not in normalized catalog order",
544 locations=(str(location),),
545 detail="The values are complete but their order differs from the checked-in catalog.",
546 recommendation=(
547 f"Replace {target} with gco/config/accelerator_catalog.json instance_types so "
548 "future catalog refreshes produce reviewable diffs."
549 ),
550 )
551 )
552 return tuple(findings)
555def validate_watch_instance_types(
556 catalog: Catalog, cdk_path: Path = DEFAULT_CDK_PATH
557) -> tuple[Finding, ...]:
558 """Require cdk.json's capacity-history watch list to exactly match the catalog."""
559 try:
560 parsed: object = json.loads(cdk_path.read_text(encoding="utf-8"))
561 except (OSError, json.JSONDecodeError) as exc:
562 raise CatalogError(f"cannot read {cdk_path}: {exc}") from exc
563 root = _mapping(parsed, str(cdk_path))
564 context = _mapping(root.get("context"), f"{cdk_path}: context")
565 historical = _mapping(context.get("historical"), f"{cdk_path}: context.historical")
566 watched = _string_list(
567 historical.get("watch_instance_types"),
568 f"{cdk_path}: context.historical.watch_instance_types",
569 )
570 return _watch_list_findings(
571 catalog,
572 watched,
573 location=cdk_path,
574 code_prefix="watch-list",
575 subject="capacity-history watch list",
576 target="context.historical.watch_instance_types",
577 peer="ConfigLoader.get_capacity_history_config()",
578 )
581def _load_config_loader_watch_instance_types(
582 config_loader_path: Path = DEFAULT_CONFIG_LOADER_PATH,
583) -> tuple[str, ...]:
584 """Read ConfigLoader's literal fallback without importing CDK or boto3."""
585 try:
586 source = config_loader_path.read_text(encoding="utf-8")
587 tree = ast.parse(source, filename=str(config_loader_path))
588 except (OSError, SyntaxError) as exc:
589 raise CatalogError(f"cannot parse {config_loader_path}: {exc}") from exc
591 classes = [
592 node for node in tree.body if isinstance(node, ast.ClassDef) and node.name == "ConfigLoader"
593 ]
594 if len(classes) != 1:
595 raise CatalogError(f"{config_loader_path}: expected exactly one ConfigLoader class")
596 methods = [
597 node
598 for node in classes[0].body
599 if isinstance(node, ast.FunctionDef) and node.name == "get_capacity_history_config"
600 ]
601 if len(methods) != 1:
602 raise CatalogError(
603 f"{config_loader_path}: expected exactly one get_capacity_history_config method"
604 )
606 default_configs: list[ast.Dict] = []
607 for statement in methods[0].body:
608 if not isinstance(statement, (ast.AnnAssign, ast.Assign)):
609 continue
610 targets = (
611 (statement.target,)
612 if isinstance(statement, ast.AnnAssign)
613 else tuple(statement.targets)
614 )
615 if (
616 len(targets) == 1
617 and isinstance(targets[0], ast.Name)
618 and targets[0].id == "default_config"
619 and isinstance(statement.value, ast.Dict)
620 ):
621 default_configs.append(statement.value)
622 if len(default_configs) != 1:
623 raise CatalogError(
624 f"{config_loader_path}: expected one literal default_config in "
625 "ConfigLoader.get_capacity_history_config()"
626 )
628 watch_values: list[ast.expr] = []
629 for key, value in zip(default_configs[0].keys, default_configs[0].values, strict=True):
630 if isinstance(key, ast.Constant) and key.value == "watch_instance_types":
631 watch_values.append(value)
632 if len(watch_values) != 1:
633 raise CatalogError(
634 f"{config_loader_path}: expected one default_config watch_instance_types value"
635 )
636 try:
637 parsed_watch_values: object = ast.literal_eval(watch_values[0])
638 except (SyntaxError, TypeError, ValueError) as exc:
639 raise CatalogError(
640 f"{config_loader_path}: default_config watch_instance_types must be a literal list"
641 ) from exc
642 return _string_list(
643 parsed_watch_values,
644 f"{config_loader_path}: ConfigLoader.get_capacity_history_config() "
645 "default_config.watch_instance_types",
646 )
649def validate_config_loader_watch_instance_types(
650 catalog: Catalog,
651 config_loader_path: Path = DEFAULT_CONFIG_LOADER_PATH,
652) -> tuple[Finding, ...]:
653 """Require ConfigLoader's fallback watch list to exactly match the catalog."""
654 watched = _load_config_loader_watch_instance_types(config_loader_path)
655 return _watch_list_findings(
656 catalog,
657 watched,
658 location=config_loader_path,
659 code_prefix="config-loader-watch-list",
660 subject="ConfigLoader capacity-history default watch list",
661 target="ConfigLoader.get_capacity_history_config() default watch_instance_types",
662 peer="cdk.json context.historical.watch_instance_types",
663 )
666@dataclass(frozen=True)
667class InstancePool:
668 """Named set of instance types scored together for Spot Placement Scores.
670 AWS documents that ``GetSpotPlacementScores`` needs at least three instance
671 types (or ``InstanceRequirements``) to return meaningful scores; querying a
672 single type yields artificially depressed values. Members are grouped by
673 accelerator class and per-instance accelerator memory so a workload can
674 plausibly run on any member without change. Pools may overlap; snapshot
675 attribution uses the first pool in ``INSTANCE_POOLS`` order that contains
676 the instance type (see ``pool_for_instance_type``).
677 """
679 name: str
680 members: tuple[str, ...]
681 description: str = ""
683 def __post_init__(self) -> None:
684 _string(self.name, "instance pool name")
685 if not self.members:
686 raise CatalogError(f"instance pool {self.name} has no members")
687 for member in self.members:
688 _family_for_instance_type(_string(member, f"instance pool {self.name} member"))
691#: Spot Placement Score pools over ``historical.watch_instance_types``.
692#:
693#: Definition order is meaningful: where pools overlap, the first pool that
694#: contains a type wins snapshot attribution. Membership follows real per-size
695#: accelerator layouts (for example ``g5.16xlarge`` carries one A10G while
696#: ``g5.12xlarge`` carries four, and ``g5g.16xlarge`` carries two T4Gs, unlike
697#: the single-GPU smaller g5g sizes). Graviton (arm64) types never share a pool
698#: with x86_64 types because images are not interchangeable across
699#: architectures.
700INSTANCE_POOLS: tuple[InstancePool, ...] = (
701 InstancePool(
702 name="single-gpu-t4-16gb",
703 members=(
704 "g4dn.xlarge",
705 "g4dn.2xlarge",
706 "g4dn.4xlarge",
707 "g4dn.8xlarge",
708 "g4dn.16xlarge",
709 ),
710 description="One NVIDIA T4 (16 GB) per instance on the single-GPU x86_64 g4dn sizes.",
711 ),
712 InstancePool(
713 name="single-gpu-arm-16gb",
714 members=(
715 "g5g.xlarge",
716 "g5g.2xlarge",
717 "g5g.4xlarge",
718 "g5g.8xlarge",
719 ),
720 description=(
721 "One NVIDIA T4G (16 GB) per Graviton g5g instance; arm64 images keep "
722 "this pool separate from every x86_64 pool."
723 ),
724 ),
725 InstancePool(
726 name="single-gpu-24gb",
727 members=(
728 "g5.xlarge",
729 "g5.2xlarge",
730 "g5.4xlarge",
731 "g5.8xlarge",
732 "g5.16xlarge",
733 "g6.xlarge",
734 "g6.2xlarge",
735 "g6.4xlarge",
736 "g6.8xlarge",
737 "g6.16xlarge",
738 "gr6.4xlarge",
739 "gr6.8xlarge",
740 ),
741 description=(
742 "One 24 GB mid-range NVIDIA GPU per x86_64 instance: A10G on g5, L4 on "
743 "g6 and the RAM-heavy gr6 sizes."
744 ),
745 ),
746 InstancePool(
747 name="single-gpu-fractional-l4",
748 members=(
749 "g6f.large",
750 "g6f.xlarge",
751 "g6f.2xlarge",
752 "g6f.4xlarge",
753 "gr6f.4xlarge",
754 ),
755 description=(
756 "Fractional shares of one NVIDIA L4 (24 GB) on g6f and gr6f; sized by "
757 "GPU fraction rather than GPU count."
758 ),
759 ),
760 InstancePool(
761 name="single-gpu-48gb",
762 members=(
763 "g6e.xlarge",
764 "g6e.2xlarge",
765 "g6e.4xlarge",
766 "g6e.8xlarge",
767 "g6e.16xlarge",
768 ),
769 description="One NVIDIA L40S (48 GB) per instance on the single-GPU g6e sizes.",
770 ),
771 InstancePool(
772 name="single-gpu-gen7",
773 members=(
774 "g7.2xlarge",
775 "g7.4xlarge",
776 "g7.8xlarge",
777 ),
778 description="One current-generation NVIDIA GPU per instance on the small g7 sizes.",
779 ),
780 InstancePool(
781 name="single-gpu-gen7-48gb",
782 members=(
783 "g7e.2xlarge",
784 "g7e.4xlarge",
785 "g7e.8xlarge",
786 ),
787 description=(
788 "One current-generation 48 GB-class NVIDIA GPU per instance on the small g7e sizes."
789 ),
790 ),
791 InstancePool(
792 name="multi-gpu-4x",
793 members=(
794 "g5.12xlarge",
795 "g5.24xlarge",
796 "g6.12xlarge",
797 "g6.24xlarge",
798 "g6e.12xlarge",
799 "g6e.24xlarge",
800 "g7.12xlarge",
801 "g7.24xlarge",
802 "g7e.12xlarge",
803 "g7e.24xlarge",
804 ),
805 description=(
806 "Four datacenter NVIDIA GPUs per instance: the 12xlarge and 24xlarge "
807 "sizes across g5, g6, g6e, g7, and g7e."
808 ),
809 ),
810 InstancePool(
811 name="multi-gpu-8x",
812 members=(
813 "g5.48xlarge",
814 "g6.48xlarge",
815 "g6e.48xlarge",
816 "g7.48xlarge",
817 "g7e.48xlarge",
818 ),
819 description=(
820 "Eight datacenter NVIDIA GPUs per instance: the 48xlarge sizes across "
821 "g5, g6, g6e, g7, and g7e."
822 ),
823 ),
824 InstancePool(
825 name="hpc-8x-training",
826 members=(
827 "p4d.24xlarge",
828 "p4de.24xlarge",
829 "p5.48xlarge",
830 "p5e.48xlarge",
831 "p5en.48xlarge",
832 "p6-b200.48xlarge",
833 "p6-b300.48xlarge",
834 ),
835 description=(
836 "Eight EFA-attached NVIDIA training GPUs per instance (A100, H100, "
837 "H200, B200, B300). Per-GPU memory spans 40 GB upward across members, "
838 "so confirm model fit before substituting within the pool."
839 ),
840 ),
841 InstancePool(
842 name="inferentia1",
843 members=(
844 "inf1.xlarge",
845 "inf1.2xlarge",
846 "inf1.6xlarge",
847 "inf1.24xlarge",
848 ),
849 description=(
850 "AWS Inferentia (first generation) instances from one to sixteen "
851 "accelerators; interchangeable for Neuron inference that fits one "
852 "accelerator."
853 ),
854 ),
855 InstancePool(
856 name="inferentia2",
857 members=(
858 "inf2.xlarge",
859 "inf2.8xlarge",
860 "inf2.24xlarge",
861 "inf2.48xlarge",
862 ),
863 description=(
864 "AWS Inferentia2 instances from one to twelve accelerators; "
865 "interchangeable for Neuron inference that fits one accelerator."
866 ),
867 ),
868 InstancePool(
869 name="trainium",
870 members=(
871 "trn1.32xlarge",
872 "trn1n.32xlarge",
873 "trn2.48xlarge",
874 ),
875 description="Sixteen-accelerator Trainium (trn1, trn1n) and Trainium2 training instances.",
876 ),
877)
879#: Watch-list types deliberately outside every pool. These still get spot
880#: pricing and Capacity Block observation from the capacity poller, but no
881#: Spot Placement Score: each lacks two interchangeable peers in the watch
882#: list, and padding a pool with unrelated types just to reach the AWS
883#: three-type minimum would make the score describe hardware the workload
884#: cannot actually run on — a worse lie than having no score at all.
885#:
886#: - g4dn.12xlarge (4x T4) and g4dn.metal (8x T4): no other multi-GPU 16 GB types.
887#: - g5g.16xlarge and g5g.metal (2x T4G): a two-member arm64 pool is invalid.
888#: - p3dn.24xlarge: deprecated V100 family; not interchangeable with active pools.
889#: - p5.4xlarge: the only single-GPU H100 size in the list.
890#: - trn1.2xlarge and trn2.3xlarge: single-accelerator Trainium sizes from
891#: different chip generations do not make an interchangeable trio.
892UNPOOLED_INSTANCE_TYPES: tuple[str, ...] = (
893 "g4dn.12xlarge",
894 "g4dn.metal",
895 "g5g.16xlarge",
896 "g5g.metal",
897 "p3dn.24xlarge",
898 "p5.4xlarge",
899 "trn1.2xlarge",
900 "trn2.3xlarge",
901)
903_POOLS_LOCATION = "scripts/accelerator_catalog.py (INSTANCE_POOLS)"
906def pool_for_instance_type(
907 instance_type: str,
908 pools: tuple[InstancePool, ...] = INSTANCE_POOLS,
909) -> InstancePool | None:
910 """Return the first pool in definition order containing ``instance_type``.
912 Pools may overlap, but a capacity snapshot records exactly one pool per
913 instance type, so attribution must be deterministic: definition order in
914 ``INSTANCE_POOLS`` decides. Returns ``None`` for unpooled types.
915 """
916 for pool in pools:
917 if instance_type in pool.members:
918 return pool
919 return None
922def validate_instance_pools(
923 catalog: Catalog,
924 pools: tuple[InstancePool, ...] = INSTANCE_POOLS,
925 unpooled_instance_types: tuple[str, ...] = UNPOOLED_INSTANCE_TYPES,
926) -> tuple[Finding, ...]:
927 """Enforce the Spot Placement Score pool policy against the catalog.
929 Every pool needs at least three distinct members, every member must be a
930 watched catalog type, and every watched type must be either pooled or
931 explicitly declared unpooled, so new catalog entries force a reviewed
932 pooling decision instead of silently going unscored.
933 """
934 findings: list[Finding] = []
935 watched = set(catalog.instance_types)
937 name_counts: dict[str, int] = {}
938 for pool in pools:
939 name_counts[pool.name] = name_counts.get(pool.name, 0) + 1
940 duplicate_names = sorted(name for name, count in name_counts.items() if count > 1)
941 if duplicate_names:
942 findings.append(
943 Finding(
944 code="instance-pool-duplicate-name",
945 title="Instance pools declare duplicate pool names",
946 locations=(_POOLS_LOCATION,),
947 detail=f"Duplicated pool name(s): {', '.join(duplicate_names)}.",
948 recommendation=(
949 "Rename or merge the duplicated pools; snapshot attribution and "
950 "configuration errors must name exactly one pool."
951 ),
952 )
953 )
955 for pool in pools:
956 duplicate_members = sorted(
957 member for member in set(pool.members) if pool.members.count(member) > 1
958 )
959 if duplicate_members:
960 findings.append(
961 Finding(
962 code="instance-pool-duplicate-member",
963 title=f"Pool {pool.name} lists duplicate member types",
964 locations=(_POOLS_LOCATION,),
965 detail=f"Duplicate member(s): {', '.join(duplicate_members)}.",
966 recommendation=(
967 f"Remove the duplicate entries from {pool.name}; duplicates "
968 "must not count toward the three-distinct-type minimum."
969 ),
970 )
971 )
972 distinct_members = set(pool.members)
973 if len(distinct_members) < 3:
974 findings.append(
975 Finding(
976 code="instance-pool-too-small",
977 title=f"Pool {pool.name} has fewer than three distinct member types",
978 locations=(_POOLS_LOCATION,),
979 detail=(
980 f"Pool {pool.name} declares {len(distinct_members)} distinct "
981 "member type(s), but GetSpotPlacementScores needs at least "
982 "three instance types to return meaningful scores."
983 ),
984 recommendation=(
985 f"Add interchangeable types to {pool.name} (comparable "
986 "accelerator class and per-instance accelerator memory) or "
987 "move its members to UNPOOLED_INSTANCE_TYPES with a rationale."
988 ),
989 )
990 )
991 unknown_members = sorted(distinct_members - watched)
992 if unknown_members:
993 findings.append(
994 Finding(
995 code="instance-pool-unknown-member",
996 title=f"Pool {pool.name} contains types outside the watch list",
997 locations=(_POOLS_LOCATION,),
998 detail=(
999 f"Member(s) not in the catalog watch list: {', '.join(unknown_members)}."
1000 ),
1001 recommendation=(
1002 "Pools score only observed capacity: add the type to the "
1003 f"reviewed catalog and watch lists first, or remove it from {pool.name}."
1004 ),
1005 )
1006 )
1008 pooled = {member for pool in pools for member in pool.members}
1009 declared_unpooled = set(unpooled_instance_types)
1010 uncovered = sorted(watched - pooled - declared_unpooled)
1011 if uncovered:
1012 findings.append(
1013 Finding(
1014 code="instance-pool-uncovered-type",
1015 title="Watched instance types have no reviewed pooling decision",
1016 locations=(_POOLS_LOCATION,),
1017 detail=(
1018 f"{len(uncovered)} watched type(s) are neither pooled nor declared "
1019 f"unpooled: {', '.join(uncovered)}."
1020 ),
1021 recommendation=(
1022 "Add each type to an interchangeable pool, or add it to "
1023 "UNPOOLED_INSTANCE_TYPES with a rationale so it visibly skips "
1024 "Spot Placement Scores."
1025 ),
1026 )
1027 )
1028 stale_unpooled = sorted((declared_unpooled & pooled) | (declared_unpooled - watched))
1029 if stale_unpooled:
1030 findings.append(
1031 Finding(
1032 code="instance-pool-stale-unpooled",
1033 title="UNPOOLED_INSTANCE_TYPES is out of date",
1034 locations=(_POOLS_LOCATION,),
1035 detail=(f"Entries are pooled or no longer watched: {', '.join(stale_unpooled)}."),
1036 recommendation=(
1037 "Keep UNPOOLED_INSTANCE_TYPES limited to watched types that no "
1038 "pool contains; remove entries that are pooled or retired."
1039 ),
1040 )
1041 )
1042 return tuple(findings)
1045def validate_repository(
1046 *,
1047 catalog_path: Path = DEFAULT_CATALOG_PATH,
1048 manifests_path: Path = DEFAULT_MANIFESTS_PATH,
1049 cdk_path: Path = DEFAULT_CDK_PATH,
1050 config_loader_path: Path = DEFAULT_CONFIG_LOADER_PATH,
1051 pools: tuple[InstancePool, ...] = INSTANCE_POOLS,
1052 unpooled_instance_types: tuple[str, ...] = UNPOOLED_INSTANCE_TYPES,
1053) -> ValidationReport:
1054 """Run every deterministic repository validation without AWS access."""
1055 catalog = Catalog.load(catalog_path)
1056 nodepools = load_nodepools(manifests_path)
1057 findings = [
1058 *validate_nodepools(catalog, nodepools),
1059 *validate_watch_instance_types(catalog, cdk_path),
1060 *validate_config_loader_watch_instance_types(catalog, config_loader_path),
1061 *validate_instance_pools(catalog, pools, unpooled_instance_types),
1062 ]
1063 return ValidationReport(tuple(sorted(findings, key=Finding.sort_key)))
1066@dataclass(frozen=True)
1067class DiscoveredFamily:
1068 accelerator: Accelerator
1069 architectures: tuple[str, ...]
1072@dataclass(frozen=True)
1073class Discovery:
1074 regions: tuple[str, ...]
1075 instance_types: tuple[str, ...]
1076 families: dict[str, DiscoveredFamily]
1078 def to_mapping(self) -> dict[str, object]:
1079 return {
1080 "regions_checked": list(self.regions),
1081 "instance_types": list(self.instance_types),
1082 "families": {
1083 name: {
1084 "accelerator": family.accelerator,
1085 "architectures": list(family.architectures),
1086 }
1087 for name, family in sorted(self.families.items())
1088 },
1089 }
1092def _detect_accelerator(instance: dict[str, object]) -> Accelerator | None:
1093 gpu_info_value = instance.get("GpuInfo")
1094 if isinstance(gpu_info_value, dict):
1095 gpu_info = _mapping(gpu_info_value, "DescribeInstanceTypes.GpuInfo")
1096 gpus_value = gpu_info.get("Gpus", [])
1097 if isinstance(gpus_value, list):
1098 for gpu_value in gpus_value:
1099 if isinstance(gpu_value, dict):
1100 gpu = _mapping(gpu_value, "DescribeInstanceTypes.GpuInfo.Gpus[]")
1101 manufacturer = gpu.get("Manufacturer")
1102 if isinstance(manufacturer, str) and manufacturer.casefold() == "nvidia":
1103 return "nvidia"
1104 neuron_info_value = instance.get("NeuronInfo")
1105 if isinstance(neuron_info_value, dict):
1106 neuron_info = _mapping(neuron_info_value, "DescribeInstanceTypes.NeuronInfo")
1107 devices = neuron_info.get("NeuronDevices")
1108 if isinstance(devices, list) and devices:
1109 return "neuron"
1110 return None
1113def _instance_architectures(instance: dict[str, object]) -> tuple[str, ...]:
1114 processor = _mapping(instance.get("ProcessorInfo"), "DescribeInstanceTypes.ProcessorInfo")
1115 ec2_architectures = _string_list(
1116 processor.get("SupportedArchitectures"),
1117 "DescribeInstanceTypes.ProcessorInfo.SupportedArchitectures",
1118 )
1119 return tuple(sorted(_EC2_TO_KUBERNETES_ARCH.get(item, item) for item in ec2_architectures))
1122def discover_accelerator_catalog(
1123 *,
1124 profile: str | None = None,
1125 home_region: str = "us-east-1",
1126) -> Discovery:
1127 """Query accelerator types across enabled commercial Regions, sequentially.
1129 EC2 Describe calls use the standard non-mutating request bucket (100-token
1130 burst, 20 requests/second refill at the time this was implemented). Explicit
1131 pagination, no parallel fan-out, and adaptive retries keep this monthly scan
1132 well below that envelope and resilient to account-level concurrent traffic.
1133 """
1134 # Keep AWS SDK imports out of the offline validation path. The minimal
1135 # shell-test job intentionally installs only Python + PyYAML, while the
1136 # monthly online workflow installs boto3/botocore through the project.
1137 import boto3
1138 from botocore.config import Config
1140 session: Any = boto3.Session() if profile is None else boto3.Session(profile_name=profile)
1141 client_config = Config(
1142 connect_timeout=10,
1143 read_timeout=60,
1144 retries={"mode": "adaptive", "total_max_attempts": 10},
1145 user_agent_extra="gco-accelerator-catalog/1",
1146 )
1147 home_client: Any = session.client("ec2", region_name=home_region, config=client_config)
1148 region_response: object = home_client.describe_regions(AllRegions=True)
1149 region_mapping = _mapping(region_response, "DescribeRegions response")
1150 region_values = region_mapping.get("Regions", [])
1151 if not isinstance(region_values, list):
1152 raise CatalogError("DescribeRegions response Regions must be a list")
1153 regions: list[str] = []
1154 for region_value in region_values:
1155 region_record = _mapping(region_value, "DescribeRegions.Regions[]")
1156 status = region_record.get("OptInStatus")
1157 region_name_value = region_record.get("RegionName")
1158 if status in _ENABLED_REGION_STATUSES and isinstance(region_name_value, str):
1159 regions.append(region_name_value)
1160 regions = sorted(set(regions))
1161 if not regions:
1162 raise CatalogError("DescribeRegions returned no enabled commercial Regions")
1164 discovered_types: set[str] = set()
1165 family_accelerators: dict[str, Accelerator] = {}
1166 family_architectures: dict[str, set[str]] = {}
1167 for region_name in regions:
1168 client: Any = session.client("ec2", region_name=region_name, config=client_config)
1169 paginator: Any = client.get_paginator("describe_instance_types")
1170 pages: Any = paginator.paginate(PaginationConfig={"PageSize": 100})
1171 for page_value in pages:
1172 page = _mapping(page_value, f"DescribeInstanceTypes response in {region_name}")
1173 instance_values = page.get("InstanceTypes", [])
1174 if not isinstance(instance_values, list):
1175 raise CatalogError(
1176 f"DescribeInstanceTypes response InstanceTypes must be a list in {region_name}"
1177 )
1178 for instance_value in instance_values:
1179 instance = _mapping(
1180 instance_value,
1181 f"DescribeInstanceTypes.InstanceTypes[] in {region_name}",
1182 )
1183 accelerator = _detect_accelerator(instance)
1184 if accelerator is None:
1185 continue
1186 instance_type = _string(
1187 instance.get("InstanceType"), "DescribeInstanceTypes.InstanceType"
1188 )
1189 family = _family_for_instance_type(instance_type)
1190 existing_accelerator = family_accelerators.setdefault(family, accelerator)
1191 if existing_accelerator != accelerator:
1192 raise CatalogError(
1193 f"EC2 returned conflicting accelerator classes for family {family}"
1194 )
1195 family_architectures.setdefault(family, set()).update(
1196 _instance_architectures(instance)
1197 )
1198 discovered_types.add(instance_type)
1200 families = {
1201 name: DiscoveredFamily(
1202 accelerator=family_accelerators[name],
1203 architectures=tuple(sorted(family_architectures[name])),
1204 )
1205 for name in sorted(family_accelerators)
1206 }
1207 return Discovery(
1208 regions=tuple(regions),
1209 instance_types=tuple(sorted(discovered_types)),
1210 families=families,
1211 )
1214@dataclass(frozen=True)
1215class CatalogDrift:
1216 added: tuple[str, ...]
1217 removed: tuple[str, ...]
1218 metadata_changes: tuple[str, ...]
1219 regions_checked: int
1221 @property
1222 def count(self) -> int:
1223 return len(self.added) + len(self.removed) + len(self.metadata_changes)
1225 @property
1226 def has_drift(self) -> bool:
1227 return self.count > 0
1229 def to_markdown(self) -> str:
1230 status = "ACTION REQUIRED" if self.has_drift else "CURRENT"
1231 lines = [
1232 "## Online EC2 accelerator catalog drift",
1233 "",
1234 f"**Status: {status}.** Checked {self.regions_checked} enabled commercial Regions "
1235 "sequentially with adaptive retries.",
1236 "",
1237 "| Change | Count |",
1238 "|--------|------:|",
1239 f"| New instance types | {len(self.added)} |",
1240 f"| No-longer-returned instance types | {len(self.removed)} |",
1241 f"| Family metadata changes | {len(self.metadata_changes)} |",
1242 ]
1243 if not self.has_drift:
1244 lines.extend(
1245 [
1246 "",
1247 "The checked-in catalog matches the EC2 union for NVIDIA GPU and AWS Neuron "
1248 "instance types.",
1249 ]
1250 )
1251 return "\n".join(lines) + "\n"
1252 if self.added:
1253 lines.extend(["", "### New EC2 instance types", ""])
1254 lines.extend(f"- `{item}`" for item in self.added)
1255 lines.extend(
1256 [
1257 "",
1258 "Review each new family/size, update family lifecycle and generation metadata, "
1259 "then run `python scripts/accelerator_catalog.py refresh`.",
1260 ]
1261 )
1262 if self.removed:
1263 lines.extend(["", "### Instance types no longer returned", ""])
1264 lines.extend(f"- `{item}`" for item in self.removed)
1265 lines.extend(
1266 [
1267 "",
1268 "Confirm this is durable across Regions before removing a type. If its family "
1269 "is retired, mark the family deprecated or end-of-life with replacements.",
1270 ]
1271 )
1272 if self.metadata_changes:
1273 lines.extend(["", "### Family metadata changes", ""])
1274 lines.extend(f"- {item}" for item in self.metadata_changes)
1275 return "\n".join(lines) + "\n"
1277 def summary_mapping(self) -> dict[str, object]:
1278 return {
1279 "status": "drift" if self.has_drift else "current",
1280 "drift_count": self.count,
1281 "added_count": len(self.added),
1282 "removed_count": len(self.removed),
1283 "metadata_change_count": len(self.metadata_changes),
1284 "regions_checked": self.regions_checked,
1285 }
1288def compare_catalog(catalog: Catalog, discovery: Discovery) -> CatalogDrift:
1289 expected = set(catalog.instance_types)
1290 actual = set(discovery.instance_types)
1291 metadata_changes: list[str] = []
1292 for name, discovered in sorted(discovery.families.items()):
1293 policy = catalog.families.get(name)
1294 if policy is None:
1295 metadata_changes.append(
1296 f"New family `{name}` requires reviewed track, generation, and lifecycle policy "
1297 f"(accelerator={discovered.accelerator}, "
1298 f"architectures={list(discovered.architectures)})."
1299 )
1300 continue
1301 if policy.accelerator != discovered.accelerator:
1302 metadata_changes.append(
1303 f"`{name}` accelerator changed: catalog={policy.accelerator}, "
1304 f"EC2={discovered.accelerator}."
1305 )
1306 if policy.architectures != discovered.architectures:
1307 metadata_changes.append(
1308 f"`{name}` architectures changed: catalog={list(policy.architectures)}, "
1309 f"EC2={list(discovered.architectures)}."
1310 )
1311 return CatalogDrift(
1312 added=tuple(sorted(actual - expected)),
1313 removed=tuple(sorted(expected - actual)),
1314 metadata_changes=tuple(metadata_changes),
1315 regions_checked=len(discovery.regions),
1316 )
1319def refresh_catalog(catalog: Catalog, discovery: Discovery, output_path: Path) -> None:
1320 """Write discovered types and a UTC timestamp after policy validation."""
1321 unknown_families = sorted(set(discovery.families) - set(catalog.families))
1322 if unknown_families:
1323 details = ", ".join(unknown_families)
1324 raise CatalogError(
1325 "refusing to refresh with unreviewed families: "
1326 f"{details}. Add explicit track/generation/lifecycle policy first."
1327 )
1328 metadata_drift = compare_catalog(catalog, discovery).metadata_changes
1329 if metadata_drift:
1330 raise CatalogError(
1331 "refusing to refresh while family metadata differs from EC2: "
1332 + " ".join(metadata_drift)
1333 )
1334 refreshed = Catalog(
1335 schema_version=catalog.schema_version,
1336 last_refreshed_at=_current_utc_timestamp(),
1337 source=catalog.source,
1338 families=catalog.families,
1339 instance_types=discovery.instance_types,
1340 )
1341 output_path.write_text(json.dumps(refreshed.to_mapping(), indent=2) + "\n")
1344def _write_or_print(content: str, output: Path | None) -> None:
1345 if output is None:
1346 print(content, end="")
1347 else:
1348 output.write_text(content)
1351def _build_parser() -> argparse.ArgumentParser:
1352 parser = argparse.ArgumentParser(description=__doc__)
1353 subparsers = parser.add_subparsers(dest="command", required=True)
1355 validate = subparsers.add_parser("validate", help="run deterministic offline validation")
1356 validate.add_argument("--catalog", type=Path, default=DEFAULT_CATALOG_PATH)
1357 validate.add_argument("--manifests", type=Path, default=DEFAULT_MANIFESTS_PATH)
1358 validate.add_argument("--cdk-config", type=Path, default=DEFAULT_CDK_PATH)
1359 validate.add_argument("--config-loader", type=Path, default=DEFAULT_CONFIG_LOADER_PATH)
1360 validate.add_argument("--format", choices=("text", "markdown"), default="text")
1361 validate.add_argument("--output", type=Path)
1363 for name, help_text in (
1364 ("capture", "print the live enabled-Region accelerator union"),
1365 ("check-online", "compare the checked-in catalog with live EC2"),
1366 ("refresh", "replace catalog instance types with the reviewed live union"),
1367 ):
1368 command = subparsers.add_parser(name, help=help_text)
1369 command.add_argument("--catalog", type=Path, default=DEFAULT_CATALOG_PATH)
1370 command.add_argument("--profile")
1371 command.add_argument("--home-region", default="us-east-1")
1372 if name == "check-online":
1373 command.add_argument("--report", type=Path)
1374 command.add_argument("--json-summary", action="store_true")
1375 if name == "refresh":
1376 command.add_argument("--output", type=Path, default=DEFAULT_CATALOG_PATH)
1378 return parser
1381def main(argv: list[str] | None = None) -> int:
1382 args = _build_parser().parse_args(argv)
1383 try:
1384 if args.command == "validate":
1385 report = validate_repository(
1386 catalog_path=args.catalog,
1387 manifests_path=args.manifests,
1388 cdk_path=args.cdk_config,
1389 config_loader_path=args.config_loader,
1390 )
1391 content = report.to_markdown() if args.format == "markdown" else report.to_text()
1392 _write_or_print(content, args.output)
1393 return 0 if report.ok else 1
1395 discovery = discover_accelerator_catalog(
1396 profile=args.profile,
1397 home_region=args.home_region,
1398 )
1399 if args.command == "capture":
1400 print(json.dumps(discovery.to_mapping(), indent=2))
1401 return 0
1403 catalog = Catalog.load(args.catalog)
1404 if args.command == "check-online":
1405 drift = compare_catalog(catalog, discovery)
1406 if args.report is not None:
1407 args.report.write_text(drift.to_markdown())
1408 if args.json_summary:
1409 print(json.dumps(drift.summary_mapping(), sort_keys=True))
1410 else:
1411 print(
1412 f"accelerator catalog: status="
1413 f"{'drift' if drift.has_drift else 'current'} "
1414 f"drift_count={drift.count} regions_checked={drift.regions_checked}"
1415 )
1416 return 1 if drift.has_drift else 0
1418 if args.command == "refresh":
1419 refresh_catalog(catalog, discovery, args.output)
1420 print(
1421 f"Refreshed {args.output} with {len(discovery.instance_types)} instance types "
1422 f"from {len(discovery.regions)} enabled Regions."
1423 )
1424 return 0
1426 raise CatalogError(f"unsupported command: {args.command}")
1427 except Exception as exc:
1428 print(f"accelerator catalog error: {exc}", file=sys.stderr)
1429 return 2
1432if __name__ == "__main__":
1433 raise SystemExit(main())