Coverage for gco/services/health_api.py: 82.93%
138 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +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_check_task = asyncio.create_task(background_health_monitor())
93 logger.info("Health monitoring started")
95 # Start webhook dispatcher for job event notifications
96 try:
97 webhook_dispatcher = create_webhook_dispatcher_from_env()
98 await webhook_dispatcher.start()
99 logger.info("Webhook dispatcher started")
100 except Exception as e:
101 logger.warning(f"Failed to start webhook dispatcher: {e}")
102 # Don't fail startup if webhook dispatcher fails - it's not critical
103 webhook_dispatcher = None
105 except Exception as e:
106 logger.error(f"Failed to start health monitoring: {e}")
107 raise
109 yield
111 # Shutdown
112 logger.info("Shutting down Health API Service")
113 if health_check_task: 113 ↛ 117line 113 didn't jump to line 117 because the condition on line 113 was always true
114 health_check_task.cancel()
115 with contextlib.suppress(asyncio.CancelledError):
116 await health_check_task
117 if webhook_dispatcher: 117 ↛ 118line 117 didn't jump to line 118 because the condition on line 117 was never true
118 await webhook_dispatcher.stop()
119 logger.info("Webhook dispatcher stopped")
120 logger.info("Health monitoring stopped")
123# Create FastAPI app with lifespan management
124app = FastAPI(
125 title="GCO Health Monitor API",
126 description="Health monitoring service for GCO (Global Capacity Orchestrator on AWS) EKS clusters",
127 version="1.0.0",
128 lifespan=lifespan,
129)
131# Add authentication middleware
132app.add_middleware(AuthenticationMiddleware)
134# Expose Prometheus /metrics for the in-cluster observability scrape. The auth
135# middleware exempts /metrics, so the cluster Prometheus reaches it over the
136# existing service port without credentials.
137from gco.services.service_metrics import mount_metrics # noqa: E402
139mount_metrics(app, "health-monitor")
142async def background_health_monitor() -> None:
143 """
144 Background task that continuously monitors cluster health
145 and publishes metrics to CloudWatch
146 """
147 global current_health_status
149 while True:
150 try:
151 if health_monitor is None: 151 ↛ 152line 151 didn't jump to line 152 because the condition on line 151 was never true
152 logger.warning("Health monitor not initialized, waiting...")
153 await asyncio.sleep(10)
154 continue
155 current_health_status = await health_monitor.get_health_status()
156 logger.debug(f"Health status updated: {current_health_status.status}")
158 # Periodically sync ALB hostname in SSM (self-healing)
159 await health_monitor.sync_alb_registration()
161 # Publish metrics to CloudWatch for dashboard visibility
162 if health_metrics and current_health_status:
163 try:
164 health_metrics.publish_resource_utilization(
165 cpu_percent=current_health_status.resource_utilization.cpu,
166 memory_percent=current_health_status.resource_utilization.memory,
167 gpu_percent=current_health_status.resource_utilization.gpu,
168 active_jobs=current_health_status.active_jobs,
169 )
170 # Also publish health status
171 threshold_violations = (
172 current_health_status.get_threshold_violations()
173 if hasattr(current_health_status, "get_threshold_violations")
174 else []
175 )
176 health_metrics.publish_health_status(
177 is_healthy=(current_health_status.status == "healthy"),
178 threshold_violations=threshold_violations,
179 )
180 logger.debug("Published health metrics to CloudWatch")
181 except Exception as e:
182 logger.warning(f"Failed to publish health metrics to CloudWatch: {e}")
184 # Sleep for 30 seconds before next check
185 await asyncio.sleep(30)
187 except asyncio.CancelledError:
188 logger.info("Background health monitoring cancelled")
189 break
190 except Exception as e:
191 logger.error(f"Error in background health monitoring: {e}")
192 await asyncio.sleep(10) # Shorter sleep on error
195@app.get("/")
196async def root() -> dict[str, Any]:
197 """Root endpoint with basic service information"""
198 return {
199 "service": "GCO Health Monitor API",
200 "version": "1.0.0",
201 "status": "running",
202 "endpoints": {
203 "health": "/api/v1/health",
204 "metrics": "/api/v1/metrics",
205 "status": "/api/v1/status",
206 },
207 }
210@app.get("/api/v1/health")
211async def health_check() -> JSONResponse:
212 """
213 Primary health check endpoint for ALB health checks
214 Returns 200 if cluster is healthy, 503 if unhealthy
215 """
216 global current_health_status
218 try:
219 # If we don't have a current status, get one immediately
220 if current_health_status is None:
221 if health_monitor is None:
222 raise HTTPException(status_code=503, detail="Health monitor not initialized")
223 current_health_status = await health_monitor.get_health_status()
225 # Check if status is too old (more than 2 minutes)
226 if current_health_status: 226 ↛ 233line 226 didn't jump to line 233 because the condition on line 226 was always true
227 age_seconds = (datetime.now() - current_health_status.timestamp).total_seconds()
228 if age_seconds > 120 and health_monitor is not None: # 2 minutes
229 logger.warning(f"Health status is {age_seconds:.0f} seconds old, refreshing")
230 current_health_status = await health_monitor.get_health_status()
232 # Return appropriate HTTP status based on health
233 if current_health_status.status == "healthy":
234 return JSONResponse(
235 status_code=200,
236 content={
237 "status": "healthy",
238 "timestamp": current_health_status.timestamp.isoformat(),
239 "cluster_id": current_health_status.cluster_id,
240 "region": current_health_status.region,
241 },
242 )
243 return JSONResponse(
244 status_code=503,
245 content={
246 "status": "unhealthy",
247 "timestamp": current_health_status.timestamp.isoformat(),
248 "cluster_id": current_health_status.cluster_id,
249 "region": current_health_status.region,
250 "message": current_health_status.message,
251 },
252 )
254 except Exception as e:
255 logger.error(f"Health check failed: {e}")
256 return JSONResponse(
257 status_code=503,
258 content={
259 "status": "unhealthy",
260 "timestamp": datetime.now().isoformat(),
261 "error": "health monitor unavailable",
262 },
263 )
266@app.get("/api/v1/metrics")
267async def get_metrics() -> dict[str, Any]:
268 """
269 Detailed metrics endpoint with resource utilization information
270 """
271 global current_health_status
273 try:
274 # Get fresh metrics if needed
275 if current_health_status is None:
276 if health_monitor is None:
277 raise HTTPException(status_code=503, detail="Health monitor not initialized")
278 current_health_status = await health_monitor.get_health_status()
280 return {
281 "cluster_id": current_health_status.cluster_id,
282 "region": current_health_status.region,
283 "timestamp": current_health_status.timestamp.isoformat(),
284 "status": current_health_status.status,
285 "resource_utilization": {
286 "cpu_percent": round(current_health_status.resource_utilization.cpu, 2),
287 "memory_percent": round(current_health_status.resource_utilization.memory, 2),
288 "gpu_percent": round(current_health_status.resource_utilization.gpu, 2),
289 },
290 "thresholds": {
291 "cpu_threshold": current_health_status.thresholds.cpu_threshold,
292 "memory_threshold": current_health_status.thresholds.memory_threshold,
293 "gpu_threshold": current_health_status.thresholds.gpu_threshold,
294 },
295 "active_jobs": current_health_status.active_jobs,
296 "message": current_health_status.message,
297 "threshold_violations": (
298 current_health_status.get_threshold_violations()
299 if hasattr(current_health_status, "get_threshold_violations")
300 else []
301 ),
302 }
304 except Exception as e:
305 logger.error(f"Failed to get metrics: {e}")
306 raise HTTPException(status_code=500, detail=f"Failed to get metrics: {e!s}") from e
309@app.get("/api/v1/status")
310async def get_status() -> dict[str, Any]:
311 """
312 Service status endpoint with operational information
313 """
315 # Get webhook dispatcher metrics if available
316 webhook_metrics = None
317 if webhook_dispatcher: 317 ↛ 318line 317 didn't jump to line 318 because the condition on line 317 was never true
318 webhook_metrics = webhook_dispatcher.get_metrics()
320 service_status = {
321 "service": "GCO Health Monitor API",
322 "version": "1.0.0",
323 "uptime_seconds": None, # Could be implemented with start time tracking
324 "health_monitor_initialized": health_monitor is not None,
325 "background_task_running": health_check_task is not None and not health_check_task.done(),
326 "last_health_check": (
327 current_health_status.timestamp.isoformat() if current_health_status else None
328 ),
329 "webhook_dispatcher": {
330 "enabled": webhook_dispatcher is not None,
331 "running": webhook_metrics.get("running", False) if webhook_metrics else False,
332 "deliveries_total": (
333 webhook_metrics.get("deliveries_total", 0) if webhook_metrics else 0
334 ),
335 "deliveries_success": (
336 webhook_metrics.get("deliveries_success", 0) if webhook_metrics else 0
337 ),
338 "deliveries_failed": (
339 webhook_metrics.get("deliveries_failed", 0) if webhook_metrics else 0
340 ),
341 "cached_jobs": webhook_metrics.get("cached_jobs", 0) if webhook_metrics else 0,
342 },
343 "environment": {
344 "cluster_name": os.getenv("CLUSTER_NAME", "unknown"),
345 "region": os.getenv("REGION", "unknown"),
346 "cpu_threshold": os.getenv("CPU_THRESHOLD", "80"),
347 "memory_threshold": os.getenv("MEMORY_THRESHOLD", "85"),
348 "gpu_threshold": os.getenv("GPU_THRESHOLD", "90"),
349 },
350 }
352 return service_status
355@app.get("/healthz")
356async def kubernetes_health_check() -> dict[str, str]:
357 """
358 Kubernetes-style health check endpoint
359 Simple endpoint that returns 200 if the service is running
360 """
361 return {"status": "ok"}
364@app.get("/readyz")
365async def kubernetes_readiness_check() -> dict[str, str]:
366 """
367 Kubernetes-style readiness check endpoint
368 Returns 200 if the service is ready to serve traffic
369 """
371 if health_monitor is None:
372 raise HTTPException(status_code=503, detail="Health monitor not ready")
374 return {"status": "ready"}
377# Error handlers
378@app.exception_handler(Exception)
379async def global_exception_handler(request: Request, exc: Exception) -> JSONResponse:
380 """Global exception handler for unhandled errors"""
381 logger.error(f"Unhandled exception: {exc}")
382 return JSONResponse(
383 status_code=500,
384 content={
385 "error": "Internal server error",
386 "detail": str(exc) if os.getenv("DEBUG") else "An unexpected error occurred",
387 },
388 )
391def create_app() -> FastAPI:
392 """Factory function to create the FastAPI app"""
393 return app
396if __name__ == "__main__":
397 import uvicorn
399 # Configuration from environment variables
400 host = os.getenv("HOST", "0.0.0.0") # nosec B104 — must bind all interfaces inside K8s pod
401 port = int(os.getenv("PORT", "8080"))
402 log_level = os.getenv("LOG_LEVEL", "info").lower()
404 logger.info(f"Starting Health API on {host}:{port}")
406 uvicorn.run(
407 "gco.services.health_api:app", host=host, port=port, log_level=log_level, reload=False
408 )