Coverage for gco / services / health_api.py: 100.00%
150 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +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 datetime
34from typing import Any
36from fastapi import FastAPI, HTTPException, Request
37from fastapi.responses import JSONResponse
39from gco.models import HealthStatus
40from gco.services.auth_middleware import AuthenticationMiddleware
41from gco.services.health_monitor import HealthMonitor, create_health_monitor_from_env
42from gco.services.metrics_publisher import HealthMonitorMetrics
43from gco.services.structured_logging import configure_structured_logging
44from gco.services.webhook_dispatcher import (
45 WebhookDispatcher,
46 create_webhook_dispatcher_from_env,
47)
49logging.basicConfig(
50 level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
51)
52logger = logging.getLogger(__name__)
54# Global health monitor instance
55health_monitor: HealthMonitor | None = None
56health_metrics: HealthMonitorMetrics | None = None
57webhook_dispatcher: WebhookDispatcher | None = None
58current_health_status: HealthStatus | None = None
59health_check_task = None
62@asynccontextmanager
63async def lifespan(app: FastAPI) -> AsyncIterator[None]:
64 """
65 Application lifespan manager - starts and stops background health monitoring
66 and webhook dispatcher.
67 """
68 global health_monitor, health_metrics, health_check_task, webhook_dispatcher
70 # Startup
71 logger.info("Starting Health API Service")
72 try:
73 health_monitor = create_health_monitor_from_env()
75 # Enable structured JSON logging now that we know cluster_id and region
76 configure_structured_logging(
77 service_name="health-api",
78 cluster_id=health_monitor.cluster_id,
79 region=health_monitor.region,
80 )
81 # Initialize metrics publisher for CloudWatch custom metrics
82 # Non-fatal: if credentials aren't available yet (e.g., Pod Identity agent
83 # still starting), we skip metrics but keep serving health checks.
84 try:
85 health_metrics = HealthMonitorMetrics(
86 cluster_name=health_monitor.cluster_id,
87 region=health_monitor.region,
88 )
89 except Exception as e:
90 logger.warning(f"Failed to initialize CloudWatch metrics publisher: {e}")
91 health_metrics = None
92 health_task = asyncio.create_task(background_health_monitor())
93 health_check_task = health_task
94 logger.info("Health monitoring started")
96 # Start webhook dispatcher for job event notifications
97 try:
98 webhook_dispatcher = create_webhook_dispatcher_from_env()
99 await webhook_dispatcher.start()
100 logger.info("Webhook dispatcher started")
101 except Exception as e:
102 logger.warning(f"Failed to start webhook dispatcher: {e}")
103 # Don't fail startup if webhook dispatcher fails - it's not critical
104 webhook_dispatcher = None
106 except Exception as e:
107 logger.error(f"Failed to start health monitoring: {e}")
108 raise
110 try:
111 yield
112 finally:
113 # Shutdown must also run when a lifespan consumer raises.
114 logger.info("Shutting down Health API Service")
115 health_task.cancel()
116 with contextlib.suppress(asyncio.CancelledError):
117 await health_task
118 if webhook_dispatcher:
119 await webhook_dispatcher.stop()
120 logger.info("Webhook dispatcher stopped")
121 logger.info("Health monitoring stopped")
124# Create FastAPI app with lifespan management
125app = FastAPI(
126 title="GCO Health Monitor API",
127 description="Health monitoring service for GCO (Global Capacity Orchestrator on AWS) EKS clusters",
128 version="1.0.0",
129 lifespan=lifespan,
130)
132# Add authentication middleware
133app.add_middleware(AuthenticationMiddleware)
135# Expose Prometheus /metrics for the in-cluster observability scrape. The auth
136# middleware exempts /metrics, so the cluster Prometheus reaches it over the
137# existing service port without credentials.
138from gco.services.service_metrics import mount_metrics # noqa: E402
140mount_metrics(app, "health-monitor")
143async def background_health_monitor() -> None:
144 """
145 Background task that continuously monitors cluster health
146 and publishes metrics to CloudWatch
147 """
148 global current_health_status
150 while True:
151 try:
152 if health_monitor is None:
153 logger.warning("Health monitor not initialized, waiting...")
154 await asyncio.sleep(10)
155 continue
156 current_health_status = await health_monitor.get_health_status()
157 logger.debug(f"Health status updated: {current_health_status.status}")
159 # Periodically sync ALB hostname in SSM (self-healing)
160 await health_monitor.sync_alb_registration()
162 # Publish metrics to CloudWatch for dashboard visibility
163 if health_metrics and current_health_status:
164 try:
165 health_metrics.publish_resource_utilization(
166 cpu_percent=current_health_status.resource_utilization.cpu,
167 memory_percent=current_health_status.resource_utilization.memory,
168 gpu_percent=current_health_status.resource_utilization.gpu,
169 active_jobs=current_health_status.active_jobs,
170 )
171 # Also publish health status
172 threshold_violations = (
173 current_health_status.get_threshold_violations()
174 if hasattr(current_health_status, "get_threshold_violations")
175 else []
176 )
177 health_metrics.publish_health_status(
178 is_healthy=(current_health_status.status == "healthy"),
179 threshold_violations=threshold_violations,
180 )
181 logger.debug("Published health metrics to CloudWatch")
182 except Exception as e:
183 logger.warning(f"Failed to publish health metrics to CloudWatch: {e}")
185 # Sleep for 30 seconds before next check
186 await asyncio.sleep(30)
188 except asyncio.CancelledError:
189 logger.info("Background health monitoring cancelled")
190 break
191 except Exception as e:
192 logger.error(f"Error in background health monitoring: {e}")
193 await asyncio.sleep(10) # Shorter sleep on error
196@app.get("/")
197async def root() -> dict[str, Any]:
198 """Root endpoint with basic service information"""
199 return {
200 "service": "GCO Health Monitor API",
201 "version": "1.0.0",
202 "status": "running",
203 "endpoints": {
204 "health": "/api/v1/health",
205 "metrics": "/api/v1/metrics",
206 "status": "/api/v1/status",
207 },
208 }
211@app.get("/api/v1/health")
212async def health_check() -> JSONResponse:
213 """
214 Primary health check endpoint for ALB health checks
215 Returns 200 if cluster is healthy, 503 if unhealthy
216 """
217 global current_health_status
219 try:
220 status = current_health_status
221 # If we don't have a current status, get one immediately
222 if status is None:
223 if health_monitor is None:
224 raise HTTPException(status_code=503, detail="Health monitor not initialized")
225 status = await health_monitor.get_health_status()
226 current_health_status = status
228 # Check if status is too old (more than 2 minutes)
229 age_seconds = (datetime.now() - status.timestamp).total_seconds()
230 if age_seconds > 120 and health_monitor is not None: # 2 minutes
231 logger.warning(f"Health status is {age_seconds:.0f} seconds old, refreshing")
232 status = await health_monitor.get_health_status()
233 current_health_status = status
235 # Return appropriate HTTP status based on health
236 if status.status == "healthy":
237 return JSONResponse(
238 status_code=200,
239 content={
240 "status": "healthy",
241 "timestamp": status.timestamp.isoformat(),
242 "cluster_id": status.cluster_id,
243 "region": status.region,
244 },
245 )
246 return JSONResponse(
247 status_code=503,
248 content={
249 "status": "unhealthy",
250 "timestamp": status.timestamp.isoformat(),
251 "cluster_id": status.cluster_id,
252 "region": status.region,
253 "message": status.message,
254 },
255 )
257 except Exception as e:
258 logger.error(f"Health check failed: {e}")
259 return JSONResponse(
260 status_code=503,
261 content={
262 "status": "unhealthy",
263 "timestamp": datetime.now().isoformat(),
264 "error": "health monitor unavailable",
265 },
266 )
269@app.get("/api/v1/metrics")
270async def get_metrics() -> dict[str, Any]:
271 """
272 Detailed metrics endpoint with resource utilization information
273 """
274 global current_health_status
276 try:
277 # Get fresh metrics if needed
278 if current_health_status is None:
279 if health_monitor is None:
280 raise HTTPException(status_code=503, detail="Health monitor not initialized")
281 current_health_status = await health_monitor.get_health_status()
283 return {
284 "cluster_id": current_health_status.cluster_id,
285 "region": current_health_status.region,
286 "timestamp": current_health_status.timestamp.isoformat(),
287 "status": current_health_status.status,
288 "resource_utilization": {
289 "cpu_percent": round(current_health_status.resource_utilization.cpu, 2),
290 "memory_percent": round(current_health_status.resource_utilization.memory, 2),
291 "gpu_percent": round(current_health_status.resource_utilization.gpu, 2),
292 },
293 "thresholds": {
294 "cpu_threshold": current_health_status.thresholds.cpu_threshold,
295 "memory_threshold": current_health_status.thresholds.memory_threshold,
296 "gpu_threshold": current_health_status.thresholds.gpu_threshold,
297 },
298 "active_jobs": current_health_status.active_jobs,
299 "message": current_health_status.message,
300 "threshold_violations": (
301 current_health_status.get_threshold_violations()
302 if hasattr(current_health_status, "get_threshold_violations")
303 else []
304 ),
305 }
307 except Exception as e:
308 logger.error(f"Failed to get metrics: {e}")
309 raise HTTPException(status_code=500, detail=f"Failed to get metrics: {e!s}") from e
312@app.get("/api/v1/status")
313async def get_status() -> dict[str, Any]:
314 """
315 Service status endpoint with operational information
316 """
318 # Get webhook dispatcher metrics if available
319 webhook_metrics = None
320 if webhook_dispatcher:
321 webhook_metrics = webhook_dispatcher.get_metrics()
323 service_status = {
324 "service": "GCO Health Monitor API",
325 "version": "1.0.0",
326 "uptime_seconds": None, # Could be implemented with start time tracking
327 "health_monitor_initialized": health_monitor is not None,
328 "background_task_running": health_check_task is not None and not health_check_task.done(),
329 "last_health_check": (
330 current_health_status.timestamp.isoformat() if current_health_status else None
331 ),
332 "webhook_dispatcher": {
333 "enabled": webhook_dispatcher is not None,
334 "running": webhook_metrics.get("running", False) if webhook_metrics else False,
335 "deliveries_total": (
336 webhook_metrics.get("deliveries_total", 0) if webhook_metrics else 0
337 ),
338 "deliveries_success": (
339 webhook_metrics.get("deliveries_success", 0) if webhook_metrics else 0
340 ),
341 "deliveries_failed": (
342 webhook_metrics.get("deliveries_failed", 0) if webhook_metrics else 0
343 ),
344 "cached_jobs": webhook_metrics.get("cached_jobs", 0) if webhook_metrics else 0,
345 },
346 "environment": {
347 "cluster_name": os.getenv("CLUSTER_NAME", "unknown"),
348 "region": os.getenv("REGION", "unknown"),
349 "cpu_threshold": os.getenv("CPU_THRESHOLD", "80"),
350 "memory_threshold": os.getenv("MEMORY_THRESHOLD", "85"),
351 "gpu_threshold": os.getenv("GPU_THRESHOLD", "90"),
352 },
353 }
355 return service_status
358@app.get("/healthz")
359async def kubernetes_health_check() -> dict[str, str]:
360 """
361 Kubernetes-style health check endpoint
362 Simple endpoint that returns 200 if the service is running
363 """
364 return {"status": "ok"}
367@app.get("/readyz")
368async def kubernetes_readiness_check() -> dict[str, str]:
369 """
370 Kubernetes-style readiness check endpoint
371 Returns 200 if the service is ready to serve traffic
372 """
374 if health_monitor is None:
375 raise HTTPException(status_code=503, detail="Health monitor not ready")
377 return {"status": "ready"}
380# Error handlers
381@app.exception_handler(Exception)
382async def global_exception_handler(request: Request, exc: Exception) -> JSONResponse:
383 """Global exception handler for unhandled errors"""
384 logger.error(f"Unhandled exception: {exc}")
385 return JSONResponse(
386 status_code=500,
387 content={
388 "error": "Internal server error",
389 "detail": str(exc) if os.getenv("DEBUG") else "An unexpected error occurred",
390 },
391 )
394def create_app() -> FastAPI:
395 """Factory function to create the FastAPI app"""
396 return app
399# The pod manifest gives the kubelet terminationGracePeriodSeconds > preStop +
400# this budget, so Uvicorn can finish in-flight requests before SIGKILL. The
401# same variable drives the TLS sidecar's drain (gco.services.tls_proxy).
402DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS = 20
405def _run_server() -> None:
406 """Run Uvicorn with the same drain budget declared by the pod manifest."""
407 import uvicorn
409 host = os.getenv("HOST", "0.0.0.0") # nosec B104 — container listener
410 port = int(os.getenv("PORT", "8080"))
411 log_level = os.getenv("LOG_LEVEL", "info").lower()
412 graceful_shutdown_seconds = int(
413 os.getenv(
414 "GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS",
415 str(DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS),
416 )
417 )
419 logger.info("Starting Health API on %s:%d", host, port)
421 uvicorn.run(
422 "gco.services.health_api:app",
423 host=host,
424 port=port,
425 log_level=log_level,
426 reload=False,
427 timeout_graceful_shutdown=graceful_shutdown_seconds,
428 )
431if __name__ == "__main__":
432 _run_server()