Coverage for cli / capacity / history.py: 100.00%
205 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""
2DynamoDB-backed time-series store for historical capacity signals.
4This is the storage and query layer for the Historical Capacity Surface.
5A scheduled poller (lambda/capacity-poller) snapshots capacity signals --
6spot placement score, spot price, AZ coverage, queue depth, and capacity
7block availability -- for a watched set of instance types across regions and
8writes them here. The CLI (gco capacity history ...) and the Bedrock capacity
9advisor read them back for temporal querying and prompt enrichment.
11Table schema (a single global table; see GCOGlobalStack._create_capacity_poller):
13 pk (partition) = "{instance_type}#{region}"
14 sk (sort) = ISO-8601 UTC timestamp of the snapshot
16 GSI "by-timestamp": pk = instance_type, sk = timestamp
17 (cross-region trend queries for one instance type)
19Item attributes: instance_type, region, timestamp, spot_score,
20spot_score_at_10, spot_score_at_50, spot_price, az_count, queue_depth,
21capacity_blocks_available, capacity_blocks_total,
22capacity_blocks_long_available, capacity_blocks_long_total, spot_pool (the
23instance pool the Spot Placement Scores were requested for, when pooled), and
24ttl (epoch seconds for DynamoDB auto-expiry, default 90 days). The
25``spot_score*`` family holds one field per configured Spot Placement Score
26target capacity (see SUPPORTED_SPOT_SCORE_TARGET_CAPACITIES); ``spot_score``
27is target capacity 1, keeping every pre-existing snapshot readable without
28migration. The ``*_long_*`` fields track availability of extended-term blocks
29(the poller's long-duration probe, default 63 days) separately from the
30soonest short block.
32Numbers are stored as DynamoDB Decimal (the resource API rejects float) and
33re-hydrated to int/float on read. The poller Lambda is self-contained and
34writes the same item shape directly with boto3; this module is the canonical
35writer/reader used by the CLI and advisor.
36"""
38from __future__ import annotations
40import logging
41import os
42import statistics
43from datetime import UTC, datetime, timedelta
44from decimal import Decimal
45from typing import Any
47import boto3
48from boto3.dynamodb.conditions import Key
50logger = logging.getLogger(__name__)
52DEFAULT_TABLE_NAME = "gco-capacity-history"
53DEFAULT_RETENTION_DAYS = 90
54GSI_BY_TIMESTAMP = "by-timestamp"
56# Spot Placement Score target capacities the schema supports, in display
57# order. The set is closed on purpose: METRIC_FIELDS is a flat, statically
58# known tuple iterated by get_statistics and get_temporal_patterns, so every
59# capacity needs a pre-declared field. ``spot_score_target_capacities`` in
60# cdk.json selects a subset of this set; the config validator, the poller, and
61# the CLI all derive field names through metric_field_for_target_capacity so
62# the naming rule lives in exactly one place.
63SUPPORTED_SPOT_SCORE_TARGET_CAPACITIES: tuple[int, ...] = (1, 10, 50)
66def metric_field_for_target_capacity(target_capacity: int) -> str:
67 """Map a Spot Placement Score target capacity to its metric field name.
69 Capacity 1 maps to the pre-existing ``spot_score`` field so every snapshot
70 written before multi-capacity collection stays readable without migration;
71 capacity N > 1 maps to ``spot_score_at_{N}``. Raises ValueError for a
72 capacity outside SUPPORTED_SPOT_SCORE_TARGET_CAPACITIES, naming the value
73 and the supported set, because unsupported capacities have no statically
74 declared metric field.
75 """
76 # bool is a subclass of int and True == 1, so reject it explicitly rather
77 # than letting a leaked flag silently masquerade as target capacity 1.
78 if isinstance(target_capacity, bool) or (
79 target_capacity not in SUPPORTED_SPOT_SCORE_TARGET_CAPACITIES
80 ):
81 raise ValueError(
82 f"unsupported Spot Placement Score target capacity {target_capacity!r}; "
83 f"supported target capacities: {list(SUPPORTED_SPOT_SCORE_TARGET_CAPACITIES)}"
84 )
85 if target_capacity == 1:
86 return "spot_score"
87 return f"spot_score_at_{target_capacity}"
90# The numeric metrics tracked per snapshot. Statistics and temporal-pattern
91# aggregation iterate over this tuple, so adding a metric is a one-line change.
92# The ``spot_score*`` block declares one field per supported Spot Placement
93# Score target capacity (via the naming rule above); ``spot_score`` keeps its
94# leading position so existing column ordering stays readable. The
95# ``capacity_blocks_long_*`` pair mirrors the short-duration block metrics
96# but for the poller's long-duration probe (default 63 days), so trend/alerting
97# queries can distinguish soonest-available blocks from extended-term ones.
98METRIC_FIELDS: tuple[str, ...] = (
99 "spot_score",
100 "spot_score_at_10",
101 "spot_score_at_50",
102 "spot_price",
103 "az_count",
104 "queue_depth",
105 "capacity_blocks_available",
106 "capacity_blocks_total",
107 "capacity_blocks_long_available",
108 "capacity_blocks_long_total",
109)
111# weekday() -> name. Monday is 0, matching datetime.weekday().
112DAY_NAMES: tuple[str, ...] = (
113 "Monday",
114 "Tuesday",
115 "Wednesday",
116 "Thursday",
117 "Friday",
118 "Saturday",
119 "Sunday",
120)
123def _utc_now() -> datetime:
124 return datetime.now(UTC)
127def make_pk(instance_type: str, region: str) -> str:
128 """Build the partition key for a (instance_type, region) series."""
129 return f"{instance_type}#{region}"
132def _parse_iso(value: str) -> datetime | None:
133 """Parse an ISO-8601 timestamp, tolerating a trailing Z and naive values."""
134 try:
135 dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
136 except ValueError, AttributeError:
137 return None
138 if dt.tzinfo is None:
139 dt = dt.replace(tzinfo=UTC)
140 return dt
143def _to_dynamo(value: Any) -> Any:
144 """Convert a Python value into a DynamoDB-storable type.
146 The DynamoDB resource API rejects float, so numbers must be Decimal.
147 Floats are routed through Decimal(str(x)) so the decimal string round-trips
148 without binary-float artifacts. bool is checked before the numeric branch
149 because bool is a subclass of int.
150 """
151 if isinstance(value, bool):
152 return value
153 if isinstance(value, float):
154 return Decimal(str(value))
155 if isinstance(value, dict):
156 return {k: _to_dynamo(v) for k, v in value.items()}
157 if isinstance(value, list):
158 return [_to_dynamo(v) for v in value]
159 return value
162def _from_dynamo(value: Any) -> Any:
163 """Convert DynamoDB types back to plain Python (Decimal -> int/float)."""
164 if isinstance(value, Decimal):
165 return int(value) if value == int(value) else float(value)
166 if isinstance(value, dict):
167 return {k: _from_dynamo(v) for k, v in value.items()}
168 if isinstance(value, list):
169 return [_from_dynamo(v) for v in value]
170 return value
173def _percentile(sorted_values: list[float], pct: float) -> float:
174 """Linear-interpolation percentile (pct in [0, 100]) over sorted values."""
175 if not sorted_values:
176 raise ValueError("percentile of empty sequence")
177 if len(sorted_values) == 1:
178 return float(sorted_values[0])
179 rank = (pct / 100.0) * (len(sorted_values) - 1)
180 lo = int(rank)
181 hi = min(lo + 1, len(sorted_values) - 1)
182 frac = rank - lo
183 return float(sorted_values[lo] + (sorted_values[hi] - sorted_values[lo]) * frac)
186def flatten_capacity_data(capacity_data: dict[str, Any]) -> list[dict[str, Any]]:
187 """Flatten gather_capacity_data() output into per-(instance_type, region) records.
189 Returns one record per (instance_type, region) pair that carries at least
190 one signal. Each record holds the keys instance_type, region, timestamp
191 plus whatever subset of METRIC_FIELDS can be derived. Metrics that cannot
192 be derived are omitted (the statistics layer treats a missing metric as
193 absent, never as zero).
194 """
195 timestamp = capacity_data.get("timestamp") or _utc_now().isoformat()
197 queue_by_region: dict[str, Any] = {}
198 for metric in capacity_data.get("cluster_metrics", []) or []:
199 region = metric.get("region")
200 if region is not None and metric.get("queue_depth") is not None:
201 queue_by_region[region] = metric["queue_depth"]
203 spot_data = capacity_data.get("spot_data", {}) or {}
204 capacity_blocks = capacity_data.get("capacity_blocks", {}) or {}
206 pairs: set[tuple[str, str]] = set()
207 for itype, regions in spot_data.items():
208 for region in regions or {}:
209 pairs.add((itype, region))
210 for itype, regions in capacity_blocks.items():
211 for region in regions or {}:
212 pairs.add((itype, region))
214 records: list[dict[str, Any]] = []
215 for itype, region in sorted(pairs):
216 spot_info = (spot_data.get(itype, {}) or {}).get(region, {}) or {}
217 prices = spot_info.get("prices", []) or []
218 scores = spot_info.get("placement_scores", {}) or {}
220 record: dict[str, Any] = {
221 "instance_type": itype,
222 "region": region,
223 "timestamp": timestamp,
224 }
226 regional_score = scores.get("regional")
227 if regional_score is not None:
228 record["spot_score"] = regional_score
230 current_prices = [p["current"] for p in prices if p.get("current") is not None]
231 if current_prices:
232 record["spot_price"] = round(sum(current_prices) / len(current_prices), 6)
233 record["az_count"] = len(current_prices)
235 if region in queue_by_region:
236 record["queue_depth"] = queue_by_region[region]
238 blocks = (capacity_blocks.get(itype, {}) or {}).get(region, []) or []
239 if blocks:
240 record["capacity_blocks_available"] = len(blocks)
241 record["capacity_blocks_total"] = len(blocks)
243 if any(field in record for field in METRIC_FIELDS):
244 records.append(record)
246 return records
249def _resolve_global_region() -> str:
250 """Resolve the region where the capacity-history table lives.
252 The table is created by the global stack, so it lives in the GCO global
253 region. Resolve that from the CLI config (which reads cdk.json and
254 GCO_GLOBAL_REGION) so callers don't need to set DYNAMODB_REGION. Falls back
255 to us-east-1 only if the config cannot be loaded.
256 """
257 try:
258 from cli.config import get_config
260 return get_config().global_region or "us-east-1"
261 except Exception:
262 return "us-east-1"
265def _resolve_default_table_name() -> str:
266 """Resolve the capacity-history table name from the configured project (#139).
268 The global stack creates the table as ``{project_name}-capacity-history``
269 (see ``GCOGlobalStack._create_capacity_poller``), so derive the same name
270 from the CLI config (which reads cdk.json / ``GCO_PROJECT_NAME``). Falls
271 back to the default ``gco-capacity-history`` when the config can't be
272 loaded. For the default ``gco`` project this is byte-identical.
273 """
274 try:
275 from cli.config import get_config
277 project = get_config().project_name
278 if project:
279 return f"{project}-capacity-history"
280 except Exception as exc:
281 logger.debug("Falling back to default capacity history table name: %s", exc)
282 return DEFAULT_TABLE_NAME
285class CapacityHistoryStore:
286 """DynamoDB-backed time-series store for capacity snapshots.
288 Table name resolves from the CAPACITY_HISTORY_TABLE_NAME env var (default
289 gco-capacity-history). Region resolves from an explicit argument, then
290 DYNAMODB_REGION or REGION, then the configured GCO global region (where the
291 global stack creates the table), falling back to us-east-1. Retention
292 defaults to 90 days and feeds the per-item ttl.
293 """
295 def __init__(
296 self,
297 table_name: str | None = None,
298 region: str | None = None,
299 retention_days: int | None = None,
300 ):
301 self.table_name = (
302 table_name or os.getenv("CAPACITY_HISTORY_TABLE_NAME") or _resolve_default_table_name()
303 )
304 self._region = (
305 region
306 or os.getenv("DYNAMODB_REGION")
307 or os.getenv("REGION")
308 or _resolve_global_region()
309 )
310 if retention_days is not None:
311 self.retention_days = retention_days
312 else:
313 self.retention_days = int(
314 os.getenv("CAPACITY_HISTORY_RETENTION_DAYS", str(DEFAULT_RETENTION_DAYS))
315 )
316 self._dynamodb = boto3.resource("dynamodb", region_name=self._region)
317 self._table = self._dynamodb.Table(self.table_name)
319 def put_snapshot(
320 self,
321 instance_type: str,
322 region: str,
323 metrics: dict[str, Any],
324 *,
325 spot_pool: str | None = None,
326 timestamp: str | None = None,
327 now: datetime | None = None,
328 ) -> dict[str, Any]:
329 """Persist a single capacity snapshot and return the stored item.
331 metrics is filtered to METRIC_FIELDS; None values are dropped so an
332 absent metric is never stored as zero. spot_pool names the instance
333 pool the snapshot's Spot Placement Scores were requested for; it is a
334 string attribution attribute, not a metric, and is omitted when the
335 instance type is unpooled (no score was collected).
336 """
337 now = now or _utc_now()
338 ts = timestamp or now.isoformat()
339 ttl_epoch = int((now + timedelta(days=self.retention_days)).timestamp())
341 item: dict[str, Any] = {
342 "pk": make_pk(instance_type, region),
343 "sk": ts,
344 "instance_type": instance_type,
345 "region": region,
346 "timestamp": ts,
347 "ttl": ttl_epoch,
348 }
349 if spot_pool is not None:
350 item["spot_pool"] = spot_pool
351 for field in METRIC_FIELDS:
352 value = metrics.get(field)
353 if value is not None:
354 item[field] = value
356 self._table.put_item(Item=_to_dynamo(item))
357 stored: dict[str, Any] = _from_dynamo(item)
358 return stored
360 def record(self, capacity_data: dict[str, Any], *, now: datetime | None = None) -> int:
361 """Flatten gather_capacity_data() output and persist one item per
362 (instance_type, region) snapshot. Returns the number of items written.
363 """
364 now = now or _utc_now()
365 written = 0
366 for rec in flatten_capacity_data(capacity_data):
367 self.put_snapshot(
368 instance_type=rec["instance_type"],
369 region=rec["region"],
370 metrics={field: rec[field] for field in METRIC_FIELDS if field in rec},
371 timestamp=rec.get("timestamp"),
372 now=now,
373 )
374 written += 1
375 return written
377 def get_trend(
378 self,
379 instance_type: str,
380 region: str,
381 hours_back: int = 168,
382 ) -> list[dict[str, Any]]:
383 """Return snapshots for one (instance_type, region) within the window,
384 oldest first. Default window is 7 days (168 hours).
385 """
386 cutoff = (_utc_now() - timedelta(hours=hours_back)).isoformat()
387 items: list[dict[str, Any]] = []
388 kwargs: dict[str, Any] = {
389 "KeyConditionExpression": (
390 Key("pk").eq(make_pk(instance_type, region)) & Key("sk").gte(cutoff)
391 ),
392 "ScanIndexForward": True,
393 }
394 while True:
395 resp = self._table.query(**kwargs)
396 items.extend(_from_dynamo(i) for i in resp.get("Items", []))
397 last_key = resp.get("LastEvaluatedKey")
398 if not last_key:
399 break
400 kwargs["ExclusiveStartKey"] = last_key
401 return items
403 def get_statistics(
404 self,
405 instance_type: str,
406 region: str,
407 hours_back: int = 168,
408 ) -> dict[str, Any]:
409 """Compute p25/p50/p75/min/max/stddev (plus count/mean) per metric over
410 the window. Metrics with no data points are omitted.
411 """
412 trend = self.get_trend(instance_type, region, hours_back)
413 stats: dict[str, Any] = {
414 "instance_type": instance_type,
415 "region": region,
416 "hours_back": hours_back,
417 "sample_count": len(trend),
418 "metrics": {},
419 }
420 for field in METRIC_FIELDS:
421 values = [float(rec[field]) for rec in trend if rec.get(field) is not None]
422 if not values:
423 continue
424 values.sort()
425 stats["metrics"][field] = {
426 "count": len(values),
427 "min": values[0],
428 "max": values[-1],
429 "mean": round(statistics.fmean(values), 6),
430 "p25": round(_percentile(values, 25), 6),
431 "p50": round(_percentile(values, 50), 6),
432 "p75": round(_percentile(values, 75), 6),
433 "stddev": round(statistics.stdev(values), 6) if len(values) > 1 else 0.0,
434 }
435 return stats
437 def get_temporal_patterns(
438 self,
439 instance_type: str,
440 region: str,
441 hours_back: int = 168,
442 metric: str = "spot_score",
443 ) -> dict[str, Any]:
444 """Group snapshots by day-of-week and hour, returning the average of
445 metric (default spot_score) per (day, hour) slot, plus a best_windows
446 list sorted by descending average for prompt enrichment.
447 """
448 trend = self.get_trend(instance_type, region, hours_back)
450 buckets: dict[tuple[int, int], list[float]] = {}
451 for rec in trend:
452 value = rec.get(metric)
453 ts = rec.get("timestamp")
454 if value is None or not ts:
455 continue
456 dt = _parse_iso(str(ts))
457 if dt is None:
458 continue
459 buckets.setdefault((dt.weekday(), dt.hour), []).append(float(value))
461 patterns: dict[str, dict[int, dict[str, float]]] = {}
462 best_windows: list[dict[str, Any]] = []
463 for (dow, hour), values in buckets.items():
464 avg = round(statistics.fmean(values), 4)
465 day = DAY_NAMES[dow]
466 patterns.setdefault(day, {})[hour] = {"avg": avg, "count": len(values)}
467 best_windows.append({"day": day, "hour": hour, "avg": avg, "count": len(values)})
469 best_windows.sort(key=lambda window: window["avg"], reverse=True)
470 return {
471 "instance_type": instance_type,
472 "region": region,
473 "metric": metric,
474 "patterns": patterns,
475 "best_windows": best_windows,
476 }
478 def get_regions_with_data(
479 self,
480 instance_type: str,
481 hours_back: int = 168,
482 ) -> list[str]:
483 """Return the distinct regions that have snapshots for an instance type.
485 Queries the ``by-timestamp`` GSI (pk=instance_type) within the window
486 and collects the distinct ``region`` values, sorted alphabetically.
487 Powers cross-region queries such as ``gco capacity predict --all-regions``.
488 """
489 cutoff = (_utc_now() - timedelta(hours=hours_back)).isoformat()
490 regions: set[str] = set()
491 kwargs: dict[str, Any] = {
492 "IndexName": GSI_BY_TIMESTAMP,
493 "KeyConditionExpression": (
494 Key("instance_type").eq(instance_type) & Key("sk").gte(cutoff)
495 ),
496 "ProjectionExpression": "#r",
497 "ExpressionAttributeNames": {"#r": "region"},
498 }
499 while True:
500 resp = self._table.query(**kwargs)
501 for item in resp.get("Items", []):
502 value = item.get("region")
503 if value:
504 regions.add(str(value))
505 last_key = resp.get("LastEvaluatedKey")
506 if not last_key:
507 break
508 kwargs["ExclusiveStartKey"] = last_key
509 return sorted(regions)
512def get_capacity_history_store(
513 table_name: str | None = None,
514 region: str | None = None,
515 retention_days: int | None = None,
516) -> CapacityHistoryStore:
517 """Factory for CapacityHistoryStore."""
518 return CapacityHistoryStore(table_name=table_name, region=region, retention_days=retention_days)