Coverage for lambda / capacity-poller / handler.py: 100.00%
277 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Capacity poller Lambda for the Historical Capacity Surface.
3Invoked on a schedule by an EventBridge rule (see GCOGlobalStack._create_capacity_poller in gco/stacks/global_stack.py).
4Snapshots capacity signals via read-only EC2 APIs and writes one item per
5watched (instance_type, region) pair into the capacity-history DynamoDB table.
7This handler is self-contained (boto3 + stdlib only) and does not import the
8CLI/gco packages, matching the convention used by the other GCO Lambdas. It writes
9the same DynamoDB item shape that cli/capacity/history.CapacityHistoryStore reads
10back.
12Control flow is phased because the signals have different shapes:
14 Phase 0 — region enablement. A client in the Lambda's default Region calls
15 ``DescribeRegions(AllRegions=True, RegionNames=[region])`` for each
16 configured Region. Only the authoritative ``not-opted-in`` state is
17 skipped. Missing/malformed responses, permission failures, throttling,
18 and transport errors remain ``unknown`` and fail open to polling so an
19 operational probe failure can never masquerade as deliberately absent
20 capacity. Explicitly not-enabled Regions are counted in the return
21 payload and receive no snapshots.
22 Phase 1 — Spot Placement Scores. AWS documents that
23 ``GetSpotPlacementScores`` needs at least three instance types for a
24 meaningful answer, so scores are requested per *instance pool* (a
25 reviewed set of interchangeable types; see INSTANCE_POOLS in
26 scripts/accelerator_catalog.py), once per (pool, target capacity),
27 with regions batched at most ``SPS_REGION_BATCH_SIZE`` per request —
28 the API returns the top 10 scored regions, so larger batches could
29 silently drop a requested region. One shared EC2 client issues every
30 SPS request; the API is cross-region regardless of endpoint.
31 Phase 2 — completeness. Expected (pool, region, capacity) combinations
32 are diffed against those received; only the gaps are re-requested,
33 for at most ``SPS_MAX_ATTEMPTS`` total passes so a persistent refusal
34 can never run the function toward its timeout.
35 Phase 3 — per-region metrics and write. Spot price, AZ count, and
36 Capacity Block offerings are inherently per-region. Failed probes stay
37 absent rather than becoming zero; when every signal for a pair fails,
38 no history item is written and the summary records an error.
40``MaxConfigLimitExceeded`` (the account has asked SPS about too many distinct
41configurations in the rolling window) is detected by error code, logged at
42warning level, counted separately in the return payload, and the affected
43score fields are omitted — an absent metric means "not obtained", never zero.
45Environment variables:
46 CAPACITY_HISTORY_TABLE_NAME DynamoDB table to write snapshots to
47 WATCH_INSTANCE_TYPES comma-separated instance types to poll
48 ENABLED_REGIONS comma-separated regions to poll
49 CAPACITY_HISTORY_RETENTION_DAYS TTL window in days (default 90)
50 CAPACITY_BLOCK_DURATION_HOURS short Capacity Block probe duration (default 24h = 1 day)
51 CAPACITY_BLOCK_LONG_DURATION_HOURS
52 long Capacity Block probe duration in hours (default 1512h = 63 days).
53 Set to 0 to skip the long probe. AWS allows durations in 1-day
54 increments up to 14 days, then 7-day increments up to 182 days.
55 SPOT_SCORE_TARGET_CAPACITIES
56 JSON array of {"target_capacity": int, "metric_field": str} objects,
57 e.g. [{"target_capacity": 1, "metric_field": "spot_score"}, ...].
58 The stack serializes this from the supported set and naming rule
59 exported by cli/capacity/history.py, so the capacity->field mapping
60 has exactly one source of truth even though this module cannot import
61 it. Default: capacity 1 -> spot_score.
62 INSTANCE_POOLS
63 JSON array of {"name": str, "members": [str, ...]} objects in
64 priority order; the stack serializes it from
65 scripts/accelerator_catalog.py INSTANCE_POOLS. Scores are requested
66 with a pool's full member list, and a watched type's snapshot records
67 the score of the first pool in this order that contains it, under the
68 ``spot_pool`` attribute. Watched types in no pool get price/Capacity
69 Block metrics but no placement score (deliberate; see
70 UNPOOLED_INSTANCE_TYPES in the catalog). Default: no pools, no SPS.
72Note: queue_depth is intentionally not collected here; it is a cluster-level signal
73that requires EKS access, which this EC2-only poller does not have. The history
74store treats a missing metric as absent, so omitting it is safe.
76The poller records two Capacity Block availability tiers per snapshot: the short
77duration (``capacity_blocks_available`` / ``capacity_blocks_total``) and the long
78duration (``capacity_blocks_long_available`` / ``capacity_blocks_long_total``), so
79history captures whether *extended-term* blocks (e.g. a 63-day P6 block) are
80available, not just the soonest 1-day block.
81"""
83from __future__ import annotations
85import json
86import logging
87import os
88import statistics
89from datetime import UTC, datetime, timedelta
90from decimal import Decimal
91from itertools import batched
92from typing import Any, Literal
94import boto3
96# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
97# Generated at (UTC): 2026-09-10T23:26:44Z
98# Generated from Git commit: 4c42b84d53d6cc01cd2b3c7e4011a43f850678b6
99# Flowchart(s) generated from this file:
100# * ``lambda_handler`` -> ``diagrams/code_diagrams/lambda/capacity-poller/handler.lambda_handler.html``
101# (PNG: ``diagrams/code_diagrams/lambda/capacity-poller/handler.lambda_handler.png``)
102# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
103# <pyflowchart-code-diagram> END
106logger = logging.getLogger()
107logger.setLevel(logging.INFO)
109DEFAULT_RETENTION_DAYS = 90
110SPOT_PRICE_LOOKBACK_DAYS = 7
112# Short Capacity Block probe: the smallest valid block (1 day), captures
113# soonest-available capacity. Long probe: an extended-term block (default 63
114# days = 9 weeks) so the history surface can answer "is a multi-week block
115# available?" for alerting and trend analysis. Both are configurable via the
116# CAPACITY_BLOCK_DURATION_HOURS / CAPACITY_BLOCK_LONG_DURATION_HOURS env vars.
117DEFAULT_BLOCK_DURATION_HOURS = 24
118DEFAULT_LONG_BLOCK_DURATION_HOURS = 63 * 24 # 1512h = 63 days = 9 weeks (a valid CB duration)
120# GetSpotPlacementScores returns the top 10 scored regions. Requesting more
121# regions than the response can name would let a region drop out silently, so
122# region batches never exceed this bound.
123SPS_REGION_BATCH_SIZE = 10
125# Total request passes for Spot Placement Scores: one initial pass plus
126# bounded retries of whatever combinations are still missing. Three passes of
127# a handful of (pool, capacity) requests complete in seconds, nowhere near the
128# 14-minute function timeout, and a combination still missing afterwards is
129# reported in the return payload rather than chased indefinitely.
130SPS_MAX_ATTEMPTS = 3
132# Fallback capacity->field mapping when SPOT_SCORE_TARGET_CAPACITIES is not
133# set: the pre-pool behavior of a single implicit target capacity of 1,
134# recorded in the original spot_score field.
135DEFAULT_TARGET_CAPACITIES: tuple[tuple[int, str], ...] = ((1, "spot_score"),)
138def _split_csv(value: str | None) -> list[str]:
139 return [item.strip() for item in (value or "").split(",") if item.strip()]
142def _to_decimal(value: Any) -> Any:
143 """Convert floats to Decimal for DynamoDB; pass other types through."""
144 if isinstance(value, bool):
145 return value
146 if isinstance(value, float):
147 return Decimal(str(value))
148 return value
151def _error_code(exc: Exception) -> str | None:
152 """Return the AWS error code carried by a botocore ClientError-shaped exception.
154 Duck-typed off the ``response`` attribute rather than importing botocore,
155 keeping this module's import surface to boto3 + stdlib.
156 """
157 response = getattr(exc, "response", None)
158 if not isinstance(response, dict):
159 return None
160 error = response.get("Error")
161 if not isinstance(error, dict):
162 return None
163 code = error.get("Code")
164 return code if isinstance(code, str) else None
167def _parse_target_capacities(raw: str | None) -> tuple[tuple[int, str], ...]:
168 """Parse SPOT_SCORE_TARGET_CAPACITIES into ((capacity, metric_field), ...).
170 The stack derives the mapping from cli/capacity/history.py, so this
171 parser only enforces shape. Malformed configuration raises rather than
172 silently collecting under wrong field names.
173 """
174 if not raw or not raw.strip():
175 return DEFAULT_TARGET_CAPACITIES
176 try:
177 parsed = json.loads(raw)
178 except json.JSONDecodeError as exc:
179 raise ValueError(f"SPOT_SCORE_TARGET_CAPACITIES is not valid JSON: {exc}") from exc
180 if not isinstance(parsed, list) or not parsed:
181 raise ValueError(
182 "SPOT_SCORE_TARGET_CAPACITIES must be a non-empty JSON array of "
183 f'{{"target_capacity", "metric_field"}} objects, got {parsed!r}'
184 )
185 capacities: list[tuple[int, str]] = []
186 for entry in parsed:
187 if not isinstance(entry, dict):
188 raise ValueError(f"SPOT_SCORE_TARGET_CAPACITIES entries must be objects, got {entry!r}")
189 capacity = entry.get("target_capacity")
190 field = entry.get("metric_field")
191 if isinstance(capacity, bool) or not isinstance(capacity, int) or capacity <= 0:
192 raise ValueError(
193 f"SPOT_SCORE_TARGET_CAPACITIES target_capacity must be a positive "
194 f"integer, got {capacity!r}"
195 )
196 if not isinstance(field, str) or not field:
197 raise ValueError(
198 f"SPOT_SCORE_TARGET_CAPACITIES metric_field must be a non-empty "
199 f"string, got {field!r}"
200 )
201 capacities.append((capacity, field))
202 return tuple(capacities)
205def _parse_instance_pools(raw: str | None) -> tuple[tuple[str, tuple[str, ...]], ...]:
206 """Parse INSTANCE_POOLS into ((pool_name, member_types), ...) in priority order.
208 A pool with fewer than three members would reintroduce the depressed-score
209 bug this poller exists to fix, so malformed pool configuration raises
210 instead of being polled.
211 """
212 if not raw or not raw.strip():
213 return ()
214 try:
215 parsed = json.loads(raw)
216 except json.JSONDecodeError as exc:
217 raise ValueError(f"INSTANCE_POOLS is not valid JSON: {exc}") from exc
218 if not isinstance(parsed, list):
219 raise ValueError(
220 f'INSTANCE_POOLS must be a JSON array of {{"name", "members"}}, got {parsed!r}'
221 )
222 pools: list[tuple[str, tuple[str, ...]]] = []
223 seen: set[str] = set()
224 for entry in parsed:
225 if not isinstance(entry, dict):
226 raise ValueError(f"INSTANCE_POOLS entries must be objects, got {entry!r}")
227 name = entry.get("name")
228 members = entry.get("members")
229 if not isinstance(name, str) or not name:
230 raise ValueError(f"INSTANCE_POOLS pool name must be a non-empty string, got {name!r}")
231 if name in seen:
232 raise ValueError(f"INSTANCE_POOLS declares pool {name!r} more than once")
233 if (
234 not isinstance(members, list)
235 or not all(isinstance(member, str) and member for member in members)
236 or len(set(members)) < 3
237 ):
238 raise ValueError(
239 f"INSTANCE_POOLS pool {name!r} must list at least three distinct "
240 f"instance types (GetSpotPlacementScores needs three for a "
241 f"meaningful score), got {members!r}"
242 )
243 seen.add(name)
244 pools.append((name, tuple(members)))
245 return tuple(pools)
248def _pool_for_instance_type(
249 pools: tuple[tuple[str, tuple[str, ...]], ...], instance_type: str
250) -> tuple[str, tuple[str, ...]] | None:
251 """Return the first pool in priority order containing instance_type, if any."""
252 for name, members in pools:
253 if instance_type in members:
254 return (name, members)
255 return None
258def _region_enablement_status(region: str) -> Literal["enabled", "not-enabled", "unknown"]:
259 """Classify account opt-in state without contacting the target endpoint.
261 ``DescribeRegions(AllRegions=True)`` is authoritative for explicit
262 ``not-opted-in`` state. Permission, throttling, transport, and malformed
263 responses remain unknown and are polled so operational failure cannot be
264 reported as intentionally absent capacity.
265 """
266 try:
267 ec2 = boto3.client("ec2")
268 response = ec2.describe_regions(AllRegions=True, RegionNames=[region])
269 rows = response.get("Regions", [])
270 match = next((row for row in rows if row.get("RegionName") == region), None)
271 if match is None:
272 logger.warning(
273 "region-enablement probe for %s returned no matching region; polling it as unknown",
274 region,
275 )
276 return "unknown"
277 opt_in_status = match.get("OptInStatus")
278 if opt_in_status == "not-opted-in":
279 logger.info("region %s is explicitly not opted in; skipping it", region)
280 return "not-enabled"
281 if opt_in_status in {"opt-in-not-required", "opted-in"}:
282 return "enabled"
283 logger.warning(
284 "region-enablement probe for %s returned unknown OptInStatus %r; polling it",
285 region,
286 opt_in_status,
287 )
288 return "unknown"
289 except Exception as exc:
290 code = _error_code(exc)
291 logger.warning(
292 "region-enablement ec2:DescribeRegions probe for %s failed (%s); polling it as "
293 "unknown so the per-region calls expose any real operational failure",
294 region,
295 code or type(exc).__name__,
296 )
297 return "unknown"
300def _regional_scores_from_response(response: dict[str, Any], regions: set[str]) -> dict[str, int]:
301 """Extract region -> regional score from one GetSpotPlacementScores page.
303 AZ-level records (carrying AvailabilityZoneId) and regions outside the
304 requested batch are ignored.
305 """
306 scores: dict[str, int] = {}
307 for rec in response.get("SpotPlacementScores", []):
308 if "AvailabilityZoneId" in rec:
309 continue
310 region = rec.get("Region")
311 if region in regions and rec.get("Score") is not None:
312 scores[region] = int(rec["Score"])
313 return scores
316def _collect_spot_placement_scores(
317 ec2: Any,
318 pools: tuple[tuple[str, tuple[str, ...]], ...],
319 target_capacities: tuple[tuple[int, str], ...],
320 regions: list[str],
321) -> tuple[dict[tuple[str, str, int], int], dict[str, int]]:
322 """Collect pooled Spot Placement Scores for every (pool, region, capacity).
324 Issues one request per (pool, target capacity, region batch) against a
325 single EC2 client, then re-requests only the missing combinations for at
326 most SPS_MAX_ATTEMPTS total passes. Returns the score mapping and the
327 counters for the structured summary; a MaxConfigLimitExceeded refusal is
328 logged distinctly, counted, and leaves its combinations absent.
329 """
330 scores: dict[tuple[str, str, int], int] = {}
331 counters = {
332 "requests_issued": 0,
333 "config_limit_refusals": 0,
334 }
335 expected = {
336 (name, region, capacity)
337 for name, _members in pools
338 for region in regions
339 for capacity, _field in target_capacities
340 }
342 for attempt in range(1, SPS_MAX_ATTEMPTS + 1):
343 requested_this_pass = False
344 for name, members in pools:
345 for capacity, _field in target_capacities:
346 missing_regions = [
347 region for region in regions if (name, region, capacity) not in scores
348 ]
349 if not missing_regions:
350 continue
351 if attempt > 1:
352 logger.info(
353 "retrying spot placement scores (attempt %d/%d) for pool=%s "
354 "capacity=%d regions=%s",
355 attempt,
356 SPS_MAX_ATTEMPTS,
357 name,
358 capacity,
359 missing_regions,
360 )
361 for batch in batched(missing_regions, SPS_REGION_BATCH_SIZE, strict=False):
362 requested_this_pass = True
363 batch_set = set(batch)
364 kwargs: dict[str, Any] = {
365 "InstanceTypes": list(members),
366 "TargetCapacity": capacity,
367 "TargetCapacityUnitType": "units",
368 "RegionNames": list(batch),
369 "SingleAvailabilityZone": False,
370 }
371 try:
372 while True:
373 counters["requests_issued"] += 1
374 response = ec2.get_spot_placement_scores(**kwargs)
375 for region, score in _regional_scores_from_response(
376 response, batch_set
377 ).items():
378 scores[(name, region, capacity)] = score
379 next_token = response.get("NextToken")
380 if not next_token:
381 break
382 kwargs["NextToken"] = next_token
383 except Exception as exc:
384 if _error_code(exc) == "MaxConfigLimitExceeded":
385 counters["config_limit_refusals"] += 1
386 logger.warning(
387 "spot placement scores REFUSED (MaxConfigLimitExceeded) for "
388 "pool=%s capacity=%d regions=%s: the account has queried too "
389 "many distinct SPS configurations in the rolling window; the "
390 "affected score fields are omitted from this cycle's snapshots",
391 name,
392 capacity,
393 list(batch),
394 )
395 else:
396 logger.warning(
397 "spot placement scores failed for pool=%s capacity=%d "
398 "regions=%s: %s",
399 name,
400 capacity,
401 list(batch),
402 exc,
403 )
404 if not requested_this_pass:
405 break
407 missing = sorted(expected - set(scores))
408 if missing:
409 logger.warning(
410 "spot placement scores missing after %d attempt(s) for %d combination(s): %s",
411 SPS_MAX_ATTEMPTS,
412 len(missing),
413 missing,
414 )
415 counters["combinations_expected"] = len(expected)
416 counters["combinations_received"] = len(expected) - len(missing)
417 counters["combinations_missing_after_retry"] = len(missing)
418 return scores, counters
421def _spot_price_summary(ec2: Any, instance_type: str) -> tuple[float | None, int | None]:
422 """Return (mean latest price, AZ count), preserving probe failure as None."""
423 end = datetime.now(UTC)
424 start = end - timedelta(days=SPOT_PRICE_LOOKBACK_DAYS)
425 try:
426 resp = ec2.describe_spot_price_history(
427 InstanceTypes=[instance_type],
428 ProductDescriptions=["Linux/UNIX"],
429 StartTime=start,
430 EndTime=end,
431 )
432 except Exception as exc:
433 logger.warning("spot price history failed for %s: %s", instance_type, exc)
434 return None, None
435 latest_by_az: dict[str, float] = {}
436 for item in resp.get("SpotPriceHistory", []):
437 az = item.get("AvailabilityZone")
438 if az and az not in latest_by_az:
439 latest_by_az[az] = float(item["SpotPrice"])
440 if not latest_by_az:
441 return None, 0
442 return round(statistics.fmean(latest_by_az.values()), 6), len(latest_by_az)
445def _capacity_block_summary(
446 ec2: Any, instance_type: str, duration_hours: int = DEFAULT_BLOCK_DURATION_HOURS
447) -> tuple[int | None, int | None]:
448 """Return offering/instance counts, or ``(None, None)`` on probe failure.
450 ``duration_hours`` is the Capacity Block duration to probe; the poller calls
451 this once for the short tier and once for the long tier. A successful empty
452 response is ``(0, 0)``; failure stays absent so it cannot become false
453 zero-capacity history.
454 """
455 try:
456 resp = ec2.describe_capacity_block_offerings(
457 InstanceType=instance_type,
458 InstanceCount=1,
459 CapacityDurationHours=duration_hours,
460 )
461 except Exception as exc:
462 logger.debug(
463 "capacity block offerings unavailable for %s (%sh): %s",
464 instance_type,
465 duration_hours,
466 exc,
467 )
468 return None, None
469 offerings = resp.get("CapacityBlockOfferings", [])
470 total = sum(int(o.get("InstanceCount", 0)) for o in offerings)
471 return len(offerings), total
474def _build_item(
475 instance_type: str,
476 region: str,
477 now: datetime,
478 retention_days: int,
479 spot_scores: dict[str, int],
480 spot_pool: str | None,
481 spot_price: float | None,
482 az_count: int | None,
483 blocks_available: int | None,
484 blocks_total: int | None,
485 long_blocks_available: int | None = None,
486 long_blocks_total: int | None = None,
487) -> dict[str, Any]:
488 """Assemble a DynamoDB item matching the CapacityHistoryStore schema.
490 ``spot_scores`` maps metric field name (spot_score, spot_score_at_N) to
491 the pooled score value; ``spot_pool`` names the pool those scores were
492 requested for and is recorded only when at least one score was obtained,
493 so refused or unpooled snapshots carry neither the fields nor a dangling
494 attribution. Per-Region probe values remain optional: ``None`` means the
495 API call failed and the field is omitted, while a successful empty
496 Capacity Block response is recorded as a real zero.
497 """
498 ts = now.isoformat()
499 item: dict[str, Any] = {
500 "pk": f"{instance_type}#{region}",
501 "sk": ts,
502 "instance_type": instance_type,
503 "region": region,
504 "timestamp": ts,
505 "ttl": int((now + timedelta(days=retention_days)).timestamp()),
506 }
507 for field, score in spot_scores.items():
508 item[field] = score
509 if spot_scores and spot_pool is not None:
510 item["spot_pool"] = spot_pool
511 if spot_price is not None:
512 item["spot_price"] = _to_decimal(spot_price)
513 if az_count is not None:
514 item["az_count"] = az_count
515 if blocks_available is not None:
516 item["capacity_blocks_available"] = blocks_available
517 if blocks_total is not None:
518 item["capacity_blocks_total"] = blocks_total
519 # Long-duration tier is omitted entirely when the long probe is disabled
520 # (CAPACITY_BLOCK_LONG_DURATION_HOURS=0), so the store treats it as absent
521 # rather than recording a misleading zero.
522 if long_blocks_available is not None:
523 item["capacity_blocks_long_available"] = long_blocks_available
524 if long_blocks_total is not None:
525 item["capacity_blocks_long_total"] = long_blocks_total
526 return item
529def lambda_handler(event: dict[str, Any], context: Any) -> dict[str, Any]:
530 """Snapshot capacity signals for every watched (instance_type, region) pair."""
531 table_name = os.environ.get("CAPACITY_HISTORY_TABLE_NAME")
532 if not table_name:
533 raise ValueError("CAPACITY_HISTORY_TABLE_NAME environment variable is required")
535 instance_types = _split_csv(os.environ.get("WATCH_INSTANCE_TYPES"))
536 configured_regions = _split_csv(os.environ.get("ENABLED_REGIONS"))
537 retention_days = int(
538 os.environ.get("CAPACITY_HISTORY_RETENTION_DAYS", str(DEFAULT_RETENTION_DAYS))
539 )
540 block_duration_hours = int(
541 os.environ.get("CAPACITY_BLOCK_DURATION_HOURS", str(DEFAULT_BLOCK_DURATION_HOURS))
542 )
543 long_block_duration_hours = int(
544 os.environ.get("CAPACITY_BLOCK_LONG_DURATION_HOURS", str(DEFAULT_LONG_BLOCK_DURATION_HOURS))
545 )
546 target_capacities = _parse_target_capacities(os.environ.get("SPOT_SCORE_TARGET_CAPACITIES"))
547 pools = _parse_instance_pools(os.environ.get("INSTANCE_POOLS"))
549 if not instance_types:
550 logger.warning("WATCH_INSTANCE_TYPES is empty; nothing to poll")
551 if not configured_regions:
552 logger.warning("ENABLED_REGIONS is empty; nothing to poll")
554 table = boto3.resource("dynamodb").Table(table_name)
555 now = datetime.now(UTC)
556 written = 0
557 errors = 0
559 # Phase 0 — region enablement pre-check. A not-enabled region would fail
560 # every API call in ways the per-type error isolation below would swallow,
561 # which is exactly the "absent data that is really a failure" class this
562 # poller must not produce.
563 regions: list[str] = []
564 regions_skipped_not_enabled: list[str] = []
565 regions_enablement_unknown: list[str] = []
566 for region in configured_regions:
567 enablement = _region_enablement_status(region)
568 if enablement == "not-enabled":
569 regions_skipped_not_enabled.append(region)
570 continue
571 regions.append(region)
572 if enablement == "unknown":
573 regions_enablement_unknown.append(region)
575 # Phase 1 + 2 — pooled, batched, multi-capacity SPS with bounded
576 # completeness retry. Only pools containing at least one watched type are
577 # requested, but each request carries the pool's full member list: the
578 # score is a property of the whole pool, and shrinking the list to the
579 # watched subset would change (and potentially depress) the measurement.
580 watched = set(instance_types)
581 relevant_pools = tuple(
582 (name, members) for name, members in pools if watched.intersection(members)
583 )
584 unpooled_watch_types = sorted(
585 itype for itype in watched if _pool_for_instance_type(pools, itype) is None
586 )
587 if unpooled_watch_types:
588 logger.info(
589 "%d watched instance type(s) belong to no instance pool and get no placement "
590 "score (price and Capacity Block metrics are still recorded): %s",
591 len(unpooled_watch_types),
592 unpooled_watch_types,
593 )
594 sps_client = boto3.client("ec2")
595 scores, sps_counters = _collect_spot_placement_scores(
596 sps_client, relevant_pools, target_capacities, regions
597 )
599 # The long probe is enabled when its duration is positive and differs from
600 # the short probe; when equal we reuse the short result to avoid a redundant
601 # API call, and when <= 0 we skip it entirely (long fields stay absent).
602 long_probe_enabled = long_block_duration_hours > 0
604 # Phase 3 — per-region price and Capacity Block metrics, then assemble and
605 # write. This loop is unchanged in shape from the pre-pool poller; only
606 # the score fields now come from the phase-1 pool collection.
607 for region in regions:
608 ec2 = boto3.client("ec2", region_name=region)
609 for instance_type in instance_types:
610 try:
611 pool = _pool_for_instance_type(pools, instance_type)
612 spot_scores: dict[str, int] = {}
613 spot_pool: str | None = None
614 if pool is not None:
615 pool_name, _members = pool
616 spot_pool = pool_name
617 for capacity, field in target_capacities:
618 value = scores.get((pool_name, region, capacity))
619 if value is not None:
620 spot_scores[field] = value
621 spot_price, az_count = _spot_price_summary(ec2, instance_type)
622 blocks_available, blocks_total = _capacity_block_summary(
623 ec2, instance_type, block_duration_hours
624 )
625 long_available: int | None = None
626 long_total: int | None = None
627 if long_probe_enabled:
628 if long_block_duration_hours == block_duration_hours:
629 long_available, long_total = blocks_available, blocks_total
630 else:
631 long_available, long_total = _capacity_block_summary(
632 ec2, instance_type, long_block_duration_hours
633 )
634 signal_obtained = bool(spot_scores) or any(
635 value is not None
636 for value in (
637 spot_price,
638 az_count,
639 blocks_available,
640 blocks_total,
641 long_available,
642 long_total,
643 )
644 )
645 if not signal_obtained:
646 errors += 1
647 logger.warning(
648 "all capacity probes failed for %s/%s; skipping history write "
649 "rather than recording false zero capacity",
650 instance_type,
651 region,
652 )
653 continue
654 item = _build_item(
655 instance_type,
656 region,
657 now,
658 retention_days,
659 spot_scores,
660 spot_pool,
661 spot_price,
662 az_count,
663 blocks_available,
664 blocks_total,
665 long_available,
666 long_total,
667 )
668 table.put_item(Item=item)
669 written += 1
670 except Exception as exc:
671 errors += 1
672 logger.exception("failed to record %s/%s: %s", instance_type, region, exc)
674 summary: dict[str, Any] = {
675 "written": written,
676 "errors": errors,
677 "timestamp": now.isoformat(),
678 "regions_polled": regions,
679 "regions_skipped_not_enabled": regions_skipped_not_enabled,
680 "regions_enablement_unknown": regions_enablement_unknown,
681 "sps": {
682 **sps_counters,
683 "pools": len(relevant_pools),
684 "target_capacities": [capacity for capacity, _field in target_capacities],
685 "unpooled_watch_types": len(unpooled_watch_types),
686 },
687 }
688 logger.info(
689 "capacity poll complete: written=%d errors=%d sps_requests=%d sps_received=%d/%d "
690 "sps_missing=%d config_limit_refusals=%d regions_skipped=%d "
691 "regions_enablement_unknown=%d",
692 written,
693 errors,
694 summary["sps"]["requests_issued"],
695 summary["sps"]["combinations_received"],
696 summary["sps"]["combinations_expected"],
697 summary["sps"]["combinations_missing_after_retry"],
698 summary["sps"]["config_limit_refusals"],
699 len(regions_skipped_not_enabled),
700 len(regions_enablement_unknown),
701 )
702 return summary