Coverage for gco / services / health_api.py: 100.00%
156 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""
2Health API Service for GCO (Global Capacity Orchestrator on AWS).
4This FastAPI service exposes health status endpoints for:
5- ALB health checks (/healthz, /readyz)
6- Detailed health status (/api/v1/health)
7- Resource utilization metrics (/api/v1/metrics)
8- Service status information (/api/v1/status)
10The service runs a background task that continuously monitors cluster health
11and caches the results for fast response times on health check endpoints.
13Endpoints:
14 GET /healthz - Kubernetes liveness probe (always 200 if running)
15 GET /readyz - Kubernetes readiness probe (200 if health monitor ready)
16 GET /api/v1/health - Detailed health status (200 if healthy, 503 if not)
17 GET /api/v1/metrics - Resource utilization metrics
18 GET /api/v1/status - Service operational status
20Environment Variables:
21 HOST: Bind address (default: 0.0.0.0)
22 PORT: Listen port (default: 8080)
23 LOG_LEVEL: Logging level (default: info)
24 CLUSTER_NAME, REGION, *_THRESHOLD: See health_monitor.py
25"""
27import asyncio
28import contextlib
29import logging
30import os
31from collections.abc import AsyncIterator
32from contextlib import asynccontextmanager
33from datetime import UTC, datetime
34from typing import Any
36from fastapi import FastAPI, HTTPException, Request
37from fastapi.responses import JSONResponse
39from gco.models import HealthStatus
40from gco.services import tracing
41from gco.services.auth_middleware import AuthenticationMiddleware
42from gco.services.health_monitor import HealthMonitor, create_health_monitor_from_env
43from gco.services.metrics_publisher import HealthMonitorMetrics
44from gco.services.structured_logging import configure_structured_logging
45from gco.services.webhook_dispatcher import (
46 WebhookDispatcher,
47 create_webhook_dispatcher_from_env,
48)
50logging.basicConfig(
51 level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
52)
53logger = logging.getLogger(__name__)
55# Global health monitor instance
56health_monitor: HealthMonitor | None = None
57health_metrics: HealthMonitorMetrics | None = None
58webhook_dispatcher: WebhookDispatcher | None = None
59current_health_status: HealthStatus | None = None
60health_check_task = None
63@asynccontextmanager
64async def lifespan(app: FastAPI) -> AsyncIterator[None]:
65 """
66 Application lifespan manager - starts and stops background health monitoring
67 and webhook dispatcher.
68 """
69 global health_monitor, health_metrics, health_check_task, webhook_dispatcher
71 # Startup
72 logger.info("Starting Health API Service")
73 try:
74 health_monitor = create_health_monitor_from_env()
76 # Enable structured JSON logging now that we know cluster_id and region
77 configure_structured_logging(
78 service_name="health-api",
79 cluster_id=health_monitor.cluster_id,
80 region=health_monitor.region,
81 )
82 # Initialize metrics publisher for CloudWatch custom metrics
83 # Non-fatal: if credentials aren't available yet (e.g., Pod Identity agent
84 # still starting), we skip metrics but keep serving health checks.
85 try:
86 health_metrics = HealthMonitorMetrics(
87 cluster_name=health_monitor.cluster_id,
88 region=health_monitor.region,
89 )
90 except Exception as e:
91 logger.warning(f"Failed to initialize CloudWatch metrics publisher: {e}")
92 health_metrics = None
93 health_task = asyncio.create_task(background_health_monitor())
94 health_check_task = health_task
95 logger.info("Health monitoring started")
97 # Start webhook dispatcher for job event notifications
98 try:
99 webhook_dispatcher = create_webhook_dispatcher_from_env()
100 await webhook_dispatcher.start()
101 logger.info("Webhook dispatcher started")
102 except Exception as e:
103 logger.warning(f"Failed to start webhook dispatcher: {e}")
104 # Don't fail startup if webhook dispatcher fails - it's not critical
105 webhook_dispatcher = None
107 except Exception as e:
108 logger.error(f"Failed to start health monitoring: {e}")
109 raise
111 try:
112 yield
113 finally:
114 # Shutdown must also run when a lifespan consumer raises.
115 logger.info("Shutting down Health API Service")
116 health_task.cancel()
117 with contextlib.suppress(asyncio.CancelledError):
118 await health_task
119 if webhook_dispatcher:
120 await webhook_dispatcher.stop()
121 logger.info("Webhook dispatcher stopped")
122 logger.info("Health monitoring stopped")
123 tracing.shutdown_tracing()
126# Create FastAPI app with lifespan management
127app = FastAPI(
128 title="GCO Health Monitor API",
129 description="Health monitoring service for GCO (Global Capacity Orchestrator on AWS) EKS clusters",
130 version="1.0.0",
131 lifespan=lifespan,
132)
134# Server spans for every route except the probe and scrape paths; inert unless
135# GCO_TRACING_ENABLED=true. The webhook dispatcher in this process is never
136# traced (its URLs carry credentials).
137tracing.configure_tracing("health-monitor")
138tracing.instrument_fastapi_app(app)
140# Add authentication middleware
141app.add_middleware(AuthenticationMiddleware)
143# Expose Prometheus /metrics for the in-cluster observability scrape. The auth
144# middleware exempts /metrics, so the cluster Prometheus reaches it over the
145# existing service port without credentials.
146from gco.services.service_metrics import mount_metrics # noqa: E402
148mount_metrics(app, "health-monitor")
151async def background_health_monitor() -> None:
152 """
153 Background task that continuously monitors cluster health
154 and publishes metrics to CloudWatch
155 """
156 global current_health_status
158 while True:
159 try:
160 if health_monitor is None:
161 logger.warning("Health monitor not initialized, waiting...")
162 await asyncio.sleep(10)
163 continue
164 current_health_status = await health_monitor.get_health_status()
165 logger.debug(f"Health status updated: {current_health_status.status}")
167 # Periodically sync ALB hostname in SSM (self-healing)
168 await health_monitor.sync_alb_registration()
170 # Publish metrics to CloudWatch for dashboard visibility
171 if health_metrics and current_health_status:
172 try:
173 health_metrics.publish_resource_utilization(
174 cpu_percent=current_health_status.resource_utilization.cpu,
175 memory_percent=current_health_status.resource_utilization.memory,
176 gpu_percent=current_health_status.resource_utilization.gpu,
177 active_jobs=current_health_status.active_jobs,
178 )
179 # Also publish health status
180 threshold_violations = (
181 current_health_status.get_threshold_violations()
182 if hasattr(current_health_status, "get_threshold_violations")
183 else []
184 )
185 health_metrics.publish_health_status(
186 is_healthy=(current_health_status.status == "healthy"),
187 threshold_violations=threshold_violations,
188 )
189 logger.debug("Published health metrics to CloudWatch")
190 except Exception as e:
191 logger.warning(f"Failed to publish health metrics to CloudWatch: {e}")
193 # Sleep for 30 seconds before next check
194 await asyncio.sleep(30)
196 except asyncio.CancelledError:
197 logger.info("Background health monitoring cancelled")
198 break
199 except Exception as e:
200 logger.error(f"Error in background health monitoring: {e}")
201 await asyncio.sleep(10) # Shorter sleep on error
204@app.get("/")
205async def root() -> dict[str, Any]:
206 """Root endpoint with basic service information"""
207 return {
208 "service": "GCO Health Monitor API",
209 "version": "1.0.0",
210 "status": "running",
211 "endpoints": {
212 "health": "/api/v1/health",
213 "metrics": "/api/v1/metrics",
214 "status": "/api/v1/status",
215 },
216 }
219@app.get("/api/v1/health")
220async def health_check() -> JSONResponse:
221 """
222 Primary health check endpoint for ALB health checks
223 Returns 200 if cluster is healthy, 503 if unhealthy
224 """
225 global current_health_status
227 try:
228 status = current_health_status
229 # If we don't have a current status, get one immediately
230 if status is None:
231 if health_monitor is None:
232 raise HTTPException(status_code=503, detail="Health monitor not initialized")
233 status = await health_monitor.get_health_status()
234 current_health_status = status
236 # Check if status is too old (more than 2 minutes)
237 age_seconds = (datetime.now(UTC) - status.timestamp).total_seconds()
238 if age_seconds > 120 and health_monitor is not None: # 2 minutes
239 logger.warning(f"Health status is {age_seconds:.0f} seconds old, refreshing")
240 status = await health_monitor.get_health_status()
241 current_health_status = status
243 # Return appropriate HTTP status based on health
244 if status.status == "healthy":
245 return JSONResponse(
246 status_code=200,
247 content={
248 "status": "healthy",
249 "timestamp": status.timestamp.isoformat(),
250 "cluster_id": status.cluster_id,
251 "region": status.region,
252 },
253 )
254 return JSONResponse(
255 status_code=503,
256 content={
257 "status": "unhealthy",
258 "timestamp": status.timestamp.isoformat(),
259 "cluster_id": status.cluster_id,
260 "region": status.region,
261 "message": status.message,
262 },
263 )
265 except Exception as e:
266 logger.error(f"Health check failed: {e}")
267 return JSONResponse(
268 status_code=503,
269 content={
270 "status": "unhealthy",
271 "timestamp": datetime.now(UTC).isoformat(),
272 "error": "health monitor unavailable",
273 },
274 )
277@app.get("/api/v1/metrics")
278async def get_metrics() -> dict[str, Any]:
279 """
280 Detailed metrics endpoint with resource utilization information
281 """
282 global current_health_status
284 try:
285 # Get fresh metrics if needed
286 if current_health_status is None:
287 if health_monitor is None:
288 raise HTTPException(status_code=503, detail="Health monitor not initialized")
289 current_health_status = await health_monitor.get_health_status()
291 return {
292 "cluster_id": current_health_status.cluster_id,
293 "region": current_health_status.region,
294 "timestamp": current_health_status.timestamp.isoformat(),
295 "status": current_health_status.status,
296 "resource_utilization": {
297 "cpu_percent": round(current_health_status.resource_utilization.cpu, 2),
298 "memory_percent": round(current_health_status.resource_utilization.memory, 2),
299 "gpu_percent": round(current_health_status.resource_utilization.gpu, 2),
300 },
301 "thresholds": {
302 "cpu_threshold": current_health_status.thresholds.cpu_threshold,
303 "memory_threshold": current_health_status.thresholds.memory_threshold,
304 "gpu_threshold": current_health_status.thresholds.gpu_threshold,
305 },
306 "active_jobs": current_health_status.active_jobs,
307 "message": current_health_status.message,
308 "threshold_violations": (
309 current_health_status.get_threshold_violations()
310 if hasattr(current_health_status, "get_threshold_violations")
311 else []
312 ),
313 }
315 except Exception as e:
316 logger.error(f"Failed to get metrics: {e}")
317 raise HTTPException(status_code=500, detail=f"Failed to get metrics: {e!s}") from e
320@app.get("/api/v1/status")
321async def get_status() -> dict[str, Any]:
322 """
323 Service status endpoint with operational information
324 """
326 # Get webhook dispatcher metrics if available
327 webhook_metrics = None
328 if webhook_dispatcher:
329 webhook_metrics = webhook_dispatcher.get_metrics()
331 service_status = {
332 "service": "GCO Health Monitor API",
333 "version": "1.0.0",
334 "uptime_seconds": None, # Could be implemented with start time tracking
335 "health_monitor_initialized": health_monitor is not None,
336 "background_task_running": health_check_task is not None and not health_check_task.done(),
337 "last_health_check": (
338 current_health_status.timestamp.isoformat() if current_health_status else None
339 ),
340 "webhook_dispatcher": {
341 "enabled": webhook_dispatcher is not None,
342 "running": webhook_metrics.get("running", False) if webhook_metrics else False,
343 "deliveries_total": (
344 webhook_metrics.get("deliveries_total", 0) if webhook_metrics else 0
345 ),
346 "deliveries_success": (
347 webhook_metrics.get("deliveries_success", 0) if webhook_metrics else 0
348 ),
349 "deliveries_failed": (
350 webhook_metrics.get("deliveries_failed", 0) if webhook_metrics else 0
351 ),
352 "cached_jobs": webhook_metrics.get("cached_jobs", 0) if webhook_metrics else 0,
353 },
354 "environment": {
355 "cluster_name": os.getenv("CLUSTER_NAME", "unknown"),
356 "region": os.getenv("REGION", "unknown"),
357 "cpu_threshold": os.getenv("CPU_THRESHOLD", "80"),
358 "memory_threshold": os.getenv("MEMORY_THRESHOLD", "85"),
359 "gpu_threshold": os.getenv("GPU_THRESHOLD", "90"),
360 },
361 }
363 return service_status
366@app.get("/healthz")
367async def kubernetes_health_check() -> dict[str, str]:
368 """
369 Kubernetes-style health check endpoint
370 Simple endpoint that returns 200 if the service is running
371 """
372 return {"status": "ok"}
375@app.get("/readyz")
376async def kubernetes_readiness_check() -> dict[str, str]:
377 """
378 Kubernetes-style readiness check endpoint
379 Returns 200 if the service is ready to serve traffic
380 """
382 if health_monitor is None:
383 raise HTTPException(status_code=503, detail="Health monitor not ready")
385 return {"status": "ready"}
388# Error handlers
389@app.exception_handler(Exception)
390async def global_exception_handler(request: Request, exc: Exception) -> JSONResponse:
391 """Global exception handler for unhandled errors"""
392 logger.error(f"Unhandled exception: {exc}")
393 return JSONResponse(
394 status_code=500,
395 content={
396 "error": "Internal server error",
397 "detail": str(exc) if os.getenv("DEBUG") else "An unexpected error occurred",
398 },
399 )
402def create_app() -> FastAPI:
403 """Factory function to create the FastAPI app"""
404 return app
407# The pod manifest gives the kubelet terminationGracePeriodSeconds > preStop +
408# this budget, so Uvicorn can finish in-flight requests before SIGKILL. The
409# same variable drives the TLS sidecar's drain (gco.services.tls_proxy).
410DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS = 20
413def _run_server() -> None:
414 """Run Uvicorn with the same drain budget declared by the pod manifest."""
415 import uvicorn
417 from gco.services.uvicorn_runtime import describe_uvicorn_runtime
419 host = os.getenv("HOST", "0.0.0.0") # container listener
420 port = int(os.getenv("PORT", "8080"))
421 log_level = os.getenv("LOG_LEVEL", "info").lower()
422 graceful_shutdown_seconds = int(
423 os.getenv(
424 "GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS",
425 str(DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS),
426 )
427 )
429 loop_impl, http_impl = describe_uvicorn_runtime()
430 logger.info(
431 "Starting Health API on %s:%d (uvicorn loop=%s http=%s)", host, port, loop_impl, http_impl
432 )
434 uvicorn.run(
435 "gco.services.health_api:app",
436 host=host,
437 port=port,
438 log_level=log_level,
439 reload=False,
440 timeout_graceful_shutdown=graceful_shutdown_seconds,
441 )
444if __name__ == "__main__":
445 _run_server()