Coverage for gco / services / cost_monitor.py: 100.00%
292 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +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.
20The bucket itself is *discovered*, not configured: its CloudFormation-generated
21name is published by the monitoring stack as an SSM parameter in the monitoring
22region (``COST_REPORT_BUCKET_PARAMETER`` / ``COST_REPORT_BUCKET_PARAMETER_REGION``),
23and that stack deploys after the regional stack this service runs in. Until the
24parameter exists the scheduled pass skips and the API answers 503; nothing here
25ever reconstructs a bucket name. ``COST_REPORT_BUCKET`` remains an explicit
26override for kind/CI.
28OpenCost is reached over verified HTTPS through the ``opencost-tls`` Service,
29a TLS sidecar in the OpenCost pod that serves a certificate issued by the GCO
30internal CA; the client trusts that CA alone (see
31:mod:`gco.services.internal_tls`). An ``http://`` ``OPENCOST_BASE_URL``
32override (kind/CI) opens no TLS session and needs no CA bundle.
33"""
35from __future__ import annotations
37import importlib
38import io
39import logging
40import math
41import os
42import time
43import uuid
44from collections.abc import Callable
45from dataclasses import dataclass, field
46from datetime import UTC, datetime, timedelta
47from typing import Any
49import boto3
50import httpx2
51from botocore.config import Config
53from gco.services import internal_tls, tracing
55logger = logging.getLogger(__name__)
57#: In-cluster OpenCost API behind the OpenCost pod's TLS sidecar.
58DEFAULT_OPENCOST_BASE_URL = "https://opencost-tls.monitoring.svc.cluster.local:9443"
60#: Normalized report row fields, in Parquet column order. This is the
61#: write-side contract of the monitoring stack's Glue table
62#: (``gco/stacks/monitoring_stack.py::_create_cost_analytics``) — the two
63#: must stay in lockstep or Athena reads misaligned columns.
64ALLOCATION_REPORT_FIELDS: tuple[str, ...] = (
65 "window_start",
66 "window_end",
67 "cluster",
68 "namespace",
69 "cpu_core_hours",
70 "cpu_cost",
71 "ram_gib_hours",
72 "ram_cost",
73 "gpu_hours",
74 "gpu_cost",
75 "pv_cost",
76 "network_cost",
77 "load_balancer_cost",
78 "shared_cost",
79 "external_cost",
80 "total_cost",
81 "total_efficiency",
82)
84_GIB = 1024.0**3
86#: Prefixes must mirror gco/stacks/constants.py (the service image does not
87#: ship the CDK stacks package, so the values are duplicated deliberately —
88#: a synth-side test asserts the two stay in lockstep).
89SCHEDULED_PREFIX = "reports"
90ADHOC_PREFIX = "adhoc"
92_MIN_WINDOW_MINUTES = 5
93_MAX_WINDOW_HOURS = 7 * 24
96class OpenCostUnavailableError(RuntimeError):
97 """Raised when the OpenCost API cannot be reached or answers abnormally."""
100class ReportWriteError(RuntimeError):
101 """Raised when a generated report cannot be persisted to S3."""
104class CostReportBucketUnavailableError(RuntimeError):
105 """Raised while the cost report bucket's identity is not yet resolvable.
107 Expected on a fresh deploy-all: the regional cost-monitor boots before the
108 monitoring stack publishes the bucket. Callers treat it as "not ready yet"
109 (skip the scheduled pass, answer HTTP 503) rather than as a write failure.
110 """
113#: How long a resolved bucket name is served before SSM is consulted again.
114#: A monitoring-stack redeploy that replaces the bucket updates the parameter;
115#: one GetParameter per refresh keeps the service converging on it.
116DEFAULT_BUCKET_REFRESH_SECONDS = 900.0
119class CostReportBucketLocator:
120 """Resolve the cost report bucket name from its published SSM parameter.
122 The monitoring stack writes ``/<project>/cost-report-bucket/name`` in the
123 monitoring region; the regional cost-monitor role holds ``ssm:GetParameter``
124 on exactly that parameter. Resolution is lazy and cached: the first
125 successful read is served for ``refresh_seconds``, a refresh that fails
126 keeps serving the last known name (a transient SSM error must not stall
127 report writes), and a read that never succeeded raises
128 :class:`CostReportBucketUnavailableError` so callers can back off.
129 """
131 def __init__(
132 self,
133 parameter_name: str,
134 region: str,
135 *,
136 ssm_client: Any | None = None,
137 refresh_seconds: float = DEFAULT_BUCKET_REFRESH_SECONDS,
138 clock: Callable[[], float] = time.monotonic,
139 ) -> None:
140 if not parameter_name.strip():
141 raise ValueError("parameter_name must not be empty")
142 if not region.strip():
143 raise ValueError("region must not be empty")
144 self.parameter_name = parameter_name
145 self.region = region
146 self.refresh_seconds = max(float(refresh_seconds), 0.0)
147 self._clock = clock
148 self._ssm = ssm_client or boto3.client(
149 "ssm",
150 region_name=region,
151 config=Config(
152 connect_timeout=5,
153 read_timeout=10,
154 retries={"max_attempts": 3, "mode": "standard"},
155 ),
156 )
157 self._cached: str | None = None
158 self._cached_at: float | None = None
160 @property
161 def cached(self) -> str | None:
162 """The last successfully resolved bucket name, or ``None`` before one."""
163 return self._cached
165 def resolve(self) -> str:
166 """Return the bucket name, reading SSM when the cache is empty or stale."""
167 now = self._clock()
168 if (
169 self._cached is not None
170 and self._cached_at is not None
171 and now - self._cached_at < self.refresh_seconds
172 ):
173 return self._cached
174 try:
175 response = self._ssm.get_parameter(Name=self.parameter_name)
176 except Exception as exc: # absence and transport errors converge
177 if self._cached is not None:
178 logger.warning(
179 "Refreshing cost report bucket from %s (%s) failed; keeping %s: %s",
180 self.parameter_name,
181 self.region,
182 self._cached,
183 exc,
184 )
185 self._cached_at = now
186 return self._cached
187 raise CostReportBucketUnavailableError(
188 f"Cost report bucket parameter {self.parameter_name} in {self.region} "
189 f"is not readable yet: {exc}"
190 ) from exc
191 parameter = response.get("Parameter") if isinstance(response, dict) else None
192 value = str((parameter or {}).get("Value") or "").strip()
193 if not value:
194 raise CostReportBucketUnavailableError(
195 f"Cost report bucket parameter {self.parameter_name} in {self.region} is empty"
196 )
197 if value != self._cached:
198 logger.info(
199 "Resolved cost report bucket %s from %s (%s)",
200 value,
201 self.parameter_name,
202 self.region,
203 )
204 self._cached = value
205 self._cached_at = now
206 return value
209def _compact_ts(moment: datetime) -> str:
210 """Render a UTC timestamp as a compact S3-key-safe token."""
211 return moment.astimezone(UTC).strftime("%Y%m%dT%H%M%SZ")
214def _as_float(value: Any) -> float:
215 """Coerce OpenCost numeric fields defensively; absent/bad values are 0.
217 Non-finite values (NaN and ±infinity — ``json.loads`` accepts both) are
218 coerced to 0 too: one poisoned row would otherwise contaminate every
219 Athena aggregate over the table.
220 """
221 try:
222 result = float(value)
223 except TypeError, ValueError:
224 return 0.0
225 return result if math.isfinite(result) else 0.0
228@dataclass(frozen=True)
229class ReportResult:
230 """Outcome of one generated allocation report."""
232 s3_key: str
233 row_count: int
234 total_cost: float
235 window_start: str
236 window_end: str
237 rows: list[dict[str, Any]] = field(default_factory=list)
239 def summary(self) -> dict[str, Any]:
240 """Operator-facing summary without the full row payload."""
241 return {
242 "s3_key": self.s3_key,
243 "row_count": self.row_count,
244 "total_cost": round(self.total_cost, 6),
245 "window_start": self.window_start,
246 "window_end": self.window_end,
247 }
250class OpenCostClient:
251 """Minimal HTTP client for the in-cluster OpenCost allocation API.
253 Every call builds its own ``httpx2.Client`` and closes it before
254 returning. The TLS trust is looked up then, through
255 :func:`~gco.services.internal_tls.verify_for_url`, whose cache hands back
256 the same context until the projected ``ca.crt`` changes and a fresh one
257 afterwards: a rotated internal CA is trusted from the next call without
258 restarting the pod, where a long-lived client would keep verifying
259 against the bundle it loaded first and report OpenCost unavailable until
260 a restart. Nothing is built at construction, so building the service
261 never touches the CA bundle: a bundle that is not mounted yet surfaces as
262 "OpenCost unhealthy" (and a failed allocation query) on each call until
263 it appears, instead of a crash at startup. The calls are a probe per
264 status request and one allocation query per report, so a connection per
265 call costs nothing that matters, and the scheduled reporter and the API
266 handlers can call in from worker threads without sharing any state.
267 """
269 def __init__(self, base_url: str, timeout_seconds: float = 30.0) -> None:
270 self.base_url = base_url.rstrip("/")
271 self.timeout_seconds = timeout_seconds
273 def _http(self) -> httpx2.Client:
274 """Build the client for one call, with the TLS trust current now.
276 The caller closes it. Raises
277 :class:`~gco.services.internal_tls.InternalTLSError` while an
278 ``https`` base URL has no usable internal CA bundle. The transport,
279 not the client, carries ``verify`` because a client given
280 ``transport=`` ignores its own TLS settings.
281 """
282 transport = httpx2.HTTPTransport(
283 verify=internal_tls.verify_for_url(self.base_url), trust_env=False
284 )
285 return httpx2.Client(
286 transport=tracing.wrap_sync_transport(transport),
287 timeout=self.timeout_seconds,
288 trust_env=False,
289 )
291 def _get(self, path: str, **kwargs: Any) -> httpx2.Response:
292 """One GET on its own client; the body is read before the client closes."""
293 with self._http() as client:
294 return client.get(f"{self.base_url}{path}", **kwargs)
296 def is_healthy(self) -> bool:
297 """Return whether OpenCost answers its /healthz probe."""
298 try:
299 response = self._get("/healthz")
300 except internal_tls.InternalTLSError as exc:
301 logger.warning("OpenCost TLS trust is unavailable: %s", exc)
302 return False
303 except httpx2.HTTPError:
304 return False
305 return response.status_code == 200
307 def get_allocation(
308 self,
309 window_start: datetime,
310 window_end: datetime,
311 *,
312 aggregate: str = "namespace",
313 ) -> dict[str, dict[str, Any]]:
314 """Fetch one accumulated allocation set for ``[window_start, window_end)``.
316 Returns a mapping of allocation name (namespace, by default) to the
317 raw OpenCost allocation object. Raises
318 :class:`OpenCostUnavailableError` on transport errors, non-200
319 responses, or a malformed body — the caller decides whether that
320 fails a scheduled pass or an API request.
321 """
322 window = (
323 f"{window_start.astimezone(UTC).strftime('%Y-%m-%dT%H:%M:%SZ')},"
324 f"{window_end.astimezone(UTC).strftime('%Y-%m-%dT%H:%M:%SZ')}"
325 )
326 try:
327 response = self._get(
328 "/allocation/compute",
329 params={
330 "window": window,
331 "aggregate": aggregate,
332 "accumulate": "true",
333 },
334 )
335 except internal_tls.InternalTLSError as exc:
336 raise OpenCostUnavailableError(f"OpenCost TLS trust is unavailable: {exc}") from exc
337 except httpx2.HTTPError as exc:
338 raise OpenCostUnavailableError(f"OpenCost request failed: {exc}") from exc
339 if response.status_code != 200:
340 raise OpenCostUnavailableError(
341 f"OpenCost allocation query returned HTTP {response.status_code}"
342 )
343 try:
344 payload = response.json()
345 except ValueError as exc:
346 raise OpenCostUnavailableError("OpenCost returned a non-JSON body") from exc
347 data = payload.get("data")
348 if not isinstance(data, list):
349 raise OpenCostUnavailableError("OpenCost allocation response omitted data")
350 merged: dict[str, dict[str, Any]] = {}
351 for allocation_set in data:
352 if not isinstance(allocation_set, dict):
353 continue
354 for name, allocation in allocation_set.items():
355 if isinstance(allocation, dict):
356 merged[str(name)] = allocation
357 return merged
360def allocations_to_rows(
361 allocations: dict[str, dict[str, Any]],
362 *,
363 cluster: str,
364 window_start: datetime,
365 window_end: datetime,
366) -> list[dict[str, Any]]:
367 """Normalize raw OpenCost allocations into the stable report row schema.
369 One row per allocation name (namespace). The ``__idle__`` and
370 ``__unallocated__`` synthetic allocations OpenCost emits are kept —
371 idle cost is exactly the visibility a cost report exists to provide —
372 but rows are sorted by descending total cost for human-readable output.
373 """
374 start_iso = window_start.astimezone(UTC).isoformat()
375 end_iso = window_end.astimezone(UTC).isoformat()
376 rows: list[dict[str, Any]] = []
377 for name, allocation in allocations.items():
378 rows.append(
379 {
380 "window_start": start_iso,
381 "window_end": end_iso,
382 "cluster": cluster,
383 "namespace": name,
384 "cpu_core_hours": _as_float(allocation.get("cpuCoreHours")),
385 "cpu_cost": _as_float(allocation.get("cpuCost")),
386 "ram_gib_hours": _as_float(allocation.get("ramByteHours")) / _GIB,
387 "ram_cost": _as_float(allocation.get("ramCost")),
388 "gpu_hours": _as_float(allocation.get("gpuHours")),
389 "gpu_cost": _as_float(allocation.get("gpuCost")),
390 "pv_cost": _as_float(allocation.get("pvCost")),
391 "network_cost": _as_float(allocation.get("networkCost")),
392 "load_balancer_cost": _as_float(allocation.get("loadBalancerCost")),
393 "shared_cost": _as_float(allocation.get("sharedCost")),
394 "external_cost": _as_float(allocation.get("externalCost")),
395 "total_cost": _as_float(allocation.get("totalCost")),
396 "total_efficiency": _as_float(allocation.get("totalEfficiency")),
397 }
398 )
399 rows.sort(key=lambda row: row["total_cost"], reverse=True)
400 return rows
403def preload_report_writer() -> bool:
404 """Load the Parquet writer now instead of on the first report.
406 :func:`rows_to_parquet_bytes` imports ``pyarrow`` lazily, so without this
407 the first scheduled pass — a minute after start, outside the startup
408 probe's window — maps pyarrow's native libraries (about 130 MB) and runs
409 its module initialization under the GIL, in the process that answers the
410 kubelet's probes. The service calls this during startup instead, where the
411 startup probe budgets for a slow node, so every scheduled pass costs what
412 a warm one does. Returns ``False`` (and logs) when pyarrow is missing; the
413 report path then fails with :class:`ReportWriteError` as before.
414 """
415 try:
416 importlib.import_module("pyarrow.parquet")
417 except ImportError:
418 logger.warning("pyarrow is not installed; cost reports cannot be written")
419 return False
420 return True
423def rows_to_parquet_bytes(rows: list[dict[str, Any]]) -> bytes:
424 """Serialize normalized report rows to a Parquet byte payload.
426 ``pyarrow`` is imported lazily so environments that never write reports
427 (unit tests exercising only transformations, or a future reader-only
428 consumer) do not need the dependency at import time; the service loads it
429 at startup with :func:`preload_report_writer`. The window bound columns
430 are stored as real timestamps so the Glue ``timestamp`` columns read them
431 natively.
432 """
433 try:
434 import pyarrow as pa
435 import pyarrow.parquet as pq
436 except ImportError as exc: # pragma: no cover - image always ships pyarrow
437 raise ReportWriteError(
438 "pyarrow is required to write cost reports; install the "
439 "image-cost-monitor dependency group"
440 ) from exc
442 schema = pa.schema(
443 [
444 ("window_start", pa.timestamp("ms", tz="UTC")),
445 ("window_end", pa.timestamp("ms", tz="UTC")),
446 ("cluster", pa.string()),
447 ("namespace", pa.string()),
448 *[
449 (name, pa.float64())
450 for name in ALLOCATION_REPORT_FIELDS
451 if name not in {"window_start", "window_end", "cluster", "namespace"}
452 ],
453 ]
454 )
455 columns: dict[str, list[Any]] = {name: [] for name in ALLOCATION_REPORT_FIELDS}
456 for row in rows:
457 for name in ALLOCATION_REPORT_FIELDS:
458 value: Any = row.get(name)
459 if name in {"window_start", "window_end"}:
460 value = datetime.fromisoformat(str(value))
461 columns[name].append(value)
462 table = pa.Table.from_pydict(columns, schema=schema)
463 sink = io.BytesIO()
464 # pyarrow ships no type stubs; route the call through an Any-typed name so
465 # environments with pyarrow installed (tests) and without it (the mypy
466 # strict CI job) type-check identically — no conditional type: ignore.
467 write_table: Any = pq.write_table
468 write_table(table, sink)
469 return sink.getvalue()
472def scheduled_report_key(region: str, window_start: datetime, window_end: datetime) -> str:
473 """Deterministic S3 key for one scheduled report window."""
474 date_partition = window_start.astimezone(UTC).strftime("%Y-%m-%d")
475 return (
476 f"{SCHEDULED_PREFIX}/region={region}/date={date_partition}/"
477 f"allocation-{_compact_ts(window_start)}-{_compact_ts(window_end)}.parquet"
478 )
481def adhoc_report_key(region: str, window_start: datetime, window_end: datetime) -> str:
482 """Unique S3 key for one ad-hoc report (kept out of the Athena table)."""
483 date_partition = datetime.now(UTC).strftime("%Y-%m-%d")
484 return (
485 f"{ADHOC_PREFIX}/region={region}/date={date_partition}/"
486 f"allocation-{_compact_ts(window_start)}-{_compact_ts(window_end)}"
487 f"-{uuid.uuid4().hex[:8]}.parquet"
488 )
491def aligned_window(now: datetime, interval_minutes: int) -> tuple[datetime, datetime]:
492 """Return the most recent *completed* interval-aligned window.
494 For ``interval_minutes=60`` at 10:25 this yields ``[09:00, 10:00)`` —
495 aligning to interval boundaries makes the scheduled key deterministic
496 across restarts and replicas, which is what makes report writes
497 idempotent.
498 """
499 interval = timedelta(minutes=interval_minutes)
500 epoch = datetime(1970, 1, 1, tzinfo=UTC)
501 elapsed = now.astimezone(UTC) - epoch
502 completed_intervals = int(elapsed / interval)
503 window_end = epoch + interval * completed_intervals
504 return window_end - interval, window_end
507class CostMonitor:
508 """Generates OpenCost allocation reports and persists them to S3.
510 The destination bucket is either fixed (``bucket``, the kind/CI override)
511 or discovered through a :class:`CostReportBucketLocator`; exactly one of
512 the two must be supplied.
513 """
515 def __init__(
516 self,
517 *,
518 region: str,
519 cluster: str,
520 opencost: OpenCostClient,
521 bucket: str | None = None,
522 bucket_locator: CostReportBucketLocator | None = None,
523 report_interval_minutes: int = 60,
524 s3_client: Any | None = None,
525 ) -> None:
526 if (bucket is None) == (bucket_locator is None):
527 raise ValueError("exactly one of bucket or bucket_locator is required")
528 self.region = region
529 self.cluster = cluster
530 self._bucket = bucket
531 self._bucket_locator = bucket_locator
532 self.opencost = opencost
533 self.report_interval_minutes = min(max(int(report_interval_minutes), 5), 1_440)
534 self._s3 = s3_client or boto3.client(
535 "s3",
536 config=Config(
537 connect_timeout=5,
538 read_timeout=60,
539 retries={"max_attempts": 3, "mode": "standard"},
540 ),
541 )
542 self.last_scheduled_report: dict[str, Any] | None = None
543 self.last_error: str | None = None
545 # ------------------------------------------------------------------
546 # Bucket identity
547 # ------------------------------------------------------------------
549 @property
550 def bucket(self) -> str | None:
551 """The report bucket, or ``None`` until SSM discovery has succeeded."""
552 if self._bucket is not None:
553 return self._bucket
554 assert self._bucket_locator is not None
555 return self._bucket_locator.cached
557 @property
558 def bucket_source(self) -> str:
559 """``"environment"`` for a fixed bucket, ``"ssm"`` for discovery."""
560 return "environment" if self._bucket is not None else "ssm"
562 @property
563 def bucket_parameter(self) -> str | None:
564 """The SSM parameter discovery reads, or ``None`` for a fixed bucket."""
565 if self._bucket_locator is None:
566 return None
567 return self._bucket_locator.parameter_name
569 def _resolve_bucket(self) -> str:
570 """Return the bucket to use now, raising while it is undiscoverable."""
571 if self._bucket is not None:
572 return self._bucket
573 assert self._bucket_locator is not None
574 return self._bucket_locator.resolve()
576 # ------------------------------------------------------------------
577 # Report generation
578 # ------------------------------------------------------------------
580 def generate_report(
581 self,
582 window_start: datetime,
583 window_end: datetime,
584 *,
585 adhoc: bool,
586 include_rows: bool = False,
587 ) -> ReportResult:
588 """Query OpenCost for one window, write Parquet to S3, return the result.
590 Raises :class:`OpenCostUnavailableError` when OpenCost cannot answer
591 and :class:`ReportWriteError` when S3 persistence fails; callers map
592 those to a failed scheduled pass or an HTTP 502/503 respectively.
593 """
594 if window_end <= window_start:
595 raise ValueError("window_end must be after window_start")
596 if window_end - window_start > timedelta(hours=_MAX_WINDOW_HOURS):
597 raise ValueError(f"report windows are capped at {_MAX_WINDOW_HOURS} hours")
598 if window_end - window_start < timedelta(minutes=_MIN_WINDOW_MINUTES):
599 raise ValueError(f"report windows must span at least {_MIN_WINDOW_MINUTES} minutes")
601 # Resolve the destination before querying OpenCost: an undiscoverable
602 # bucket is a "not ready" signal and must not cost an allocation query.
603 bucket = self._resolve_bucket()
604 allocations = self.opencost.get_allocation(window_start, window_end)
605 rows = allocations_to_rows(
606 allocations,
607 cluster=self.cluster,
608 window_start=window_start,
609 window_end=window_end,
610 )
611 key = (
612 adhoc_report_key(self.region, window_start, window_end)
613 if adhoc
614 else scheduled_report_key(self.region, window_start, window_end)
615 )
616 payload = rows_to_parquet_bytes(rows)
617 try:
618 self._s3.put_object(Bucket=bucket, Key=key, Body=payload)
619 except Exception as exc: # boto surfaces many shapes
620 raise ReportWriteError(f"Failed to write cost report to S3: {exc}") from exc
622 return ReportResult(
623 s3_key=key,
624 row_count=len(rows),
625 total_cost=sum(row["total_cost"] for row in rows),
626 window_start=window_start.astimezone(UTC).isoformat(),
627 window_end=window_end.astimezone(UTC).isoformat(),
628 rows=rows if include_rows else [],
629 )
631 def run_scheduled_once(self, now: datetime | None = None) -> ReportResult | None:
632 """Write the report for the most recent completed aligned window.
634 Skips (returns ``None``) when that window's object already exists —
635 the previous pass, or another replica during a rollout, already
636 persisted it — and when the bucket has not been published yet (the
637 monitoring stack deploys after this region; ``last_error`` records
638 the wait so ``status()`` explains the missing reports). Other failures
639 update ``last_error`` and re-raise so the caller's loop logs and
640 retries on the next tick.
641 """
642 moment = now or datetime.now(UTC)
643 window_start, window_end = aligned_window(moment, self.report_interval_minutes)
644 key = scheduled_report_key(self.region, window_start, window_end)
645 try:
646 bucket = self._resolve_bucket()
647 except CostReportBucketUnavailableError as exc:
648 self.last_error = str(exc)
649 logger.info("Cost report bucket not published yet; skipping scheduled pass: %s", exc)
650 return None
651 if self._object_exists(bucket, key):
652 logger.debug("Scheduled cost report already present: %s", key)
653 return None
654 try:
655 result = self.generate_report(window_start, window_end, adhoc=False)
656 except Exception as exc:
657 self.last_error = str(exc)
658 raise
659 self.last_scheduled_report = result.summary()
660 self.last_error = None
661 logger.info(
662 "Wrote scheduled cost report %s (%d rows, total %.4f USD)",
663 result.s3_key,
664 result.row_count,
665 result.total_cost,
666 )
667 return result
669 def _object_exists(self, bucket: str, key: str) -> bool:
670 try:
671 self._s3.head_object(Bucket=bucket, Key=key)
672 except Exception: # 404 and transport errors both mean "write it"
673 return False
674 return True
676 # ------------------------------------------------------------------
677 # Introspection for the API surface
678 # ------------------------------------------------------------------
680 def list_reports(self, *, adhoc: bool = False, limit: int = 50) -> list[dict[str, Any]]:
681 """List this region's most recent report objects, newest first."""
682 prefix = (
683 f"{ADHOC_PREFIX}/region={self.region}/"
684 if adhoc
685 else f"{SCHEDULED_PREFIX}/region={self.region}/"
686 )
687 bounded_limit = min(max(int(limit), 1), 1_000)
688 bucket = self._resolve_bucket()
689 paginator = self._s3.get_paginator("list_objects_v2")
690 objects: list[dict[str, Any]] = []
691 for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
692 for entry in page.get("Contents", []):
693 objects.append(
694 {
695 "key": entry["Key"],
696 "size_bytes": int(entry.get("Size", 0)),
697 "last_modified": (
698 entry["LastModified"].astimezone(UTC).isoformat()
699 if entry.get("LastModified")
700 else None
701 ),
702 }
703 )
704 objects.sort(key=lambda item: str(item["last_modified"] or ""), reverse=True)
705 return objects[:bounded_limit]
707 def status(self) -> dict[str, Any]:
708 """Operator-facing service status, including OpenCost health.
710 ``opencost_returning_data`` performs a live one-hour allocation probe
711 — this is the signal release validation gates on, so a healthy-but-
712 empty OpenCost (e.g. Prometheus scrape broken) fails validation
713 rather than silently producing empty reports.
714 """
715 opencost_healthy = self.opencost.is_healthy()
716 returning_data = False
717 allocation_names: list[str] = []
718 if opencost_healthy:
719 try:
720 now = datetime.now(UTC)
721 allocations = self.opencost.get_allocation(now - timedelta(hours=1), now)
722 allocation_names = sorted(allocations)
723 returning_data = bool(allocations)
724 except OpenCostUnavailableError as exc:
725 logger.warning("OpenCost allocation probe failed: %s", exc)
726 return {
727 "service": "cost-monitor",
728 "region": self.region,
729 "cluster": self.cluster,
730 "bucket": self.bucket,
731 "bucket_source": self.bucket_source,
732 "bucket_parameter": self.bucket_parameter,
733 "report_interval_minutes": self.report_interval_minutes,
734 "opencost_healthy": opencost_healthy,
735 "opencost_returning_data": returning_data,
736 "allocation_names": allocation_names[:25],
737 "last_scheduled_report": self.last_scheduled_report,
738 "last_error": self.last_error,
739 "timestamp": datetime.now(UTC).isoformat(),
740 }
743def create_cost_monitor_from_env() -> CostMonitor:
744 """Build a :class:`CostMonitor` from the Deployment's environment.
746 ``COST_REPORT_BUCKET`` (explicit name) wins when set; otherwise
747 ``COST_REPORT_BUCKET_PARAMETER`` names the SSM parameter the monitoring
748 stack publishes, read in ``COST_REPORT_BUCKET_PARAMETER_REGION`` (defaults
749 to this region). Discovery is lazy, so constructing the monitor never
750 touches AWS.
751 """
752 region = os.getenv("REGION") or os.getenv("AWS_REGION", "")
753 if not region:
754 raise RuntimeError("REGION environment variable is required")
755 bucket = os.getenv("COST_REPORT_BUCKET", "").strip() or None
756 bucket_locator: CostReportBucketLocator | None = None
757 if bucket is None:
758 parameter_name = os.getenv("COST_REPORT_BUCKET_PARAMETER", "").strip()
759 if not parameter_name:
760 raise RuntimeError(
761 "COST_REPORT_BUCKET or COST_REPORT_BUCKET_PARAMETER environment variable is required"
762 )
763 parameter_region = os.getenv("COST_REPORT_BUCKET_PARAMETER_REGION", "").strip() or region
764 bucket_locator = CostReportBucketLocator(parameter_name, parameter_region)
765 cluster = os.getenv("CLUSTER_NAME", f"gco-{region}")
766 base_url = os.getenv("OPENCOST_BASE_URL", DEFAULT_OPENCOST_BASE_URL)
767 try:
768 interval = int(os.getenv("COST_REPORT_INTERVAL_MINUTES", "60"))
769 except ValueError:
770 interval = 60
771 return CostMonitor(
772 region=region,
773 cluster=cluster,
774 bucket=bucket,
775 bucket_locator=bucket_locator,
776 opencost=OpenCostClient(base_url),
777 report_interval_minutes=interval,
778 )