Coverage for scripts / live_release_validation / checks / opencost.py: 100.00%
245 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Cost monitoring (OpenCost) health, data, and report-pipeline checks.
3Validates through the same authenticated API surface operators use: each
4Region's ``/api/v1/cost/status`` must report a healthy OpenCost that is
5returning allocation data, and an ad-hoc ``/api/v1/cost/reports`` request
6must produce a Parquet object that is then confirmed present in the central
7cost report bucket. The bucket the service reports is compared against the
8identity the monitoring stack published to SSM (the bucket carries a
9CloudFormation-generated name; nothing reconstructs it), which also proves
10the runtime discovery the regional cost-monitor performs resolved the same
11bucket the operator surface does. Data readiness is polled with a bounded
12deadline because a freshly-deployed Prometheus needs a few scrape cycles
13before OpenCost can answer with non-empty allocations.
14"""
16from __future__ import annotations
18import re
19import time
20from typing import Any
22from gco.stacks.constants import COST_REPORT_ADHOC_PREFIX, cost_report_ssm_parameter_prefix
24from ..checks.jobs import _response_json
25from ..context import _job_transport_region
26from ..json_utils import loads_without_duplicate_keys
27from ..models import RunContext, utc_now
29#: Ceiling for the OpenCost data-readiness poll. A fresh deploy needs
30#: Prometheus up, OpenCost scraped, and at least one allocation window
31#: resolvable; measured cold-start readiness sits well inside this bound.
32_OPENCOST_READY_TIMEOUT_SECONDS = 1_200
34#: Trailing window requested for the validation ad-hoc report.
35_VALIDATION_REPORT_WINDOW_HOURS = 1
37#: The API bridge can time out while the backend finishes a cold OpenCost/
38#: Parquet/S3 operation. Live validation runs in an exclusive disposable
39#: account, so tolerate at most one ambiguous duplicate for this exact 504.
40_REPORT_MAX_ATTEMPTS = 2
41_REPORT_RETRY_DELAY_SECONDS = 15
42_MAX_REPORT_RESPONSE_EVIDENCE_CHARS = 512
43_SUCCESSFUL_REPORT_STATUS_CODES = {200, 201}
44_EXACT_BRIDGE_TIMEOUT_BODY = {
45 "error": "Gateway timeout",
46 "message": "Upstream failed after 1 attempt(s)",
47}
48_STARTED_ATTEMPT_FIELDS = {"attempt", "state", "started_at"}
49_COMPLETED_ATTEMPT_FIELDS = _STARTED_ATTEMPT_FIELDS | {
50 "ended_at",
51 "status_code",
52 "exact_bridge_timeout",
53 "response_text",
54 "retry_scheduled",
55}
56_JOURNAL_FIELDS = {"attempts", "duplicate_possible", "completed_report"}
59def _bounded_response_text(value: Any) -> str:
60 text = str(value or "")
61 return text[:_MAX_REPORT_RESPONSE_EVIDENCE_CHARS]
64def _is_exact_bridge_timeout_payload(payload: Any) -> bool:
65 return isinstance(payload, dict) and payload == _EXACT_BRIDGE_TIMEOUT_BODY
68def _is_exact_bridge_timeout(response: Any) -> bool:
69 if response.status_code != 504:
70 return False
71 try:
72 payload = loads_without_duplicate_keys(response.text)
73 except TypeError, ValueError:
74 return False
75 return _is_exact_bridge_timeout_payload(payload)
78def _is_exact_bridge_timeout_evidence(status_code: int, response_text: str) -> bool:
79 if status_code != 504:
80 return False
81 try:
82 payload = loads_without_duplicate_keys(response_text)
83 except TypeError, ValueError:
84 return False
85 return _is_exact_bridge_timeout_payload(payload)
88def _expected_report_bucket(ctx: RunContext) -> str:
89 """The cost bucket the monitoring stack published, read from SSM.
91 The bucket's physical name is CloudFormation-generated, so the published
92 ``<prefix>/name`` parameter in the monitoring region is the only
93 authority for it — the same parameter the regional cost-monitor services
94 resolve at runtime. Any failure to read it fails validation: a report
95 whose bucket cannot be matched against the published identity is not
96 evidence.
97 """
98 region = _monitoring_region(ctx)
99 parameter_name = f"{cost_report_ssm_parameter_prefix(ctx.config.project_name)}/name"
100 ssm = ctx.session.client("ssm", region_name=region)
101 try:
102 response = ssm.get_parameter(Name=parameter_name)
103 except Exception as exc: # noqa: BLE001 - absence and access failures both fail validation
104 raise RuntimeError(
105 f"Cost report bucket parameter {parameter_name} in {region} is not readable: {exc}"
106 ) from exc
107 parameter = response.get("Parameter") if isinstance(response, dict) else None
108 value = str((parameter or {}).get("Value") or "").strip()
109 if not value:
110 raise RuntimeError(f"Cost report bucket parameter {parameter_name} in {region} is empty")
111 return value
114def _validated_completed_report(
115 ctx: RunContext,
116 report: dict[str, Any],
117 region: str,
118) -> dict[str, Any]:
119 observed_region = report.get("region")
120 if observed_region != region:
121 raise RuntimeError(
122 f"Ad-hoc cost report returned Region {observed_region!r}; expected {region!r}"
123 )
124 s3_key = report.get("s3_key")
125 if not isinstance(s3_key, str) or not s3_key.strip():
126 raise RuntimeError(f"Ad-hoc cost report for {region} omitted its S3 key")
127 expected_prefix = f"{COST_REPORT_ADHOC_PREFIX}/region={region}/"
128 key_pattern = (
129 rf"{re.escape(expected_prefix)}date=\d{{4}}-\d{{2}}-\d{{2}}/"
130 r"allocation-\d{8}T\d{6}Z-\d{8}T\d{6}Z-[0-9a-f]{8}\.parquet"
131 )
132 if re.fullmatch(key_pattern, s3_key) is None:
133 raise RuntimeError(f"Ad-hoc cost report for {region} used unexpected S3 key {s3_key!r}")
134 row_count = report.get("row_count")
135 if isinstance(row_count, bool) or not isinstance(row_count, int) or row_count <= 0:
136 raise RuntimeError(f"Ad-hoc cost report for {region} contained zero allocation rows")
137 bucket = report.get("bucket")
138 if not isinstance(bucket, str) or not bucket.strip():
139 raise RuntimeError(f"Ad-hoc cost report for {region} omitted its bucket")
140 expected_bucket = _expected_report_bucket(ctx)
141 if bucket != expected_bucket:
142 raise RuntimeError(
143 f"Ad-hoc cost report for {region} used unexpected bucket {bucket!r}; "
144 f"expected {expected_bucket!r}"
145 )
146 return dict(report)
149def _validate_report_attempt(
150 attempt: dict[str, Any],
151 expected_number: int,
152 total_attempts: int,
153) -> None:
154 attempt_number = attempt.get("attempt")
155 if (
156 isinstance(attempt_number, bool)
157 or not isinstance(attempt_number, int)
158 or attempt_number != expected_number
159 ):
160 raise RuntimeError("OpenCost report-attempt checkpoint has invalid ordering")
161 state = attempt.get("state")
162 started_at = attempt.get("started_at")
163 if state not in {"started", "completed"} or not isinstance(started_at, str) or not started_at:
164 raise RuntimeError("OpenCost report-attempt checkpoint has invalid fields")
165 if state == "started":
166 if set(attempt) != _STARTED_ATTEMPT_FIELDS:
167 raise RuntimeError("OpenCost started-attempt checkpoint has invalid fields")
168 if expected_number != total_attempts:
169 raise RuntimeError("OpenCost report-attempt checkpoint has an interior start")
170 return
172 if set(attempt) != _COMPLETED_ATTEMPT_FIELDS:
173 raise RuntimeError("OpenCost completed-attempt checkpoint has invalid fields")
174 ended_at = attempt.get("ended_at")
175 status_code = attempt.get("status_code")
176 exact_bridge_timeout = attempt.get("exact_bridge_timeout")
177 response_text = attempt.get("response_text")
178 retry_scheduled = attempt.get("retry_scheduled")
179 if (
180 not isinstance(ended_at, str)
181 or not ended_at
182 or isinstance(status_code, bool)
183 or not isinstance(status_code, int)
184 or not 100 <= status_code <= 599
185 or not isinstance(exact_bridge_timeout, bool)
186 or not isinstance(response_text, str)
187 or len(response_text) > _MAX_REPORT_RESPONSE_EVIDENCE_CHARS
188 or not isinstance(retry_scheduled, bool)
189 ):
190 raise RuntimeError("OpenCost completed-attempt checkpoint has invalid fields")
191 evidenced_timeout = _is_exact_bridge_timeout_evidence(status_code, response_text)
192 if exact_bridge_timeout is not evidenced_timeout:
193 raise RuntimeError("OpenCost report-attempt checkpoint has invalid timeout evidence")
194 expected_retry = expected_number == 1 and evidenced_timeout
195 if retry_scheduled is not expected_retry:
196 raise RuntimeError("OpenCost report-attempt checkpoint has invalid retry transition")
197 if status_code in _SUCCESSFUL_REPORT_STATUS_CODES and response_text:
198 raise RuntimeError(
199 "OpenCost successful-attempt checkpoint has unexpected response evidence"
200 )
203def _validated_report_journal(
204 ctx: RunContext,
205 raw: dict[str, Any],
206 region: str,
207) -> tuple[list[dict[str, Any]], bool, dict[str, Any] | None]:
208 if set(raw) != _JOURNAL_FIELDS:
209 raise RuntimeError("OpenCost report-attempt checkpoint has invalid fields")
210 raw_attempts = raw.get("attempts")
211 duplicate_possible = raw.get("duplicate_possible")
212 completed_report = raw.get("completed_report")
213 if (
214 not isinstance(raw_attempts, list)
215 or not raw_attempts
216 or len(raw_attempts) > _REPORT_MAX_ATTEMPTS
217 or not all(isinstance(item, dict) for item in raw_attempts)
218 or not isinstance(duplicate_possible, bool)
219 or not isinstance(completed_report, (dict, type(None)))
220 ):
221 raise RuntimeError("OpenCost report-attempt checkpoint has invalid fields")
223 attempts = [dict(item) for item in raw_attempts]
224 for expected_number, attempt in enumerate(attempts, start=1):
225 _validate_report_attempt(attempt, expected_number, len(attempts))
226 if len(attempts) == 2 and (
227 attempts[0].get("state") != "completed"
228 or attempts[0].get("exact_bridge_timeout") is not True
229 or attempts[0].get("retry_scheduled") is not True
230 ):
231 raise RuntimeError("OpenCost report-attempt checkpoint has invalid retry ancestry")
233 evidenced_duplicate = any(attempt.get("exact_bridge_timeout") is True for attempt in attempts)
234 if duplicate_possible is not evidenced_duplicate:
235 raise RuntimeError("OpenCost report-attempt checkpoint has invalid duplicate evidence")
237 validated_report: dict[str, Any] | None = None
238 if completed_report is not None:
239 if (
240 not attempts
241 or attempts[-1].get("state") != "completed"
242 or attempts[-1].get("status_code") not in _SUCCESSFUL_REPORT_STATUS_CODES
243 ):
244 raise RuntimeError("OpenCost completed-report checkpoint has no successful attempt")
245 validated_report = _validated_completed_report(ctx, completed_report, region)
246 return attempts, duplicate_possible, validated_report
249def _validated_report_journal_root(
250 ctx: RunContext,
251 root: Any,
252 *,
253 allow_empty: bool = False,
254) -> dict[str, tuple[list[dict[str, Any]], bool, dict[str, Any] | None]]:
255 if not isinstance(root, dict) or (not root and not allow_empty):
256 raise RuntimeError("OpenCost report-attempt checkpoint root is malformed")
257 expected_regions = set(ctx.deployment_regions)
258 validated: dict[
259 str,
260 tuple[list[dict[str, Any]], bool, dict[str, Any] | None],
261 ] = {}
262 for sibling_region, raw in root.items():
263 if not isinstance(sibling_region, str) or sibling_region not in expected_regions:
264 raise RuntimeError(
265 f"OpenCost report-attempt checkpoint has unexpected Region {sibling_region!r}"
266 )
267 if not isinstance(raw, dict):
268 raise RuntimeError(
269 f"OpenCost report-attempt checkpoint for {sibling_region} is malformed"
270 )
271 validated[sibling_region] = _validated_report_journal(
272 ctx,
273 raw,
274 sibling_region,
275 )
276 return validated
279def _load_report_journal(
280 ctx: RunContext,
281 region: str,
282) -> tuple[list[dict[str, Any]], bool, dict[str, Any] | None]:
283 """Load and validate every durable per-Region non-idempotent journal."""
284 if region not in ctx.deployment_regions:
285 raise RuntimeError(f"OpenCost report requested unexpected Region {region!r}")
286 with ctx.state_lock:
287 state = ctx.checkpoint.state
288 if "opencost_report_attempts" not in state:
289 return [], False, None
290 validated = _validated_report_journal_root(
291 ctx,
292 state["opencost_report_attempts"],
293 )
294 return validated.get(region, ([], False, None))
297def _persist_report_attempts(
298 ctx: RunContext,
299 region: str,
300 attempts: list[dict[str, Any]],
301 *,
302 duplicate_possible: bool,
303 completed_report: dict[str, Any] | None = None,
304) -> None:
305 if region not in ctx.deployment_regions:
306 raise RuntimeError(f"OpenCost report requested unexpected Region {region!r}")
307 record = {
308 "attempts": [dict(item) for item in attempts],
309 "duplicate_possible": duplicate_possible,
310 "completed_report": (dict(completed_report) if completed_report is not None else None),
311 }
312 _validated_report_journal(ctx, record, region)
313 with ctx.state_lock:
314 state = ctx.checkpoint.state.setdefault("opencost_report_attempts", {})
315 _validated_report_journal_root(ctx, state, allow_empty=True)
316 state[region] = record
317 ctx.persist_callback(ctx.checkpoint)
320def _cost_monitoring_configured(ctx: RunContext) -> bool:
321 """Return whether the checked-in cdk.json enables the cost pipeline."""
322 cost_block = ctx.cdk_context.get("cost_monitoring")
323 cost_enabled = True
324 if isinstance(cost_block, dict) and "enabled" in cost_block:
325 cost_enabled = bool(cost_block["enabled"])
326 observability_block = ctx.cdk_context.get("cluster_observability")
327 observability_enabled = True
328 if isinstance(observability_block, dict) and "enabled" in observability_block:
329 observability_enabled = bool(observability_block["enabled"])
330 return cost_enabled and observability_enabled
333def _monitoring_region(ctx: RunContext) -> str:
334 regions = ctx.cdk_context.get("deployment_regions") or {}
335 monitoring = regions.get("monitoring") if isinstance(regions, dict) else None
336 return str(monitoring or ctx.config.global_region)
339def _get_cost_status(ctx: RunContext, region: str) -> dict[str, Any]:
340 """Fetch one Region's /api/v1/cost/status through its authorized transport."""
341 response = ctx.aws_client.make_authenticated_request(
342 method="GET",
343 path="/api/v1/cost/status",
344 target_region=_job_transport_region(ctx, region),
345 )
346 if not response.ok:
347 raise RuntimeError(
348 f"Cost status for {region} failed: {response.status_code} {response.text}"
349 )
350 status = _response_json(response, f"Cost status for {region}")
351 observed_region = str(status.get("region") or "")
352 if observed_region and observed_region != region:
353 raise RuntimeError(
354 f"Cost status transport returned Region {observed_region!r}; expected {region!r}"
355 )
356 return status
359def _wait_for_opencost_data(ctx: RunContext, region: str) -> dict[str, Any]:
360 """Poll until OpenCost is healthy and returning allocation data.
362 Fails the action when the bounded deadline passes with OpenCost either
363 unhealthy or answering with empty allocations — both mean the deployed
364 cost pipeline cannot produce trustworthy reports.
365 """
366 deadline = time.monotonic() + _OPENCOST_READY_TIMEOUT_SECONDS
367 last_status: dict[str, Any] = {}
368 while True:
369 last_status = _get_cost_status(ctx, region)
370 if bool(last_status.get("opencost_healthy")) and bool(
371 last_status.get("opencost_returning_data")
372 ):
373 return last_status
374 if time.monotonic() >= deadline:
375 raise RuntimeError(
376 f"OpenCost in {region} did not become healthy with allocation data "
377 f"within {_OPENCOST_READY_TIMEOUT_SECONDS}s: "
378 f"healthy={last_status.get('opencost_healthy')} "
379 f"returning_data={last_status.get('opencost_returning_data')} "
380 f"last_error={last_status.get('last_error')}"
381 )
382 time.sleep(ctx.settings.poll_interval_seconds)
385def _generate_validation_report(ctx: RunContext, region: str) -> dict[str, Any]:
386 """Request one report under a crash-safe two-attempt durable journal."""
387 attempts, duplicate_possible, completed_report = _load_report_journal(ctx, region)
388 if completed_report is not None:
389 return {
390 **completed_report,
391 "request_attempts": attempts,
392 "duplicate_possible": duplicate_possible,
393 }
395 if attempts:
396 previous = attempts[-1]
397 if previous.get("state") == "started":
398 raise RuntimeError(
399 f"OpenCost report attempt {previous.get('attempt')} for {region} has an "
400 "ambiguous in-flight outcome; automatic replay is forbidden"
401 )
402 if previous.get("status_code") in _SUCCESSFUL_REPORT_STATUS_CODES:
403 raise RuntimeError(
404 f"OpenCost report attempt {previous.get('attempt')} for {region} has a "
405 "successful HTTP outcome but no validated report; automatic replay is forbidden"
406 )
407 if len(attempts) >= _REPORT_MAX_ATTEMPTS:
408 raise RuntimeError(
409 f"OpenCost report retry budget for {region} is exhausted: "
410 f"{previous.get('status_code')} {previous.get('response_text', '')}"
411 )
412 if (
413 previous.get("exact_bridge_timeout") is not True
414 or previous.get("retry_scheduled") is not True
415 ):
416 raise RuntimeError(
417 f"Prior OpenCost report attempt for {region} is not safely retryable: "
418 f"{previous.get('status_code')} {previous.get('response_text', '')}"
419 )
420 time.sleep(_REPORT_RETRY_DELAY_SECONDS)
422 for attempt_number in range(len(attempts) + 1, _REPORT_MAX_ATTEMPTS + 1):
423 attempt: dict[str, Any] = {
424 "attempt": attempt_number,
425 "state": "started",
426 "started_at": utc_now(),
427 }
428 attempts.append(attempt)
429 # Persist the non-idempotent boundary before the network call. A crash
430 # with this state is ambiguous and deliberately blocks automatic replay.
431 _persist_report_attempts(
432 ctx,
433 region,
434 attempts,
435 duplicate_possible=duplicate_possible,
436 )
437 response = ctx.aws_client.make_authenticated_request(
438 method="POST",
439 path="/api/v1/cost/reports",
440 body={"window_hours": _VALIDATION_REPORT_WINDOW_HOURS, "include_rows": False},
441 target_region=_job_transport_region(ctx, region),
442 )
443 exact_bridge_timeout = _is_exact_bridge_timeout(response)
444 retry_scheduled = attempt_number == 1 and exact_bridge_timeout
445 response_text = (
446 ""
447 if response.status_code in _SUCCESSFUL_REPORT_STATUS_CODES
448 else _bounded_response_text(response.text)
449 )
450 attempt.update(
451 {
452 "state": "completed",
453 "ended_at": utc_now(),
454 "status_code": response.status_code,
455 "exact_bridge_timeout": exact_bridge_timeout,
456 "response_text": response_text,
457 "retry_scheduled": retry_scheduled,
458 }
459 )
460 duplicate_possible = duplicate_possible or exact_bridge_timeout
461 # Persist every HTTP outcome before parsing or validating its body. A
462 # successful response without completed_report is terminal on resume:
463 # it proves the POST returned but cannot safely authorize a replay.
464 _persist_report_attempts(
465 ctx,
466 region,
467 attempts,
468 duplicate_possible=duplicate_possible,
469 )
471 if retry_scheduled:
472 # The upstream may finish after the bridge's 28-second deadline.
473 # A second request can therefore create one additional ad-hoc
474 # object; preserve that ambiguity explicitly instead of pretending
475 # this validation-only retry is idempotent.
476 time.sleep(_REPORT_RETRY_DELAY_SECONDS)
477 continue
479 if response.status_code not in _SUCCESSFUL_REPORT_STATUS_CODES:
480 raise RuntimeError(
481 f"Ad-hoc cost report for {region} failed: {response.status_code} {response.text}"
482 )
483 payload = _response_json(response, f"Ad-hoc cost report for {region}")
484 report = payload.get("report")
485 if not isinstance(report, dict):
486 raise RuntimeError(f"Ad-hoc cost report for {region} omitted its S3 key")
487 completed_report = _validated_completed_report(
488 ctx,
489 {
490 **report,
491 "region": payload.get("region"),
492 "bucket": payload.get("bucket"),
493 },
494 region,
495 )
496 _persist_report_attempts(
497 ctx,
498 region,
499 attempts,
500 duplicate_possible=duplicate_possible,
501 completed_report=completed_report,
502 )
503 return {
504 **completed_report,
505 "request_attempts": attempts,
506 "duplicate_possible": duplicate_possible,
507 }
509 raise AssertionError("OpenCost report attempt loop exhausted without returning")
512def _verify_report_object(ctx: RunContext, report: dict[str, Any]) -> dict[str, Any]:
513 """Confirm the provenance-validated Parquet object actually exists in S3."""
514 region = report.get("region")
515 if not isinstance(region, str) or region not in ctx.deployment_regions:
516 raise RuntimeError(f"Ad-hoc cost report has invalid Region {region!r}")
517 validated_report = _validated_completed_report(ctx, report, region)
518 s3 = ctx.session.client("s3", region_name=_monitoring_region(ctx))
519 key = str(validated_report["s3_key"])
520 bucket = str(validated_report["bucket"])
521 try:
522 head = s3.head_object(Bucket=bucket, Key=key)
523 except Exception as exc: # noqa: BLE001 - absence and access failures both fail validation
524 raise RuntimeError(
525 f"Cost report object s3://{bucket}/{key} is not readable: {exc}"
526 ) from exc
527 size = int(head.get("ContentLength") or 0)
528 if size <= 0:
529 raise RuntimeError(f"Cost report object s3://{bucket}/{key} is empty")
530 return {"bucket": bucket, "key": key, "size_bytes": size}