Coverage for gco/services/spot_price_gate.py: 96.83%
98 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +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
28logger = logging.getLogger(__name__)
30#: EC2 instance type shape (``g5.xlarge``, ``p6-b200.48xlarge``, ``trn2.3xlarge``).
31INSTANCE_TYPE_PATTERN = re.compile(r"^[a-z][a-z0-9-]{0,29}\.[a-z0-9]{1,20}$")
33#: Spot price caps accepted at submission time (USD/hour).
34MIN_SPOT_PRICE = 0.0001
35MAX_SPOT_PRICE = 1_000.0
37_PRICE_CACHE_TTL_SECONDS = 60.0
38_PRICE_LOOKBACK_HOURS = 4
40#: Minimum seconds between persisted gate observations per job. In-memory
41#: evaluation still happens every pass; only the DynamoDB write is throttled.
42OBSERVATION_WRITE_INTERVAL_SECONDS = 60.0
45def validate_spot_gate_fields(max_price: float | None, instance_type: str | None) -> str | None:
46 """Validate the submission-time gate pair; return an error or ``None``.
48 The two fields are all-or-nothing: a cap without an instance type is
49 unenforceable, and an instance type without a cap is meaningless.
50 """
51 if max_price is None and instance_type is None:
52 return None
53 if max_price is None or instance_type is None:
54 return "max_spot_price and spot_instance_type must be provided together"
55 if not (MIN_SPOT_PRICE <= max_price <= MAX_SPOT_PRICE):
56 return f"max_spot_price must be between {MIN_SPOT_PRICE} and {MAX_SPOT_PRICE} USD/hour"
57 if not INSTANCE_TYPE_PATTERN.fullmatch(instance_type):
58 return "spot_instance_type is not a valid EC2 instance type"
59 return None
62@dataclass(frozen=True)
63class SpotGateDecision:
64 """Outcome of evaluating one job's spot price gate."""
66 gated: bool
67 instance_type: str
68 max_price: float
69 observed_price: float | None
70 reason: str
73class SpotPriceGate:
74 """TTL-cached regional spot price lookups plus per-job gate evaluation."""
76 def __init__(
77 self,
78 region: str,
79 *,
80 ec2_client: Any | None = None,
81 cache_ttl_seconds: float = _PRICE_CACHE_TTL_SECONDS,
82 ) -> None:
83 self.region = region
84 self.cache_ttl_seconds = float(cache_ttl_seconds)
85 self._ec2 = ec2_client
86 self._cache: dict[str, tuple[float, float | None]] = {}
88 def _client(self) -> Any:
89 if self._ec2 is None: 89 ↛ 90line 89 didn't jump to line 90 because the condition on line 89 was never true
90 import boto3
91 from botocore.config import Config
93 self._ec2 = boto3.client(
94 "ec2",
95 region_name=self.region,
96 config=Config(
97 connect_timeout=3,
98 read_timeout=10,
99 retries={"max_attempts": 2, "mode": "standard"},
100 ),
101 )
102 return self._ec2
104 def current_min_spot_price(self, instance_type: str) -> float | None:
105 """Return the lowest current spot price across AZs, or ``None``.
107 ``None`` means the price could not be determined (no offerings in the
108 region, API error, malformed response). Callers treat ``None`` as
109 "defer the job" — an unknown price must never dispatch a price-capped
110 job.
111 """
112 cached = self._cache.get(instance_type)
113 now = time.monotonic()
114 if cached is not None and now - cached[0] < self.cache_ttl_seconds:
115 return cached[1]
117 price = self._fetch_min_spot_price(instance_type)
118 self._cache[instance_type] = (now, price)
119 return price
121 def _fetch_min_spot_price(self, instance_type: str) -> float | None:
122 end = datetime.now(UTC)
123 try:
124 response = self._client().describe_spot_price_history(
125 InstanceTypes=[instance_type],
126 ProductDescriptions=["Linux/UNIX"],
127 StartTime=end - timedelta(hours=_PRICE_LOOKBACK_HOURS),
128 EndTime=end,
129 )
130 except Exception as exc: # noqa: BLE001 - lookup failures defer, never dispatch
131 logger.warning(
132 "Spot price lookup failed for %s in %s: %s",
133 instance_type,
134 self.region,
135 exc,
136 )
137 return None
139 # DescribeSpotPriceHistory returns newest-first per AZ; keep each
140 # AZ's most recent price and take the minimum across AZs.
141 latest_by_az: dict[str, float] = {}
142 for entry in response.get("SpotPriceHistory", []):
143 az = str(entry.get("AvailabilityZone") or "")
144 if not az or az in latest_by_az:
145 continue
146 try:
147 latest_by_az[az] = float(entry["SpotPrice"])
148 except KeyError, TypeError, ValueError:
149 continue
150 if not latest_by_az:
151 return None
152 return min(latest_by_az.values())
154 def evaluate(self, job: dict[str, Any]) -> SpotGateDecision | None:
155 """Evaluate one queue record's gate; ``None`` means the job is ungated.
157 Malformed gate fields on a stored record gate the job closed (with a
158 descriptive reason) rather than dispatching a job whose cap cannot be
159 honored.
160 """
161 raw_price = job.get("spot_max_price")
162 instance_type = job.get("spot_instance_type")
163 if raw_price is None and instance_type is None:
164 return None
165 try:
166 max_price = float(str(raw_price))
167 except TypeError, ValueError:
168 max_price = float("nan")
169 # A cap must be a finite number: NaN means the stored field is
170 # unparseable, and an infinite cap would wave every price through —
171 # both gate closed rather than dispatching a job whose cap cannot be
172 # honored.
173 if not isinstance(instance_type, str) or not instance_type or not math.isfinite(max_price):
174 return SpotGateDecision(
175 gated=True,
176 instance_type=str(instance_type or ""),
177 max_price=0.0,
178 observed_price=None,
179 reason="spot gate fields are malformed; refusing to dispatch",
180 )
182 observed = self.current_min_spot_price(instance_type)
183 if observed is None:
184 return SpotGateDecision(
185 gated=True,
186 instance_type=instance_type,
187 max_price=max_price,
188 observed_price=None,
189 reason=(
190 f"current spot price for {instance_type} in {self.region} "
191 "is unavailable; deferring"
192 ),
193 )
194 if observed > max_price:
195 return SpotGateDecision(
196 gated=True,
197 instance_type=instance_type,
198 max_price=max_price,
199 observed_price=observed,
200 reason=(
201 f"spot price {observed:.4f} USD/h for {instance_type} in "
202 f"{self.region} is above the {max_price:.4f} USD/h cap"
203 ),
204 )
205 return SpotGateDecision(
206 gated=False,
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} clears the {max_price:.4f} USD/h cap"
213 ),
214 )
217def should_persist_observation(job: dict[str, Any], now: datetime | None = None) -> bool:
218 """Throttle DynamoDB gate-observation writes per job.
220 Evaluation happens on every worker pass; persisting every observation
221 would add one write per gated job per pass for no operator benefit. Only
222 write when the record has no observation yet or the last one is older
223 than :data:`OBSERVATION_WRITE_INTERVAL_SECONDS`.
224 """
225 checked_at = job.get("spot_gate_checked_at")
226 if not checked_at:
227 return True
228 try:
229 last = datetime.fromisoformat(str(checked_at))
230 except ValueError:
231 return True
232 moment = now or datetime.now(UTC)
233 return (moment - last).total_seconds() >= OBSERVATION_WRITE_INTERVAL_SECONDS