Coverage for cli / capacity / checker.py: 100.00%
813 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""
2Single-region EC2 capacity checker using real AWS signals.
4This is the core capacity intelligence module (~1265 lines). It queries multiple
5AWS APIs to build a comprehensive picture of GPU/accelerator availability in a
6single region. The MultiRegionCapacityChecker in multi_region.py calls this for
7each region in parallel.
9Data Sources:
10 - EC2 GetSpotPlacementScores: likelihood of getting spot capacity (1-10 score)
11 - EC2 DescribeSpotPriceHistory: current and historical spot prices (7-day window)
12 - EC2 DescribeInstanceTypes: full compute characteristics — vCPU/cores/threads,
13 memory, every accelerator class, EFA and network limits, local NVMe and EBS,
14 placement-group support, purchase options, platform capabilities
15 - EC2 DescribeInstanceTypeOfferings: which instance types are available in the region
16 - EC2 DescribeCapacityBlockOfferings: purchasable Capacity Blocks for ML workloads
17 - EC2 PurchaseCapacityBlock: (optional) purchase a Capacity Block by offering ID
19Key Classes:
20 CapacityChecker: Main class. Instantiated with a region and optional GCOConfig.
21 - check_capacity(instance_type) → CapacityEstimate
22 - get_spot_prices(instance_type) → list[SpotPriceInfo]
23 - get_instance_info(instance_type, region) → InstanceTypeInfo
24 - check_capacity_blocks(instance_type, count, duration) → list[dict]
26Output Models (defined in models.py):
27 - CapacityEstimate: spot score, price, trend, on-demand price, instance specs
28 - SpotPriceInfo: AZ, price, timestamp
29 - InstanceTypeInfo: every compute characteristic EC2 reports for a type
31Instance-type characteristics are always resolved live. A checked-in
32GPU_INSTANCE_SPECS table used to short-circuit DescribeInstanceTypes here; it was
33removed because a hand-maintained catalog hid new accelerator families until
34someone edited it, and a stale number fed straight into NodePool sizing and
35capacity scoring. The trade is one EC2 call per lookup, and that lookups now
36require credentials and a region that offers the type.
37"""
39from __future__ import annotations
41import functools
42import importlib.util
43import json
44import logging
45import statistics
46import sys
47from concurrent.futures import ThreadPoolExecutor, as_completed
48from datetime import UTC, datetime, timedelta
49from pathlib import Path
50from typing import Any
52import boto3
53from botocore.config import Config
54from botocore.exceptions import BotoCoreError, ClientError
56from cli.config import GCOConfig, get_config
58from . import blocks
59from .models import (
60 CapacityCheckError,
61 CapacityEstimate,
62 InstanceTypeInfo,
63 SpotPriceInfo,
64 instance_type_info_from_ec2,
65)
67logger = logging.getLogger(__name__)
70class SpotPlacementConfigLimitError(Exception):
71 """GetSpotPlacementScores refused the request with ``MaxConfigLimitExceeded``.
73 The account has asked about too many distinct Spot placement
74 configurations in the rolling 24-hour window. This is a refusal, not a
75 capacity signal: it must never be collapsed into an empty score mapping,
76 which callers could mistake for "this instance type has no capacity
77 data". Callers degrade to non-SPS signals and say the score was
78 unavailable.
79 """
82# Checkout-relative location of the pool catalog module, used when a plain
83# ``import scripts`` fails because the repository root is not on sys.path
84# (the installed ``gco`` console entrypoint). Patchable in tests.
85_POOL_CATALOG_PATH = Path(__file__).resolve().parents[2] / "scripts" / "accelerator_catalog.py"
88@functools.cache
89def _pool_catalog_lookup() -> Any | None:
90 """Return ``scripts.accelerator_catalog.pool_for_instance_type`` or ``None``.
92 The pool catalog lives beside the repository's maintenance tooling, not in
93 the installed ``cli``/``gco`` packages. Two situations need handling:
95 - The ``gco`` console entrypoint does not put the repository root on
96 ``sys.path``, so a plain ``import scripts`` fails even when running from
97 a checkout. The catalog module is self-contained (stdlib + PyYAML), so
98 load it directly from the file that sits two levels above this package.
99 - An installed wheel outside a checkout has no ``scripts/`` directory at
100 all; return ``None`` and let callers treat every type as unpooled
101 rather than issue a known-invalid single-type request.
103 Cached because the catalog is static for the life of the process and the
104 lookup runs on every capacity check.
105 """
106 try:
107 from scripts.accelerator_catalog import pool_for_instance_type
109 return pool_for_instance_type
110 except ImportError:
111 pass
112 catalog_path = _POOL_CATALOG_PATH
113 if not catalog_path.is_file():
114 return None
115 module_name = "_gco_pool_catalog"
116 try:
117 spec = importlib.util.spec_from_file_location(module_name, catalog_path)
118 if spec is None or spec.loader is None:
119 return None
120 module = importlib.util.module_from_spec(spec)
121 # dataclass creation resolves cls.__module__ through sys.modules, so
122 # the module must be registered before its body executes.
123 sys.modules[module_name] = module
124 try:
125 spec.loader.exec_module(module)
126 except BaseException:
127 sys.modules.pop(module_name, None)
128 raise
129 return module.pool_for_instance_type
130 except Exception as exc:
131 logger.debug("failed to load the instance pool catalog from %s: %s", catalog_path, exc)
132 return None
135# Capacity Block offering API error codes that mean "this instance type / region
136# simply doesn't have Capacity Blocks" rather than a real failure — expected and
137# handled quietly. ``Unsupported`` / ``UnsupportedOperation`` / ``InvalidAction``
138# cover regions where the Capacity Block API isn't available at all (e.g.
139# DescribeCapacityBlockOfferings returns ``InvalidAction`` in eu-west-1).
140_CB_EXPECTED_ERROR_CODES = frozenset(
141 {
142 "Unsupported",
143 "UnsupportedOperation",
144 "InvalidAction",
145 "InvalidParameterValue",
146 "InvalidParameterCombination",
147 }
148)
150# Adaptive client-side retry config for DescribeCapacityBlockOfferings. The
151# find_capacity_blocks sweep fans out region x duration probes in parallel, which
152# can trip the API's per-account request-rate limit (RequestLimitExceeded).
153# Adaptive mode adds client-side rate limiting plus exponential backoff, and a
154# higher attempt budget gives each probe room to succeed under throttling rather
155# than silently returning an empty (misleading) result for a throttled region.
156_CB_RETRY_CONFIG = Config(retries={"max_attempts": 10, "mode": "adaptive"})
158# A conservative cap on the parallel fan-out so a wide region x duration sweep
159# can't open an unbounded number of threads / sockets at once.
160_MAX_SEARCH_WORKERS = 12
163def _instance_desc(instance_type: str, gpu_count: int, gpu_type: str, total_gpu_mem: float) -> str:
164 """Build a human-readable instance description."""
165 if gpu_count > 0 and gpu_type:
166 mem_str = f", {total_gpu_mem:.0f}GB" if total_gpu_mem else ""
167 return f"{instance_type} ({gpu_count}x {gpu_type}{mem_str})"
168 return instance_type
171def _offering_fee(offering: dict[str, Any]) -> float:
172 """Sort key: an offering's upfront fee in USD, with missing fees sorting last."""
173 fee = offering.get("upfront_fee_usd")
174 if fee is None:
175 fee = blocks.parse_upfront_fee(offering.get("upfront_fee"))
176 return fee if fee is not None else float("inf")
179class CapacityChecker:
180 """
181 Checks EC2 capacity availability using real AWS capacity signals.
183 Uses:
184 - Spot Placement Score API for spot capacity estimates
185 - EC2 describe-instance-type-offerings for regional availability
186 - Spot price history for pricing trends
187 - On-demand pricing API
188 """
190 def __init__(self, config: GCOConfig | None = None):
191 self.config = config or get_config()
192 self._session = boto3.Session()
193 self._pricing_cache: dict[str, Any] = {}
194 self._offerings_cache: dict[str, set[str]] = {}
195 # Instance descriptions are now always resolved from EC2, and callers
196 # like _enrich_reservation_pricing ask per row. Memoize per
197 # (type, region) so listing 50 reservations of one type stays one call.
198 self._instance_info_cache: dict[tuple[str, str], InstanceTypeInfo | None] = {}
200 def get_instance_info(
201 self, instance_type: str, region: str | None = None
202 ) -> InstanceTypeInfo | None:
203 """Describe an instance type's compute characteristics from EC2.
205 Always resolved live. There is no checked-in specification table behind
206 this any more: the previous 25-entry GPU catalog short-circuited the API,
207 so a new accelerator family stayed invisible until someone hand-edited
208 it, and a wrong number propagated silently into NodePool sizing and
209 capacity scores. Every field now comes from
210 ``ec2:DescribeInstanceTypes`` via :func:`instance_type_info_from_ec2`.
212 ``DescribeInstanceTypes`` is region-scoped and returns a type only where
213 it is offered, so ``region`` matters. It defaults to the configured
214 default region rather than a hardcoded one; a type that region does not
215 offer returns ``None`` even though it may exist elsewhere. Use
216 ``check_instance_available_in_region`` or ``recommend-region`` to find
217 where a type is offered.
219 Friendly aliases are normalized first (``p6-b200`` ->
220 ``p6-b200.48xlarge``) so this agrees with the rest of the capacity
221 surface.
223 Returns ``None`` when the type is unknown, is not offered in the target
224 region, or cannot be described. The distinction is logged; callers
225 needing to tell those cases apart should use
226 :meth:`validate_instance_type`.
227 """
228 canonical, _ = blocks.normalize_instance_type(instance_type)
229 target_region = region or self.config.default_region
231 try:
232 ec2 = self._session.client("ec2", region_name=target_region)
233 response = ec2.describe_instance_types(InstanceTypes=[canonical])
234 except ClientError as e:
235 logger.debug(
236 "Failed to describe instance type %s in %s: %s", canonical, target_region, e
237 )
238 return None
239 except Exception as e:
240 logger.warning(
241 "Unexpected error getting instance info for %s in %s: %s",
242 canonical,
243 target_region,
244 e,
245 )
246 return None
248 records = response.get("InstanceTypes") or []
249 if not records:
250 logger.debug(
251 "Instance type %s is not offered in %s (it may exist in another region)",
252 canonical,
253 target_region,
254 )
255 return None
256 return instance_type_info_from_ec2(records[0], region=target_region)
258 def validate_instance_type(self, instance_type: str) -> dict[str, Any]:
259 """Classify an instance type as valid / invalid, with friendly normalization.
261 The offering APIs collapse "unknown instance type" and "valid type with
262 zero offerings" into the same empty result. This method separates them so
263 callers can tell a typo from genuine unavailability:
265 * EC2 ``DescribeInstanceTypes`` is consulted for every type; an
266 ``InvalidInstanceType`` marks it invalid, an empty result means the
267 type is not offered in the region consulted, and a transient API error
268 leaves it valid-but-unverified with a note.
269 * Friendly aliases are expanded (``p6-b200`` -> ``p6-b200.48xlarge``,
270 ``p6-b300`` -> ``p6-b300.48xlarge``) and UltraServer-only families
271 (the Grace-Blackwell ``gb200``/``gb300`` superchips, sold only as
272 ``P6e-GB`` UltraServers) are flagged invalid-for-``InstanceType`` with
273 guidance toward the UltraServer search flow.
275 Returns a dict: ``requested``, ``instance_type`` (canonical), ``valid``,
276 ``known``, ``note``, ``gpu_count``.
278 ``known`` means "EC2 returned a description for this type in the region
279 consulted". It used to mean "present in our hardcoded GPU catalog"; that
280 catalog is gone, so the flag now reflects live confirmation. It is
281 reported for information (as ``known_instance_type``) and never gates
282 anything.
284 Note the three-way outcome, which callers do depend on: a bogus type is
285 ``valid=False``; a real type the consulted region does not offer stays
286 ``valid=True`` with an explanatory ``note`` and ``known=False``; and a
287 confirmed type is ``valid=True, known=True``. Collapsing the middle case
288 into ``valid=False`` would reject a p5 in a region that merely lacks it,
289 which is what the removed catalog used to paper over.
290 """
291 canonical, note = blocks.normalize_instance_type(instance_type)
292 result: dict[str, Any] = {
293 "requested": instance_type,
294 "instance_type": canonical,
295 "valid": True,
296 "known": False,
297 "note": note,
298 "gpu_count": None,
299 }
301 # UltraServer-only families are real accelerators but not standalone EC2
302 # instance types, so InstanceType-based searches can never resolve them.
303 if (instance_type or "").strip().lower() in blocks.NON_STANDALONE_INSTANCE_NOTES:
304 result["valid"] = False
305 return result
307 # Ask EC2, separating "invalid type" from "not offered here" from
308 # "API error". An empty result is the middle case: the type is real but
309 # this region does not offer it, so it must not be reported invalid.
310 region = self.config.default_region
311 try:
312 ec2 = self._session.client("ec2", region_name=region)
313 response = ec2.describe_instance_types(InstanceTypes=[canonical])
314 types = response.get("InstanceTypes", [])
315 if not types:
316 result["note"] = note or (
317 f"Instance type is not offered in {region}; it may exist in another region."
318 )
319 return result
320 result["known"] = True
321 gpu_info = types[0].get("GpuInfo") or {}
322 # Sum across models rather than reading Gpus[0] so a heterogeneous
323 # accelerator list is not undercounted.
324 result["gpu_count"] = sum(
325 int(gpu.get("Count") or 0) for gpu in (gpu_info.get("Gpus") or [])
326 )
327 except ClientError as e:
328 code = e.response.get("Error", {}).get("Code", "")
329 if code in ("InvalidInstanceType", "InvalidParameterValue"):
330 result["valid"] = False
331 else:
332 logger.debug("Could not verify instance type %s: %s", canonical, e)
333 result["note"] = note or f"Could not verify instance type via EC2 ({code or e})."
334 except Exception as e:
335 logger.debug("Unexpected error validating instance type %s: %s", canonical, e)
336 result["note"] = note or f"Could not verify instance type: {e}"
337 return result
339 def check_instance_available_in_region(self, instance_type: str, region: str) -> bool:
340 """Check whether an instance type is offered in a region.
342 Returns ``True``/``False`` only from a *successful* offerings lookup, so a
343 ``False`` genuinely means "not offered in this region". If the underlying
344 DescribeInstanceTypeOfferings call fails (throttling, expired/invalid
345 credentials, denied permissions, region not opted in) this raises
346 :class:`CapacityCheckError` instead of silently returning ``False`` — a
347 failed check must not be reported to the user as "not available".
348 """
349 cache_key = f"{region}"
350 if cache_key not in self._offerings_cache:
351 try:
352 ec2 = self._session.client("ec2", region_name=region)
353 paginator = ec2.get_paginator("describe_instance_type_offerings")
354 offerings = set()
355 for page in paginator.paginate(LocationType="region"):
356 for offering in page["InstanceTypeOfferings"]:
357 offerings.add(offering["InstanceType"])
358 self._offerings_cache[cache_key] = offerings
359 except (ClientError, BotoCoreError) as e:
360 logger.warning("Failed to check instance offerings in %s: %s", region, e)
361 raise CapacityCheckError(
362 f"Could not check instance availability in {region}: {e}"
363 ) from e
365 return instance_type in self._offerings_cache[cache_key]
367 def get_availability_zones(self, region: str) -> list[str]:
368 """Get availability zones for a region."""
369 try:
370 ec2 = self._session.client("ec2", region_name=region)
371 response = ec2.describe_availability_zones(
372 Filters=[{"Name": "state", "Values": ["available"]}]
373 )
374 return [az["ZoneName"] for az in response["AvailabilityZones"]]
375 except ClientError as e:
376 logger.warning("Failed to get availability zones for %s: %s", region, e)
377 return []
378 except Exception as e:
379 logger.warning("Unexpected error getting AZs for %s: %s", region, e)
380 return []
382 def get_az_coverage(self, instance_type: str, region: str) -> float | None:
383 """Get the fraction of AZs in a region that offer this instance type.
385 Returns a value between 0.0 and 1.0, or None if we can't determine it.
386 Constrained instances are often available in fewer AZs.
387 """
388 try:
389 ec2 = self._session.client("ec2", region_name=region)
390 total_azs = self.get_availability_zones(region)
391 if not total_azs:
392 return None
394 paginator = ec2.get_paginator("describe_instance_type_offerings")
395 offering_azs = set()
396 for page in paginator.paginate(
397 LocationType="availability-zone",
398 Filters=[{"Name": "instance-type", "Values": [instance_type]}],
399 ):
400 for offering in page["InstanceTypeOfferings"]:
401 offering_azs.add(offering["Location"])
403 return len(offering_azs) / len(total_azs) if total_azs else None
404 except Exception as e:
405 logger.warning("Failed to get AZ coverage for %s in %s: %s", instance_type, region, e)
406 return None
408 @staticmethod
409 def instance_pool_for(instance_type: str) -> tuple[str, tuple[str, ...]] | None:
410 """Return (pool name, member types) for the first catalog pool containing the type.
412 AWS documents that ``GetSpotPlacementScores`` needs at least three
413 instance types to return meaningful scores, so every SPS request this
414 checker makes is scoped to a reviewed instance pool from
415 ``scripts/accelerator_catalog.py`` (the same catalog the capacity
416 poller uses). Returns ``None`` when no pool contains the type — and,
417 as a deliberate degradation, when the pool catalog is unavailable (an
418 installed wheel running outside a repository checkout ships without
419 ``scripts/``): an honest "no score obtained" beats a known-invalid
420 single-type request in both cases.
421 """
422 lookup = _pool_catalog_lookup()
423 if lookup is None:
424 logger.debug(
425 "instance pool catalog (scripts/accelerator_catalog.py) is unavailable; "
426 "treating %s as unpooled and skipping Spot Placement Scores",
427 instance_type,
428 )
429 return None
430 pool = lookup(instance_type)
431 if pool is None:
432 return None
433 return (pool.name, pool.members)
435 def get_spot_placement_score(
436 self, instance_type: str, region: str, target_capacity: int = 1
437 ) -> dict[str, int]:
438 """
439 Get the Spot Placement Score for the instance pool containing a type.
441 The Spot Placement Score (1-10) indicates the likelihood of getting
442 spot capacity. Higher scores mean better availability. The request
443 carries the full member list of the first catalog pool containing
444 ``instance_type`` — never the single type, which AWS documents as
445 returning misleadingly low scores — so the score describes the pool
446 at ``target_capacity``, not one instance type.
448 Returns:
449 Dict mapping AZ to score (1-10). Empty when no pool contains the
450 type (no request is made) or the API reports the type/region as
451 unsupported.
453 Raises:
454 SpotPlacementConfigLimitError: the account has queried too many
455 distinct SPS configurations in the rolling window
456 (``MaxConfigLimitExceeded``). Surfaced as a distinct condition
457 because "refused" must never be mistaken for "no capacity
458 data exists".
459 """
460 pool = self.instance_pool_for(instance_type)
461 if pool is None:
462 logger.info(
463 "no instance pool contains %s; skipping the Spot Placement Score request "
464 "(a single-type request returns misleadingly low scores)",
465 instance_type,
466 )
467 return {}
468 pool_name, members = pool
469 try:
470 ec2 = self._session.client("ec2", region_name=region)
472 response = ec2.get_spot_placement_scores(
473 InstanceTypes=list(members),
474 TargetCapacity=target_capacity,
475 TargetCapacityUnitType="units",
476 RegionNames=[region],
477 SingleAvailabilityZone=False,
478 )
480 scores = {}
481 for recommendation in response.get("SpotPlacementScores", []):
482 # Regional score
483 if "AvailabilityZoneId" not in recommendation:
484 scores["regional"] = recommendation.get("Score", 0)
485 else:
486 az_id = recommendation["AvailabilityZoneId"]
487 scores[az_id] = recommendation.get("Score", 0)
489 return scores
491 except ClientError as e:
492 error_code = e.response.get("Error", {}).get("Code", "")
493 if error_code == "MaxConfigLimitExceeded":
494 logger.warning(
495 "Spot Placement Score request refused (MaxConfigLimitExceeded) for "
496 "pool %s (%s) in %s: the account has queried too many distinct SPS "
497 "configurations in the rolling 24h window",
498 pool_name,
499 instance_type,
500 region,
501 )
502 raise SpotPlacementConfigLimitError(
503 f"Spot Placement Score request for pool {pool_name} in {region} was "
504 "refused: the account reached its Spot placement configuration limit "
505 "(MaxConfigLimitExceeded)"
506 ) from e
507 if error_code in ("InvalidParameterValue", "UnsupportedOperation"):
508 return {}
509 raise
510 except Exception as e:
511 logger.warning(
512 "Failed to get spot placement scores for %s in %s: %s", instance_type, region, e
513 )
514 return {}
516 def get_spot_price_history(
517 self, instance_type: str, region: str, days: int = 7
518 ) -> list[SpotPriceInfo]:
519 """Get spot price history for an instance type."""
520 ec2 = self._session.client("ec2", region_name=region)
522 end_time = datetime.now(UTC)
523 start_time = end_time - timedelta(days=days)
525 try:
526 response = ec2.describe_spot_price_history(
527 InstanceTypes=[instance_type],
528 ProductDescriptions=["Linux/UNIX"],
529 StartTime=start_time,
530 EndTime=end_time,
531 )
533 # Group by availability zone
534 az_prices: dict[str, list[float]] = {}
535 for item in response["SpotPriceHistory"]:
536 az = item["AvailabilityZone"]
537 price = float(item["SpotPrice"])
538 if az not in az_prices:
539 az_prices[az] = []
540 az_prices[az].append(price)
542 results = []
543 for az, prices in az_prices.items():
544 current = prices[0]
545 avg = statistics.mean(prices)
546 min_price = min(prices)
547 max_price = max(prices)
549 if avg > 0:
550 std_dev = statistics.stdev(prices) if len(prices) > 1 else 0
551 cv = std_dev / avg
552 stability = max(0, 1 - cv)
553 else:
554 stability = 0
556 results.append(
557 SpotPriceInfo(
558 instance_type=instance_type,
559 availability_zone=az,
560 current_price=current,
561 avg_price_7d=avg,
562 min_price_7d=min_price,
563 max_price_7d=max_price,
564 price_stability=stability,
565 )
566 )
568 return results
570 except ClientError as e:
571 if "InvalidParameterValue" in str(e):
572 return []
573 raise
575 def get_on_demand_price(self, instance_type: str, region: str) -> float | None:
576 """Get on-demand price for an instance type."""
577 cache_key = f"{instance_type}:{region}"
578 if cache_key in self._pricing_cache:
579 cached_value = self._pricing_cache[cache_key]
580 return float(cached_value) if cached_value is not None else None
582 try:
583 pricing = self._session.client("pricing", region_name="us-east-1")
585 # Filter on the regionCode product attribute — the AWS region
586 # identifier itself — never on the human-readable location name.
587 # The former ten-entry region-name map silently returned None
588 # for every region outside it, and Price List location strings
589 # are not derivable from any published source (eu-north-1 is
590 # "EU (Stockholm)", not "Europe (Stockholm)"), so a name map is
591 # inherently fragile. regionCode removes the class of bug:
592 # verified to return identical prices for all ten formerly
593 # mapped regions and correct prices for previously failing ones.
594 response = pricing.get_products(
595 ServiceCode="AmazonEC2",
596 Filters=[
597 {"Type": "TERM_MATCH", "Field": "instanceType", "Value": instance_type},
598 {"Type": "TERM_MATCH", "Field": "regionCode", "Value": region},
599 {"Type": "TERM_MATCH", "Field": "operatingSystem", "Value": "Linux"},
600 {"Type": "TERM_MATCH", "Field": "tenancy", "Value": "Shared"},
601 {"Type": "TERM_MATCH", "Field": "preInstalledSw", "Value": "NA"},
602 {"Type": "TERM_MATCH", "Field": "capacitystatus", "Value": "Used"},
603 ],
604 MaxResults=1,
605 )
607 if response["PriceList"]:
608 price_data = json.loads(response["PriceList"][0])
609 terms = price_data.get("terms", {}).get("OnDemand", {})
610 for term in terms.values():
611 for price_dim in term.get("priceDimensions", {}).values():
612 price = float(price_dim["pricePerUnit"]["USD"])
613 self._pricing_cache[cache_key] = price
614 return price
616 except ClientError as e:
617 logger.warning(
618 "Failed to get on-demand price for %s in %s: %s", instance_type, region, e
619 )
620 except Exception as e:
621 logger.warning(
622 "Unexpected error getting pricing for %s in %s: %s", instance_type, region, e
623 )
625 return None
627 def estimate_capacity(
628 self, instance_type: str, region: str, capacity_type: str = "both"
629 ) -> list[CapacityEstimate]:
630 """
631 Estimate capacity availability using real AWS signals.
633 Args:
634 instance_type: EC2 instance type
635 region: AWS region
636 capacity_type: "spot", "on-demand", or "both"
638 Returns:
639 List of CapacityEstimate objects
640 """
641 estimates = []
643 # Check if instance type is available in region
644 if not self.check_instance_available_in_region(instance_type, region):
645 return [
646 CapacityEstimate(
647 instance_type=instance_type,
648 region=region,
649 availability_zone=None,
650 capacity_type="both",
651 availability="unavailable",
652 confidence=1.0,
653 recommendation=f"{instance_type} is not available in {region}",
654 details={"reason": "Instance type not offered in region"},
655 )
656 ]
658 instance_info = self.get_instance_info(instance_type)
660 if capacity_type in ("spot", "both"):
661 spot_estimates = self._estimate_spot_capacity(instance_type, region, instance_info)
662 estimates.extend(spot_estimates)
664 if capacity_type in ("on-demand", "both"):
665 # Pass spot placement scores to on-demand estimator as a scarcity signal
666 spot_scores = {}
667 if capacity_type == "on-demand":
668 try:
669 spot_scores = self.get_spot_placement_score(instance_type, region)
670 except SpotPlacementConfigLimitError:
671 # The refusal is already logged distinctly; the on-demand
672 # estimate degrades to its other scarcity signals.
673 spot_scores = {}
674 # If we already fetched spot estimates, extract the scores from them
675 if spot_estimates := [e for e in estimates if e.capacity_type == "spot"]:
676 spot_scores = {
677 e.availability_zone or "unknown": e.details.get("spot_placement_score", 0)
678 for e in spot_estimates
679 if e.details.get("spot_placement_score") is not None
680 }
681 # Also gather spot price data for price-ratio signal
682 spot_prices = self.get_spot_price_history(instance_type, region)
683 od_estimate = self._estimate_on_demand_capacity(
684 instance_type, region, instance_info, spot_scores, spot_prices
685 )
686 if od_estimate:
687 estimates.append(od_estimate)
689 return estimates
691 def _estimate_spot_capacity(
692 self, instance_type: str, region: str, instance_info: InstanceTypeInfo | None
693 ) -> list[CapacityEstimate]:
694 """Estimate spot capacity using pooled Spot Placement Scores and price history."""
695 estimates = []
697 # Spot Placement Score (primary signal), requested for the instance
698 # pool containing the type. Three distinct non-score outcomes exist
699 # and must stay distinguishable: the type is in no pool (no request
700 # made), the account's configuration limit refused the request, or
701 # the API had nothing to say. None of them may borrow the confidence
702 # reserved for a real score.
703 pool = self.instance_pool_for(instance_type)
704 sps_target_capacity = 1
705 placement_scores: dict[str, int] = {}
706 if pool is None:
707 sps_note = (
708 f"no placement score was obtained: {instance_type} belongs to no "
709 "instance pool, and a single-type request returns misleadingly "
710 "low scores"
711 )
712 else:
713 sps_note = "no placement score was obtained"
714 try:
715 placement_scores = self.get_spot_placement_score(
716 instance_type, region, sps_target_capacity
717 )
718 except SpotPlacementConfigLimitError:
719 sps_note = (
720 "no placement score was obtained: the account reached its Spot "
721 "placement configuration limit (MaxConfigLimitExceeded)"
722 )
724 # Get spot prices for pricing info
725 spot_prices = self.get_spot_price_history(instance_type, region)
726 price_by_az = {sp.availability_zone: sp for sp in spot_prices}
728 on_demand_price = self.get_on_demand_price(instance_type, region)
730 # Get AZs in the region
731 azs = self.get_availability_zones(region)
733 if placement_scores:
734 # Use Spot Placement Score as primary signal
735 pool_name = pool[0] if pool is not None else "unknown"
736 regional_score = placement_scores.get("regional", 0)
738 for az in azs:
739 # Try to get AZ-specific score, fall back to regional
740 az_id = az # Note: might need to map zone name to zone ID
741 score = placement_scores.get(az_id, regional_score)
743 # Convert score (1-10) to availability
744 if score >= 8:
745 availability = "high"
746 recommendation = "Excellent spot availability"
747 elif score >= 5:
748 availability = "medium"
749 recommendation = "Good spot availability, some interruption risk"
750 elif score >= 3:
751 availability = "low"
752 recommendation = "Limited spot capacity, consider alternatives"
753 else:
754 availability = "low"
755 recommendation = "Very limited spot capacity"
757 spot_info = price_by_az.get(az)
758 price = spot_info.current_price if spot_info else None
760 details: dict[str, Any] = {
761 "spot_placement_score": score,
762 # The score describes the pool's fleet at the stated
763 # target capacity, not this one instance type.
764 "score_interpretation": (
765 f"{score}/10 for instance pool {pool_name} "
766 f"at target capacity {sps_target_capacity}"
767 ),
768 "spot_pool": pool_name,
769 "sps_target_capacity": sps_target_capacity,
770 }
772 if spot_info:
773 details["current_price"] = spot_info.current_price
774 details["avg_price_7d"] = spot_info.avg_price_7d
775 details["price_stability"] = f"{spot_info.price_stability:.2f}"
777 if on_demand_price and price:
778 savings = (1 - price / on_demand_price) * 100
779 details["savings_vs_on_demand"] = f"{savings:.1f}%"
780 details["on_demand_price"] = on_demand_price
782 estimates.append(
783 CapacityEstimate(
784 instance_type=instance_type,
785 region=region,
786 availability_zone=az,
787 capacity_type="spot",
788 availability=availability,
789 confidence=0.85, # Spot Placement Score is reliable
790 price_per_hour=price,
791 recommendation=recommendation,
792 details=details,
793 )
794 )
796 elif spot_prices:
797 # Fall back to price-based estimation if no placement score
798 for spot_info in spot_prices:
799 # Use price stability as a proxy (less reliable)
800 if spot_info.price_stability > 0.8:
801 availability = "medium"
802 recommendation = "Spot prices stable, likely available"
803 elif spot_info.price_stability > 0.5:
804 availability = "low"
805 recommendation = "Spot prices volatile, capacity uncertain"
806 else:
807 availability = "low"
808 recommendation = "High price volatility, limited capacity likely"
810 details = {
811 "current_price": spot_info.current_price,
812 "avg_price_7d": spot_info.avg_price_7d,
813 "price_stability": f"{spot_info.price_stability:.2f}",
814 "note": f"Estimate based on price history; {sps_note}",
815 }
817 if on_demand_price:
818 savings = (1 - spot_info.current_price / on_demand_price) * 100
819 details["savings_vs_on_demand"] = f"{savings:.1f}%"
821 estimates.append(
822 CapacityEstimate(
823 instance_type=instance_type,
824 region=region,
825 availability_zone=spot_info.availability_zone,
826 capacity_type="spot",
827 availability=availability,
828 confidence=0.5, # Lower confidence without placement score
829 price_per_hour=spot_info.current_price,
830 recommendation=f"{recommendation} ({sps_note})",
831 details=details,
832 )
833 )
835 if not estimates:
836 estimates.append(
837 CapacityEstimate(
838 instance_type=instance_type,
839 region=region,
840 availability_zone=None,
841 capacity_type="spot",
842 availability="unknown",
843 confidence=0.1,
844 recommendation=(
845 f"No spot data available for {instance_type} in {region}; {sps_note}"
846 ),
847 details={"reason": f"No spot price history, and {sps_note}"},
848 )
849 )
851 return estimates
853 def _estimate_on_demand_capacity(
854 self,
855 instance_type: str,
856 region: str,
857 instance_info: InstanceTypeInfo | None,
858 spot_placement_scores: dict[str, int] | None = None,
859 spot_prices: list[SpotPriceInfo] | None = None,
860 ) -> CapacityEstimate | None:
861 """Estimate on-demand capacity using live signals for ALL instance types.
863 Uses spot placement scores, instance size (vCPUs, memory, GPUs),
864 pricing, and spot-to-on-demand price ratios as universal scarcity
865 signals — no hardcoded instance families or GPU type lists.
866 """
867 on_demand_price = self.get_on_demand_price(instance_type, region)
869 is_offered = self.check_instance_available_in_region(instance_type, region)
871 if not is_offered:
872 return CapacityEstimate(
873 instance_type=instance_type,
874 region=region,
875 availability_zone=None,
876 capacity_type="on-demand",
877 availability="unavailable",
878 confidence=1.0,
879 recommendation=f"{instance_type} is not offered in {region}",
880 details={"reason": "Instance type not offered in region"},
881 )
883 # Fetch spot placement scores if not provided (on-demand only mode)
884 if spot_placement_scores is None:
885 try:
886 spot_placement_scores = self.get_spot_placement_score(instance_type, region)
887 except SpotPlacementConfigLimitError:
888 # Logged distinctly at the source; degrade to the other
889 # universal scarcity signals rather than failing the estimate.
890 spot_placement_scores = {}
892 if spot_prices is None:
893 spot_prices = self.get_spot_price_history(instance_type, region)
895 az_coverage = self.get_az_coverage(instance_type, region)
897 availability, confidence, recommendation = self._assess_on_demand_availability(
898 instance_type,
899 instance_info,
900 on_demand_price,
901 spot_placement_scores,
902 spot_prices,
903 az_coverage,
904 )
906 if on_demand_price:
907 recommendation += f" Price: ${on_demand_price:.4f}/hr."
908 else:
909 confidence -= 0.1
910 recommendation += " Pricing data unavailable."
912 details: dict[str, Any] = {
913 "price_per_hour": on_demand_price,
914 "is_gpu": instance_info.is_gpu if instance_info else False,
915 }
916 if spot_placement_scores:
917 scores = [s for s in spot_placement_scores.values() if s > 0]
918 if scores:
919 details["avg_spot_placement_score"] = round(sum(scores) / len(scores), 1)
921 return CapacityEstimate(
922 instance_type=instance_type,
923 region=region,
924 availability_zone=None,
925 capacity_type="on-demand",
926 availability=availability,
927 confidence=confidence,
928 price_per_hour=on_demand_price,
929 recommendation=recommendation,
930 details=details,
931 )
933 @staticmethod
934 def _assess_on_demand_availability(
935 instance_type: str,
936 instance_info: InstanceTypeInfo | None,
937 on_demand_price: float | None,
938 spot_placement_scores: dict[str, int] | None = None,
939 spot_prices: list[SpotPriceInfo] | None = None,
940 az_coverage: float | None = None,
941 ) -> tuple[str, float, str]:
942 """Assess on-demand availability using only live market signals.
944 Five live signals, zero hardcoded instance families:
945 1. Spot placement score — AWS's own capacity assessment (1-10)
946 2. Spot-to-on-demand price ratio — when spot approaches on-demand price,
947 the spot market has very little excess capacity
948 3. Spot price volatility — unstable prices reflect capacity fluctuations
949 4. AZ coverage — fraction of AZs that offer this instance type;
950 constrained instances are often available in fewer AZs
951 5. Spot price availability — how many AZs have spot price data;
952 missing price data in some AZs suggests limited capacity there
954 Confidence scales with the number of live signals available.
955 When no signals exist, returns "unknown" rather than guessing.
957 Returns:
958 Tuple of (availability, confidence, recommendation)
959 """
960 price = on_demand_price or 0
961 gpu_count = instance_info.gpu_count if instance_info else 0
962 gpu_type = (instance_info.gpu_type or "") if instance_info else ""
963 total_gpu_mem = instance_info.gpu_memory_gib if instance_info else 0
965 # --- Signal 1: Spot placement score ---
966 avg_spot_score = 0.0
967 has_spot_score = False
968 if spot_placement_scores:
969 scores = [s for s in spot_placement_scores.values() if s > 0]
970 if scores:
971 avg_spot_score = sum(scores) / len(scores)
972 has_spot_score = True
974 # --- Signal 2 & 3: Spot price ratio and volatility ---
975 avg_spot_ratio = 0.0
976 avg_stability = 1.0
977 has_price_signal = False
978 if spot_prices and price > 0:
979 ratios = [sp.current_price / price for sp in spot_prices if sp.current_price > 0]
980 if ratios:
981 avg_spot_ratio = sum(ratios) / len(ratios)
982 has_price_signal = True
983 stabilities = [sp.price_stability for sp in spot_prices]
984 avg_stability = sum(stabilities) / len(stabilities)
986 # --- Signal 4: AZ coverage (passed in from caller) ---
987 has_az_signal = az_coverage is not None
989 # --- Combine live signals into scarcity (0.0 - 1.0) ---
990 scarcity = 0.0
991 signal_count = 0
993 if has_spot_score:
994 signal_count += 1
995 if avg_spot_score <= 2:
996 scarcity += 0.5
997 elif avg_spot_score <= 4:
998 scarcity += 0.3
999 elif avg_spot_score <= 6:
1000 scarcity += 0.15
1002 if has_price_signal:
1003 signal_count += 1
1004 # Spot price near on-demand = spot market has minimal excess capacity
1005 if avg_spot_ratio >= 0.9:
1006 scarcity += 0.3
1007 elif avg_spot_ratio >= 0.7:
1008 scarcity += 0.15
1009 elif avg_spot_ratio >= 0.5:
1010 scarcity += 0.05
1012 # Price instability = capacity fluctuations
1013 if avg_stability < 0.6:
1014 scarcity += 0.1
1015 elif avg_stability < 0.8:
1016 scarcity += 0.05
1018 if has_az_signal and az_coverage is not None:
1019 signal_count += 1
1020 # Available in fewer than half the AZs = constrained
1021 if az_coverage <= 0.3:
1022 scarcity += 0.2
1023 elif az_coverage <= 0.5:
1024 scarcity += 0.1
1026 # --- Confidence scales with signal count ---
1027 confidence = min(0.5 + (signal_count * 0.12), 0.9)
1029 # --- Map scarcity to availability ---
1030 desc = _instance_desc(instance_type, gpu_count, gpu_type, total_gpu_mem)
1032 if signal_count == 0:
1033 # No live data — be honest about it
1034 return (
1035 "unknown",
1036 0.3,
1037 f"No live capacity signals available for {instance_type}."
1038 " Unable to assess on-demand availability.",
1039 )
1041 if scarcity >= 0.6:
1042 return (
1043 "low",
1044 confidence,
1045 f"On-demand {desc} is extremely scarce based on live capacity signals."
1046 " Capacity reservations or Capacity Blocks are strongly recommended.",
1047 )
1049 if scarcity >= 0.35:
1050 return (
1051 "low",
1052 confidence,
1053 f"On-demand {desc} has limited availability based on live capacity signals."
1054 " Consider capacity reservations.",
1055 )
1057 if scarcity >= 0.15:
1058 return (
1059 "medium",
1060 confidence,
1061 f"On-demand {instance_type} may have constrained availability"
1062 " based on current market conditions.",
1063 )
1065 return (
1066 "high",
1067 confidence,
1068 f"On-demand capacity likely available for {instance_type}"
1069 " based on live capacity signals.",
1070 )
1072 def recommend_capacity_type(
1073 self, instance_type: str, region: str, fault_tolerance: str = "medium"
1074 ) -> tuple[str, str]:
1075 """
1076 Recommend spot vs on-demand based on actual capacity and requirements.
1078 Args:
1079 instance_type: EC2 instance type
1080 region: AWS region
1081 fault_tolerance: "high" (can handle interruptions),
1082 "medium" (some tolerance),
1083 "low" (needs stability)
1085 Returns:
1086 Tuple of (recommended_capacity_type, explanation)
1087 """
1088 estimates = self.estimate_capacity(instance_type, region, "both")
1090 spot_estimates = [e for e in estimates if e.capacity_type == "spot"]
1091 od_estimates = [e for e in estimates if e.capacity_type == "on-demand"]
1093 # Check for unavailable
1094 if any(e.availability == "unavailable" for e in estimates):
1095 return "unavailable", f"{instance_type} is not available in {region}"
1097 # Get best spot option (highest availability)
1098 best_spot = None
1099 if spot_estimates:
1100 available_spots = [e for e in spot_estimates if e.availability != "unknown"]
1101 if available_spots:
1102 # Sort by availability (high > medium > low) then by price
1103 avail_order = {"high": 0, "medium": 1, "low": 2}
1104 best_spot = min(
1105 available_spots,
1106 key=lambda x: (avail_order.get(x.availability, 3), x.price_per_hour or 999),
1107 )
1109 od_estimate = od_estimates[0] if od_estimates else None
1111 # Decision logic based on fault tolerance and actual availability
1112 if fault_tolerance == "low":
1113 if od_estimate and od_estimate.availability in ("high", "medium"):
1114 return "on-demand", "Low fault tolerance requires stable on-demand capacity"
1115 return (
1116 "on-demand",
1117 "On-demand recommended but capacity may be limited; consider capacity reservation",
1118 )
1120 if best_spot:
1121 if best_spot.availability == "high":
1122 savings = ""
1123 if best_spot.price_per_hour and od_estimate and od_estimate.price_per_hour:
1124 pct = (1 - best_spot.price_per_hour / od_estimate.price_per_hour) * 100
1125 savings = f" (save ~{pct:.0f}%)"
1126 return "spot", f"High spot availability (score-based){savings}"
1128 if best_spot.availability == "medium":
1129 if fault_tolerance == "high":
1130 return "spot", "Medium spot availability acceptable with high fault tolerance"
1131 return (
1132 "on-demand",
1133 "Spot availability is medium; on-demand recommended for reliability",
1134 )
1136 # ``unknown`` values were filtered above; high and medium already
1137 # returned, so low is the only remaining availability in the model.
1138 if fault_tolerance == "high":
1139 return (
1140 "spot",
1141 "Low spot availability but acceptable with high fault tolerance",
1142 )
1143 return "on-demand", "Spot capacity is limited; on-demand recommended"
1145 # Default to on-demand
1146 return "on-demand", "On-demand recommended (spot availability unknown or limited)"
1148 # -------------------------------------------------------------------------
1149 # Capacity Reservations (ODCRs) and Capacity Blocks for ML
1150 # -------------------------------------------------------------------------
1152 def list_capacity_reservations(
1153 self,
1154 region: str,
1155 instance_type: str | None = None,
1156 state: str | None = "active",
1157 *,
1158 include_pricing: bool = False,
1159 ) -> list[dict[str, Any]]:
1160 """
1161 List EC2 On-Demand Capacity Reservations (ODCRs) in a region.
1163 Args:
1164 region: AWS region to query
1165 instance_type: Filter by instance type (optional)
1166 state: Filter by state — "active" (default), or None for all
1167 include_pricing: Enrich each reservation with On-Demand pricing
1168 (per-instance-hour, whole-reservation per-hour, per-GPU-hour).
1169 Adds one Pricing API call per distinct instance type (cached), so
1170 it is opt-in and off by default to keep the plain list fast.
1172 Returns:
1173 List of reservation dictionaries with availability details
1174 """
1175 ec2 = self._session.client("ec2", region_name=region)
1177 filters: list[dict[str, Any]] = []
1178 if state:
1179 filters.append({"Name": "state", "Values": [state]})
1180 if instance_type:
1181 filters.append({"Name": "instance-type", "Values": [instance_type]})
1183 reservations: list[dict[str, Any]] = []
1184 try:
1185 paginator = ec2.get_paginator("describe_capacity_reservations")
1186 page_kwargs: dict[str, Any] = {}
1187 if filters:
1188 page_kwargs["Filters"] = filters
1190 for page in paginator.paginate(**page_kwargs):
1191 for cr in page.get("CapacityReservations", []):
1192 total = cr.get("TotalInstanceCount", 0)
1193 available = cr.get("AvailableInstanceCount", 0)
1194 used = total - available
1196 entry = {
1197 "type": "odcr",
1198 "reservation_id": cr.get("CapacityReservationId"),
1199 "instance_type": cr.get("InstanceType"),
1200 "availability_zone": cr.get("AvailabilityZone"),
1201 "region": region,
1202 "state": cr.get("State"),
1203 "total_instances": total,
1204 "available_instances": available,
1205 "used_instances": used,
1206 "utilization_pct": round(used / total * 100, 1) if total else 0,
1207 "instance_platform": cr.get("InstancePlatform"),
1208 "tenancy": cr.get("Tenancy"),
1209 "instance_match_criteria": cr.get("InstanceMatchCriteria"),
1210 "start_date": (
1211 cr["StartDate"].isoformat() if cr.get("StartDate") else None
1212 ),
1213 "end_date": (cr["EndDate"].isoformat() if cr.get("EndDate") else None),
1214 "end_date_type": cr.get("EndDateType"),
1215 "tags": {t["Key"]: t["Value"] for t in cr.get("Tags", [])},
1216 }
1217 if include_pricing:
1218 self._enrich_reservation_pricing(entry, region)
1219 reservations.append(entry)
1220 except ClientError as e:
1221 logger.debug("Failed to list capacity reservations in %s: %s", region, e)
1223 return reservations
1225 def _gpus_per_instance(self, instance_type: str, region: str) -> int | None:
1226 """GPUs per instance for per-GPU-hour pricing, memoized per type+region.
1228 Returns ``None`` when the type cannot be described, which callers treat
1229 as "omit the per-GPU figure" rather than as zero GPUs — reporting a
1230 per-GPU-hour price of infinity, or silently dividing by a wrong count,
1231 would be worse than omitting it.
1232 """
1233 key = (instance_type, region)
1234 if key not in self._instance_info_cache:
1235 self._instance_info_cache[key] = self.get_instance_info(instance_type, region=region)
1236 info = self._instance_info_cache[key]
1237 return info.gpu_count if info else None
1239 def _enrich_reservation_pricing(self, reservation: dict[str, Any], region: str) -> None:
1240 """Add On-Demand pricing keys to a reservation dict, in place.
1242 ODCRs bill at the On-Demand rate for the reserved instance type, so the
1243 per-instance-hour / per-hour / per-GPU-hour figures mirror the Capacity
1244 Block pricing surface (see :func:`blocks.compute_reservation_pricing`) and
1245 let a caller rank and compare reservations the same way it ranks blocks.
1246 Pricing that can't be resolved is recorded as ``None`` — never fatal.
1247 """
1248 instance_type = reservation.get("instance_type")
1249 if not instance_type:
1250 return
1251 try:
1252 on_demand = self.get_on_demand_price(instance_type, region)
1253 except Exception as e: # pricing is supplementary; never fail the listing
1254 logger.debug("On-demand price lookup failed for %s in %s: %s", instance_type, region, e)
1255 on_demand = None
1257 gpus_per_instance = self._gpus_per_instance(str(instance_type), region)
1259 pricing = blocks.compute_reservation_pricing(
1260 on_demand, reservation.get("total_instances"), gpus_per_instance
1261 )
1262 reservation["on_demand_price_per_hour"] = (
1263 round(float(on_demand), 4) if on_demand is not None else None
1264 )
1265 reservation["gpus_per_instance"] = gpus_per_instance
1266 reservation.update(pricing)
1268 def _build_block_offering(
1269 self,
1270 offering: dict[str, Any],
1271 region: str,
1272 requested_count: int,
1273 gpus_per_instance: int | None,
1274 requested_duration_hours: int,
1275 ) -> dict[str, Any]:
1276 """Shape a raw DescribeCapacityBlockOfferings entry into an enriched dict.
1278 Reports the offering's *actual* duration (the API returns blocks whose
1279 duration is the closest match to the request, not necessarily equal) and
1280 adds per-hour / per-GPU-hour pricing derived from the upfront fee.
1281 ``upfront_fee`` is the raw API value (a string); ``upfront_fee_usd`` is the
1282 parsed float used for ranking and display.
1283 """
1284 start_date = offering.get("StartDate")
1285 end_date = offering.get("EndDate")
1287 actual_duration = offering.get("CapacityBlockDurationHours")
1288 if actual_duration is None:
1289 minutes = offering.get("CapacityBlockDurationMinutes")
1290 actual_duration = round(minutes / 60) if minutes else requested_duration_hours
1292 count = offering.get("InstanceCount")
1293 if count is None:
1294 count = requested_count
1296 pricing = blocks.compute_offering_pricing(
1297 offering.get("UpfrontFee"), actual_duration, count, gpus_per_instance
1298 )
1300 return {
1301 "type": "capacity_block",
1302 "offering_id": offering.get("CapacityBlockOfferingId"),
1303 "instance_type": offering.get("InstanceType"),
1304 "availability_zone": offering.get("AvailabilityZone"),
1305 "region": region,
1306 "instance_count": count,
1307 "duration_hours": actual_duration,
1308 "duration_days": blocks.hours_to_days(actual_duration),
1309 "start_date": start_date.isoformat() if start_date else None,
1310 "end_date": end_date.isoformat() if end_date else None,
1311 "upfront_fee": offering.get("UpfrontFee"),
1312 "upfront_fee_usd": pricing["upfront_fee_usd"],
1313 "price_per_hour": pricing["price_per_hour"],
1314 "price_per_instance_hour": pricing["price_per_instance_hour"],
1315 "price_per_gpu_hour": pricing["price_per_gpu_hour"],
1316 "gpus_per_instance": gpus_per_instance,
1317 "currency": offering.get("CurrencyCode", "USD"),
1318 "tenancy": offering.get("Tenancy"),
1319 }
1321 def list_capacity_block_offerings(
1322 self,
1323 region: str,
1324 instance_type: str,
1325 instance_count: int = 1,
1326 duration_hours: int = 24,
1327 *,
1328 earliest_start: datetime | None = None,
1329 latest_start: datetime | None = None,
1330 gpus_per_instance: int | None = None,
1331 ) -> list[dict[str, Any]]:
1332 """
1333 List available Capacity Block offerings for ML workloads.
1335 Capacity Blocks provide guaranteed GPU capacity for a fixed duration
1336 at a known price — ideal for training jobs with predictable runtimes.
1337 Queries a single duration in a single region; the date window and the
1338 multi-duration / multi-region sweep are layered on top by
1339 :meth:`find_capacity_blocks`.
1341 Args:
1342 region: AWS region to query
1343 instance_type: GPU instance type (e.g. p5.48xlarge, p4d.24xlarge)
1344 instance_count: Number of instances needed
1345 duration_hours: Desired block duration in hours (must be a supported value)
1346 earliest_start: Only return blocks starting on/after this datetime
1347 (EC2 StartDateRange). Lets callers ask "blocks starting near D1".
1348 latest_start: Only return blocks starting on/before this datetime
1349 (EC2 EndDateRange).
1350 gpus_per_instance: GPUs per instance, used for per-GPU-hour pricing.
1351 Resolved from the instance specs when omitted.
1353 Returns:
1354 List of available capacity block offerings (enriched with pricing).
1355 All matching pages are followed via NextToken.
1356 """
1357 ec2 = self._session.client("ec2", region_name=region, config=_CB_RETRY_CONFIG)
1358 offerings: list[dict[str, Any]] = []
1360 if gpus_per_instance is None:
1361 info = self.get_instance_info(instance_type)
1362 gpus_per_instance = info.gpu_count if info else None
1364 api_kwargs: dict[str, Any] = {
1365 "InstanceType": instance_type,
1366 "InstanceCount": instance_count,
1367 "CapacityDurationHours": duration_hours,
1368 }
1369 if earliest_start is not None:
1370 api_kwargs["StartDateRange"] = earliest_start
1371 if latest_start is not None:
1372 api_kwargs["EndDateRange"] = latest_start
1374 try:
1375 next_token: str | None = None
1376 while True:
1377 if next_token:
1378 api_kwargs["NextToken"] = next_token
1379 response = ec2.describe_capacity_block_offerings(**api_kwargs)
1380 for offering in response.get("CapacityBlockOfferings", []):
1381 offerings.append(
1382 self._build_block_offering(
1383 offering, region, instance_count, gpus_per_instance, duration_hours
1384 )
1385 )
1386 next_token = response.get("NextToken")
1387 if not next_token:
1388 break
1389 except ClientError as e:
1390 error_code = e.response.get("Error", {}).get("Code", "")
1391 if error_code in _CB_EXPECTED_ERROR_CODES:
1392 pass # Type/region doesn't support Capacity Blocks — expected
1393 else:
1394 logger.warning("Failed to list capacity block offerings in %s: %s", region, e)
1395 except BotoCoreError as e:
1396 # Endpoint resolution / connection errors surface here when the
1397 # Capacity Block API isn't available in a region.
1398 logger.warning("Capacity Block API unavailable in %s: %s", region, e)
1400 return offerings
1402 def get_capacity_block_trend(
1403 self,
1404 instance_type: str,
1405 region: str,
1406 ) -> float:
1407 """
1408 Estimate capacity block availability trend via time-series regression.
1410 Queries offerings across the maximum 182-day (26-week) window, buckets
1411 them into weekly bins by start date, and fits a linear regression to
1412 the offering counts per week. The normalized slope indicates whether
1413 capacity is growing or shrinking over time.
1415 Returns:
1416 Trend score from -1.0 to 1.0:
1417 > 0 = capacity growing (offerings increasing week-over-week)
1418 = 0 = stable or no data
1419 < 0 = capacity shrinking (offerings decreasing week-over-week)
1420 """
1421 ec2 = self._session.client("ec2", region_name=region, config=_CB_RETRY_CONFIG)
1423 now = datetime.now(UTC)
1424 far_end = now + timedelta(days=182)
1426 try:
1427 response = ec2.describe_capacity_block_offerings(
1428 InstanceType=instance_type,
1429 InstanceCount=1,
1430 CapacityDurationHours=24, # Minimum duration for broadest results
1431 StartDateRange=now,
1432 EndDateRange=far_end,
1433 )
1434 except ClientError as e:
1435 error_code = e.response.get("Error", {}).get("Code", "")
1436 if error_code not in _CB_EXPECTED_ERROR_CODES:
1437 logger.warning(
1438 "Capacity block trend query failed for %s in %s: %s",
1439 instance_type,
1440 region,
1441 e,
1442 )
1443 return 0.0
1444 except BotoCoreError as e:
1445 logger.warning(
1446 "Capacity block trend query failed for %s in %s: %s", instance_type, region, e
1447 )
1448 return 0.0
1450 offerings = response.get("CapacityBlockOfferings", [])
1451 if not offerings:
1452 return 0.0
1454 # Bucket offerings into weekly bins (week 0 = this week, week 25 = ~6 months out)
1455 num_weeks = 26
1456 bins = [0] * num_weeks
1457 for o in offerings:
1458 start = o.get("StartDate")
1459 if start is None:
1460 continue
1461 delta_days = (start - now).total_seconds() / 86400.0
1462 week_idx = int(delta_days / 7)
1463 if 0 <= week_idx < num_weeks:
1464 bins[week_idx] += 1
1466 # Need at least 2 non-zero bins to detect a meaningful trend
1467 non_zero = sum(1 for b in bins if b > 0)
1468 if non_zero < 2:
1469 return 0.0
1471 # Linear regression: slope of offerings-per-week over time
1472 # Using least-squares: slope = Σ((x-x̄)(y-ȳ)) / Σ((x-x̄)²)
1473 n = len(bins)
1474 x_mean = (n - 1) / 2.0
1475 y_mean = statistics.mean(bins)
1477 numerator = sum((i - x_mean) * (bins[i] - y_mean) for i in range(n))
1478 denominator = sum((i - x_mean) ** 2 for i in range(n))
1479 slope = numerator / denominator
1481 # Normalize slope to -1..1 range relative to the mean offering count.
1482 # A slope of +y_mean per 26 weeks would be a doubling → maps to ~1.0.
1483 normalized = slope * num_weeks / (y_mean * 2) if y_mean > 0 else 0.0
1485 return round(max(-1.0, min(1.0, normalized)), 4)
1487 def list_all_reservations(
1488 self,
1489 instance_type: str | None = None,
1490 regions: list[str] | None = None,
1491 ) -> dict[str, Any]:
1492 """
1493 List all capacity reservations (ODCRs) across deployed regions.
1495 Args:
1496 instance_type: Filter by instance type (optional)
1497 regions: Regions to query (defaults to deployed GCO regions)
1499 Returns:
1500 Summary dict with reservations grouped by region
1501 """
1502 if not regions:
1503 from cli.aws_client import get_aws_client
1505 aws_client = get_aws_client(self.config)
1506 stacks = aws_client.discover_regional_stacks()
1507 regions = list(stacks.keys()) if stacks else [self.config.default_region]
1509 all_reservations: list[dict[str, Any]] = []
1510 for region in regions:
1511 all_reservations.extend(
1512 self.list_capacity_reservations(region, instance_type=instance_type)
1513 )
1515 total_reserved = sum(r["total_instances"] for r in all_reservations)
1516 total_available = sum(r["available_instances"] for r in all_reservations)
1518 return {
1519 "regions_checked": regions,
1520 "instance_type_filter": instance_type,
1521 "total_reservations": len(all_reservations),
1522 "total_reserved_instances": total_reserved,
1523 "total_available_instances": total_available,
1524 "reservations": all_reservations,
1525 }
1527 def find_capacity_reservations(
1528 self,
1529 instance_type: str | None = None,
1530 regions: list[str] | None = None,
1531 *,
1532 min_count: int = 1,
1533 state: str | None = "active",
1534 include_pricing: bool = True,
1535 max_workers: int | None = None,
1536 ) -> dict[str, Any]:
1537 """Sweep regions for existing ODCRs in one parallel, ranked call.
1539 The ODCR counterpart to :meth:`find_capacity_blocks`. It fans out across
1540 every requested region in parallel, normalizes a friendly instance-type
1541 alias (``p6-b200`` -> ``p6-b200.48xlarge``), enriches each reservation with
1542 On-Demand pricing, and returns a single consolidated report ranked
1543 most-available-first (then cheapest per-GPU-hour). Where
1544 :meth:`list_all_reservations` simply aggregates region by region, this
1545 answers "where do I already have free reserved capacity for this instance
1546 type?" across many regions at once.
1548 Args:
1549 instance_type: Instance type or friendly alias to filter by. Omit to
1550 return every reservation (no type filter).
1551 regions: Regions to search in parallel (any regions, not just
1552 deployed); defaults to the deployed GCO regions when omitted.
1553 min_count: Minimum available instances for the summary to consider the
1554 search satisfied (does not filter the returned list).
1555 state: Reservation state filter ("active" by default; None for all).
1556 include_pricing: Enrich each reservation with On-Demand pricing.
1557 max_workers: Override the parallel fan-out width.
1559 Returns:
1560 A consolidated report dict (see keys assembled below).
1561 """
1562 canonical = instance_type
1563 note: str | None = None
1564 valid = True
1565 known = False
1566 if instance_type:
1567 validation = self.validate_instance_type(instance_type)
1568 canonical = validation["instance_type"]
1569 note = validation["note"]
1570 valid = validation["valid"]
1571 known = validation["known"]
1573 if not regions:
1574 from cli.aws_client import get_aws_client
1576 aws_client = get_aws_client(self.config)
1577 stacks = aws_client.discover_regional_stacks()
1578 regions = list(stacks.keys()) if stacks else [self.config.default_region]
1580 report: dict[str, Any] = {
1581 "instance_type": canonical,
1582 "requested_instance_type": instance_type,
1583 "valid_instance_type": valid,
1584 "known_instance_type": known,
1585 "note": note,
1586 "min_count": min_count,
1587 "state": state,
1588 "regions_checked": list(regions),
1589 "reservations_found": 0,
1590 "reservations": [],
1591 "ranked": [],
1592 "best": None,
1593 "total_reserved_instances": 0,
1594 "total_available_instances": 0,
1595 "regions_with_reservations": [],
1596 }
1598 if instance_type and not valid:
1599 report["recommendation"] = (
1600 note or f"'{instance_type}' is not a recognized EC2 instance type."
1601 )
1602 return report
1604 def _probe(region: str) -> list[dict[str, Any]]:
1605 try:
1606 return self.list_capacity_reservations(
1607 region,
1608 instance_type=canonical,
1609 state=state,
1610 include_pricing=include_pricing,
1611 )
1612 except Exception as e:
1613 logger.warning("Reservation probe failed for %s in %s: %s", canonical, region, e)
1614 return []
1616 collected: list[dict[str, Any]] = []
1617 workers = max(1, min(max_workers or _MAX_SEARCH_WORKERS, len(regions)))
1618 if len(regions) == 1:
1619 collected.extend(_probe(regions[0]))
1620 else:
1621 with ThreadPoolExecutor(max_workers=workers) as executor:
1622 for result in executor.map(_probe, regions):
1623 collected.extend(result)
1625 ranked = blocks.rank_reservations(collected)
1626 best = ranked[0] if ranked else None
1627 total_reserved = sum(r.get("total_instances") or 0 for r in collected)
1628 total_available = sum(r.get("available_instances") or 0 for r in collected)
1630 report.update(
1631 {
1632 "reservations_found": len(collected),
1633 "reservations": blocks.sort_reservations(collected),
1634 "ranked": ranked,
1635 "best": best,
1636 "total_reserved_instances": total_reserved,
1637 "total_available_instances": total_available,
1638 "regions_with_reservations": sorted(
1639 {r["region"] for r in collected if r.get("region")}
1640 ),
1641 "recommendation": self._summarize_reservation_search(
1642 canonical, regions, collected, best, min_count, note
1643 ),
1644 }
1645 )
1646 return report
1648 @staticmethod
1649 def _summarize_reservation_search(
1650 instance_type: str | None,
1651 regions: list[str],
1652 reservations: list[dict[str, Any]],
1653 best: dict[str, Any] | None,
1654 min_count: int,
1655 note: str | None,
1656 ) -> str:
1657 """Build a one-line human recommendation for a consolidated ODCR search."""
1658 label = instance_type or "any instance type"
1659 if not reservations:
1660 msg = (
1661 f"No active On-Demand Capacity Reservations for {label} across "
1662 f"{len(regions)} region(s). Create one with "
1663 "'gco capacity create-reservation', or search purchasable Capacity "
1664 "Blocks with 'gco capacity find-blocks'."
1665 )
1666 return f"{note} {msg}" if note else msg
1668 total_available = sum(r.get("available_instances") or 0 for r in reservations)
1669 regions_with = sorted({r["region"] for r in reservations if r.get("region")})
1670 parts = [
1671 f"Found {len(reservations)} reservation(s) for {label} across "
1672 f"{len(regions_with)} region(s); {total_available} instance(s) available."
1673 ]
1674 if total_available < min_count:
1675 parts.append(
1676 f"Fewer than the {min_count} requested are free — consider creating "
1677 "another reservation or a Capacity Block."
1678 )
1679 if best and (best.get("available_instances") or 0) > 0:
1680 gpu_hr = best.get("price_per_gpu_hour")
1681 gpu_hr_str = f", ${gpu_hr}/GPU-hr" if gpu_hr is not None else ""
1682 parts.append(
1683 f"Most available: {best.get('available_instances')}/"
1684 f"{best.get('total_instances')} in {best.get('region')}/"
1685 f"{best.get('availability_zone')} ({best.get('reservation_id')}{gpu_hr_str})."
1686 )
1687 msg = " ".join(parts)
1688 return f"{note} {msg}" if note else msg
1690 @staticmethod
1691 def _summarize_block_search(
1692 instance_type: str,
1693 regions: list[str],
1694 offerings: list[dict[str, Any]],
1695 best: dict[str, Any] | None,
1696 longest: dict[str, Any] | None,
1697 note: str | None,
1698 ) -> str:
1699 """Build a one-line human recommendation for a consolidated block search."""
1700 if not offerings:
1701 msg = (
1702 f"No Capacity Block offerings for {instance_type} across "
1703 f"{len(regions)} region(s) in the requested window. Try a wider "
1704 "date range, a shorter duration, more regions, or check back later."
1705 )
1706 return f"{note} {msg}" if note else msg
1708 regions_with = sorted({o["region"] for o in offerings if o.get("region")})
1709 parts = [
1710 f"Found {len(offerings)} Capacity Block offering(s) for {instance_type} "
1711 f"across {len(regions_with)} region(s)."
1712 ]
1713 if best:
1714 price = best.get("price_per_gpu_hour")
1715 price_str = f", from ${price}/GPU-hr" if price is not None else ""
1716 parts.append(
1717 f"Cheapest: {best.get('region')}/{best.get('availability_zone')} "
1718 f"{best.get('duration_days')}d starting "
1719 f"{(best.get('start_date') or '')[:16]}{price_str}."
1720 )
1721 if longest and longest is not best:
1722 parts.append(
1723 f"Longest: {longest.get('duration_days')}d in "
1724 f"{longest.get('region')}/{longest.get('availability_zone')}."
1725 )
1726 msg = " ".join(parts)
1727 return f"{note} {msg}" if note else msg
1729 def find_capacity_blocks(
1730 self,
1731 instance_type: str,
1732 regions: list[str] | None = None,
1733 *,
1734 instance_count: int = 1,
1735 duration_hours: int | None = None,
1736 duration_days: int | None = None,
1737 min_duration_hours: int | None = None,
1738 min_duration_days: int | None = None,
1739 max_duration_hours: int | None = None,
1740 max_duration_days: int | None = None,
1741 earliest_start: str | datetime | None = None,
1742 latest_start: str | datetime | None = None,
1743 find_longest: bool = False,
1744 max_workers: int | None = None,
1745 ) -> dict[str, Any]:
1746 """Sweep regions x durations x a date window for Capacity Blocks in one call.
1748 This is the high-level search that answers "find 1x p6-b200.48xlarge
1749 across us-east-1/us-east-2/us-west-2/eu-west-1 for durations 1-63 days
1750 starting 2026-07-01..2026-07-10" without manual multi-call sweeping.
1752 Because EC2 requires an exact ``CapacityDurationHours`` per query, a
1753 duration *range* (or ``find_longest``) expands to every valid Capacity
1754 Block duration in the range, and each (region, duration) pair is probed
1755 in parallel. Offerings are de-duplicated across probes, grouped/sorted by
1756 region + AZ + start date, and ranked cheapest-first by per-GPU-hour price.
1758 Args:
1759 instance_type: GPU instance type or friendly alias (e.g. p6-b200).
1760 regions: Explicit regions to search (any regions, not just deployed).
1761 Defaults to the deployed GCO regions when omitted.
1762 instance_count: Instances per block.
1763 duration_hours / duration_days: A single target duration.
1764 min_duration_hours / min_duration_days: Lower bound of a duration range.
1765 max_duration_hours / max_duration_days: Upper bound of a duration range.
1766 earliest_start / latest_start: Date (YYYY-MM-DD) or ISO datetime window
1767 for the block start, threaded to StartDateRange / EndDateRange.
1768 find_longest: Sweep the full duration ladder (within any range given)
1769 and surface the longest available block.
1770 max_workers: Override the parallel fan-out width.
1772 Returns:
1773 A consolidated report dict (see keys assembled below).
1774 """
1775 validation = self.validate_instance_type(instance_type)
1776 canonical = validation["instance_type"]
1777 gpus_per_instance = validation["gpu_count"]
1779 parsed_earliest = blocks.parse_date_input(earliest_start)
1780 parsed_latest = blocks.parse_date_input(latest_start)
1782 durations = blocks.resolve_search_durations(
1783 duration_hours=blocks.coerce_hours(duration_hours, duration_days),
1784 min_duration_hours=blocks.coerce_hours(min_duration_hours, min_duration_days),
1785 max_duration_hours=blocks.coerce_hours(max_duration_hours, max_duration_days),
1786 find_longest=find_longest,
1787 )
1789 if not regions:
1790 from cli.aws_client import get_aws_client
1792 aws_client = get_aws_client(self.config)
1793 stacks = aws_client.discover_regional_stacks()
1794 regions = list(stacks.keys()) if stacks else [self.config.default_region]
1796 report: dict[str, Any] = {
1797 "instance_type": canonical,
1798 "requested_instance_type": instance_type,
1799 "valid_instance_type": validation["valid"],
1800 "known_instance_type": validation["known"],
1801 "note": validation["note"],
1802 "instance_count": instance_count,
1803 "regions_checked": list(regions),
1804 "durations_probed_hours": durations,
1805 "durations_probed_days": [blocks.hours_to_days(h) for h in durations],
1806 "date_window": {
1807 "earliest_start": parsed_earliest.isoformat() if parsed_earliest else None,
1808 "latest_start": parsed_latest.isoformat() if parsed_latest else None,
1809 },
1810 "offerings_found": 0,
1811 "offerings": [],
1812 "ranked": [],
1813 "best": None,
1814 "longest": None,
1815 "regions_with_offerings": [],
1816 }
1818 if not validation["valid"]:
1819 report["recommendation"] = (
1820 validation["note"]
1821 or f"'{instance_type}' is not a valid standalone EC2 instance type "
1822 "for Capacity Blocks."
1823 )
1824 return report
1826 probes = [(region, dur) for region in regions for dur in durations]
1827 collected: list[dict[str, Any]] = []
1828 workers = max(1, min(max_workers or _MAX_SEARCH_WORKERS, len(probes)))
1830 def _probe(region: str, dur: int) -> list[dict[str, Any]]:
1831 try:
1832 return self.list_capacity_block_offerings(
1833 region,
1834 canonical,
1835 instance_count=instance_count,
1836 duration_hours=dur,
1837 earliest_start=parsed_earliest,
1838 latest_start=parsed_latest,
1839 gpus_per_instance=gpus_per_instance,
1840 )
1841 except Exception as e:
1842 logger.warning(
1843 "Capacity block probe failed for %s in %s (%sh): %s",
1844 canonical,
1845 region,
1846 dur,
1847 e,
1848 )
1849 return []
1851 with ThreadPoolExecutor(max_workers=workers) as executor:
1852 futures = [executor.submit(_probe, region, dur) for region, dur in probes]
1853 for future in as_completed(futures):
1854 collected.extend(future.result())
1856 unique = blocks.dedupe_offerings(collected)
1857 ranked = blocks.rank_offerings(unique)
1858 best = ranked[0] if ranked else None
1859 longest = blocks.longest_offering(unique)
1861 report.update(
1862 {
1863 "offerings_found": len(unique),
1864 "offerings": blocks.sort_offerings(unique),
1865 "ranked": ranked,
1866 "best": best,
1867 "longest": longest,
1868 "regions_with_offerings": sorted({o["region"] for o in unique if o.get("region")}),
1869 "recommendation": self._summarize_block_search(
1870 canonical, regions, unique, best, longest, validation["note"]
1871 ),
1872 }
1873 )
1874 return report
1876 def check_reservation_availability(
1877 self,
1878 instance_type: str,
1879 min_count: int = 1,
1880 include_capacity_blocks: bool = True,
1881 block_duration_hours: int = 24,
1882 *,
1883 regions: list[str] | None = None,
1884 block_duration_days: int | None = None,
1885 earliest_start: str | datetime | None = None,
1886 latest_start: str | datetime | None = None,
1887 max_workers: int | None = None,
1888 ) -> dict[str, Any]:
1889 """
1890 Check if capacity reservations or blocks have available instances.
1892 Checks both ODCRs (existing reservations) and Capacity Block offerings
1893 (purchasable guaranteed capacity) for a given instance type, across one
1894 or many regions queried in parallel.
1896 Args:
1897 instance_type: EC2 instance type to check
1898 min_count: Minimum number of available instances needed
1899 include_capacity_blocks: Also check Capacity Block offerings
1900 block_duration_hours: Duration for capacity block search (hours)
1901 regions: Regions to check in parallel (any regions, not just
1902 deployed); falls back to the deployed regions when omitted.
1903 block_duration_days: Block duration in days (overrides hours when set).
1904 earliest_start / latest_start: Date window for block start
1905 (StartDateRange / EndDateRange).
1906 max_workers: Override the parallel fan-out width.
1908 Returns:
1909 Dictionary with ODCR availability and capacity block offerings
1910 """
1911 if regions:
1912 target_regions = list(regions)
1913 else:
1914 from cli.aws_client import get_aws_client
1916 aws_client = get_aws_client(self.config)
1917 stacks = aws_client.discover_regional_stacks()
1918 target_regions = list(stacks.keys()) if stacks else [self.config.default_region]
1920 effective_duration = blocks.snap_duration_hours(
1921 blocks.coerce_hours(block_duration_hours, block_duration_days) or 24
1922 )
1923 parsed_earliest = blocks.parse_date_input(earliest_start)
1924 parsed_latest = blocks.parse_date_input(latest_start)
1926 def _check_region(r: str) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
1927 reservations = self.list_capacity_reservations(r, instance_type=instance_type)
1928 region_blocks: list[dict[str, Any]] = []
1929 if include_capacity_blocks:
1930 region_blocks = self.list_capacity_block_offerings(
1931 r,
1932 instance_type=instance_type,
1933 instance_count=min_count,
1934 duration_hours=effective_duration,
1935 earliest_start=parsed_earliest,
1936 latest_start=parsed_latest,
1937 )
1938 return reservations, region_blocks
1940 odcr_results: list[dict[str, Any]] = []
1941 block_offerings: list[dict[str, Any]] = []
1942 total_available = 0
1943 total_reserved = 0
1945 workers = max(1, min(max_workers or _MAX_SEARCH_WORKERS, len(target_regions)))
1946 if len(target_regions) == 1:
1947 region_results = [_check_region(target_regions[0])]
1948 else:
1949 with ThreadPoolExecutor(max_workers=workers) as executor:
1950 region_results = list(executor.map(_check_region, target_regions))
1952 for reservations, region_blocks in region_results:
1953 for res in reservations:
1954 avail = res["available_instances"]
1955 total_available += avail
1956 total_reserved += res["total_instances"]
1957 if avail > 0:
1958 odcr_results.append(res)
1959 block_offerings.extend(region_blocks)
1961 block_offerings = blocks.sort_offerings(block_offerings)
1962 has_odcr = total_available >= min_count
1963 has_blocks = len(block_offerings) > 0
1965 # Build recommendation
1966 if has_odcr:
1967 recommendation = (
1968 f"ODCR capacity available: {total_available} instances "
1969 f"across {len(odcr_results)} reservation(s)"
1970 )
1971 elif has_blocks:
1972 cheapest = min(block_offerings, key=_offering_fee)
1973 fee_display = cheapest.get("upfront_fee_usd")
1974 if fee_display is None:
1975 fee_display = cheapest.get("upfront_fee", "?")
1976 recommendation = (
1977 f"No ODCR capacity, but {len(block_offerings)} Capacity Block offering(s) "
1978 f"available (from ${fee_display} for {effective_duration}h)"
1979 )
1980 else:
1981 recommendation = (
1982 "No reserved capacity or block offerings found. "
1983 "Consider on-demand or spot, or request a Capacity Block "
1984 "for a different duration/region."
1985 )
1987 return {
1988 "instance_type": instance_type,
1989 "min_count_requested": min_count,
1990 "regions_checked": target_regions,
1991 "odcr": {
1992 "total_reserved_instances": total_reserved,
1993 "total_available_instances": total_available,
1994 "has_availability": has_odcr,
1995 "reservations": odcr_results,
1996 },
1997 "capacity_blocks": {
1998 "offerings_found": len(block_offerings),
1999 "has_offerings": has_blocks,
2000 "duration_hours": effective_duration,
2001 "date_window": {
2002 "earliest_start": parsed_earliest.isoformat() if parsed_earliest else None,
2003 "latest_start": parsed_latest.isoformat() if parsed_latest else None,
2004 },
2005 "offerings": block_offerings,
2006 },
2007 "recommendation": recommendation,
2008 }
2010 def purchase_capacity_block(
2011 self,
2012 offering_id: str,
2013 region: str,
2014 dry_run: bool = False,
2015 ) -> dict[str, Any]:
2016 """
2017 Purchase a Capacity Block offering by its ID.
2019 Args:
2020 offering_id: Capacity Block offering ID (cb-xxx) from list_capacity_block_offerings
2021 region: AWS region where the offering exists
2022 dry_run: If True, validate the offering without purchasing
2024 Returns:
2025 Dictionary with the created capacity reservation details
2026 """
2027 ec2 = self._session.client("ec2", region_name=region)
2029 if dry_run:
2030 try:
2031 # AWS signals an authorized DryRun with DryRunOperation. A
2032 # normal return violates that protocol and must never fall
2033 # through to the real purchase call.
2034 ec2.purchase_capacity_block(
2035 CapacityBlockOfferingId=offering_id,
2036 InstancePlatform="Linux/UNIX",
2037 DryRun=True,
2038 )
2039 except ClientError as e:
2040 error_code = e.response.get("Error", {}).get("Code", "")
2041 if error_code == "DryRunOperation":
2042 return {
2043 "success": True,
2044 "dry_run": True,
2045 "offering_id": offering_id,
2046 "region": region,
2047 "message": "Dry run succeeded — offering is valid and purchasable",
2048 }
2049 error_msg = e.response.get("Error", {}).get("Message", str(e))
2050 return {
2051 "success": False,
2052 "dry_run": True,
2053 "offering_id": offering_id,
2054 "region": region,
2055 "error_code": error_code,
2056 "error": error_msg,
2057 }
2058 return {
2059 "success": False,
2060 "dry_run": True,
2061 "offering_id": offering_id,
2062 "region": region,
2063 "error_code": "DryRunProtocolError",
2064 "error": "AWS returned normally for a DryRun request; no purchase was attempted.",
2065 }
2067 try:
2068 response = ec2.purchase_capacity_block(
2069 CapacityBlockOfferingId=offering_id,
2070 InstancePlatform="Linux/UNIX",
2071 )
2073 reservation = response.get("CapacityReservation", {})
2074 reservation_id = reservation.get("CapacityReservationId", "")
2075 instance_type = reservation.get("InstanceType", "")
2076 az = reservation.get("AvailabilityZone", "")
2077 total = reservation.get("TotalInstanceCount", 0)
2078 start = reservation.get("StartDate")
2079 end = reservation.get("EndDate")
2081 return {
2082 "success": True,
2083 "dry_run": False,
2084 "reservation_id": reservation_id,
2085 "offering_id": offering_id,
2086 "instance_type": instance_type,
2087 "availability_zone": az,
2088 "region": region,
2089 "total_instances": total,
2090 "start_date": start.isoformat() if start else None,
2091 "end_date": end.isoformat() if end else None,
2092 "state": reservation.get("State", ""),
2093 }
2094 except ClientError as e:
2095 error_code = e.response.get("Error", {}).get("Code", "")
2096 error_msg = e.response.get("Error", {}).get("Message", str(e))
2097 return {
2098 "success": False,
2099 "dry_run": False,
2100 "offering_id": offering_id,
2101 "region": region,
2102 "error_code": error_code,
2103 "error": error_msg,
2104 }
2106 def create_capacity_reservation(
2107 self,
2108 instance_type: str,
2109 region: str,
2110 availability_zone: str,
2111 instance_count: int = 1,
2112 *,
2113 instance_platform: str = "Linux/UNIX",
2114 tenancy: str = "default",
2115 instance_match_criteria: str = "open",
2116 end_date: str | datetime | None = None,
2117 ebs_optimized: bool = False,
2118 dry_run: bool = False,
2119 ) -> dict[str, Any]:
2120 """Create a new On-Demand Capacity Reservation (ODCR).
2122 The ODCR counterpart to :meth:`purchase_capacity_block`: it reserves
2123 On-Demand capacity for an instance type in a specific Availability Zone.
2124 Unlike a Capacity Block (a fixed-term block bought by offering id), an ODCR
2125 is open-ended by default and billed at the On-Demand rate until cancelled.
2126 Friendly instance-type aliases are normalized (``p6-b200`` ->
2127 ``p6-b200.48xlarge``).
2129 Args:
2130 instance_type: EC2 instance type or friendly alias.
2131 region: AWS region.
2132 availability_zone: Target AZ (e.g. us-east-1a).
2133 instance_count: Number of instances to reserve.
2134 instance_platform: Platform/OS (default "Linux/UNIX").
2135 tenancy: "default" or "dedicated".
2136 instance_match_criteria: "open" (any matching instance) or "targeted".
2137 end_date: Optional end date (ISO string or datetime). When set the
2138 reservation is "limited" and auto-releases then; omit for
2139 "unlimited".
2140 ebs_optimized: Reserve EBS-optimized capacity.
2141 dry_run: Validate permissions/parameters without creating (no cost).
2143 Returns:
2144 Dict with the created reservation's details, or an error payload.
2145 """
2146 validation = self.validate_instance_type(instance_type)
2147 canonical = validation["instance_type"]
2148 if not validation["valid"]:
2149 return {
2150 "success": False,
2151 "dry_run": dry_run,
2152 "instance_type": canonical,
2153 "region": region,
2154 "error_code": "InvalidInstanceType",
2155 "error": (
2156 validation["note"] or f"'{instance_type}' is not a valid EC2 instance type."
2157 ),
2158 }
2160 ec2 = self._session.client("ec2", region_name=region)
2161 parsed_end = blocks.parse_date_input(end_date)
2162 api_kwargs: dict[str, Any] = {
2163 "InstanceType": canonical,
2164 "InstancePlatform": instance_platform,
2165 "AvailabilityZone": availability_zone,
2166 "InstanceCount": instance_count,
2167 "Tenancy": tenancy,
2168 "InstanceMatchCriteria": instance_match_criteria,
2169 "EbsOptimized": ebs_optimized,
2170 }
2171 if parsed_end is not None:
2172 api_kwargs["EndDate"] = parsed_end
2173 api_kwargs["EndDateType"] = "limited"
2174 else:
2175 api_kwargs["EndDateType"] = "unlimited"
2177 if dry_run:
2178 try:
2179 ec2.create_capacity_reservation(DryRun=True, **api_kwargs)
2180 except ClientError as e:
2181 error_code = e.response.get("Error", {}).get("Code", "")
2182 if error_code == "DryRunOperation":
2183 return {
2184 "success": True,
2185 "dry_run": True,
2186 "instance_type": canonical,
2187 "availability_zone": availability_zone,
2188 "region": region,
2189 "instance_count": instance_count,
2190 "message": "Dry run succeeded — reservation parameters are valid.",
2191 }
2192 error_msg = e.response.get("Error", {}).get("Message", str(e))
2193 return {
2194 "success": False,
2195 "dry_run": True,
2196 "instance_type": canonical,
2197 "region": region,
2198 "error_code": error_code,
2199 "error": error_msg,
2200 }
2201 return {
2202 "success": False,
2203 "dry_run": True,
2204 "instance_type": canonical,
2205 "availability_zone": availability_zone,
2206 "region": region,
2207 "instance_count": instance_count,
2208 "error_code": "DryRunProtocolError",
2209 "error": "AWS returned normally for a DryRun request; no reservation was created.",
2210 }
2212 try:
2213 response = ec2.create_capacity_reservation(**api_kwargs)
2214 cr = response.get("CapacityReservation", {})
2215 return {
2216 "success": True,
2217 "dry_run": False,
2218 "reservation_id": cr.get("CapacityReservationId", ""),
2219 "instance_type": cr.get("InstanceType", canonical),
2220 "availability_zone": cr.get("AvailabilityZone", availability_zone),
2221 "region": region,
2222 "total_instances": cr.get("TotalInstanceCount", instance_count),
2223 "available_instances": cr.get("AvailableInstanceCount"),
2224 "state": cr.get("State", ""),
2225 "tenancy": cr.get("Tenancy", tenancy),
2226 "instance_match_criteria": cr.get("InstanceMatchCriteria", instance_match_criteria),
2227 "start_date": cr["StartDate"].isoformat() if cr.get("StartDate") else None,
2228 "end_date": cr["EndDate"].isoformat() if cr.get("EndDate") else None,
2229 "end_date_type": cr.get("EndDateType"),
2230 }
2231 except ClientError as e:
2232 error_code = e.response.get("Error", {}).get("Code", "")
2233 error_msg = e.response.get("Error", {}).get("Message", str(e))
2234 return {
2235 "success": False,
2236 "dry_run": False,
2237 "instance_type": canonical,
2238 "region": region,
2239 "error_code": error_code,
2240 "error": error_msg,
2241 }
2243 def cancel_capacity_reservation(
2244 self,
2245 reservation_id: str,
2246 region: str,
2247 dry_run: bool = False,
2248 ) -> dict[str, Any]:
2249 """Cancel an On-Demand Capacity Reservation by id.
2251 Releases reserved capacity so it stops incurring On-Demand charges. Only
2252 ODCRs can be cancelled this way; a Capacity Block runs for its fixed term.
2253 Cancelling does not terminate instances already running against the
2254 reservation — they simply revert to normal On-Demand billing.
2256 Args:
2257 reservation_id: Capacity Reservation id (cr-...).
2258 region: AWS region where the reservation exists.
2259 dry_run: Validate permissions without cancelling.
2261 Returns:
2262 Dict describing the outcome.
2263 """
2264 ec2 = self._session.client("ec2", region_name=region)
2266 if dry_run:
2267 try:
2268 ec2.cancel_capacity_reservation(CapacityReservationId=reservation_id, DryRun=True)
2269 except ClientError as e:
2270 error_code = e.response.get("Error", {}).get("Code", "")
2271 if error_code == "DryRunOperation":
2272 return {
2273 "success": True,
2274 "dry_run": True,
2275 "reservation_id": reservation_id,
2276 "region": region,
2277 "message": "Dry run succeeded — reservation can be cancelled.",
2278 }
2279 error_msg = e.response.get("Error", {}).get("Message", str(e))
2280 return {
2281 "success": False,
2282 "dry_run": True,
2283 "reservation_id": reservation_id,
2284 "region": region,
2285 "error_code": error_code,
2286 "error": error_msg,
2287 }
2288 return {
2289 "success": False,
2290 "dry_run": True,
2291 "reservation_id": reservation_id,
2292 "region": region,
2293 "error_code": "DryRunProtocolError",
2294 "error": "AWS returned normally for a DryRun request; no reservation was cancelled.",
2295 }
2297 try:
2298 response = ec2.cancel_capacity_reservation(CapacityReservationId=reservation_id)
2299 return {
2300 "success": bool(response.get("Return", True)),
2301 "dry_run": False,
2302 "reservation_id": reservation_id,
2303 "region": region,
2304 "message": f"Capacity reservation {reservation_id} cancelled.",
2305 }
2306 except ClientError as e:
2307 error_code = e.response.get("Error", {}).get("Code", "")
2308 error_msg = e.response.get("Error", {}).get("Message", str(e))
2309 return {
2310 "success": False,
2311 "dry_run": False,
2312 "reservation_id": reservation_id,
2313 "region": region,
2314 "error_code": error_code,
2315 "error": error_msg,
2316 }
2318 def recommend_region_for_job(
2319 self,
2320 gpu_required: bool = False,
2321 min_gpus: int = 0,
2322 instance_type: str | None = None,
2323 gpu_count: int = 0,
2324 ) -> dict[str, Any]:
2325 """
2326 Recommend the optimal region for job placement.
2328 Delegates to MultiRegionCapacityChecker for cross-region analysis.
2330 Args:
2331 gpu_required: Whether the job requires GPUs
2332 min_gpus: Minimum number of GPUs required
2333 instance_type: Specific instance type for workload-aware scoring
2334 gpu_count: Number of GPUs required
2336 Returns:
2337 Dictionary with recommended region and justification
2338 """
2339 from .multi_region import MultiRegionCapacityChecker
2341 checker = MultiRegionCapacityChecker(self.config)
2342 return checker.recommend_region_for_job(
2343 gpu_required, min_gpus, instance_type=instance_type, gpu_count=gpu_count
2344 )
2347def get_capacity_checker(config: GCOConfig | None = None) -> CapacityChecker:
2348 """Get a configured capacity checker instance."""
2349 return CapacityChecker(config)