Coverage for gco / services / api_routes / cost.py: 100.00%

1"""Cost reporting endpoints — the authenticated /api/v1/cost/* surface. 

2 

3The manifest API is the cluster's authenticated ingress (HMAC middleware + 

4IAM-authorized API Gateway in front), so cost reporting is exposed here and 

5proxied to the internal ``cost-monitor`` ClusterIP service, which owns the 

6OpenCost queries and the S3 report pipeline. Keeping the cost-monitor 

7unexposed preserves its single-writer isolation while giving operators one 

8API host for every control-plane call: 

9 

10- ``GET /api/v1/cost/status`` — service + OpenCost health for this region. 

11- ``GET /api/v1/cost/reports`` — recent scheduled/ad-hoc report objects. 

12- ``POST /api/v1/cost/reports`` — generate an ad-hoc report now. 

13 

14When cost monitoring is disabled the cost-monitor Deployment does not exist, 

15so the proxy maps connection failures to a clear 503. 

16 

17The hop is verified HTTPS: the cost-monitor's TLS sidecar serves a leaf 

18certificate issued by the GCO internal CA, and this client trusts that CA 

19alone (see :mod:`gco.services.internal_tls`). A missing or unusable CA bundle 

20fails closed through the same 503, never by falling back to plaintext or to 

21the public trust store. 

22""" 

23 

24from __future__ import annotations 

25 

26import logging 

27import os 

28from datetime import UTC, datetime 

29from typing import Any 

30 

31import httpx2 

32from fastapi import APIRouter, HTTPException, Query 

33from fastapi.responses import JSONResponse, Response 

34from pydantic import BaseModel, Field 

35 

36from gco.services import internal_tls, tracing 

37 

38router = APIRouter(prefix="/api/v1/cost", tags=["Cost"]) 

39logger = logging.getLogger(__name__) 

40 

41# The cost-monitor Service publishes only its TLS sidecar port (8443); the 

42# application itself binds loopback. The VPC CNI's network policy enforcement 

43# wants Service port == container port, and the manifest processor's egress 

44# rule names 8443. 

45_DEFAULT_COST_MONITOR_URL = "https://cost-monitor.gco-system.svc.cluster.local:8443" 

46_PROXY_TIMEOUT_SECONDS = 30.0 

47_REPORT_TIMEOUT_SECONDS = 120.0 

48 

49_DISABLED_DETAIL = ( 

50 "Cost monitoring is unavailable in this region. Enable cost_monitoring " 

51 "(and cluster_observability) in cdk.json and redeploy, or check the " 

52 "cost-monitor Deployment in gco-system." 

53) 

54 

55 

56class CostReportRequest(BaseModel): 

57 """Request body for POST /api/v1/cost/reports.""" 

58 

59 window_hours: int = Field( 

60 24, 

61 ge=1, 

62 le=168, 

63 description="Trailing window the ad-hoc report covers, in hours", 

64 ) 

65 include_rows: bool = Field( 

66 False, 

67 description="Include the normalized allocation rows in the response", 

68 ) 

69 

70 

71def _cost_monitor_base_url() -> str: 

72 return os.getenv("COST_MONITOR_URL", _DEFAULT_COST_MONITOR_URL).rstrip("/") 

73 

74 

75async def _cost_monitor_request( 

76 method: str, path: str, *, timeout: float, **kwargs: Any 

77) -> httpx2.Response: 

78 """Send one request to the cost monitor; unreachable or untrusted maps to 503. 

79 

80 Each call builds its own short-lived client: this is a low-traffic 

81 operator surface, so a per-call TLS handshake is cheaper than owning a 

82 long-lived pool's lifecycle, and the verifying SSL context itself is cached 

83 by :mod:`gco.services.internal_tls` (no CA parse per call). The transport, 

84 not the client, carries ``verify`` because a client given ``transport=`` 

85 ignores its own TLS settings. 

86 """ 

87 url = f"{_cost_monitor_base_url()}{path}" 

88 try: 

89 transport = httpx2.AsyncHTTPTransport( 

90 verify=internal_tls.verify_for_url(url), trust_env=False 

91 ) 

92 async with httpx2.AsyncClient( 

93 transport=tracing.wrap_async_transport(transport), 

94 timeout=timeout, 

95 trust_env=False, 

96 ) as client: 

97 return await client.request(method, url, **kwargs) 

98 except internal_tls.InternalTLSError as exc: 

99 logger.error("Cost monitor TLS trust is unavailable for %s: %s", url, exc) 

100 raise HTTPException(status_code=503, detail=_DISABLED_DETAIL) from exc 

101 except httpx2.HTTPError as exc: 

102 logger.warning("Cost monitor unreachable at %s: %s", url, exc) 

103 raise HTTPException(status_code=503, detail=_DISABLED_DETAIL) from exc 

104 

105 

106async def _proxy_get(path: str, params: dict[str, Any]) -> dict[str, Any]: 

107 response = await _cost_monitor_request( 

108 "GET", path, timeout=_PROXY_TIMEOUT_SECONDS, params=params 

109 ) 

110 return _relay_json(response) 

111 

112 

113def _relay_json(response: httpx2.Response) -> dict[str, Any]: 

114 """Return the cost-monitor JSON body, propagating its error statuses.""" 

115 try: 

116 payload = response.json() 

117 except ValueError as exc: 

118 raise HTTPException( 

119 status_code=502, detail="Cost monitor returned a non-JSON body" 

120 ) from exc 

121 if response.status_code >= 400: 

122 detail = payload.get("detail") if isinstance(payload, dict) else None 

123 raise HTTPException( 

124 status_code=response.status_code, 

125 detail=str(detail or "Cost monitor request failed"), 

126 ) 

127 if not isinstance(payload, dict): 

128 raise HTTPException(status_code=502, detail="Cost monitor returned a non-object body") 

129 return payload 

130 

131 

132@router.get("/status") 

133async def get_cost_status() -> Response: 

134 """Cost monitoring status for this region, including OpenCost health.""" 

135 payload = await _proxy_get("/internal/status", {}) 

136 return JSONResponse(status_code=200, content=payload) 

137 

138 

139@router.get("/reports") 

140async def list_cost_reports( 

141 adhoc: bool = Query(False, description="List ad-hoc instead of scheduled reports"), 

142 limit: int = Query(50, ge=1, le=1000, description="Maximum objects returned"), 

143) -> Response: 

144 """List this region's most recent cost report objects in S3.""" 

145 payload = await _proxy_get("/internal/reports", {"adhoc": str(adhoc).lower(), "limit": limit}) 

146 return JSONResponse(status_code=200, content=payload) 

147 

148 

149@router.post("/reports") 

150async def generate_cost_report(request: CostReportRequest) -> Response: 

151 """Generate an ad-hoc cost report for the trailing window.""" 

152 response = await _cost_monitor_request( 

153 "POST", 

154 "/internal/reports", 

155 timeout=_REPORT_TIMEOUT_SECONDS, 

156 json={ 

157 "window_hours": request.window_hours, 

158 "include_rows": request.include_rows, 

159 }, 

160 ) 

161 payload = _relay_json(response) 

162 payload.setdefault("timestamp", datetime.now(UTC).isoformat()) 

163 return JSONResponse(status_code=201, content=payload)