Coverage for gco / services / cost_api.py: 100.00%
107 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +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.
17The report bucket is discovered from SSM (see :mod:`gco.services.cost_monitor`);
18until the monitoring stack has published it, the report endpoints answer 503
19and ``/internal/status`` shows ``bucket: null`` with the wait in ``last_error``.
20"""
22from __future__ import annotations
24import asyncio
25import logging
26import os
27from collections.abc import AsyncIterator
28from contextlib import asynccontextmanager
29from datetime import UTC, datetime, timedelta
30from typing import Any
32from fastapi import FastAPI, HTTPException, Query
33from pydantic import BaseModel, Field
35from gco.services.cost_monitor import (
36 CostMonitor,
37 CostReportBucketUnavailableError,
38 OpenCostUnavailableError,
39 ReportWriteError,
40 create_cost_monitor_from_env,
41)
42from gco.services.structured_logging import configure_structured_logging
44logging.basicConfig(
45 level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
46)
47logger = logging.getLogger(__name__)
49#: Populated by the lifespan handler; read by the route handlers.
50cost_monitor: CostMonitor | None = None
52_SCHEDULER_TICK_SECONDS = 60.0
55class AdhocReportRequest(BaseModel):
56 """Request body for POST /internal/reports."""
58 window_hours: int = Field(
59 24,
60 ge=1,
61 le=168,
62 description="Trailing window the report covers, in hours",
63 )
64 include_rows: bool = Field(
65 False,
66 description="Include the normalized allocation rows in the response",
67 )
70def _check_monitor() -> CostMonitor:
71 if cost_monitor is None:
72 raise HTTPException(status_code=503, detail="Cost monitor not initialized")
73 return cost_monitor
76async def _scheduled_report_loop(monitor: CostMonitor, stop: asyncio.Event) -> None:
77 """Write the aligned interval report; failures retry on the next tick."""
78 while not stop.is_set():
79 try:
80 await asyncio.to_thread(monitor.run_scheduled_once)
81 except asyncio.CancelledError:
82 raise
83 except Exception as exc: # noqa: BLE001 - the loop must survive any pass failure
84 logger.warning("Scheduled cost report pass failed: %s", exc)
85 try:
86 await asyncio.wait_for(stop.wait(), timeout=_SCHEDULER_TICK_SECONDS)
87 except TimeoutError:
88 continue
91@asynccontextmanager
92async def lifespan(app: FastAPI) -> AsyncIterator[None]:
93 """Initialize the monitor and run the scheduled reporter until shutdown."""
94 global cost_monitor
96 logger.info("Starting Cost Monitor Service")
97 cost_monitor = create_cost_monitor_from_env()
98 configure_structured_logging(
99 service_name="cost-monitor",
100 cluster_id=cost_monitor.cluster,
101 region=cost_monitor.region,
102 )
103 stop = asyncio.Event()
104 loop_task = asyncio.create_task(
105 _scheduled_report_loop(cost_monitor, stop),
106 name="cost-monitor-scheduled-reports",
107 )
108 app.state.scheduled_report_task = loop_task
109 try:
110 yield
111 finally:
112 stop.set()
113 try:
114 await asyncio.wait_for(loop_task, timeout=30)
115 except TimeoutError:
116 loop_task.cancel()
117 logger.info("Shutting down Cost Monitor Service")
120app = FastAPI(
121 title="GCO Cost Monitor",
122 description="Scheduled and on-demand OpenCost allocation reporting",
123 version="1.0.0",
124 lifespan=lifespan,
125)
127from gco.services.service_metrics import mount_metrics # noqa: E402
129mount_metrics(app, "cost-monitor")
132@app.get("/healthz", tags=["Health"])
133async def kubernetes_health_check() -> dict[str, str]:
134 """Kubernetes-style liveness probe."""
135 return {"status": "ok"}
138@app.get("/readyz", tags=["Health"])
139async def kubernetes_readiness_check() -> dict[str, str]:
140 """Readiness requires the monitor plus a live scheduled-report task."""
141 if cost_monitor is None:
142 raise HTTPException(status_code=503, detail="Cost monitor not ready")
143 task = getattr(app.state, "scheduled_report_task", None)
144 if task is not None and task.done():
145 raise HTTPException(status_code=503, detail="Scheduled reporter stopped unexpectedly")
146 return {"status": "ready"}
149@app.get("/internal/status", tags=["Cost"])
150async def get_status() -> dict[str, Any]:
151 """Service status including OpenCost health and the data-returning probe."""
152 monitor = _check_monitor()
153 return await asyncio.to_thread(monitor.status)
156@app.get("/internal/reports", tags=["Cost"])
157async def list_reports(
158 adhoc: bool = Query(False, description="List ad-hoc instead of scheduled reports"),
159 limit: int = Query(50, ge=1, le=1000, description="Maximum objects returned"),
160) -> dict[str, Any]:
161 """List this region's most recent report objects, newest first."""
162 monitor = _check_monitor()
163 try:
164 reports = await asyncio.to_thread(monitor.list_reports, adhoc=adhoc, limit=limit)
165 except CostReportBucketUnavailableError as exc:
166 raise HTTPException(status_code=503, detail=str(exc)) from exc
167 except Exception as exc: # noqa: BLE001 - surface S3 failures as 502
168 raise HTTPException(status_code=502, detail=f"Failed to list reports: {exc}") from exc
169 return {
170 "timestamp": datetime.now(UTC).isoformat(),
171 "region": monitor.region,
172 "bucket": monitor.bucket,
173 "count": len(reports),
174 "reports": reports,
175 }
178@app.post("/internal/reports", tags=["Cost"], status_code=201)
179async def generate_adhoc_report(request: AdhocReportRequest) -> dict[str, Any]:
180 """Generate one ad-hoc allocation report for the trailing window."""
181 monitor = _check_monitor()
182 window_end = datetime.now(UTC)
183 window_start = window_end - timedelta(hours=request.window_hours)
184 try:
185 result = await asyncio.to_thread(
186 monitor.generate_report,
187 window_start,
188 window_end,
189 adhoc=True,
190 include_rows=request.include_rows,
191 )
192 except (CostReportBucketUnavailableError, OpenCostUnavailableError) as exc:
193 raise HTTPException(status_code=503, detail=str(exc)) from exc
194 except ReportWriteError as exc:
195 raise HTTPException(status_code=502, detail=str(exc)) from exc
196 except ValueError as exc:
197 raise HTTPException(status_code=422, detail=str(exc)) from exc
198 body: dict[str, Any] = {
199 "timestamp": datetime.now(UTC).isoformat(),
200 "region": monitor.region,
201 "bucket": monitor.bucket,
202 "report": result.summary(),
203 }
204 if request.include_rows:
205 body["rows"] = result.rows
206 return body
209def create_app() -> FastAPI:
210 """Factory function to create the FastAPI app."""
211 return app
214# The pod manifest gives the kubelet terminationGracePeriodSeconds > preStop +
215# this budget, so Uvicorn can finish in-flight requests before SIGKILL. The
216# same variable drives the TLS sidecar's drain (gco.services.tls_proxy).
217DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS = 20
220def _run_server() -> None:
221 """Run Uvicorn with the same drain budget declared by the pod manifest."""
222 import uvicorn
224 host = os.getenv("HOST", "0.0.0.0") # nosec B104 — container listener
225 port = int(os.getenv("PORT", "8080"))
226 log_level = os.getenv("LOG_LEVEL", "info").lower()
227 graceful_shutdown_seconds = int(
228 os.getenv(
229 "GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS",
230 str(DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS),
231 )
232 )
234 logger.info("Starting Cost Monitor API on %s:%d", host, port)
236 uvicorn.run(
237 "gco.services.cost_api:app",
238 host=host,
239 port=port,
240 log_level=log_level,
241 reload=False,
242 timeout_graceful_shutdown=graceful_shutdown_seconds,
243 )
246if __name__ == "__main__":
247 _run_server()