""" oracle_node_manager.py — Gestore Nodo E (Oracle Cloud) Funzionalità: - Monitoraggio stato del server Oracle - Sincronizzazione automatica con cluster principale - Gestione task distribuiti - Health check e failover - Logging e metriche Utilizzo: from backend.api.oracle_node_manager import OracleNodeManager manager = OracleNodeManager() await manager.sync_with_cluster() """ import os import asyncio import logging from .oci_gateway_manager import oci_gateway import json from typing import Optional, Dict, List, Any from datetime import datetime, timedelta from enum import Enum import httpx import paramiko _logger = logging.getLogger("oracle_node_manager") # ── Configurazione Oracle Node E ─────────────────────────────────────────── ORACLE_NODE_E_ENABLED = os.getenv("ORACLE_NODE_E_ENABLED", "true").lower() == "true" ORACLE_NODE_E_IP = os.getenv("ORACLE_NODE_E_IP", "80.225.89.217") ORACLE_NODE_E_USER = os.getenv("ORACLE_NODE_E_USER", "opc") ORACLE_NODE_E_SSH_KEY_PATH = os.getenv("ORACLE_NODE_E_SSH_KEY_PATH", "/app/secrets/oracle_private_key.pem") ORACLE_NODE_E_PORT = int(os.getenv("ORACLE_NODE_E_PORT", "8080")) ORACLE_NODE_E_ROLE = os.getenv("ORACLE_NODE_E_ROLE", "Compute/AI Workload") ORACLE_NODE_E_REGION = os.getenv("ORACLE_NODE_E_REGION", "Italy Northwest (Milan)") # ── Enumerazione Stato Nodo ──────────────────────────────────────────────── class NodeStatus(str, Enum): HEALTHY = "healthy" DEGRADED = "degraded" UNHEALTHY = "unhealthy" OFFLINE = "offline" # ── SSH Client per Oracle ────────────────────────────────────────────────── class OracleSSHClient: """Client SSH per gestire il server Oracle.""" def __init__(self, host: str, user: str, key_path: str): self.host = host self.user = user self.key_path = key_path self.client = None async def connect(self) -> bool: """Connette al server Oracle via SSH.""" try: self.client = paramiko.SSHClient() self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # Carica chiave privata if not os.path.exists(self.key_path): _logger.error(f"Chiave SSH non trovata: {self.key_path}") return False pkey = paramiko.RSAKey.from_private_key_file(self.key_path) self.client.connect(self.host, username=self.user, pkey=pkey, timeout=10) _logger.info(f"Connesso a Oracle Node E: {self.host}") return True except Exception as exc: _logger.error(f"Errore connessione SSH: {exc}") return False async def execute_command(self, cmd: str) -> tuple[bool, str]: """Esegue un comando remoto.""" try: if not self.client: return False, "Non connesso" stdin, stdout, stderr = self.client.exec_command(cmd, timeout=30) output = stdout.read().decode('utf-8') error = stderr.read().decode('utf-8') return True, output if output else error except Exception as exc: _logger.error(f"Errore esecuzione comando: {exc}") return False, str(exc) async def disconnect(self): """Disconnette dal server.""" if self.client: self.client.close() _logger.info("Disconnesso da Oracle Node E") # ── Oracle Node Manager ──────────────────────────────────────────────────── class OracleNodeManager: """Gestore del Nodo E (Oracle Cloud).""" def __init__(self): self.ssh_client = OracleSSHClient(ORACLE_NODE_E_IP, ORACLE_NODE_E_USER, ORACLE_NODE_E_SSH_KEY_PATH) self.status = NodeStatus.OFFLINE self._last_sync = None self._metrics = { "cpu_usage": 0.0, "memory_usage": 0.0, "disk_usage": 0.0, "uptime_seconds": 0, "sync_lag_seconds": 0, } async def initialize(self) -> bool: """Inizializza la connessione al Nodo E.""" if not ORACLE_NODE_E_ENABLED: _logger.info("Oracle Node E disabilitato") return False connected = await self.ssh_client.connect() if connected: self.status = NodeStatus.HEALTHY return True else: self.status = NodeStatus.OFFLINE return False async def health_check(self) -> Dict[str, Any]: # Tentativo tramite Gateway se la connessione diretta fallisce gateway_res = await oci_gateway.get_gateway_status() if gateway_res.get("ok"): _logger.info("Connessione via OCI API Gateway attiva") """Esegue health check del Nodo E.""" if not self.ssh_client.client: return { "ok": False, "status": NodeStatus.OFFLINE.value, "error": "Non connesso", } try: # Verifica backend backend_ok, _ = await self.ssh_client.execute_command( f"curl -s http://localhost:{ORACLE_NODE_E_PORT}/api/health || echo 'FAIL'" ) # Verifica risorse cpu_ok, cpu_output = await self.ssh_client.execute_command("top -bn1 | grep 'Cpu(s)' | awk '{print $2}'") mem_ok, mem_output = await self.ssh_client.execute_command("free | grep Mem | awk '{printf \"%.1f\", ($3/$2)*100}'") disk_ok, disk_output = await self.ssh_client.execute_command("df -h / | tail -1 | awk '{print $5}'") # Aggiorna metriche self._metrics["cpu_usage"] = float(cpu_output.strip()) if cpu_ok else 0.0 self._metrics["memory_usage"] = float(mem_output.strip()) if mem_ok else 0.0 self._metrics["disk_usage"] = float(disk_output.strip().rstrip('%')) if disk_ok else 0.0 # Determina stato if backend_ok and self._metrics["cpu_usage"] < 80 and self._metrics["memory_usage"] < 85: self.status = NodeStatus.HEALTHY elif self._metrics["cpu_usage"] < 95 and self._metrics["memory_usage"] < 95: self.status = NodeStatus.DEGRADED else: self.status = NodeStatus.UNHEALTHY return { "ok": True, "status": self.status.value, "metrics": self._metrics, "timestamp": datetime.now().isoformat(), } except Exception as exc: _logger.error(f"Errore health check: {exc}") self.status = NodeStatus.UNHEALTHY return { "ok": False, "status": NodeStatus.UNHEALTHY.value, "error": str(exc), } async def sync_with_cluster(self) -> Dict[str, Any]: """Sincronizza il Nodo E con il cluster principale.""" if not self.ssh_client.client: return {"ok": False, "error": "Non connesso"} try: # Esegui script di sincronizzazione remoto sync_cmd = "cd /home/opc/ai-cluster/repo && node scripts/sync-cluster-state.mjs --node E" success, output = await self.ssh_client.execute_command(sync_cmd) if success: self._last_sync = datetime.now() _logger.info(f"Sincronizzazione completata: {output}") return { "ok": True, "message": "Sincronizzazione completata", "output": output, "timestamp": datetime.now().isoformat(), } else: return { "ok": False, "error": output, } except Exception as exc: _logger.error(f"Errore sincronizzazione: {exc}") return { "ok": False, "error": str(exc), } async def get_logs(self, lines: int = 100) -> str: """Recupera i log del backend remoto.""" if not self.ssh_client.client: return "Non connesso" try: cmd = f"tail -n {lines} /home/opc/ai-cluster/backend.log" success, output = await self.ssh_client.execute_command(cmd) return output if success else "Errore lettura log" except Exception as exc: return f"Errore: {str(exc)}" async def restart_services(self) -> Dict[str, Any]: """Riavvia i servizi sul Nodo E.""" if not self.ssh_client.client: return {"ok": False, "error": "Non connesso"} try: # Ferma servizi await self.ssh_client.execute_command("pkill -f 'npm start' || true") await asyncio.sleep(2) # Avvia servizi start_cmd = "cd /home/opc/ai-cluster/repo/backend && nohup npm start > /home/opc/ai-cluster/backend.log 2>&1 &" success, output = await self.ssh_client.execute_command(start_cmd) await asyncio.sleep(3) # Verifica health health = await self.health_check() return { "ok": True, "message": "Servizi riavviati", "health": health, "timestamp": datetime.now().isoformat(), } except Exception as exc: _logger.error(f"Errore riavvio servizi: {exc}") return { "ok": False, "error": str(exc), } async def get_metrics(self) -> Dict[str, Any]: """Ritorna metriche del Nodo E.""" return { "node": "E", "ip": ORACLE_NODE_E_IP, "role": ORACLE_NODE_E_ROLE, "region": ORACLE_NODE_E_REGION, "status": self.status.value, "metrics": self._metrics, "last_sync": self._last_sync.isoformat() if self._last_sync else None, "timestamp": datetime.now().isoformat(), } async def cleanup(self): """Pulizia risorse.""" await self.ssh_client.disconnect() # ── Istanza Globale ─────────────────────────────────────────────────────── _oracle_manager = OracleNodeManager() # ── Funzioni Pubbliche ──────────────────────────────────────────────────── async def initialize_oracle_node() -> bool: """Inizializza il Nodo E.""" return await _oracle_manager.initialize() async def get_oracle_health() -> Dict[str, Any]: """Ritorna health check del Nodo E.""" return await _oracle_manager.health_check() async def sync_oracle_cluster() -> Dict[str, Any]: """Sincronizza il Nodo E con il cluster.""" return await _oracle_manager.sync_with_cluster() async def get_oracle_metrics() -> Dict[str, Any]: """Ritorna metriche del Nodo E.""" return await _oracle_manager.get_metrics() async def restart_oracle_services() -> Dict[str, Any]: """Riavvia servizi sul Nodo E.""" return await _oracle_manager.restart_services()