Coverage for gco / services / cost_api.py: 100.00%
115 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +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 import tracing
36from gco.services.cost_monitor import (
37 CostMonitor,
38 CostReportBucketUnavailableError,
39 OpenCostUnavailableError,
40 ReportWriteError,
41 create_cost_monitor_from_env,
42 preload_report_writer,
43)
44from gco.services.structured_logging import configure_structured_logging
46logging.basicConfig(
47 level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
48)
49logger = logging.getLogger(__name__)
51#: Populated by the lifespan handler; read by the route handlers.
52cost_monitor: CostMonitor | None = None
54_SCHEDULER_TICK_SECONDS = 60.0
57class AdhocReportRequest(BaseModel):
58 """Request body for POST /internal/reports."""
60 window_hours: int = Field(
61 24,
62 ge=1,
63 le=168,
64 description="Trailing window the report covers, in hours",
65 )
66 include_rows: bool = Field(
67 False,
68 description="Include the normalized allocation rows in the response",
69 )
72def _check_monitor() -> CostMonitor:
73 if cost_monitor is None:
74 raise HTTPException(status_code=503, detail="Cost monitor not initialized")
75 return cost_monitor
78async def _scheduled_report_loop(monitor: CostMonitor, stop: asyncio.Event) -> None:
79 """Write the aligned interval report; failures retry on the next tick."""
80 while not stop.is_set():
81 try:
82 await asyncio.to_thread(monitor.run_scheduled_once)
83 except asyncio.CancelledError:
84 raise
85 except Exception as exc: # the loop must survive any pass failure
86 logger.warning("Scheduled cost report pass failed: %s", exc)
87 try:
88 await asyncio.wait_for(stop.wait(), timeout=_SCHEDULER_TICK_SECONDS)
89 except TimeoutError:
90 continue
93@asynccontextmanager
94async def lifespan(app: FastAPI) -> AsyncIterator[None]:
95 """Initialize the monitor and run the scheduled reporter until shutdown."""
96 global cost_monitor
98 logger.info("Starting Cost Monitor Service")
99 # Before the server listens: the startup probe covers this window, the
100 # liveness probe does not cover the first scheduled pass.
101 preload_report_writer()
102 # The scheduled loop runs on the instance this lifespan built, whatever
103 # the module global holds by then. It holds no upstream connections to
104 # release: every OpenCost call builds and closes its own client.
105 monitor = create_cost_monitor_from_env()
106 cost_monitor = monitor
107 configure_structured_logging(
108 service_name="cost-monitor",
109 cluster_id=monitor.cluster,
110 region=monitor.region,
111 )
112 stop = asyncio.Event()
113 loop_task = asyncio.create_task(
114 _scheduled_report_loop(monitor, stop),
115 name="cost-monitor-scheduled-reports",
116 )
117 app.state.scheduled_report_task = loop_task
118 try:
119 yield
120 finally:
121 stop.set()
122 try:
123 await asyncio.wait_for(loop_task, timeout=30)
124 except TimeoutError:
125 loop_task.cancel()
126 # Last, so spans from the final scheduled pass are exported too.
127 tracing.shutdown_tracing()
128 logger.info("Shutting down Cost Monitor Service")
131app = FastAPI(
132 title="GCO Cost Monitor",
133 description="Scheduled and on-demand OpenCost allocation reporting",
134 version="1.0.0",
135 lifespan=lifespan,
136)
138# Server spans for every route except the probe and scrape paths, plus the
139# OpenCost client spans under them; inert unless GCO_TRACING_ENABLED=true.
140tracing.configure_tracing("cost-monitor")
141tracing.instrument_fastapi_app(app)
143from gco.services.service_metrics import mount_metrics # noqa: E402
145mount_metrics(app, "cost-monitor")
148@app.get("/healthz", tags=["Health"])
149async def kubernetes_health_check() -> dict[str, str]:
150 """Kubernetes-style liveness probe."""
151 return {"status": "ok"}
154@app.get("/readyz", tags=["Health"])
155async def kubernetes_readiness_check() -> dict[str, str]:
156 """Readiness requires the monitor plus a live scheduled-report task."""
157 if cost_monitor is None:
158 raise HTTPException(status_code=503, detail="Cost monitor not ready")
159 task = getattr(app.state, "scheduled_report_task", None)
160 if task is not None and task.done():
161 raise HTTPException(status_code=503, detail="Scheduled reporter stopped unexpectedly")
162 return {"status": "ready"}
165@app.get("/internal/status", tags=["Cost"])
166async def get_status() -> dict[str, Any]:
167 """Service status including OpenCost health and the data-returning probe."""
168 monitor = _check_monitor()
169 return await asyncio.to_thread(monitor.status)
172@app.get("/internal/reports", tags=["Cost"])
173async def list_reports(
174 adhoc: bool = Query(False, description="List ad-hoc instead of scheduled reports"),
175 limit: int = Query(50, ge=1, le=1000, description="Maximum objects returned"),
176) -> dict[str, Any]:
177 """List this region's most recent report objects, newest first."""
178 monitor = _check_monitor()
179 try:
180 reports = await asyncio.to_thread(monitor.list_reports, adhoc=adhoc, limit=limit)
181 except CostReportBucketUnavailableError as exc:
182 raise HTTPException(status_code=503, detail=str(exc)) from exc
183 except Exception as exc: # surface S3 failures as 502
184 raise HTTPException(status_code=502, detail=f"Failed to list reports: {exc}") from exc
185 return {
186 "timestamp": datetime.now(UTC).isoformat(),
187 "region": monitor.region,
188 "bucket": monitor.bucket,
189 "count": len(reports),
190 "reports": reports,
191 }
194@app.post("/internal/reports", tags=["Cost"], status_code=201)
195async def generate_adhoc_report(request: AdhocReportRequest) -> dict[str, Any]:
196 """Generate one ad-hoc allocation report for the trailing window."""
197 monitor = _check_monitor()
198 window_end = datetime.now(UTC)
199 window_start = window_end - timedelta(hours=request.window_hours)
200 try:
201 result = await asyncio.to_thread(
202 monitor.generate_report,
203 window_start,
204 window_end,
205 adhoc=True,
206 include_rows=request.include_rows,
207 )
208 except (CostReportBucketUnavailableError, OpenCostUnavailableError) as exc:
209 raise HTTPException(status_code=503, detail=str(exc)) from exc
210 except ReportWriteError as exc:
211 raise HTTPException(status_code=502, detail=str(exc)) from exc
212 except ValueError as exc:
213 raise HTTPException(status_code=422, detail=str(exc)) from exc
214 body: dict[str, Any] = {
215 "timestamp": datetime.now(UTC).isoformat(),
216 "region": monitor.region,
217 "bucket": monitor.bucket,
218 "report": result.summary(),
219 }
220 if request.include_rows:
221 body["rows"] = result.rows
222 return body
225def create_app() -> FastAPI:
226 """Factory function to create the FastAPI app."""
227 return app
230# The pod manifest gives the kubelet terminationGracePeriodSeconds > preStop +
231# this budget, so Uvicorn can finish in-flight requests before SIGKILL. The
232# same variable drives the TLS sidecar's drain (gco.services.tls_proxy).
233DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS = 20
236def _run_server() -> None:
237 """Run Uvicorn with the same drain budget declared by the pod manifest."""
238 import uvicorn
240 from gco.services.uvicorn_runtime import describe_uvicorn_runtime
242 host = os.getenv("HOST", "0.0.0.0") # container listener
243 port = int(os.getenv("PORT", "8080"))
244 log_level = os.getenv("LOG_LEVEL", "info").lower()
245 graceful_shutdown_seconds = int(
246 os.getenv(
247 "GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS",
248 str(DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS),
249 )
250 )
252 loop_impl, http_impl = describe_uvicorn_runtime()
253 logger.info(
254 "Starting Cost Monitor API on %s:%d (uvicorn loop=%s http=%s)",
255 host,
256 port,
257 loop_impl,
258 http_impl,
259 )
261 uvicorn.run(
262 "gco.services.cost_api:app",
263 host=host,
264 port=port,
265 log_level=log_level,
266 reload=False,
267 timeout_graceful_shutdown=graceful_shutdown_seconds,
268 )
271if __name__ == "__main__":
272 _run_server()