Coverage for src / lexigram / admin / controllers / pool_health.py: 17%
105 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-13 22:14 +0800
1"""Pool Health Controller for monitoring connection pools.
3Provides HTTP endpoints for checking pool health, viewing statistics,
4and managing connection pools in the admin interface.
5"""
7from __future__ import annotations
9from typing import TYPE_CHECKING
11from starlette.requests import Request
12from starlette.responses import JSONResponse, Response
14from lexigram.admin.controllers.base import AdminController
15from lexigram.logging import get_logger
17if TYPE_CHECKING:
18 from lexigram.admin.engine.renderer import AdminRenderer
19 from lexigram.contracts.core import TaskManagerProtocol
20 from lexigram.contracts.infra import PoolManagerProtocol
21from lexigram.di.decorators import inject
23logger = get_logger(__name__)
26@inject
27class PoolHealthController(AdminController):
28 """Controller for connection pool health monitoring.
30 Routes:
31 - GET /admin/pools/health - Get health status for all pools
32 - GET /admin/pools/health/{name} - Get health for specific pool
33 - POST /admin/pools/{name}/reconnect - Force reconnect a pool
34 """
36 def __init__(
37 self,
38 renderer: AdminRenderer | None = None,
39 pool_manager: PoolManagerProtocol | None = None,
40 task_manager: TaskManagerProtocol | None = None,
41 ) -> None:
42 """Initialize the pool health controller.
44 Args:
45 renderer: AdminRenderer instance (DI-injected, optional).
46 pool_manager: PoolManager instance for querying pool state (optional).
47 task_manager: TaskManagerProtocol instance (optional).
48 """
49 super().__init__(renderer=renderer, task_manager=task_manager)
50 self._pool_manager = pool_manager
52 def _require_pool_manager(self) -> JSONResponse | None:
53 """Return a 503 response if no pool manager is available, otherwise None."""
54 if self._pool_manager is None:
55 return JSONResponse(
56 {"error": "Pool manager is not available"},
57 status_code=503,
58 )
59 return None
61 async def get_all_health(self, request: Request) -> JSONResponse:
62 """Get health status for all connection pools."""
63 if (missing := self._require_pool_manager()) is not None:
64 return missing
65 try:
66 all_stats = self._pool_manager.get_stats() # type: ignore[union-attr]
67 pools_data = {}
68 for name, stats in all_stats.items():
69 is_healthy = stats.pool_utilization < 90.0
70 pools_data[name] = {
71 "name": name,
72 "is_healthy": is_healthy,
73 "stats": stats.__dict__,
74 "last_check": stats.last_health_check,
75 "error": None if is_healthy else "High utilization",
76 }
78 total = len(pools_data)
79 healthy = sum(1 for p in pools_data.values() if p["is_healthy"])
81 return JSONResponse(
82 {
83 "pools": pools_data,
84 "summary": {
85 "total_pools": total,
86 "healthy_pools": healthy,
87 "unhealthy_pools": total - healthy,
88 },
89 },
90 )
91 except Exception as e: # noqa: BLE001 — controller boundary; unexpected errors become HTTP 500 responses
92 logger.exception("Failed to get all pool health")
93 return JSONResponse(
94 {"error": "Failed to get pool health", "detail": str(e)},
95 status_code=500,
96 )
98 async def get_pool_health(self, request: Request) -> JSONResponse:
99 """Get health status for a specific pool."""
100 pool_name = request.path_params.get("name")
101 if not pool_name:
102 return JSONResponse({"error": "Pool name required"}, status_code=400)
103 if (missing := self._require_pool_manager()) is not None:
104 return missing
105 try:
106 pool = await self._pool_manager.get_pool(pool_name) # type: ignore[union-attr]
107 stats = pool.get_stats()
108 is_healthy = stats.pool_utilization < 90.0
109 return JSONResponse(
110 {
111 "name": pool_name,
112 "is_healthy": is_healthy,
113 "stats": stats.__dict__,
114 "last_check": stats.last_health_check,
115 "error": None if is_healthy else "High utilization",
116 },
117 )
118 except KeyError:
119 return JSONResponse(
120 {"error": f"Pool '{pool_name}' not found"},
121 status_code=404,
122 )
123 except Exception as e: # noqa: BLE001 — controller boundary; unexpected errors become HTTP 500 responses
124 logger.exception("Failed to get health for pool %s", pool_name)
125 return JSONResponse(
126 {"error": "Failed to get pool health", "detail": str(e)},
127 status_code=500,
128 )
130 async def reconnect_pool(self, request: Request) -> JSONResponse:
131 """Force reconnect a specific pool."""
132 pool_name = request.path_params.get("name")
133 if not pool_name:
134 return JSONResponse({"error": "Pool name required"}, status_code=400)
135 if (missing := self._require_pool_manager()) is not None:
136 return missing
137 try:
138 pool = await self._pool_manager.get_pool(pool_name) # type: ignore[union-attr]
139 await pool.close()
140 return JSONResponse(
141 {"message": f"Pool '{pool_name}' closed for reconnection"},
142 )
143 except KeyError:
144 return JSONResponse(
145 {"error": f"Pool '{pool_name}' not found"},
146 status_code=404,
147 )
148 except Exception as e: # noqa: BLE001 — controller boundary; unexpected errors become HTTP 500 responses
149 logger.exception("Failed to reconnect pool %s", pool_name)
150 return JSONResponse(
151 {"error": "Failed to reconnect pool", "detail": str(e)},
152 status_code=500,
153 )
155 async def get_pool_stats_summary(self, request: Request) -> JSONResponse:
156 """Get aggregated statistics across all pools."""
157 if (missing := self._require_pool_manager()) is not None:
158 return missing
159 try:
160 all_stats = self._pool_manager.get_stats() # type: ignore[union-attr]
161 total_connections = sum(s.total_connections for s in all_stats.values())
162 active_connections = sum(s.active_connections for s in all_stats.values())
163 idle_connections = sum(s.idle_connections for s in all_stats.values())
164 total_created = sum(s.created_connections for s in all_stats.values())
165 total_destroyed = sum(s.destroyed_connections for s in all_stats.values())
167 utilizations = [s.pool_utilization for s in all_stats.values()]
168 avg_utilization = (
169 sum(utilizations) / len(utilizations) if utilizations else 0.0
170 )
171 max_utilization = max(utilizations) if utilizations else 0.0
173 return JSONResponse(
174 {
175 "total_connections": total_connections,
176 "active_connections": active_connections,
177 "idle_connections": idle_connections,
178 "total_created": total_created,
179 "total_destroyed": total_destroyed,
180 "avg_utilization": avg_utilization,
181 "max_utilization": max_utilization,
182 },
183 )
184 except Exception as e: # noqa: BLE001 — controller boundary; unexpected errors become HTTP 500 responses
185 logger.exception("Failed to compute pool stats summary")
186 return JSONResponse(
187 {"error": "Failed to get pool stats", "detail": str(e)},
188 status_code=500,
189 )
191 async def get_prometheus_metrics(self, request: Request) -> Response:
192 """Export pool metrics in Prometheus format."""
193 if (missing := self._require_pool_manager()) is not None:
194 return missing
195 try:
196 all_stats = self._pool_manager.get_stats() # type: ignore[union-attr]
197 lines = []
198 for name, stats in all_stats.items():
199 labels = f'pool="{name}"'
200 lines.append(
201 f"pool_active_connections{{{labels}}} {stats.active_connections}",
202 )
203 lines.append(
204 f"pool_total_connections{{{labels}}} {stats.total_connections}",
205 )
206 lines.append(f"pool_utilization{{{labels}}} {stats.pool_utilization}")
208 return Response(
209 "\n".join(lines) + "\n",
210 media_type="text/plain; version=0.0.4",
211 )
212 except Exception as e: # noqa: BLE001 — controller boundary; unexpected errors become HTTP 500 responses
213 logger.exception("Failed to export metrics")
214 return JSONResponse(
215 {"error": "Failed to export metrics", "detail": str(e)},
216 status_code=500,
217 )
219 async def get_json_metrics(self, request: Request) -> JSONResponse:
220 """Export pool metrics in JSON format."""
221 if (missing := self._require_pool_manager()) is not None:
222 return missing
223 try:
224 all_stats = self._pool_manager.get_stats() # type: ignore[union-attr]
225 metrics_json = {name: stats.__dict__ for name, stats in all_stats.items()}
226 return JSONResponse(metrics_json)
227 except Exception as e: # noqa: BLE001 — controller boundary; unexpected errors become HTTP 500 responses
228 logger.exception("Failed to export metrics as JSON")
229 return JSONResponse(
230 {"error": "Failed to export metrics", "detail": str(e)},
231 status_code=500,
232 )