Coverage for gco / services / cost_monitor.py: 100.00%
271 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +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.
27"""
29from __future__ import annotations
31import io
32import logging
33import math
34import os
35import time
36import uuid
37from collections.abc import Callable
38from dataclasses import dataclass, field
39from datetime import UTC, datetime, timedelta
40from typing import Any
42import boto3
43import httpx
44from botocore.config import Config
46logger = logging.getLogger(__name__)
48#: Normalized report row fields, in Parquet column order. This is the
49#: write-side contract of the monitoring stack's Glue table
50#: (``gco/stacks/monitoring_stack.py::_create_cost_analytics``) — the two
51#: must stay in lockstep or Athena reads misaligned columns.
52ALLOCATION_REPORT_FIELDS: tuple[str, ...] = (
53 "window_start",
54 "window_end",
55 "cluster",
56 "namespace",
57 "cpu_core_hours",
58 "cpu_cost",
59 "ram_gib_hours",
60 "ram_cost",
61 "gpu_hours",
62 "gpu_cost",
63 "pv_cost",
64 "network_cost",
65 "load_balancer_cost",
66 "shared_cost",
67 "external_cost",
68 "total_cost",
69 "total_efficiency",
70)
72_GIB = 1024.0**3
74#: Prefixes must mirror gco/stacks/constants.py (the service image does not
75#: ship the CDK stacks package, so the values are duplicated deliberately —
76#: a synth-side test asserts the two stay in lockstep).
77SCHEDULED_PREFIX = "reports"
78ADHOC_PREFIX = "adhoc"
80_MIN_WINDOW_MINUTES = 5
81_MAX_WINDOW_HOURS = 7 * 24
84class OpenCostUnavailableError(RuntimeError):
85 """Raised when the OpenCost API cannot be reached or answers abnormally."""
88class ReportWriteError(RuntimeError):
89 """Raised when a generated report cannot be persisted to S3."""
92class CostReportBucketUnavailableError(RuntimeError):
93 """Raised while the cost report bucket's identity is not yet resolvable.
95 Expected on a fresh deploy-all: the regional cost-monitor boots before the
96 monitoring stack publishes the bucket. Callers treat it as "not ready yet"
97 (skip the scheduled pass, answer HTTP 503) rather than as a write failure.
98 """
101#: How long a resolved bucket name is served before SSM is consulted again.
102#: A monitoring-stack redeploy that replaces the bucket updates the parameter;
103#: one GetParameter per refresh keeps the service converging on it.
104DEFAULT_BUCKET_REFRESH_SECONDS = 900.0
107class CostReportBucketLocator:
108 """Resolve the cost report bucket name from its published SSM parameter.
110 The monitoring stack writes ``/<project>/cost-report-bucket/name`` in the
111 monitoring region; the regional cost-monitor role holds ``ssm:GetParameter``
112 on exactly that parameter. Resolution is lazy and cached: the first
113 successful read is served for ``refresh_seconds``, a refresh that fails
114 keeps serving the last known name (a transient SSM error must not stall
115 report writes), and a read that never succeeded raises
116 :class:`CostReportBucketUnavailableError` so callers can back off.
117 """
119 def __init__(
120 self,
121 parameter_name: str,
122 region: str,
123 *,
124 ssm_client: Any | None = None,
125 refresh_seconds: float = DEFAULT_BUCKET_REFRESH_SECONDS,
126 clock: Callable[[], float] = time.monotonic,
127 ) -> None:
128 if not parameter_name.strip():
129 raise ValueError("parameter_name must not be empty")
130 if not region.strip():
131 raise ValueError("region must not be empty")
132 self.parameter_name = parameter_name
133 self.region = region
134 self.refresh_seconds = max(float(refresh_seconds), 0.0)
135 self._clock = clock
136 self._ssm = ssm_client or boto3.client(
137 "ssm",
138 region_name=region,
139 config=Config(
140 connect_timeout=5,
141 read_timeout=10,
142 retries={"max_attempts": 3, "mode": "standard"},
143 ),
144 )
145 self._cached: str | None = None
146 self._cached_at: float | None = None
148 @property
149 def cached(self) -> str | None:
150 """The last successfully resolved bucket name, or ``None`` before one."""
151 return self._cached
153 def resolve(self) -> str:
154 """Return the bucket name, reading SSM when the cache is empty or stale."""
155 now = self._clock()
156 if (
157 self._cached is not None
158 and self._cached_at is not None
159 and now - self._cached_at < self.refresh_seconds
160 ):
161 return self._cached
162 try:
163 response = self._ssm.get_parameter(Name=self.parameter_name)
164 except Exception as exc: # noqa: BLE001 - absence and transport errors converge
165 if self._cached is not None:
166 logger.warning(
167 "Refreshing cost report bucket from %s (%s) failed; keeping %s: %s",
168 self.parameter_name,
169 self.region,
170 self._cached,
171 exc,
172 )
173 self._cached_at = now
174 return self._cached
175 raise CostReportBucketUnavailableError(
176 f"Cost report bucket parameter {self.parameter_name} in {self.region} "
177 f"is not readable yet: {exc}"
178 ) from exc
179 parameter = response.get("Parameter") if isinstance(response, dict) else None
180 value = str((parameter or {}).get("Value") or "").strip()
181 if not value:
182 raise CostReportBucketUnavailableError(
183 f"Cost report bucket parameter {self.parameter_name} in {self.region} is empty"
184 )
185 if value != self._cached:
186 logger.info(
187 "Resolved cost report bucket %s from %s (%s)",
188 value,
189 self.parameter_name,
190 self.region,
191 )
192 self._cached = value
193 self._cached_at = now
194 return value
197def _compact_ts(moment: datetime) -> str:
198 """Render a UTC timestamp as a compact S3-key-safe token."""
199 return moment.astimezone(UTC).strftime("%Y%m%dT%H%M%SZ")
202def _as_float(value: Any) -> float:
203 """Coerce OpenCost numeric fields defensively; absent/bad values are 0.
205 Non-finite values (NaN and ±infinity — ``json.loads`` accepts both) are
206 coerced to 0 too: one poisoned row would otherwise contaminate every
207 Athena aggregate over the table.
208 """
209 try:
210 result = float(value)
211 except TypeError, ValueError:
212 return 0.0
213 return result if math.isfinite(result) else 0.0
216@dataclass(frozen=True)
217class ReportResult:
218 """Outcome of one generated allocation report."""
220 s3_key: str
221 row_count: int
222 total_cost: float
223 window_start: str
224 window_end: str
225 rows: list[dict[str, Any]] = field(default_factory=list)
227 def summary(self) -> dict[str, Any]:
228 """Operator-facing summary without the full row payload."""
229 return {
230 "s3_key": self.s3_key,
231 "row_count": self.row_count,
232 "total_cost": round(self.total_cost, 6),
233 "window_start": self.window_start,
234 "window_end": self.window_end,
235 }
238class OpenCostClient:
239 """Minimal HTTP client for the in-cluster OpenCost allocation API."""
241 def __init__(self, base_url: str, timeout_seconds: float = 30.0) -> None:
242 self.base_url = base_url.rstrip("/")
243 self.timeout_seconds = timeout_seconds
245 def is_healthy(self) -> bool:
246 """Return whether OpenCost answers its /healthz probe."""
247 try:
248 response = httpx.get(
249 f"{self.base_url}/healthz",
250 timeout=self.timeout_seconds,
251 )
252 except httpx.HTTPError:
253 return False
254 return response.status_code == 200
256 def get_allocation(
257 self,
258 window_start: datetime,
259 window_end: datetime,
260 *,
261 aggregate: str = "namespace",
262 ) -> dict[str, dict[str, Any]]:
263 """Fetch one accumulated allocation set for ``[window_start, window_end)``.
265 Returns a mapping of allocation name (namespace, by default) to the
266 raw OpenCost allocation object. Raises
267 :class:`OpenCostUnavailableError` on transport errors, non-200
268 responses, or a malformed body — the caller decides whether that
269 fails a scheduled pass or an API request.
270 """
271 window = (
272 f"{window_start.astimezone(UTC).strftime('%Y-%m-%dT%H:%M:%SZ')},"
273 f"{window_end.astimezone(UTC).strftime('%Y-%m-%dT%H:%M:%SZ')}"
274 )
275 try:
276 response = httpx.get(
277 f"{self.base_url}/allocation/compute",
278 params={
279 "window": window,
280 "aggregate": aggregate,
281 "accumulate": "true",
282 },
283 timeout=self.timeout_seconds,
284 )
285 except httpx.HTTPError as exc:
286 raise OpenCostUnavailableError(f"OpenCost request failed: {exc}") from exc
287 if response.status_code != 200:
288 raise OpenCostUnavailableError(
289 f"OpenCost allocation query returned HTTP {response.status_code}"
290 )
291 try:
292 payload = response.json()
293 except ValueError as exc:
294 raise OpenCostUnavailableError("OpenCost returned a non-JSON body") from exc
295 data = payload.get("data")
296 if not isinstance(data, list):
297 raise OpenCostUnavailableError("OpenCost allocation response omitted data")
298 merged: dict[str, dict[str, Any]] = {}
299 for allocation_set in data:
300 if not isinstance(allocation_set, dict):
301 continue
302 for name, allocation in allocation_set.items():
303 if isinstance(allocation, dict):
304 merged[str(name)] = allocation
305 return merged
308def allocations_to_rows(
309 allocations: dict[str, dict[str, Any]],
310 *,
311 cluster: str,
312 window_start: datetime,
313 window_end: datetime,
314) -> list[dict[str, Any]]:
315 """Normalize raw OpenCost allocations into the stable report row schema.
317 One row per allocation name (namespace). The ``__idle__`` and
318 ``__unallocated__`` synthetic allocations OpenCost emits are kept —
319 idle cost is exactly the visibility a cost report exists to provide —
320 but rows are sorted by descending total cost for human-readable output.
321 """
322 start_iso = window_start.astimezone(UTC).isoformat()
323 end_iso = window_end.astimezone(UTC).isoformat()
324 rows: list[dict[str, Any]] = []
325 for name, allocation in allocations.items():
326 rows.append(
327 {
328 "window_start": start_iso,
329 "window_end": end_iso,
330 "cluster": cluster,
331 "namespace": name,
332 "cpu_core_hours": _as_float(allocation.get("cpuCoreHours")),
333 "cpu_cost": _as_float(allocation.get("cpuCost")),
334 "ram_gib_hours": _as_float(allocation.get("ramByteHours")) / _GIB,
335 "ram_cost": _as_float(allocation.get("ramCost")),
336 "gpu_hours": _as_float(allocation.get("gpuHours")),
337 "gpu_cost": _as_float(allocation.get("gpuCost")),
338 "pv_cost": _as_float(allocation.get("pvCost")),
339 "network_cost": _as_float(allocation.get("networkCost")),
340 "load_balancer_cost": _as_float(allocation.get("loadBalancerCost")),
341 "shared_cost": _as_float(allocation.get("sharedCost")),
342 "external_cost": _as_float(allocation.get("externalCost")),
343 "total_cost": _as_float(allocation.get("totalCost")),
344 "total_efficiency": _as_float(allocation.get("totalEfficiency")),
345 }
346 )
347 rows.sort(key=lambda row: row["total_cost"], reverse=True)
348 return rows
351def rows_to_parquet_bytes(rows: list[dict[str, Any]]) -> bytes:
352 """Serialize normalized report rows to a Parquet byte payload.
354 ``pyarrow`` is imported lazily so environments that never write reports
355 (unit tests exercising only transformations, or a future reader-only
356 consumer) do not need the dependency at import time. The window bound
357 columns are stored as real timestamps so the Glue ``timestamp`` columns
358 read them natively.
359 """
360 try:
361 import pyarrow as pa
362 import pyarrow.parquet as pq
363 except ImportError as exc: # pragma: no cover - image always ships pyarrow
364 raise ReportWriteError(
365 "pyarrow is required to write cost reports; install the "
366 "image-cost-monitor dependency group"
367 ) from exc
369 schema = pa.schema(
370 [
371 ("window_start", pa.timestamp("ms", tz="UTC")),
372 ("window_end", pa.timestamp("ms", tz="UTC")),
373 ("cluster", pa.string()),
374 ("namespace", pa.string()),
375 *[
376 (name, pa.float64())
377 for name in ALLOCATION_REPORT_FIELDS
378 if name not in {"window_start", "window_end", "cluster", "namespace"}
379 ],
380 ]
381 )
382 columns: dict[str, list[Any]] = {name: [] for name in ALLOCATION_REPORT_FIELDS}
383 for row in rows:
384 for name in ALLOCATION_REPORT_FIELDS:
385 value: Any = row.get(name)
386 if name in {"window_start", "window_end"}:
387 value = datetime.fromisoformat(str(value))
388 columns[name].append(value)
389 table = pa.Table.from_pydict(columns, schema=schema)
390 sink = io.BytesIO()
391 # pyarrow ships no type stubs; route the call through an Any-typed name so
392 # environments with pyarrow installed (tests) and without it (the mypy
393 # strict CI job) type-check identically — no conditional type: ignore.
394 write_table: Any = pq.write_table
395 write_table(table, sink)
396 return sink.getvalue()
399def scheduled_report_key(region: str, window_start: datetime, window_end: datetime) -> str:
400 """Deterministic S3 key for one scheduled report window."""
401 date_partition = window_start.astimezone(UTC).strftime("%Y-%m-%d")
402 return (
403 f"{SCHEDULED_PREFIX}/region={region}/date={date_partition}/"
404 f"allocation-{_compact_ts(window_start)}-{_compact_ts(window_end)}.parquet"
405 )
408def adhoc_report_key(region: str, window_start: datetime, window_end: datetime) -> str:
409 """Unique S3 key for one ad-hoc report (kept out of the Athena table)."""
410 date_partition = datetime.now(UTC).strftime("%Y-%m-%d")
411 return (
412 f"{ADHOC_PREFIX}/region={region}/date={date_partition}/"
413 f"allocation-{_compact_ts(window_start)}-{_compact_ts(window_end)}"
414 f"-{uuid.uuid4().hex[:8]}.parquet"
415 )
418def aligned_window(now: datetime, interval_minutes: int) -> tuple[datetime, datetime]:
419 """Return the most recent *completed* interval-aligned window.
421 For ``interval_minutes=60`` at 10:25 this yields ``[09:00, 10:00)`` —
422 aligning to interval boundaries makes the scheduled key deterministic
423 across restarts and replicas, which is what makes report writes
424 idempotent.
425 """
426 interval = timedelta(minutes=interval_minutes)
427 epoch = datetime(1970, 1, 1, tzinfo=UTC)
428 elapsed = now.astimezone(UTC) - epoch
429 completed_intervals = int(elapsed / interval)
430 window_end = epoch + interval * completed_intervals
431 return window_end - interval, window_end
434class CostMonitor:
435 """Generates OpenCost allocation reports and persists them to S3.
437 The destination bucket is either fixed (``bucket``, the kind/CI override)
438 or discovered through a :class:`CostReportBucketLocator`; exactly one of
439 the two must be supplied.
440 """
442 def __init__(
443 self,
444 *,
445 region: str,
446 cluster: str,
447 opencost: OpenCostClient,
448 bucket: str | None = None,
449 bucket_locator: CostReportBucketLocator | None = None,
450 report_interval_minutes: int = 60,
451 s3_client: Any | None = None,
452 ) -> None:
453 if (bucket is None) == (bucket_locator is None):
454 raise ValueError("exactly one of bucket or bucket_locator is required")
455 self.region = region
456 self.cluster = cluster
457 self._bucket = bucket
458 self._bucket_locator = bucket_locator
459 self.opencost = opencost
460 self.report_interval_minutes = min(max(int(report_interval_minutes), 5), 1_440)
461 self._s3 = s3_client or boto3.client(
462 "s3",
463 config=Config(
464 connect_timeout=5,
465 read_timeout=60,
466 retries={"max_attempts": 3, "mode": "standard"},
467 ),
468 )
469 self.last_scheduled_report: dict[str, Any] | None = None
470 self.last_error: str | None = None
472 # ------------------------------------------------------------------
473 # Bucket identity
474 # ------------------------------------------------------------------
476 @property
477 def bucket(self) -> str | None:
478 """The report bucket, or ``None`` until SSM discovery has succeeded."""
479 if self._bucket is not None:
480 return self._bucket
481 assert self._bucket_locator is not None
482 return self._bucket_locator.cached
484 @property
485 def bucket_source(self) -> str:
486 """``"environment"`` for a fixed bucket, ``"ssm"`` for discovery."""
487 return "environment" if self._bucket is not None else "ssm"
489 @property
490 def bucket_parameter(self) -> str | None:
491 """The SSM parameter discovery reads, or ``None`` for a fixed bucket."""
492 if self._bucket_locator is None:
493 return None
494 return self._bucket_locator.parameter_name
496 def _resolve_bucket(self) -> str:
497 """Return the bucket to use now, raising while it is undiscoverable."""
498 if self._bucket is not None:
499 return self._bucket
500 assert self._bucket_locator is not None
501 return self._bucket_locator.resolve()
503 # ------------------------------------------------------------------
504 # Report generation
505 # ------------------------------------------------------------------
507 def generate_report(
508 self,
509 window_start: datetime,
510 window_end: datetime,
511 *,
512 adhoc: bool,
513 include_rows: bool = False,
514 ) -> ReportResult:
515 """Query OpenCost for one window, write Parquet to S3, return the result.
517 Raises :class:`OpenCostUnavailableError` when OpenCost cannot answer
518 and :class:`ReportWriteError` when S3 persistence fails; callers map
519 those to a failed scheduled pass or an HTTP 502/503 respectively.
520 """
521 if window_end <= window_start:
522 raise ValueError("window_end must be after window_start")
523 if window_end - window_start > timedelta(hours=_MAX_WINDOW_HOURS):
524 raise ValueError(f"report windows are capped at {_MAX_WINDOW_HOURS} hours")
525 if window_end - window_start < timedelta(minutes=_MIN_WINDOW_MINUTES):
526 raise ValueError(f"report windows must span at least {_MIN_WINDOW_MINUTES} minutes")
528 # Resolve the destination before querying OpenCost: an undiscoverable
529 # bucket is a "not ready" signal and must not cost an allocation query.
530 bucket = self._resolve_bucket()
531 allocations = self.opencost.get_allocation(window_start, window_end)
532 rows = allocations_to_rows(
533 allocations,
534 cluster=self.cluster,
535 window_start=window_start,
536 window_end=window_end,
537 )
538 key = (
539 adhoc_report_key(self.region, window_start, window_end)
540 if adhoc
541 else scheduled_report_key(self.region, window_start, window_end)
542 )
543 payload = rows_to_parquet_bytes(rows)
544 try:
545 self._s3.put_object(Bucket=bucket, Key=key, Body=payload)
546 except Exception as exc: # noqa: BLE001 - boto surfaces many shapes
547 raise ReportWriteError(f"Failed to write cost report to S3: {exc}") from exc
549 return ReportResult(
550 s3_key=key,
551 row_count=len(rows),
552 total_cost=sum(row["total_cost"] for row in rows),
553 window_start=window_start.astimezone(UTC).isoformat(),
554 window_end=window_end.astimezone(UTC).isoformat(),
555 rows=rows if include_rows else [],
556 )
558 def run_scheduled_once(self, now: datetime | None = None) -> ReportResult | None:
559 """Write the report for the most recent completed aligned window.
561 Skips (returns ``None``) when that window's object already exists —
562 the previous pass, or another replica during a rollout, already
563 persisted it — and when the bucket has not been published yet (the
564 monitoring stack deploys after this region; ``last_error`` records
565 the wait so ``status()`` explains the missing reports). Other failures
566 update ``last_error`` and re-raise so the caller's loop logs and
567 retries on the next tick.
568 """
569 moment = now or datetime.now(UTC)
570 window_start, window_end = aligned_window(moment, self.report_interval_minutes)
571 key = scheduled_report_key(self.region, window_start, window_end)
572 try:
573 bucket = self._resolve_bucket()
574 except CostReportBucketUnavailableError as exc:
575 self.last_error = str(exc)
576 logger.info("Cost report bucket not published yet; skipping scheduled pass: %s", exc)
577 return None
578 if self._object_exists(bucket, key):
579 logger.debug("Scheduled cost report already present: %s", key)
580 return None
581 try:
582 result = self.generate_report(window_start, window_end, adhoc=False)
583 except Exception as exc:
584 self.last_error = str(exc)
585 raise
586 self.last_scheduled_report = result.summary()
587 self.last_error = None
588 logger.info(
589 "Wrote scheduled cost report %s (%d rows, total %.4f USD)",
590 result.s3_key,
591 result.row_count,
592 result.total_cost,
593 )
594 return result
596 def _object_exists(self, bucket: str, key: str) -> bool:
597 try:
598 self._s3.head_object(Bucket=bucket, Key=key)
599 except Exception: # noqa: BLE001 - 404 and transport errors both mean "write it"
600 return False
601 return True
603 # ------------------------------------------------------------------
604 # Introspection for the API surface
605 # ------------------------------------------------------------------
607 def list_reports(self, *, adhoc: bool = False, limit: int = 50) -> list[dict[str, Any]]:
608 """List this region's most recent report objects, newest first."""
609 prefix = (
610 f"{ADHOC_PREFIX}/region={self.region}/"
611 if adhoc
612 else f"{SCHEDULED_PREFIX}/region={self.region}/"
613 )
614 bounded_limit = min(max(int(limit), 1), 1_000)
615 bucket = self._resolve_bucket()
616 paginator = self._s3.get_paginator("list_objects_v2")
617 objects: list[dict[str, Any]] = []
618 for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
619 for entry in page.get("Contents", []):
620 objects.append(
621 {
622 "key": entry["Key"],
623 "size_bytes": int(entry.get("Size", 0)),
624 "last_modified": (
625 entry["LastModified"].astimezone(UTC).isoformat()
626 if entry.get("LastModified")
627 else None
628 ),
629 }
630 )
631 objects.sort(key=lambda item: str(item["last_modified"] or ""), reverse=True)
632 return objects[:bounded_limit]
634 def status(self) -> dict[str, Any]:
635 """Operator-facing service status, including OpenCost health.
637 ``opencost_returning_data`` performs a live one-hour allocation probe
638 — this is the signal release validation gates on, so a healthy-but-
639 empty OpenCost (e.g. Prometheus scrape broken) fails validation
640 rather than silently producing empty reports.
641 """
642 opencost_healthy = self.opencost.is_healthy()
643 returning_data = False
644 allocation_names: list[str] = []
645 if opencost_healthy:
646 try:
647 now = datetime.now(UTC)
648 allocations = self.opencost.get_allocation(now - timedelta(hours=1), now)
649 allocation_names = sorted(allocations)
650 returning_data = bool(allocations)
651 except OpenCostUnavailableError as exc:
652 logger.warning("OpenCost allocation probe failed: %s", exc)
653 return {
654 "service": "cost-monitor",
655 "region": self.region,
656 "cluster": self.cluster,
657 "bucket": self.bucket,
658 "bucket_source": self.bucket_source,
659 "bucket_parameter": self.bucket_parameter,
660 "report_interval_minutes": self.report_interval_minutes,
661 "opencost_healthy": opencost_healthy,
662 "opencost_returning_data": returning_data,
663 "allocation_names": allocation_names[:25],
664 "last_scheduled_report": self.last_scheduled_report,
665 "last_error": self.last_error,
666 "timestamp": datetime.now(UTC).isoformat(),
667 }
670def create_cost_monitor_from_env() -> CostMonitor:
671 """Build a :class:`CostMonitor` from the Deployment's environment.
673 ``COST_REPORT_BUCKET`` (explicit name) wins when set; otherwise
674 ``COST_REPORT_BUCKET_PARAMETER`` names the SSM parameter the monitoring
675 stack publishes, read in ``COST_REPORT_BUCKET_PARAMETER_REGION`` (defaults
676 to this region). Discovery is lazy, so constructing the monitor never
677 touches AWS.
678 """
679 region = os.getenv("REGION") or os.getenv("AWS_REGION", "")
680 if not region:
681 raise RuntimeError("REGION environment variable is required")
682 bucket = os.getenv("COST_REPORT_BUCKET", "").strip() or None
683 bucket_locator: CostReportBucketLocator | None = None
684 if bucket is None:
685 parameter_name = os.getenv("COST_REPORT_BUCKET_PARAMETER", "").strip()
686 if not parameter_name:
687 raise RuntimeError(
688 "COST_REPORT_BUCKET or COST_REPORT_BUCKET_PARAMETER environment variable is required"
689 )
690 parameter_region = os.getenv("COST_REPORT_BUCKET_PARAMETER_REGION", "").strip() or region
691 bucket_locator = CostReportBucketLocator(parameter_name, parameter_region)
692 cluster = os.getenv("CLUSTER_NAME", f"gco-{region}")
693 base_url = os.getenv(
694 "OPENCOST_BASE_URL",
695 "http://opencost.monitoring.svc.cluster.local:9003",
696 )
697 try:
698 interval = int(os.getenv("COST_REPORT_INTERVAL_MINUTES", "60"))
699 except ValueError:
700 interval = 60
701 return CostMonitor(
702 region=region,
703 cluster=cluster,
704 bucket=bucket,
705 bucket_locator=bucket_locator,
706 opencost=OpenCostClient(base_url),
707 report_interval_minutes=interval,
708 )