| import logging |
| from typing import Any |
|
|
| logger = logging.getLogger("qkernel.healing") |
|
|
| HEALING_ACTIONS = { |
| "network_latency": "restart_node", |
| "memory_leak": "clear_cache", |
| "null_pointer": "rollback_deploy", |
| "unknown": "isolate_node", |
| } |
|
|
|
|
| class SelfHealingEngine: |
| def __init__(self): |
| self._history: list[dict] = [] |
|
|
| def detect(self, metrics: dict) -> bool: |
| if metrics.get("latency", 0) > 1000: |
| return True |
| if metrics.get("error_rate", 0) > 0.2: |
| return True |
| return False |
|
|
| def analyze(self, error: str, context: dict | None = None) -> str: |
| error_lower = error.lower() |
| if "timeout" in error_lower or "latency" in error_lower: |
| return "network_latency" |
| if "memory" in error_lower: |
| return "memory_leak" |
| if "null" in error_lower or "none" in error_lower: |
| return "null_pointer" |
| return "unknown" |
|
|
| def plan(self, root_cause: str) -> str: |
| return HEALING_ACTIONS.get(root_cause, "isolate_node") |
|
|
| async def execute(self, action: str, node: Any) -> dict: |
| result = {"action": action, "success": True, "details": ""} |
|
|
| if action == "restart_node": |
| if hasattr(node, "restart"): |
| await node.restart() |
| result["details"] = "Node restarted" |
| else: |
| result["details"] = "Node has no restart method" |
|
|
| elif action == "clear_cache": |
| if hasattr(node, "cache"): |
| node.cache.clear() |
| result["details"] = "Cache cleared" |
| else: |
| result["details"] = "Node has no cache attribute" |
|
|
| elif action == "rollback_deploy": |
| if hasattr(node, "rollback"): |
| await node.rollback() |
| result["details"] = "Deployment rolled back" |
| else: |
| result["details"] = "Node has no rollback method" |
|
|
| elif action == "isolate_node": |
| if hasattr(node, "isolated"): |
| node.isolated = True |
| result["details"] = "Node isolated from traffic" |
| else: |
| result["details"] = "Node has no isolated attribute" |
|
|
| return result |
|
|
| async def handle_failure(self, node: Any, metrics: dict, error: str) -> dict: |
| if not self.detect(metrics): |
| return {"status": "healthy"} |
|
|
| root_cause = self.analyze(error) |
| action = self.plan(root_cause) |
| exec_result = await self.execute(action, node) |
|
|
| record = { |
| "error": error, |
| "root_cause": root_cause, |
| "action": action, |
| "result": exec_result, |
| } |
| self._history.append(record) |
| if len(self._history) > 100: |
| self._history = self._history[-100:] |
|
|
| logger.info( |
| "Healing: cause=%s action=%s success=%s", |
| root_cause, action, exec_result["success"], |
| ) |
|
|
| return { |
| "status": "healed" if exec_result["success"] else "failed", |
| "root_cause": root_cause, |
| "action": action, |
| } |
|
|
| def get_history(self) -> list[dict]: |
| return list(self._history) |
|
|
