import asyncio import logging import os from typing import Dict, Any import redis import psycopg2 import httpx # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') # Environment variables for database connections POSTGRES_HOST = os.getenv("POSTGRES_HOST", "postgresql-genesis-u50607.vm.elestio.app") POSTGRES_PORT = int(os.getenv("POSTGRES_PORT", "25432")) POSTGRES_USER = os.getenv("POSTGRES_USER", "genesis") POSTGRES_PASSWORD = os.getenv("POSTGRES_PASSWORD", "your_password") # Replace with actual password POSTGRES_DB = os.getenv("POSTGRES_DB", "genesis") REDIS_HOST = os.getenv("REDIS_HOST", "redis-genesis-u50607.vm.elestio.app") REDIS_PORT = int(os.getenv("REDIS_PORT", "26379")) QDRANT_HOST = os.getenv("QDRANT_HOST", "qdrant-b3knu-u50607.vm.elestio.app") QDRANT_PORT = int(os.getenv("QDRANT_PORT", "6333")) AIVA_OLLAMA_HOST = os.getenv("AIVA_OLLAMA_HOST", "localhost") AIVA_OLLAMA_PORT = int(os.getenv("AIVA_OLLAMA_PORT", "23405")) N8N_HEALTH_ENDPOINT = os.getenv("N8N_HEALTH_ENDPOINT", "http://localhost:5678/health") class HealthAggregator: """ Aggregates health status from various system components. """ def __init__(self): """ Initializes the HealthAggregator. """ self.monitors = { "redis": self.check_redis, "postgres": self.check_postgres, "qdrant": self.check_qdrant, "aiva": self.check_aiva, "n8n": self.check_n8n, } async def check_redis(self) -> Dict[str, Any]: """ Checks the health of the Redis connection. Returns: Dict[str, Any]: A dictionary containing the status and details of the Redis health check. """ try: r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=0) r.ping() return {"status": "healthy", "details": "Redis connection successful"} except redis.exceptions.ConnectionError as e: logging.error(f"Redis connection error: {e}") return {"status": "critical", "details": str(e)} async def check_postgres(self) -> Dict[str, Any]: """ Checks the health of the PostgreSQL connection. Returns: Dict[str, Any]: A dictionary containing the status and details of the PostgreSQL health check. """ try: conn = psycopg2.connect( host=POSTGRES_HOST, port=POSTGRES_PORT, user=POSTGRES_USER, password=POSTGRES_PASSWORD, dbname=POSTGRES_DB, connect_timeout=5, ) conn.close() return {"status": "healthy", "details": "PostgreSQL connection successful"} except psycopg2.Error as e: logging.error(f"PostgreSQL connection error: {e}") return {"status": "critical", "details": str(e)} async def check_qdrant(self) -> Dict[str, Any]: """ Checks the health of the Qdrant instance. Returns: Dict[str, Any]: A dictionary containing the status and details of the Qdrant health check. """ try: async with httpx.AsyncClient() as client: response = await client.get(f"http://{QDRANT_HOST}:{QDRANT_PORT}/", timeout=5) response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx) return {"status": "healthy", "details": "Qdrant is reachable"} except httpx.HTTPError as e: logging.error(f"Qdrant health check failed: {e}") return {"status": "critical", "details": str(e)} except httpx.TimeoutException as e: logging.error(f"Qdrant health check timed out: {e}") return {"status": "critical", "details": "Timeout connecting to Qdrant"} except Exception as e: logging.error(f"Unexpected error during Qdrant health check: {e}") return {"status": "critical", "details": f"Unexpected error: {e}"} async def check_aiva(self) -> Dict[str, Any]: """ Checks the health of the AIVA Ollama instance. Returns: Dict[str, Any]: A dictionary containing the status and details of the AIVA Ollama health check. """ try: async with httpx.AsyncClient() as client: response = await client.get(f"http://{AIVA_OLLAMA_HOST}:{AIVA_OLLAMA_PORT}/", timeout=5) response.raise_for_status() return {"status": "healthy", "details": "AIVA Ollama is reachable"} except httpx.HTTPError as e: logging.error(f"AIVA Ollama health check failed: {e}") return {"status": "critical", "details": str(e)} except httpx.TimeoutException as e: logging.error(f"AIVA Ollama health check timed out: {e}") return {"status": "critical", "details": "Timeout connecting to AIVA Ollama"} except Exception as e: logging.error(f"Unexpected error during AIVA Ollama health check: {e}") return {"status": "critical", "details": f"Unexpected error: {e}"} async def check_n8n(self) -> Dict[str, Any]: """ Checks the health of the n8n instance using its /health endpoint. Returns: Dict[str, Any]: A dictionary containing the status and details of the n8n health check. """ try: async with httpx.AsyncClient() as client: response = await client.get(N8N_HEALTH_ENDPOINT, timeout=5) response.raise_for_status() return {"status": "healthy", "details": "n8n is reachable"} except httpx.HTTPError as e: logging.error(f"n8n health check failed: {e}") return {"status": "critical", "details": str(e)} except httpx.TimeoutException as e: logging.error(f"n8n health check timed out: {e}") return {"status": "critical", "details": "Timeout connecting to n8n"} except Exception as e: logging.error(f"Unexpected error during n8n health check: {e}") return {"status": "critical", "details": f"Unexpected error: {e}"} async def get_health_status(self) -> Dict[str, Any]: """ Aggregates health status from all monitors. Returns: Dict[str, Any]: A dictionary containing the overall status and details of each component. """ component_statuses = await asyncio.gather( *(self.monitors[name]() for name in self.monitors) ) component_names = list(self.monitors.keys()) health_data = { component_names[i]: component_statuses[i] for i in range(len(component_names)) } overall_status = "healthy" for component, status in health_data.items(): if status["status"] == "critical": overall_status = "critical" break elif status["status"] == "degraded" and overall_status != "critical": overall_status = "degraded" return {"overall_status": overall_status, "components": health_data} # Example usage (for testing purposes): async def main(): """ Main function to demonstrate the usage of HealthAggregator. """ aggregator = HealthAggregator() health_status = await aggregator.get_health_status() print(health_status) if __name__ == "__main__": asyncio.run(main())