Coverage for src/lexigram/admin/controllers/pool_health.py: 88%

1"""Pool Health Controller for monitoring connection pools. 

2 

3Provides HTTP endpoints for checking pool health, viewing statistics, 

4and managing connection pools in the admin interface. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import TYPE_CHECKING, Any 

10 

11from starlette.requests import Request 

12from starlette.responses import JSONResponse, Response 

13 

14from lexigram.admin.auth.types import AdminSecurityEventType 

15from lexigram.admin.config import AdminRbacConfig 

16from lexigram.admin.controllers.base import AdminController 

17from lexigram.admin.rbac.super_admin import is_super_admin 

18from lexigram.logging import get_logger 

19 

20if TYPE_CHECKING: 

21 from lexigram.admin.auth.protocols import AdminAuditLogServiceProtocol 

22 from lexigram.admin.engine.renderer import AdminRenderer 

23 from lexigram.contracts.core import TaskManagerProtocol 

24 from lexigram.contracts.infra import PoolManagerProtocol 

25from lexigram.di.decorators import inject 

26 

27logger = get_logger(__name__) 

28 

29_POOL_HEALTH_VIEW_PERMISSION = "pool_health.view" 

30_POOL_HEALTH_MANAGE_PERMISSION = "pool_health.manage" 

31 

32 

33@inject 

34class PoolHealthController(AdminController): 

35 """Controller for connection pool health monitoring. 

36 

37 Routes: 

38 - GET /admin/pools/health - Get health status for all pools 

39 - GET /admin/pools/health/{name} - Get health for specific pool 

40 - POST /admin/pools/{name}/reconnect - Force reconnect a pool 

41 """ 

42 

43 def __init__( 

44 self, 

45 renderer: AdminRenderer | None = None, 

46 pool_manager: PoolManagerProtocol | None = None, 

47 task_manager: TaskManagerProtocol | None = None, 

48 audit_service: AdminAuditLogServiceProtocol | None = None, 

49 rbac_config: AdminRbacConfig | None = None, 

50 ) -> None: 

51 """Initialize the pool health controller. 

52 

53 Args: 

54 renderer: AdminRenderer instance (DI-injected, optional). 

55 pool_manager: PoolManager instance for querying pool state (optional). 

56 task_manager: TaskManagerProtocol instance (optional). 

57 audit_service: Security audit log service (optional; denials are 

58 best-effort logged, failures never break the response path). 

59 rbac_config: Optional; the resolved RBAC config whose 

60 ``super_admin_role`` grants management rights. 

61 """ 

62 if renderer is None: 

63 from lexigram.admin.engine.renderer import AdminRenderer 

64 

65 renderer = AdminRenderer() 

66 super().__init__(renderer=renderer, task_manager=task_manager) 

67 self._pool_manager = pool_manager 

68 self._audit_service = audit_service 

69 self._rbac_config = rbac_config 

70 

71 def _require_pool_manager(self) -> JSONResponse | None: 

72 """Return a 503 response if no pool manager is available, otherwise None.""" 

73 if self._pool_manager is None: 

74 return JSONResponse( 

75 {"error": "Pool manager is not available"}, 

76 status_code=503, 

77 ) 

78 return None 

79 

80 @staticmethod 

81 def _user_permissions(request: Request) -> frozenset[str]: 

82 """Return the requesting user's permission set (empty when unknown).""" 

83 user = getattr(getattr(request, "state", None), "user", None) 

84 return frozenset(getattr(user, "permissions", None) or ()) 

85 

86 def _user_is_superadmin(self, request: Request) -> bool: 

87 """Return True when the requesting user holds the superadmin role. 

88 

89 Superadmin bypasses per-spec permission gating so accounts created 

90 with an empty permission set (e.g. via the setup wizard) can still 

91 manage system operations. 

92 """ 

93 role = (self._rbac_config or AdminRbacConfig()).super_admin_role 

94 user = getattr(getattr(request, "state", None), "user", None) 

95 return is_super_admin(user, role) 

96 

97 async def _audit( 

98 self, 

99 request: Request, 

100 success: bool = True, 

101 event_type: AdminSecurityEventType = AdminSecurityEventType.SETTINGS_UPDATED, 

102 **metadata: Any, 

103 ) -> None: 

104 """Append a security event to the audit log, best-effort.""" 

105 if not self._audit_service: 

106 return 

107 try: 

108 client = getattr(request, "client", None) 

109 await self._audit_service.log_event( 

110 event_type=event_type, 

111 ip_address=getattr(client, "host", "unknown"), 

112 user_agent=request.headers.get("user-agent", "") or "", 

113 success=success, 

114 metadata=metadata, 

115 ) 

116 except Exception: # noqa: BLE001 — audit failures must not break the response path 

117 logger.warning("pool_health.audit_failed", **metadata) 

118 

119 async def get_all_health(self, request: Request) -> JSONResponse: 

120 """Get health status for all connection pools.""" 

121 if (missing := self._require_pool_manager()) is not None: 

