Coverage for cli/capacity/history.py: 90.74%
196 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +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, spot_price,
20az_count, queue_depth, capacity_blocks_available, capacity_blocks_total,
21capacity_blocks_long_available, capacity_blocks_long_total, and ttl (epoch
22seconds for DynamoDB auto-expiry, default 90 days). The ``*_long_*`` fields
23track availability of extended-term blocks (the poller's long-duration probe,
24default 63 days) separately from the soonest short block.
26Numbers are stored as DynamoDB Decimal (the resource API rejects float) and
27re-hydrated to int/float on read. The poller Lambda is self-contained and
28writes the same item shape directly with boto3; this module is the canonical
29writer/reader used by the CLI and advisor.
30"""
32from __future__ import annotations
34import logging
35import os
36import statistics
37from datetime import UTC, datetime, timedelta
38from decimal import Decimal
39from typing import Any
41import boto3
42from boto3.dynamodb.conditions import Key
44logger = logging.getLogger(__name__)
46DEFAULT_TABLE_NAME = "gco-capacity-history"
47DEFAULT_RETENTION_DAYS = 90
48GSI_BY_TIMESTAMP = "by-timestamp"
50# The numeric metrics tracked per snapshot. Statistics and temporal-pattern
51# aggregation iterate over this tuple, so adding a metric is a one-line change.
52# The ``capacity_blocks_long_*`` pair mirrors the short-duration block metrics
53# but for the poller's long-duration probe (default 63 days), so trend/alerting
54# queries can distinguish soonest-available blocks from extended-term ones.
55METRIC_FIELDS: tuple[str, ...] = (
56 "spot_score",
57 "spot_price",
58 "az_count",
59 "queue_depth",
60 "capacity_blocks_available",
61 "capacity_blocks_total",
62 "capacity_blocks_long_available",
63 "capacity_blocks_long_total",
64)
66# weekday() -> name. Monday is 0, matching datetime.weekday().
67DAY_NAMES: tuple[str, ...] = (
68 "Monday",
69 "Tuesday",
70 "Wednesday",
71 "Thursday",
72 "Friday",
73 "Saturday",
74 "Sunday",
75)
78def _utc_now() -> datetime:
79 return datetime.now(UTC)
82def make_pk(instance_type: str, region: str) -> str:
83 """Build the partition key for a (instance_type, region) series."""
84 return f"{instance_type}#{region}"
87def _parse_iso(value: str) -> datetime | None:
88 """Parse an ISO-8601 timestamp, tolerating a trailing Z and naive values."""
89 try:
90 dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
91 except ValueError, AttributeError:
92 return None
93 if dt.tzinfo is None: 93 ↛ 94line 93 didn't jump to line 94 because the condition on line 93 was never true
94 dt = dt.replace(tzinfo=UTC)
95 return dt
98def _to_dynamo(value: Any) -> Any:
99 """Convert a Python value into a DynamoDB-storable type.
101 The DynamoDB resource API rejects float, so numbers must be Decimal.
102 Floats are routed through Decimal(str(x)) so the decimal string round-trips
103 without binary-float artifacts. bool is checked before the numeric branch
104 because bool is a subclass of int.
105 """
106 if isinstance(value, bool):
107 return value
108 if isinstance(value, float):
109 return Decimal(str(value))
110 if isinstance(value, dict):
111 return {k: _to_dynamo(v) for k, v in value.items()}
112 if isinstance(value, list): 112 ↛ 113line 112 didn't jump to line 113 because the condition on line 112 was never true
113 return [_to_dynamo(v) for v in value]
114 return value
117def _from_dynamo(value: Any) -> Any:
118 """Convert DynamoDB types back to plain Python (Decimal -> int/float)."""
119 if isinstance(value, Decimal):
120 return int(value) if value == int(value) else float(value)
121 if isinstance(value, dict):
122 return {k: _from_dynamo(v) for k, v in value.items()}
123 if isinstance(value, list): 123 ↛ 124line 123 didn't jump to line 124 because the condition on line 123 was never true
124 return [_from_dynamo(v) for v in value]
125 return value
128def _percentile(sorted_values: list[float], pct: float) -> float:
129 """Linear-interpolation percentile (pct in [0, 100]) over sorted values."""
130 if not sorted_values: 130 ↛ 131line 130 didn't jump to line 131 because the condition on line 130 was never true
131 raise ValueError("percentile of empty sequence")
132 if len(sorted_values) == 1:
133 return float(sorted_values[0])
134 rank = (pct / 100.0) * (len(sorted_values) - 1)
135 lo = int(rank)
136 hi = min(lo + 1, len(sorted_values) - 1)
137 frac = rank - lo
138 return float(sorted_values[lo] + (sorted_values[hi] - sorted_values[lo]) * frac)
141def flatten_capacity_data(capacity_data: dict[str, Any]) -> list[dict[str, Any]]:
142 """Flatten gather_capacity_data() output into per-(instance_type, region) records.
144 Returns one record per (instance_type, region) pair that carries at least
145 one signal. Each record holds the keys instance_type, region, timestamp
146 plus whatever subset of METRIC_FIELDS can be derived. Metrics that cannot
147 be derived are omitted (the statistics layer treats a missing metric as
148 absent, never as zero).
149 """
150 timestamp = capacity_data.get("timestamp") or _utc_now().isoformat()
152 queue_by_region: dict[str, Any] = {}
153 for metric in capacity_data.get("cluster_metrics", []) or []:
154 region = metric.get("region")
155 if region is not None and metric.get("queue_depth") is not None: 155 ↛ 153line 155 didn't jump to line 153 because the condition on line 155 was always true
156 queue_by_region[region] = metric["queue_depth"]
158 spot_data = capacity_data.get("spot_data", {}) or {}
159 capacity_blocks = capacity_data.get("capacity_blocks", {}) or {}
161 pairs: set[tuple[str, str]] = set()
162 for itype, regions in spot_data.items():
163 for region in regions or {}:
164 pairs.add((itype, region))
165 for itype, regions in capacity_blocks.items():
166 for region in regions or {}:
167 pairs.add((itype, region))
169 records: list[dict[str, Any]] = []
170 for itype, region in sorted(pairs):
171 spot_info = (spot_data.get(itype, {}) or {}).get(region, {}) or {}
172 prices = spot_info.get("prices", []) or []
173 scores = spot_info.get("placement_scores", {}) or {}
175 record: dict[str, Any] = {
176 "instance_type": itype,
177 "region": region,
178 "timestamp": timestamp,
179 }
181 regional_score = scores.get("regional")
182 if regional_score is not None: 182 ↛ 185line 182 didn't jump to line 185 because the condition on line 182 was always true
183 record["spot_score"] = regional_score
185 current_prices = [p["current"] for p in prices if p.get("current") is not None]
186 if current_prices: 186 ↛ 190line 186 didn't jump to line 190 because the condition on line 186 was always true
187 record["spot_price"] = round(sum(current_prices) / len(current_prices), 6)
188 record["az_count"] = len(current_prices)
190 if region in queue_by_region: 190 ↛ 193line 190 didn't jump to line 193 because the condition on line 190 was always true
191 record["queue_depth"] = queue_by_region[region]
193 blocks = (capacity_blocks.get(itype, {}) or {}).get(region, []) or []
194 if blocks: 194 ↛ 198line 194 didn't jump to line 198 because the condition on line 194 was always true
195 record["capacity_blocks_available"] = len(blocks)
196 record["capacity_blocks_total"] = len(blocks)
198 if any(field in record for field in METRIC_FIELDS): 198 ↛ 170line 198 didn't jump to line 170 because the condition on line 198 was always true
199 records.append(record)
201 return records
204def _resolve_global_region() -> str:
205 """Resolve the region where the capacity-history table lives.
207 The table is created by the global stack, so it lives in the GCO global
208 region. Resolve that from the CLI config (which reads cdk.json and
209 GCO_GLOBAL_REGION) so callers don't need to set DYNAMODB_REGION. Falls back
210 to us-east-1 only if the config cannot be loaded.
211 """
212 try:
213 from cli.config import get_config
215 return get_config().global_region or "us-east-1"
216 except Exception:
217 return "us-east-1"
220def _resolve_default_table_name() -> str:
221 """Resolve the capacity-history table name from the configured project (#139).
223 The global stack creates the table as ``{project_name}-capacity-history``
224 (see ``GCOGlobalStack._create_capacity_poller``), so derive the same name
225 from the CLI config (which reads cdk.json / ``GCO_PROJECT_NAME``). Falls
226 back to the default ``gco-capacity-history`` when the config can't be
227 loaded. For the default ``gco`` project this is byte-identical.
228 """
229 try:
230 from cli.config import get_config
232 project = get_config().project_name
233 if project: 233 ↛ 237line 233 didn't jump to line 237 because the condition on line 233 was always true
234 return f"{project}-capacity-history"
235 except Exception as exc:
236 logger.debug("Falling back to default capacity history table name: %s", exc)
237 return DEFAULT_TABLE_NAME
240class CapacityHistoryStore:
241 """DynamoDB-backed time-series store for capacity snapshots.
243 Table name resolves from the CAPACITY_HISTORY_TABLE_NAME env var (default
244 gco-capacity-history). Region resolves from an explicit argument, then
245 DYNAMODB_REGION or REGION, then the configured GCO global region (where the
246 global stack creates the table), falling back to us-east-1. Retention
247 defaults to 90 days and feeds the per-item ttl.
248 """
250 def __init__(
251 self,
252 table_name: str | None = None,
253 region: str | None = None,
254 retention_days: int | None = None,
255 ):
256 self.table_name = (
257 table_name or os.getenv("CAPACITY_HISTORY_TABLE_NAME") or _resolve_default_table_name()
258 )
259 self._region = (
260 region
261 or os.getenv("DYNAMODB_REGION")
262 or os.getenv("REGION")
263 or _resolve_global_region()
264 )
265 if retention_days is not None:
266 self.retention_days = retention_days
267 else:
268 self.retention_days = int(
269 os.getenv("CAPACITY_HISTORY_RETENTION_DAYS", str(DEFAULT_RETENTION_DAYS))
270 )
271 self._dynamodb = boto3.resource("dynamodb", region_name=self._region)
272 self._table = self._dynamodb.Table(self.table_name)
274 def put_snapshot(
275 self,
276 instance_type: str,
277 region: str,
278 metrics: dict[str, Any],
279 *,
280 timestamp: str | None = None,
281 now: datetime | None = None,
282 ) -> dict[str, Any]:
283 """Persist a single capacity snapshot and return the stored item.
285 metrics is filtered to METRIC_FIELDS; None values are dropped so an
286 absent metric is never stored as zero.
287 """
288 now = now or _utc_now()
289 ts = timestamp or now.isoformat()
290 ttl_epoch = int((now + timedelta(days=self.retention_days)).timestamp())
292 item: dict[str, Any] = {
293 "pk": make_pk(instance_type, region),
294 "sk": ts,
295 "instance_type": instance_type,
296 "region": region,
297 "timestamp": ts,
298 "ttl": ttl_epoch,
299 }
300 for field in METRIC_FIELDS:
301 value = metrics.get(field)
302 if value is not None:
303 item[field] = value
305 self._table.put_item(Item=_to_dynamo(item))
306 stored: dict[str, Any] = _from_dynamo(item)
307 return stored
309 def record(self, capacity_data: dict[str, Any], *, now: datetime | None = None) -> int:
310 """Flatten gather_capacity_data() output and persist one item per
311 (instance_type, region) snapshot. Returns the number of items written.
312 """
313 now = now or _utc_now()
314 written = 0
315 for rec in flatten_capacity_data(capacity_data):
316 self.put_snapshot(
317 instance_type=rec["instance_type"],
318 region=rec["region"],
319 metrics={field: rec[field] for field in METRIC_FIELDS if field in rec},
320 timestamp=rec.get("timestamp"),
321 now=now,
322 )
323 written += 1
324 return written
326 def get_trend(
327 self,
328 instance_type: str,
329 region: str,
330 hours_back: int = 168,
331 ) -> list[dict[str, Any]]:
332 """Return snapshots for one (instance_type, region) within the window,
333 oldest first. Default window is 7 days (168 hours).
334 """
335 cutoff = (_utc_now() - timedelta(hours=hours_back)).isoformat()
336 items: list[dict[str, Any]] = []
337 kwargs: dict[str, Any] = {
338 "KeyConditionExpression": (
339 Key("pk").eq(make_pk(instance_type, region)) & Key("sk").gte(cutoff)
340 ),
341 "ScanIndexForward": True,
342 }
343 while True:
344 resp = self._table.query(**kwargs)
345 items.extend(_from_dynamo(i) for i in resp.get("Items", []))
346 last_key = resp.get("LastEvaluatedKey")
347 if not last_key:
348 break
349 kwargs["ExclusiveStartKey"] = last_key
350 return items
352 def get_statistics(
353 self,
354 instance_type: str,
355 region: str,
356 hours_back: int = 168,
357 ) -> dict[str, Any]:
358 """Compute p25/p50/p75/min/max/stddev (plus count/mean) per metric over
359 the window. Metrics with no data points are omitted.
360 """
361 trend = self.get_trend(instance_type, region, hours_back)
362 stats: dict[str, Any] = {
363 "instance_type": instance_type,
364 "region": region,
365 "hours_back": hours_back,
366 "sample_count": len(trend),
367 "metrics": {},
368 }
369 for field in METRIC_FIELDS:
370 values = [float(rec[field]) for rec in trend if rec.get(field) is not None]
371 if not values:
372 continue
373 values.sort()
374 stats["metrics"][field] = {
375 "count": len(values),
376 "min": values[0],
377 "max": values[-1],
378 "mean": round(statistics.fmean(values), 6),
379 "p25": round(_percentile(values, 25), 6),
380 "p50": round(_percentile(values, 50), 6),
381 "p75": round(_percentile(values, 75), 6),
382 "stddev": round(statistics.stdev(values), 6) if len(values) > 1 else 0.0,
383 }
384 return stats
386 def get_temporal_patterns(
387 self,
388 instance_type: str,
389 region: str,
390 hours_back: int = 168,
391 metric: str = "spot_score",
392 ) -> dict[str, Any]:
393 """Group snapshots by day-of-week and hour, returning the average of
394 metric (default spot_score) per (day, hour) slot, plus a best_windows
395 list sorted by descending average for prompt enrichment.
396 """
397 trend = self.get_trend(instance_type, region, hours_back)
399 buckets: dict[tuple[int, int], list[float]] = {}
400 for rec in trend:
401 value = rec.get(metric)
402 ts = rec.get("timestamp")
403 if value is None or not ts: 403 ↛ 404line 403 didn't jump to line 404 because the condition on line 403 was never true
404 continue
405 dt = _parse_iso(str(ts))
406 if dt is None: 406 ↛ 407line 406 didn't jump to line 407 because the condition on line 406 was never true
407 continue
408 buckets.setdefault((dt.weekday(), dt.hour), []).append(float(value))
410 patterns: dict[str, dict[int, dict[str, float]]] = {}
411 best_windows: list[dict[str, Any]] = []
412 for (dow, hour), values in buckets.items():
413 avg = round(statistics.fmean(values), 4)
414 day = DAY_NAMES[dow]
415 patterns.setdefault(day, {})[hour] = {"avg": avg, "count": len(values)}
416 best_windows.append({"day": day, "hour": hour, "avg": avg, "count": len(values)})
418 best_windows.sort(key=lambda window: window["avg"], reverse=True)
419 return {
420 "instance_type": instance_type,
421 "region": region,
422 "metric": metric,
423 "patterns": patterns,
424 "best_windows": best_windows,
425 }
427 def get_regions_with_data(
428 self,
429 instance_type: str,
430 hours_back: int = 168,
431 ) -> list[str]:
432 """Return the distinct regions that have snapshots for an instance type.
434 Queries the ``by-timestamp`` GSI (pk=instance_type) within the window
435 and collects the distinct ``region`` values, sorted alphabetically.
436 Powers cross-region queries such as ``gco capacity predict --all-regions``.
437 """
438 cutoff = (_utc_now() - timedelta(hours=hours_back)).isoformat()
439 regions: set[str] = set()
440 kwargs: dict[str, Any] = {
441 "IndexName": GSI_BY_TIMESTAMP,
442 "KeyConditionExpression": (
443 Key("instance_type").eq(instance_type) & Key("sk").gte(cutoff)
444 ),
445 "ProjectionExpression": "#r",
446 "ExpressionAttributeNames": {"#r": "region"},
447 }
448 while True:
449 resp = self._table.query(**kwargs)
450 for item in resp.get("Items", []):
451 value = item.get("region")
452 if value: 452 ↛ 450line 452 didn't jump to line 450 because the condition on line 452 was always true
453 regions.add(str(value))
454 last_key = resp.get("LastEvaluatedKey")
455 if not last_key:
456 break
457 kwargs["ExclusiveStartKey"] = last_key
458 return sorted(regions)
461def get_capacity_history_store(
462 table_name: str | None = None,
463 region: str | None = None,
464 retention_days: int | None = None,
465) -> CapacityHistoryStore:
466 """Factory for CapacityHistoryStore."""
467 return CapacityHistoryStore(table_name=table_name, region=region, retention_days=retention_days)