Coverage for gco/services/cost_api.py: 96.23%
96 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +0000
1"""Cost Monitor HTTP service.
3Runs inside the ``cost-monitor`` Deployment (gco-system) and serves:
5- ``/healthz`` / ``/readyz`` — Kubernetes probes.
6- ``/metrics`` — Prometheus metrics for the in-cluster scrape.
7- ``/internal/status`` — service + OpenCost health, including the live
8 "returning data" probe release validation gates on.
9- ``GET /internal/reports`` — recent report objects for this region.
10- ``POST /internal/reports`` — ad-hoc report generation.
12The service is cluster-internal (ClusterIP, default-deny ingress except the
13manifest processor): the *authenticated* public surface is the manifest API's
14``/api/v1/cost/*`` router, which proxies here. A background task writes the
15scheduled interval reports.
16"""
18from __future__ import annotations
20import asyncio
21import logging
22import os
23from collections.abc import AsyncIterator
24from contextlib import asynccontextmanager
25from datetime import UTC, datetime, timedelta
26from typing import Any
28from fastapi import FastAPI, HTTPException, Query
29from pydantic import BaseModel, Field
31from gco.services.cost_monitor import (
32 CostMonitor,
33 OpenCostUnavailableError,
34 ReportWriteError,
35 create_cost_monitor_from_env,
36)
37from gco.services.structured_logging import configure_structured_logging
39logging.basicConfig(
40 level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
41)
42logger = logging.getLogger(__name__)
44#: Populated by the lifespan handler; read by the route handlers.
45cost_monitor: CostMonitor | None = None
47_SCHEDULER_TICK_SECONDS = 60.0
50class AdhocReportRequest(BaseModel):
51 """Request body for POST /internal/reports."""
53 window_hours: int = Field(
54 24,
55 ge=1,
56 le=168,
57 description="Trailing window the report covers, in hours",
58 )
59 include_rows: bool = Field(
60 False,
61 description="Include the normalized allocation rows in the response",
62 )
65def _check_monitor() -> CostMonitor:
66 if cost_monitor is None:
67 raise HTTPException(status_code=503, detail="Cost monitor not initialized")
68 return cost_monitor
71async def _scheduled_report_loop(monitor: CostMonitor, stop: asyncio.Event) -> None:
72 """Write the aligned interval report; failures retry on the next tick."""
73 while not stop.is_set():
74 try:
75 await asyncio.to_thread(monitor.run_scheduled_once)
76 except asyncio.CancelledError:
77 raise
78 except Exception as exc: # noqa: BLE001 - the loop must survive any pass failure
79 logger.warning("Scheduled cost report pass failed: %s", exc)
80 try:
81 await asyncio.wait_for(stop.wait(), timeout=_SCHEDULER_TICK_SECONDS)
82 except TimeoutError:
83 continue
86@asynccontextmanager
87async def lifespan(app: FastAPI) -> AsyncIterator[None]:
88 """Initialize the monitor and run the scheduled reporter until shutdown."""
89 global cost_monitor
91 logger.info("Starting Cost Monitor Service")
92 cost_monitor = create_cost_monitor_from_env()
93 configure_structured_logging(
94 service_name="cost-monitor",
95 cluster_id=cost_monitor.cluster,
96 region=cost_monitor.region,
97 )
98 stop = asyncio.Event()
99 loop_task = asyncio.create_task(
100 _scheduled_report_loop(cost_monitor, stop),
101 name="cost-monitor-scheduled-reports",
102 )
103 app.state.scheduled_report_task = loop_task
104 try:
105 yield
106 finally:
107 stop.set()
108 try:
109 await asyncio.wait_for(loop_task, timeout=30)
110 except TimeoutError:
111 loop_task.cancel()
112 logger.info("Shutting down Cost Monitor Service")
115app = FastAPI(
116 title="GCO Cost Monitor",
117 description="Scheduled and on-demand OpenCost allocation reporting",
118 version="1.0.0",
119 lifespan=lifespan,
120)
122from gco.services.service_metrics import mount_metrics # noqa: E402
124mount_metrics(app, "cost-monitor")
127@app.get("/healthz", tags=["Health"])
128async def kubernetes_health_check() -> dict[str, str]:
129 """Kubernetes-style liveness probe."""
130 return {"status": "ok"}
133@app.get("/readyz", tags=["Health"])
134async def kubernetes_readiness_check() -> dict[str, str]:
135 """Readiness requires the monitor plus a live scheduled-report task."""
136 if cost_monitor is None:
137 raise HTTPException(status_code=503, detail="Cost monitor not ready")
138 task = getattr(app.state, "scheduled_report_task", None)
139 if task is not None and task.done():
140 raise HTTPException(status_code=503, detail="Scheduled reporter stopped unexpectedly")
141 return {"status": "ready"}
144@app.get("/internal/status", tags=["Cost"])
145async def get_status() -> dict[str, Any]:
146 """Service status including OpenCost health and the data-returning probe."""
147 monitor = _check_monitor()
148 return await asyncio.to_thread(monitor.status)
151@app.get("/internal/reports", tags=["Cost"])
152async def list_reports(
153 adhoc: bool = Query(False, description="List ad-hoc instead of scheduled reports"),
154 limit: int = Query(50, ge=1, le=1000, description="Maximum objects returned"),
155) -> dict[str, Any]:
156 """List this region's most recent report objects, newest first."""
157 monitor = _check_monitor()
158 try:
159 reports = await asyncio.to_thread(monitor.list_reports, adhoc=adhoc, limit=limit)
160 except Exception as exc: # noqa: BLE001 - surface S3 failures as 502
161 raise HTTPException(status_code=502, detail=f"Failed to list reports: {exc}") from exc
162 return {
163 "timestamp": datetime.now(UTC).isoformat(),
164 "region": monitor.region,
165 "bucket": monitor.bucket,
166 "count": len(reports),
167 "reports": reports,
168 }
171@app.post("/internal/reports", tags=["Cost"], status_code=201)
172async def generate_adhoc_report(request: AdhocReportRequest) -> dict[str, Any]:
173 """Generate one ad-hoc allocation report for the trailing window."""
174 monitor = _check_monitor()
175 window_end = datetime.now(UTC)
176 window_start = window_end - timedelta(hours=request.window_hours)
177 try:
178 result = await asyncio.to_thread(
179 monitor.generate_report,
180 window_start,
181 window_end,
182 adhoc=True,
183 include_rows=request.include_rows,
184 )
185 except OpenCostUnavailableError as exc:
186 raise HTTPException(status_code=503, detail=str(exc)) from exc
187 except ReportWriteError as exc:
188 raise HTTPException(status_code=502, detail=str(exc)) from exc
189 except ValueError as exc:
190 raise HTTPException(status_code=422, detail=str(exc)) from exc
191 body: dict[str, Any] = {
192 "timestamp": datetime.now(UTC).isoformat(),
193 "region": monitor.region,
194 "bucket": monitor.bucket,
195 "report": result.summary(),
196 }
197 if request.include_rows:
198 body["rows"] = result.rows
199 return body
202def create_app() -> FastAPI:
203 """Factory function to create the FastAPI app."""
204 return app
207if __name__ == "__main__":
208 import uvicorn
210 host = os.getenv("HOST", "0.0.0.0") # nosec B104 — must bind all interfaces inside K8s pod
211 port = int(os.getenv("PORT", "8080"))
212 logger.info("Starting Cost Monitor API on %s:%d", host, port)
213 uvicorn.run(
214 "gco.services.cost_api:app",
215 host=host,
216 port=port,
217 log_level=os.getenv("LOG_LEVEL", "info").lower(),
218 reload=False,
219 )