Coverage for gco/services/cost_monitor.py: 99.57%
195 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +0000
1"""Cost Monitor service core: OpenCost allocation reports as Parquet in S3.
3The cost-monitor Deployment (one per regional cluster) runs two surfaces on
4top of this module:
61. A scheduled reporter (driven by :mod:`gco.services.cost_api`) that writes
7 one Parquet allocation report per interval to the central cost report
8 bucket under ``reports/region=<region>/date=<YYYY-MM-DD>/`` — the layout
9 the monitoring stack's Glue table reads with partition projection.
102. An internal HTTP API the manifest processor proxies as ``/api/v1/cost/*``,
11 serving ad-hoc report generation (written under ``adhoc/`` so overlapping
12 windows never double-count in Athena) plus report listing and service
13 status.
15Report object keys for scheduled windows are **deterministic** — derived only
16from the window bounds — so a rollout overlap or retry can never produce two
17objects for one window: concurrent writers converge on the same key and the
18last write wins with identical content.
19"""
21from __future__ import annotations
23import io
24import logging
25import math
26import os
27import uuid
28from dataclasses import dataclass, field
29from datetime import UTC, datetime, timedelta
30from typing import Any
32import boto3
33import httpx
34from botocore.config import Config
36logger = logging.getLogger(__name__)
38#: Normalized report row fields, in Parquet column order. This is the
39#: write-side contract of the monitoring stack's Glue table
40#: (``gco/stacks/monitoring_stack.py::_create_cost_analytics``) — the two
41#: must stay in lockstep or Athena reads misaligned columns.
42ALLOCATION_REPORT_FIELDS: tuple[str, ...] = (
43 "window_start",
44 "window_end",
45 "cluster",
46 "namespace",
47 "cpu_core_hours",
48 "cpu_cost",
49 "ram_gib_hours",
50 "ram_cost",
51 "gpu_hours",
52 "gpu_cost",
53 "pv_cost",
54 "network_cost",
55 "load_balancer_cost",
56 "shared_cost",
57 "external_cost",
58 "total_cost",
59 "total_efficiency",
60)
62_GIB = 1024.0**3
64#: Prefixes must mirror gco/stacks/constants.py (the service image does not
65#: ship the CDK stacks package, so the values are duplicated deliberately —
66#: a synth-side test asserts the two stay in lockstep).
67SCHEDULED_PREFIX = "reports"
68ADHOC_PREFIX = "adhoc"
70_MIN_WINDOW_MINUTES = 5
71_MAX_WINDOW_HOURS = 7 * 24
74class OpenCostUnavailableError(RuntimeError):
75 """Raised when the OpenCost API cannot be reached or answers abnormally."""
78class ReportWriteError(RuntimeError):
79 """Raised when a generated report cannot be persisted to S3."""
82def _compact_ts(moment: datetime) -> str:
83 """Render a UTC timestamp as a compact S3-key-safe token."""
84 return moment.astimezone(UTC).strftime("%Y%m%dT%H%M%SZ")
87def _as_float(value: Any) -> float:
88 """Coerce OpenCost numeric fields defensively; absent/bad values are 0.
90 Non-finite values (NaN and ±infinity — ``json.loads`` accepts both) are
91 coerced to 0 too: one poisoned row would otherwise contaminate every
92 Athena aggregate over the table.
93 """
94 try:
95 result = float(value)
96 except TypeError, ValueError:
97 return 0.0
98 return result if math.isfinite(result) else 0.0
101@dataclass(frozen=True)
102class ReportResult:
103 """Outcome of one generated allocation report."""
105 s3_key: str
106 row_count: int
107 total_cost: float
108 window_start: str
109 window_end: str
110 rows: list[dict[str, Any]] = field(default_factory=list)
112 def summary(self) -> dict[str, Any]:
113 """Operator-facing summary without the full row payload."""
114 return {
115 "s3_key": self.s3_key,
116 "row_count": self.row_count,
117 "total_cost": round(self.total_cost, 6),
118 "window_start": self.window_start,
119 "window_end": self.window_end,
120 }
123class OpenCostClient:
124 """Minimal HTTP client for the in-cluster OpenCost allocation API."""
126 def __init__(self, base_url: str, timeout_seconds: float = 30.0) -> None:
127 self.base_url = base_url.rstrip("/")
128 self.timeout_seconds = timeout_seconds
130 def is_healthy(self) -> bool:
131 """Return whether OpenCost answers its /healthz probe."""
132 try:
133 response = httpx.get(
134 f"{self.base_url}/healthz",
135 timeout=self.timeout_seconds,
136 )
137 except httpx.HTTPError:
138 return False
139 return response.status_code == 200
141 def get_allocation(
142 self,
143 window_start: datetime,
144 window_end: datetime,
145 *,
146 aggregate: str = "namespace",
147 ) -> dict[str, dict[str, Any]]:
148 """Fetch one accumulated allocation set for ``[window_start, window_end)``.
150 Returns a mapping of allocation name (namespace, by default) to the
151 raw OpenCost allocation object. Raises
152 :class:`OpenCostUnavailableError` on transport errors, non-200
153 responses, or a malformed body — the caller decides whether that
154 fails a scheduled pass or an API request.
155 """
156 window = (
157 f"{window_start.astimezone(UTC).strftime('%Y-%m-%dT%H:%M:%SZ')},"
158 f"{window_end.astimezone(UTC).strftime('%Y-%m-%dT%H:%M:%SZ')}"
159 )
160 try:
161 response = httpx.get(
162 f"{self.base_url}/allocation/compute",
163 params={
164 "window": window,
165 "aggregate": aggregate,
166 "accumulate": "true",
167 },
168 timeout=self.timeout_seconds,
169 )
170 except httpx.HTTPError as exc:
171 raise OpenCostUnavailableError(f"OpenCost request failed: {exc}") from exc
172 if response.status_code != 200:
173 raise OpenCostUnavailableError(
174 f"OpenCost allocation query returned HTTP {response.status_code}"
175 )
176 try:
177 payload = response.json()
178 except ValueError as exc:
179 raise OpenCostUnavailableError("OpenCost returned a non-JSON body") from exc
180 data = payload.get("data")
181 if not isinstance(data, list):
182 raise OpenCostUnavailableError("OpenCost allocation response omitted data")
183 merged: dict[str, dict[str, Any]] = {}
184 for allocation_set in data:
185 if not isinstance(allocation_set, dict):
186 continue
187 for name, allocation in allocation_set.items():
188 if isinstance(allocation, dict): 188 ↛ 187line 188 didn't jump to line 187 because the condition on line 188 was always true
189 merged[str(name)] = allocation
190 return merged
193def allocations_to_rows(
194 allocations: dict[str, dict[str, Any]],
195 *,
196 cluster: str,
197 window_start: datetime,
198 window_end: datetime,
199) -> list[dict[str, Any]]:
200 """Normalize raw OpenCost allocations into the stable report row schema.
202 One row per allocation name (namespace). The ``__idle__`` and
203 ``__unallocated__`` synthetic allocations OpenCost emits are kept —
204 idle cost is exactly the visibility a cost report exists to provide —
205 but rows are sorted by descending total cost for human-readable output.
206 """
207 start_iso = window_start.astimezone(UTC).isoformat()
208 end_iso = window_end.astimezone(UTC).isoformat()
209 rows: list[dict[str, Any]] = []
210 for name, allocation in allocations.items():
211 rows.append(
212 {
213 "window_start": start_iso,
214 "window_end": end_iso,
215 "cluster": cluster,
216 "namespace": name,
217 "cpu_core_hours": _as_float(allocation.get("cpuCoreHours")),
218 "cpu_cost": _as_float(allocation.get("cpuCost")),
219 "ram_gib_hours": _as_float(allocation.get("ramByteHours")) / _GIB,
220 "ram_cost": _as_float(allocation.get("ramCost")),
221 "gpu_hours": _as_float(allocation.get("gpuHours")),
222 "gpu_cost": _as_float(allocation.get("gpuCost")),
223 "pv_cost": _as_float(allocation.get("pvCost")),
224 "network_cost": _as_float(allocation.get("networkCost")),
225 "load_balancer_cost": _as_float(allocation.get("loadBalancerCost")),
226 "shared_cost": _as_float(allocation.get("sharedCost")),
227 "external_cost": _as_float(allocation.get("externalCost")),
228 "total_cost": _as_float(allocation.get("totalCost")),
229 "total_efficiency": _as_float(allocation.get("totalEfficiency")),
230 }
231 )
232 rows.sort(key=lambda row: row["total_cost"], reverse=True)
233 return rows
236def rows_to_parquet_bytes(rows: list[dict[str, Any]]) -> bytes:
237 """Serialize normalized report rows to a Parquet byte payload.
239 ``pyarrow`` is imported lazily so environments that never write reports
240 (unit tests exercising only transformations, or a future reader-only
241 consumer) do not need the dependency at import time. The window bound
242 columns are stored as real timestamps so the Glue ``timestamp`` columns
243 read them natively.
244 """
245 try:
246 import pyarrow as pa
247 import pyarrow.parquet as pq
248 except ImportError as exc: # pragma: no cover - image always ships pyarrow
249 raise ReportWriteError(
250 "pyarrow is required to write cost reports; install the "
251 "image-cost-monitor dependency group"
252 ) from exc
254 schema = pa.schema(
255 [
256 ("window_start", pa.timestamp("ms", tz="UTC")),
257 ("window_end", pa.timestamp("ms", tz="UTC")),
258 ("cluster", pa.string()),
259 ("namespace", pa.string()),
260 *[
261 (name, pa.float64())
262 for name in ALLOCATION_REPORT_FIELDS
263 if name not in {"window_start", "window_end", "cluster", "namespace"}
264 ],
265 ]
266 )
267 columns: dict[str, list[Any]] = {name: [] for name in ALLOCATION_REPORT_FIELDS}
268 for row in rows:
269 for name in ALLOCATION_REPORT_FIELDS:
270 value: Any = row.get(name)
271 if name in {"window_start", "window_end"}:
272 value = datetime.fromisoformat(str(value))
273 columns[name].append(value)
274 table = pa.Table.from_pydict(columns, schema=schema)
275 sink = io.BytesIO()
276 # pyarrow ships no type stubs; route the call through an Any-typed name so
277 # environments with pyarrow installed (tests) and without it (the mypy
278 # strict CI job) type-check identically — no conditional type: ignore.
279 write_table: Any = pq.write_table
280 write_table(table, sink)
281 return sink.getvalue()
284def scheduled_report_key(region: str, window_start: datetime, window_end: datetime) -> str:
285 """Deterministic S3 key for one scheduled report window."""
286 date_partition = window_start.astimezone(UTC).strftime("%Y-%m-%d")
287 return (
288 f"{SCHEDULED_PREFIX}/region={region}/date={date_partition}/"
289 f"allocation-{_compact_ts(window_start)}-{_compact_ts(window_end)}.parquet"
290 )
293def adhoc_report_key(region: str, window_start: datetime, window_end: datetime) -> str:
294 """Unique S3 key for one ad-hoc report (kept out of the Athena table)."""
295 date_partition = datetime.now(UTC).strftime("%Y-%m-%d")
296 return (
297 f"{ADHOC_PREFIX}/region={region}/date={date_partition}/"
298 f"allocation-{_compact_ts(window_start)}-{_compact_ts(window_end)}"
299 f"-{uuid.uuid4().hex[:8]}.parquet"
300 )
303def aligned_window(now: datetime, interval_minutes: int) -> tuple[datetime, datetime]:
304 """Return the most recent *completed* interval-aligned window.
306 For ``interval_minutes=60`` at 10:25 this yields ``[09:00, 10:00)`` —
307 aligning to interval boundaries makes the scheduled key deterministic
308 across restarts and replicas, which is what makes report writes
309 idempotent.
310 """
311 interval = timedelta(minutes=interval_minutes)
312 epoch = datetime(1970, 1, 1, tzinfo=UTC)
313 elapsed = now.astimezone(UTC) - epoch
314 completed_intervals = int(elapsed / interval)
315 window_end = epoch + interval * completed_intervals
316 return window_end - interval, window_end
319class CostMonitor:
320 """Generates OpenCost allocation reports and persists them to S3."""
322 def __init__(
323 self,
324 *,
325 region: str,
326 cluster: str,
327 bucket: str,
328 opencost: OpenCostClient,
329 report_interval_minutes: int = 60,
330 s3_client: Any | None = None,
331 ) -> None:
332 self.region = region
333 self.cluster = cluster
334 self.bucket = bucket
335 self.opencost = opencost
336 self.report_interval_minutes = min(max(int(report_interval_minutes), 5), 1_440)
337 self._s3 = s3_client or boto3.client(
338 "s3",
339 config=Config(
340 connect_timeout=5,
341 read_timeout=60,
342 retries={"max_attempts": 3, "mode": "standard"},
343 ),
344 )
345 self.last_scheduled_report: dict[str, Any] | None = None
346 self.last_error: str | None = None
348 # ------------------------------------------------------------------
349 # Report generation
350 # ------------------------------------------------------------------
352 def generate_report(
353 self,
354 window_start: datetime,
355 window_end: datetime,
356 *,
357 adhoc: bool,
358 include_rows: bool = False,
359 ) -> ReportResult:
360 """Query OpenCost for one window, write Parquet to S3, return the result.
362 Raises :class:`OpenCostUnavailableError` when OpenCost cannot answer
363 and :class:`ReportWriteError` when S3 persistence fails; callers map
364 those to a failed scheduled pass or an HTTP 502/503 respectively.
365 """
366 if window_end <= window_start:
367 raise ValueError("window_end must be after window_start")
368 if window_end - window_start > timedelta(hours=_MAX_WINDOW_HOURS):
369 raise ValueError(f"report windows are capped at {_MAX_WINDOW_HOURS} hours")
370 if window_end - window_start < timedelta(minutes=_MIN_WINDOW_MINUTES):
371 raise ValueError(f"report windows must span at least {_MIN_WINDOW_MINUTES} minutes")
373 allocations = self.opencost.get_allocation(window_start, window_end)
374 rows = allocations_to_rows(
375 allocations,
376 cluster=self.cluster,
377 window_start=window_start,
378 window_end=window_end,
379 )
380 key = (
381 adhoc_report_key(self.region, window_start, window_end)
382 if adhoc
383 else scheduled_report_key(self.region, window_start, window_end)
384 )
385 payload = rows_to_parquet_bytes(rows)
386 try:
387 self._s3.put_object(Bucket=self.bucket, Key=key, Body=payload)
388 except Exception as exc: # noqa: BLE001 - boto surfaces many shapes
389 raise ReportWriteError(f"Failed to write cost report to S3: {exc}") from exc
391 return ReportResult(
392 s3_key=key,
393 row_count=len(rows),
394 total_cost=sum(row["total_cost"] for row in rows),
395 window_start=window_start.astimezone(UTC).isoformat(),
396 window_end=window_end.astimezone(UTC).isoformat(),
397 rows=rows if include_rows else [],
398 )
400 def run_scheduled_once(self, now: datetime | None = None) -> ReportResult | None:
401 """Write the report for the most recent completed aligned window.
403 Skips (returns ``None``) when that window's object already exists —
404 the previous pass, or another replica during a rollout, already
405 persisted it. Failures update ``last_error`` and re-raise so the
406 caller's loop logs and retries on the next tick.
407 """
408 moment = now or datetime.now(UTC)
409 window_start, window_end = aligned_window(moment, self.report_interval_minutes)
410 key = scheduled_report_key(self.region, window_start, window_end)
411 if self._object_exists(key):
412 logger.debug("Scheduled cost report already present: %s", key)
413 return None
414 try:
415 result = self.generate_report(window_start, window_end, adhoc=False)
416 except Exception as exc:
417 self.last_error = str(exc)
418 raise
419 self.last_scheduled_report = result.summary()
420 self.last_error = None
421 logger.info(
422 "Wrote scheduled cost report %s (%d rows, total %.4f USD)",
423 result.s3_key,
424 result.row_count,
425 result.total_cost,
426 )
427 return result
429 def _object_exists(self, key: str) -> bool:
430 try:
431 self._s3.head_object(Bucket=self.bucket, Key=key)
432 except Exception: # noqa: BLE001 - 404 and transport errors both mean "write it"
433 return False
434 return True
436 # ------------------------------------------------------------------
437 # Introspection for the API surface
438 # ------------------------------------------------------------------
440 def list_reports(self, *, adhoc: bool = False, limit: int = 50) -> list[dict[str, Any]]:
441 """List this region's most recent report objects, newest first."""
442 prefix = (
443 f"{ADHOC_PREFIX}/region={self.region}/"
444 if adhoc
445 else f"{SCHEDULED_PREFIX}/region={self.region}/"
446 )
447 bounded_limit = min(max(int(limit), 1), 1_000)
448 paginator = self._s3.get_paginator("list_objects_v2")
449 objects: list[dict[str, Any]] = []
450 for page in paginator.paginate(Bucket=self.bucket, Prefix=prefix):
451 for entry in page.get("Contents", []):
452 objects.append(
453 {
454 "key": entry["Key"],
455 "size_bytes": int(entry.get("Size", 0)),
456 "last_modified": (
457 entry["LastModified"].astimezone(UTC).isoformat()
458 if entry.get("LastModified")
459 else None
460 ),
461 }
462 )
463 objects.sort(key=lambda item: str(item["last_modified"] or ""), reverse=True)
464 return objects[:bounded_limit]
466 def status(self) -> dict[str, Any]:
467 """Operator-facing service status, including OpenCost health.
469 ``opencost_returning_data`` performs a live one-hour allocation probe
470 — this is the signal release validation gates on, so a healthy-but-
471 empty OpenCost (e.g. Prometheus scrape broken) fails validation
472 rather than silently producing empty reports.
473 """
474 opencost_healthy = self.opencost.is_healthy()
475 returning_data = False
476 allocation_names: list[str] = []
477 if opencost_healthy:
478 try:
479 now = datetime.now(UTC)
480 allocations = self.opencost.get_allocation(now - timedelta(hours=1), now)
481 allocation_names = sorted(allocations)
482 returning_data = bool(allocations)
483 except OpenCostUnavailableError as exc:
484 logger.warning("OpenCost allocation probe failed: %s", exc)
485 return {
486 "service": "cost-monitor",
487 "region": self.region,
488 "cluster": self.cluster,
489 "bucket": self.bucket,
490 "report_interval_minutes": self.report_interval_minutes,
491 "opencost_healthy": opencost_healthy,
492 "opencost_returning_data": returning_data,
493 "allocation_names": allocation_names[:25],
494 "last_scheduled_report": self.last_scheduled_report,
495 "last_error": self.last_error,
496 "timestamp": datetime.now(UTC).isoformat(),
497 }
500def create_cost_monitor_from_env() -> CostMonitor:
501 """Build a :class:`CostMonitor` from the Deployment's environment."""
502 bucket = os.getenv("COST_REPORT_BUCKET", "")
503 if not bucket:
504 raise RuntimeError("COST_REPORT_BUCKET environment variable is required")
505 region = os.getenv("REGION") or os.getenv("AWS_REGION", "")
506 if not region:
507 raise RuntimeError("REGION environment variable is required")
508 cluster = os.getenv("CLUSTER_NAME", f"gco-{region}")
509 base_url = os.getenv(
510 "OPENCOST_BASE_URL",
511 "http://opencost.monitoring.svc.cluster.local:9003",
512 )
513 try:
514 interval = int(os.getenv("COST_REPORT_INTERVAL_MINUTES", "60"))
515 except ValueError:
516 interval = 60
517 return CostMonitor(
518 region=region,
519 cluster=cluster,
520 bucket=bucket,
521 opencost=OpenCostClient(base_url),
522 report_interval_minutes=interval,
523 )