Coverage for scripts / live_release_validation / actions / opencost.py: 100.00%

1"""opencost: cost monitoring health and report-pipeline validation.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from ..checks.opencost import ( 

8 _cost_monitoring_configured, 

9 _generate_validation_report, 

10 _verify_report_object, 

11 _wait_for_opencost_data, 

12) 

13from ..models import RunContext 

14 

15 

16def action_opencost(ctx: RunContext) -> dict[str, Any]: 

17 """Require healthy, data-returning OpenCost and a working report pipeline. 

18 

19 For every deployed Region: poll ``/api/v1/cost/status`` until OpenCost is 

20 healthy *and* returning allocation data (bounded), then generate an 

21 ad-hoc report through ``/api/v1/cost/reports`` and confirm the Parquet 

22 object exists in the central cost report bucket. Any Region with an 

23 unhealthy or empty OpenCost fails validation. 

24 

25 When the checked-in ``cdk.json`` disables cost monitoring (or cluster 

26 observability, its data source), the action records that configuration 

27 and passes — validation proves the configured topology, and an opted-out 

28 pipeline has nothing to prove. 

29 """ 

30 if not _cost_monitoring_configured(ctx): 

31 return { 

32 "cost_monitoring_enabled": False, 

33 "detail": ( 

34 "cost_monitoring (or cluster_observability) is disabled in " 

35 "cdk.json; no OpenCost surface is deployed to validate" 

36 ), 

37 } 

38 

39 regions: dict[str, Any] = {} 

40 for region in ctx.deployment_regions: 

41 status = _wait_for_opencost_data(ctx, region) 

42 with ctx.state_lock: 

43 ctx.checkpoint.state.setdefault("opencost_status", {})[region] = status 

44 ctx.persist_callback(ctx.checkpoint) 

45 report = _generate_validation_report(ctx, region) 

46 object_evidence = _verify_report_object(ctx, report) 

47 regions[region] = { 

48 "opencost_healthy": bool(status.get("opencost_healthy")), 

49 "opencost_returning_data": bool(status.get("opencost_returning_data")), 

50 "allocation_names": status.get("allocation_names"), 

51 "report": report, 

52 "s3_object": object_evidence, 

53 } 

54 

55 evidence = { 

56 "cost_monitoring_enabled": True, 

57 "regions": regions, 

58 } 

59 with ctx.state_lock: 

60 ctx.checkpoint.state["opencost"] = evidence 

61 ctx.persist_callback(ctx.checkpoint) 

62 return evidence