diff --git a/app/core/health_route.py b/app/core/health_route.py index a977cee..00abe3b 100644 --- a/app/core/health_route.py +++ b/app/core/health_route.py @@ -5,80 +5,111 @@ Per RMIV5 v4.0 §T33. Provides basic Kubernetes-style health endpoints: - /live — liveness (process alive, no deps) - /ready — readiness (critical deps reachable) -Each returns a JSON body with status + per-store details. +Emits Prometheus HEALTH_CHECK_DURATION + HEALTH_CHECK_STATUS gauges per store. """ + from __future__ import annotations import asyncio +import os import time from typing import Any from fastapi import APIRouter from pydantic import BaseModel +from app.core.metrics import HEALTH_CHECK_DURATION, HEALTH_CHECK_STATUS + router = APIRouter(tags=["health"]) -class HealthResponse(BaseModel): - """Response shape for /health endpoint.""" +QDRANT_URL = os.getenv("QDRANT_URL", "http://127.0.0.1:6333") +CLICKHOUSE_URL = os.getenv("CLICKHOUSE_URL", "http://127.0.0.1:8123/") +MINIO_URL = os.getenv("MINIO_URL", "http://127.0.0.1:9000") +RETH_URL = os.getenv("RETH_URL", "http://127.0.0.1:8545") +_SHARED_HTTP_CLIENT: httpx.AsyncClient | None = None - status: str # "healthy" | "degraded" | "unhealthy" + +async def _get_http_client() -> httpx.AsyncClient: + global _SHARED_HTTP_CLIENT + if _SHARED_HTTP_CLIENT is None: + import httpx + + _SHARED_HTTP_CLIENT = httpx.AsyncClient(timeout=5.0) + return _SHARED_HTTP_CLIENT + + +class HealthResponse(BaseModel): + status: str version: str uptime_seconds: float stores: dict[str, dict[str, Any]] class LivenessResponse(BaseModel): - """Response for /live — process is alive.""" - status: str = "alive" class ReadinessResponse(BaseModel): - """Response for /ready — process can serve traffic.""" - - status: str # "ready" | "not_ready" + status: str checks: dict[str, bool] _START_TIME = time.monotonic() -async def _check_redis() -> dict[str, Any]: - """Check Redis connectivity (PING).""" +async def _check_store(name: str, coro) -> dict[str, Any]: + """Run a health check and emit Prometheus metrics.""" + start = time.perf_counter() try: - from app.core.redis import get_redis + result = await coro + elapsed = time.perf_counter() - start + healthy = result.get("healthy", False) + HEALTH_CHECK_DURATION.labels(store=name).set(elapsed) + HEALTH_CHECK_STATUS.labels(store=name).set(1 if healthy else 0) + return result + except Exception as e: + elapsed = time.perf_counter() - start + HEALTH_CHECK_DURATION.labels(store=name).set(elapsed) + HEALTH_CHECK_STATUS.labels(store=name).set(0) + return {"healthy": False, "error": str(e)[:200]} - r = get_redis() - await asyncio.wait_for(r.ping(), timeout=2.0) + +async def _check_redis() -> dict[str, Any]: + try: + from app.core.redis import get_redis_async + + client = get_redis_async() + if client is None: + return {"healthy": False, "error": "Redis pool not initialized"} + await asyncio.wait_for(client.ping(), timeout=5.0) return {"healthy": True} except Exception as e: return {"healthy": False, "error": str(e)[:200]} async def _check_postgres() -> dict[str, Any]: - """Check Postgres connectivity (SELECT 1).""" try: - from app.core.db_pool import get_pg_pool + import asyncpg - pool = get_pg_pool() - async with pool.acquire() as conn: - await asyncio.wait_for(conn.fetchval("SELECT 1"), timeout=2.0) + conn = await asyncpg.connect( + os.getenv("DATABASE_URL", "postgresql://rmi:***@rmi-postgres:5432/rmi") + ) + await asyncio.wait_for(conn.fetchval("SELECT 1"), timeout=5.0) + await conn.close() return {"healthy": True} except Exception as e: return {"healthy": False, "error": str(e)[:200]} + async def _check_neo4j() -> dict[str, Any]: - """Check Neo4j connectivity (RETURN 1 + node count).""" try: from app.core.neo4j_connection import _driver + if _driver is None: return {"healthy": False, "error": "Neo4j driver not initialized"} async with _driver.session() as session: - result = await asyncio.wait_for( - session.run("RETURN 1"), - timeout=3.0, - ) + result = await asyncio.wait_for(session.run("RETURN 1"), timeout=3.0) val = await result.single() node_count_result = await session.run("MATCH (n) RETURN count(n)") node_count = await node_count_result.single() @@ -91,14 +122,11 @@ async def _check_neo4j() -> dict[str, Any]: async def _check_qdrant() -> dict[str, Any]: - """Check Qdrant connectivity (GET /collections + collection count).""" try: - import httpx - qdrant_url = "http://localhost:6333/collections" - async with httpx.AsyncClient(timeout=3.0) as client: - resp = await client.get(qdrant_url) - data = resp.json() - collections = data.get("result", {}).get("collections", []) + client = await _get_http_client() + resp = await client.get(f'{QDRANT_URL.rstrip(chr(47))}/collections') + data = resp.json() + collections = data.get("result", {}).get("collections", []) return { "healthy": resp.status_code == 200, "details": {"collection_count": len(collections)}, @@ -108,19 +136,20 @@ async def _check_qdrant() -> dict[str, Any]: async def _check_clickhouse() -> dict[str, Any]: - """Check ClickHouse connectivity (SELECT 1 + rmi DB exists).""" try: - import httpx - async with httpx.AsyncClient(timeout=3.0) as client: - resp = await client.post( - "http://localhost:8123/", - content="SELECT 1", - ) - db_resp = await client.post( - "http://localhost:8123/", - content="SHOW DATABASES", - ) - dbs = db_resp.text.strip().split("\n") if db_resp.status_code == 200 else [] + import base64 + import os + + user = os.getenv("CLICKHOUSE_USER", "rmi") + password = os.getenv("CLICKHOUSE_PASSWORD", os.getenv("CLICKHOUSE_PASS", "")) + token = base64.b64encode(f"{user}:{password}".encode()).decode() + headers = {"Authorization": f"Basic {token}"} + client = await _get_http_client() + resp = await client.post("http://rmi-clickhouse:8123/", content="SELECT 1", headers=headers) + db_resp = await client.post( + "http://rmi-clickhouse:8123/", content="SHOW DATABASES", headers=headers + ) + dbs = db_resp.text.strip().split("\n") if db_resp.status_code == 200 else [] return { "healthy": resp.text.strip() == "1", "details": {"rmi_db_exists": "rmi" in dbs, "databases": dbs}, @@ -130,113 +159,68 @@ async def _check_clickhouse() -> dict[str, Any]: async def _check_minio() -> dict[str, Any]: - """Check MinIO connectivity (health endpoint).""" try: - import httpx - async with httpx.AsyncClient(timeout=3.0) as client: - resp = await client.get("http://localhost:9000/minio/health/live") - return { - "healthy": resp.status_code == 200, - "details": {"status_code": resp.status_code}, - } + client = await _get_http_client() + resp = await client.get("http://rmi-minio:9000/minio/health/live") + return {"healthy": resp.status_code == 200, "details": {"status_code": resp.status_code}} except Exception as e: return {"healthy": False, "error": str(e)[:200]} async def _check_reth() -> dict[str, Any]: - """Check Reth (Ethereum node) connectivity (eth_blockNumber).""" try: - - import httpx - async with httpx.AsyncClient(timeout=3.0) as client: - resp = await client.post( - "http://localhost:8545", - json={"jsonrpc": "2.0", "method": "eth_blockNumber", "params": [], "id": 1}, - ) - data = resp.json() - block_hex = data.get("result", "0x0") - block_num = int(block_hex, 16) if block_hex.startswith("0x") else 0 - return { - "healthy": block_num > 0, - "details": {"block_number": block_num}, - } + rpc_url = os.getenv("RETH_URL", "https://ethereum-rpc.publicnode.com") + client = await _get_http_client() + resp = await client.post( + rpc_url, + json={"jsonrpc": "2.0", "method": "eth_blockNumber", "params": [], "id": 1}, + ) + data = resp.json() + block_hex = data.get("result", "0x0") + block_num = int(block_hex, 16) if block_hex.startswith("0x") else 0 + return {"healthy": block_num > 0, "details": {"block_number": block_num}} except Exception as e: return {"healthy": False, "error": str(e)[:200]} - - @router.get("/live", response_model=LivenessResponse) async def liveness() -> LivenessResponse: - """Liveness probe — confirms the process is running. - - Used by Kubernetes/load balancers to decide whether to restart - the container. Does NOT check dependencies. - """ return LivenessResponse(status="alive") @router.get("/ready", response_model=ReadinessResponse) async def readiness() -> ReadinessResponse: - """Readiness probe — confirms critical deps are reachable. - - Used to decide whether to send traffic. Fails fast (2s per check). - """ - redis_ok, pg_ok = await asyncio.gather( - _check_redis(), - _check_postgres(), - return_exceptions=True, - ) - # Normalize exceptions - if isinstance(redis_ok, Exception): - redis_ok = {"healthy": False, "error": str(redis_ok)[:200]} - if isinstance(pg_ok, Exception): - pg_ok = {"healthy": False, "error": str(pg_ok)[:200]} - checks = { - "redis": bool(redis_ok.get("healthy")), - "postgres": bool(pg_ok.get("healthy")), - } + r1 = await _check_store("redis", _check_redis()) + r2 = await _check_store("postgres", _check_postgres()) + redis_ok = r1 if not isinstance(r1, Exception) else {"healthy": False} + pg_ok = r2 if not isinstance(r2, Exception) else {"healthy": False} + checks = {"redis": redis_ok.get("healthy", False), "postgres": pg_ok.get("healthy", False)} all_ok = all(checks.values()) - return ReadinessResponse( - status="ready" if all_ok else "not_ready", - checks=checks, - ) + return ReadinessResponse(status="ready" if all_ok else "not_ready", checks=checks) @router.get("/health", response_model=HealthResponse) async def health() -> HealthResponse: - """Full health check — deep validation of every store. - - Returns per-store health with latency. Slower than /ready but - gives ops full visibility. - """ started = time.monotonic() - (redis_h, pg_h, neo4j_h, qdrant_h, clickhouse_h, minio_h, reth_h) = await asyncio.gather( - _check_redis(), - _check_postgres(), - _check_neo4j(), - _check_qdrant(), - _check_clickhouse(), - _check_minio(), - _check_reth(), - return_exceptions=True, - ) - for ref in [redis_h, pg_h, neo4j_h, qdrant_h, clickhouse_h, minio_h, reth_h]: + store_checks = { + "redis": _check_store("redis", _check_redis()), + "postgres": _check_store("postgres", _check_postgres()), + "neo4j": _check_store("neo4j", _check_neo4j()), + "qdrant": _check_store("qdrant", _check_qdrant()), + "clickhouse": _check_store("clickhouse", _check_clickhouse()), + "minio": _check_store("minio", _check_minio()), + "eth_rpc": _check_store("eth_rpc", _check_reth()), + } + results = await asyncio.gather(*store_checks.values(), return_exceptions=True) + stores = {} + for (name, _), ref in zip(store_checks.items(), results, strict=False): if isinstance(ref, Exception): ref = {"healthy": False, "error": str(ref)[:200]} if "latency_ms" not in ref: ref["latency_ms"] = int((time.monotonic() - started) * 1000) + stores[name] = ref - stores = { - "redis": redis_h, - "postgres": pg_h, - "neo4j": neo4j_h, - "qdrant": qdrant_h, - "clickhouse": clickhouse_h, - "minio": minio_h, - "reth": reth_h, - } - critical_down = not (redis_h.get("healthy") and pg_h.get("healthy")) + critical_down = not (stores["redis"].get("healthy") and stores["postgres"].get("healthy")) any_down = not all(s.get("healthy", False) for s in stores.values()) status = "unhealthy" if critical_down else "degraded" if any_down else "healthy"