Coverage for gco / services / spot_price_gate.py: 100.00%
98 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Spot price gating for the central DynamoDB job queue.
3A job submitted to the central queue may carry a spot price cap:
4``spot_max_price`` (USD/hour) for ``spot_instance_type``. The regional queue
5worker consults this gate before claiming such a job — while the instance
6type's current spot price in the worker's region sits above the cap, the job
7stays queued and is re-evaluated on every worker pass. The moment pricing
8drops to or below the cap, dispatch proceeds normally.
10Price lookups use ``ec2:DescribeSpotPriceHistory`` and take the *minimum*
11current price across the region's Availability Zones — a capacity-flexible
12job can land in whichever zone currently clears its cap. Results are cached
13briefly so a busy queue never hammers the EC2 API, and lookup failures fail
14open at the per-pass level by deferring only the affected job (never by
15dispatching above the cap).
16"""
18from __future__ import annotations
20import logging
21import math
22import re
23import time
24from dataclasses import dataclass
25from datetime import UTC, datetime, timedelta
26from typing import Any
28# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
29# Generated at (UTC): 2026-09-01T14:42:56Z
30# Generated from Git commit: 89b000378ed5a912a38c06f4feab2b029936ebcc
31# Flowchart(s) generated from this file:
32# * ``SpotPriceGate.evaluate`` -> ``diagrams/code_diagrams/gco/services/spot_price_gate.SpotPriceGate_evaluate.html``
33# (PNG: ``diagrams/code_diagrams/gco/services/spot_price_gate.SpotPriceGate_evaluate.png``)
34# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
35# <pyflowchart-code-diagram> END
38logger = logging.getLogger(__name__)
40#: EC2 instance type shape (``g5.xlarge``, ``p6-b200.48xlarge``, ``trn2.3xlarge``).
41INSTANCE_TYPE_PATTERN = re.compile(r"^[a-z][a-z0-9-]{0,29}\.[a-z0-9]{1,20}$")
43#: Spot price caps accepted at submission time (USD/hour).
44MIN_SPOT_PRICE = 0.0001
45MAX_SPOT_PRICE = 1_000.0
47_PRICE_CACHE_TTL_SECONDS = 60.0
48_PRICE_LOOKBACK_HOURS = 4
50#: Minimum seconds between persisted gate observations per job. In-memory
51#: evaluation still happens every pass; only the DynamoDB write is throttled.
52OBSERVATION_WRITE_INTERVAL_SECONDS = 60.0
55def validate_spot_gate_fields(max_price: float | None, instance_type: str | None) -> str | None:
56 """Validate the submission-time gate pair; return an error or ``None``.
58 The two fields are all-or-nothing: a cap without an instance type is
59 unenforceable, and an instance type without a cap is meaningless.
60 """
61 if max_price is None and instance_type is None:
62 return None
63 if max_price is None or instance_type is None:
64 return "max_spot_price and spot_instance_type must be provided together"
65 if not (MIN_SPOT_PRICE <= max_price <= MAX_SPOT_PRICE):
66 return f"max_spot_price must be between {MIN_SPOT_PRICE} and {MAX_SPOT_PRICE} USD/hour"
67 if not INSTANCE_TYPE_PATTERN.fullmatch(instance_type):
68 return "spot_instance_type is not a valid EC2 instance type"
69 return None
72@dataclass(frozen=True)
73class SpotGateDecision:
74 """Outcome of evaluating one job's spot price gate."""
76 gated: bool
77 instance_type: str
78 max_price: float
79 observed_price: float | None
80 reason: str
83class SpotPriceGate:
84 """TTL-cached regional spot price lookups plus per-job gate evaluation."""
86 def __init__(
87 self,
88 region: str,
89 *,
90 ec2_client: Any | None = None,
91 cache_ttl_seconds: float = _PRICE_CACHE_TTL_SECONDS,
92 ) -> None:
93 self.region = region
94 self.cache_ttl_seconds = float(cache_ttl_seconds)
95 self._ec2 = ec2_client
96 self._cache: dict[str, tuple[float, float | None]] = {}
98 def _client(self) -> Any:
99 if self._ec2 is None:
100 import boto3
101 from botocore.config import Config
103 self._ec2 = boto3.client(
104 "ec2",
105 region_name=self.region,
106 config=Config(
107 connect_timeout=3,
108 read_timeout=10,
109 retries={"max_attempts": 2, "mode": "standard"},
110 ),
111 )
112 return self._ec2
114 def current_min_spot_price(self, instance_type: str) -> float | None:
115 """Return the lowest current spot price across AZs, or ``None``.
117 ``None`` means the price could not be determined (no offerings in the
118 region, API error, malformed response). Callers treat ``None`` as
119 "defer the job" — an unknown price must never dispatch a price-capped
120 job.
121 """
122 cached = self._cache.get(instance_type)
123 now = time.monotonic()
124 if cached is not None and now - cached[0] < self.cache_ttl_seconds:
125 return cached[1]
127 price = self._fetch_min_spot_price(instance_type)
128 self._cache[instance_type] = (now, price)
129 return price
131 def _fetch_min_spot_price(self, instance_type: str) -> float | None:
132 end = datetime.now(UTC)
133 try:
134 response = self._client().describe_spot_price_history(
135 InstanceTypes=[instance_type],
136 ProductDescriptions=["Linux/UNIX"],
137 StartTime=end - timedelta(hours=_PRICE_LOOKBACK_HOURS),
138 EndTime=end,
139 )
140 except Exception as exc: # noqa: BLE001 - lookup failures defer, never dispatch
141 logger.warning(
142 "Spot price lookup failed for %s in %s: %s",
143 instance_type,
144 self.region,
145 exc,
146 )
147 return None
149 # DescribeSpotPriceHistory returns newest-first per AZ; keep each
150 # AZ's most recent price and take the minimum across AZs.
151 latest_by_az: dict[str, float] = {}
152 for entry in response.get("SpotPriceHistory", []):
153 az = str(entry.get("AvailabilityZone") or "")
154 if not az or az in latest_by_az:
155 continue
156 try:
157 latest_by_az[az] = float(entry["SpotPrice"])
158 except KeyError, TypeError, ValueError:
159 continue
160 if not latest_by_az:
161 return None
162 return min(latest_by_az.values())
164 def evaluate(self, job: dict[str, Any]) -> SpotGateDecision | None:
165 """Evaluate one queue record's gate; ``None`` means the job is ungated.
167 Malformed gate fields on a stored record gate the job closed (with a
168 descriptive reason) rather than dispatching a job whose cap cannot be
169 honored.
170 """
171 raw_price = job.get("spot_max_price")
172 instance_type = job.get("spot_instance_type")
173 if raw_price is None and instance_type is None:
174 return None
175 try:
176 max_price = float(str(raw_price))
177 except TypeError, ValueError:
178 max_price = float("nan")
179 # A cap must be a finite number: NaN means the stored field is
180 # unparseable, and an infinite cap would wave every price through —
181 # both gate closed rather than dispatching a job whose cap cannot be
182 # honored.
183 if not isinstance(instance_type, str) or not instance_type or not math.isfinite(max_price):
184 return SpotGateDecision(
185 gated=True,
186 instance_type=str(instance_type or ""),
187 max_price=0.0,
188 observed_price=None,
189 reason="spot gate fields are malformed; refusing to dispatch",
190 )
192 observed = self.current_min_spot_price(instance_type)
193 if observed is None:
194 return SpotGateDecision(
195 gated=True,
196 instance_type=instance_type,
197 max_price=max_price,
198 observed_price=None,
199 reason=(
200 f"current spot price for {instance_type} in {self.region} "
201 "is unavailable; deferring"
202 ),
203 )
204 if observed > max_price:
205 return SpotGateDecision(
206 gated=True,
207 instance_type=instance_type,
208 max_price=max_price,
209 observed_price=observed,
210 reason=(
211 f"spot price {observed:.4f} USD/h for {instance_type} in "
212 f"{self.region} is above the {max_price:.4f} USD/h cap"
213 ),
214 )
215 return SpotGateDecision(
216 gated=False,
217 instance_type=instance_type,
218 max_price=max_price,
219 observed_price=observed,
220 reason=(
221 f"spot price {observed:.4f} USD/h for {instance_type} in "
222 f"{self.region} clears the {max_price:.4f} USD/h cap"
223 ),
224 )
227def should_persist_observation(job: dict[str, Any], now: datetime | None = None) -> bool:
228 """Throttle DynamoDB gate-observation writes per job.
230 Evaluation happens on every worker pass; persisting every observation
231 would add one write per gated job per pass for no operator benefit. Only
232 write when the record has no observation yet or the last one is older
233 than :data:`OBSERVATION_WRITE_INTERVAL_SECONDS`.
234 """
235 checked_at = job.get("spot_gate_checked_at")
236 if not checked_at:
237 return True
238 try:
239 last = datetime.fromisoformat(str(checked_at))
240 except ValueError:
241 return True
242 moment = now or datetime.now(UTC)
243 return (moment - last).total_seconds() >= OBSERVATION_WRITE_INTERVAL_SECONDS