122 return missing 

123 if not self._user_is_superadmin( 

124 request 

125 ) and _POOL_HEALTH_VIEW_PERMISSION not in self._user_permissions(request): 

126 await self._audit( 

127 request, 

128 success=False, 

129 event_type=AdminSecurityEventType.PERMISSION_DENIED, 

130 controller="pool_health", 

131 action="get_all_health", 

132 ) 

133 return JSONResponse({"error": "Permission denied"}, status_code=403) 

134 try: 

135 all_stats = self._pool_manager.get_stats() # type: ignore[union-attr] 

136 pools_data = {} 

137 for name, stats in all_stats.items(): 

138 is_healthy = stats.pool_utilization < 90.0 

139 pools_data[name] = { 

140 "name": name, 

141 "is_healthy": is_healthy, 

142 "stats": stats.__dict__, 

143 "last_check": stats.last_health_check, 

144 "error": None if is_healthy else "High utilization", 

145 } 

146 

147 total = len(pools_data) 

148 healthy = sum(1 for p in pools_data.values() if p["is_healthy"]) 

149 

150 return JSONResponse( 

151 { 

152 "pools": pools_data, 

153 "summary": { 

154 "total_pools": total, 

155 "healthy_pools": healthy, 

156 "unhealthy_pools": total - healthy, 

157 }, 

158 }, 

159 ) 

160 except Exception as e: # noqa: BLE001 — controller boundary; unexpected errors become HTTP 500 responses 

161 logger.exception("Failed to get all pool health") 

162 return JSONResponse( 

163 {"error": "Failed to get pool health", "detail": str(e)}, 

164 status_code=500, 

165 ) 

166 

167 async def get_pool_health(self, request: Request) -> JSONResponse: 

168 """Get health status for a specific pool.""" 

169 pool_name = request.path_params.get("name") 

170 if not pool_name: 

171 return JSONResponse({"error": "Pool name required"}, status_code=400) 

172 if (missing := self._require_pool_manager()) is not None: 

173 return missing 

174 if not self._user_is_superadmin( 

175 request 

176 ) and _POOL_HEALTH_VIEW_PERMISSION not in self._user_permissions(request): 

177 await self._audit( 

178 request, 

179 success=False, 

180 event_type=AdminSecurityEventType.PERMISSION_DENIED, 

181 controller="pool_health", 

182 action="get_pool_health", 

183 ) 

184 return JSONResponse({"error": "Permission denied"}, status_code=403) 

185 try: 

186 pool = await self._pool_manager.get_pool(pool_name) # type: ignore[union-attr] 

187 stats = pool.get_stats() 

188 is_healthy = stats.pool_utilization < 90.0 

189 return JSONResponse( 

190 { 

191 "name": pool_name, 

192 "is_healthy": is_healthy, 

193 "stats": stats.__dict__, 

194 "last_check": stats.last_health_check, 

195 "error": None if is_healthy else "High utilization", 

196 }, 

197 ) 

198 except KeyError: 

199 return JSONResponse( 

200 {"error": f"Pool '{pool_name}' not found"}, 

201 status_code=404, 

202 ) 

203 except Exception as e: # noqa: BLE001 — controller boundary; unexpected errors become HTTP 500 responses 

204 logger.exception("Failed to get health for pool %s", pool_name) 

205 return JSONResponse( 

206 {"error": "Failed to get pool health", "detail": str(e)}, 

207 status_code=500, 

208 ) 

209 

210 async def reconnect_pool(self, request: Request) -> JSONResponse: 

211 """Force reconnect a specific pool.""" 

212 pool_name = request.path_params.get("name") 

213 if not pool_name: 

214 return JSONResponse({"error": "Pool name required"}, status_code=400) 

215 if (missing := self._require_pool_manager()) is not None: 

216 return missing 

217 if not self._user_is_superadmin( 

218 request 

219 ) and _POOL_HEALTH_MANAGE_PERMISSION not in self._user_permissions(request): 

220 await self._audit( 

221 request, 

222 success=False, 

223 event_type=AdminSecurityEventType.PERMISSION_DENIED, 

224 controller="pool_health", 

225 action="reconnect_pool", 

226 ) 

227 return JSONResponse({"error": "Permission denied"}, status_code=403) 

228 try: 

229 pool = await self._pool_manager.get_pool(pool_name) # type: ignore[union-attr] 

230 await pool.close() 

231 return JSONResponse( 

232 {"message": f"Pool '{pool_name}' closed for reconnection"}, 

233 ) 

234 except KeyError: 

235 return JSONResponse( 

236 {"error": f"Pool '{pool_name}' not found"}, 

237 status_code=404, 

238 ) 

239 except Exception as e: # noqa: BLE001 — controller boundary; unexpected errors become HTTP 500 responses 

240 logger.exception("Failed to reconnect pool %s", pool_name) 

241 return JSONResponse( 

242 {"error": "Failed to reconnect pool", "detail": str(e)}, 

243 status_code=500, 

244 ) 

245 

246 async def get_pool_stats_summary(self, request: Request) -> JSONResponse: 

247 """Get aggregated statistics across all pools.""" 

248 if (missing := self._require_pool_manager()) is not None: 

249 return missing 

250 if not self._user_is_superadmin( 

251 request 

252 ) and _POOL_HEALTH_VIEW_PERMISSION not in self._user_permissions(request): 

253 await self._audit( 

254 request, 

255 success=False, 

256 event_type=AdminSecurityEventType.PERMISSION_DENIED, 

257 controller="pool_health", 

258 action="get_pool_stats_summary", 

259 ) 

260 return JSONResponse({"error": "Permission denied"}, status_code=403) 

261 try: 

262 all_stats = self._pool_manager.get_stats() # type: ignore[union-attr] 

263 total_connections = sum(s.total_connections for s in all_stats.values()) 

264 active_connections = sum(s.active_connections for s in all_stats.values()) 

265 idle_connections = sum(s.idle_connections for s in all_stats.values()) 

266 total_created = sum(s.created_connections for s in all_stats.values()) 

267 total_destroyed = sum(s.destroyed_connections for s in all_stats.values()) 

268 

269 utilizations = [s.pool_utilization for s in all_stats.values()] 

270 avg_utilization = ( 

271 sum(utilizations) / len(utilizations) if utilizations else 0.0 

272 ) 

273 max_utilization = max(utilizations) if utilizations else 0.0 

274 

275 return JSONResponse( 

276 { 

277 "total_connections": total_connections, 

278 "active_connections": active_connections, 

279 "idle_connections": idle_connections, 

280 "total_created": total_created, 

281 "total_destroyed": total_destroyed, 

282 "avg_utilization": avg_utilization, 

283 "max_utilization": max_utilization, 

284 }, 

285 ) 

286 except Exception as e: # noqa: BLE001 — controller boundary; unexpected errors become HTTP 500 responses 

287 logger.exception("Failed to compute pool stats summary") 

288 return JSONResponse( 

289 {"error": "Failed to get pool stats", "detail": str(e)}, 

290 status_code=500, 

291 ) 

292 

293 async def get_prometheus_metrics(self, request: Request) -> Response: 

294 """Export pool metrics in Prometheus format.""" 

295 if (missing := self._require_pool_manager()) is not None: 

296 return missing 

297 if not self._user_is_superadmin( 

298 request 

299 ) and _POOL_HEALTH_VIEW_PERMISSION not in self._user_permissions(request): 

300 await self._audit( 

301 request, 

302 success=False, 

303 event_type=AdminSecurityEventType.PERMISSION_DENIED, 

304 controller="pool_health", 

305 action="get_prometheus_metrics", 

306 ) 

307 return JSONResponse({"error": "Permission denied"}, status_code=403) 

308 try: 

309 all_stats = self._pool_manager.get_stats() # type: ignore[union-attr] 

310 lines = [] 

311 for name, stats in all_stats.items(): 

312 labels = f'pool="{name}"' 

313 lines.append( 

314 f"pool_active_connections{{{labels}}} {stats.active_connections}", 

315 ) 

316 lines.append( 

317 f"pool_total_connections{{{labels}}} {stats.total_connections}", 

318 ) 

319 lines.append(f"pool_utilization{{{labels}}} {stats.pool_utilization}") 

320 

321 return Response( 

322 "\n".join(lines) + "\n", 

323 media_type="text/plain; version=0.0.4", 

324 ) 

325 except Exception as e: # noqa: BLE001 — controller boundary; unexpected errors become HTTP 500 responses 

326 logger.exception("Failed to export metrics") 

327 return JSONResponse( 

328 {"error": "Failed to export metrics", "detail": str(e)}, 

329 status_code=500, 

330 ) 

331 

332 async def get_json_metrics(self, request: Request) -> JSONResponse: 

333 """Export pool metrics in JSON format.""" 

334 if (missing := self._require_pool_manager()) is not None: 

335 return missing 

336 if not self._user_is_superadmin( 

337 request 

338 ) and _POOL_HEALTH_VIEW_PERMISSION not in self._user_permissions(request): 

339 await self._audit( 

340 request, 

341 success=False, 

342 event_type=AdminSecurityEventType.PERMISSION_DENIED, 

343 controller="pool_health", 

344 action="get_json_metrics", 

345 ) 

346 return JSONResponse({"error": "Permission denied"}, status_code=403) 

347 try: 

348 all_stats = self._pool_manager.get_stats() # type: ignore[union-attr] 

349 metrics_json = {name: stats.__dict__ for name, stats in all_stats.items()} 

350 return JSONResponse(metrics_json) 

351 except Exception as e: # noqa: BLE001 — controller boundary; unexpected errors become HTTP 500 responses 

352 logger.exception("Failed to export metrics as JSON") 

353 return JSONResponse( 

354 {"error": "Failed to export metrics", "detail": str(e)}, 

355 status_code=500, 

356 )