| import time
|
| from datetime import datetime
|
| from agent_brain import NagaMLOpsAgent
|
| from pipeline_engine import MLPipelineEngine
|
| import utils
|
|
|
| class SelfHealingOrchestrator:
|
| def __init__(self):
|
| self.agent = NagaMLOpsAgent()
|
| self.engine = MLPipelineEngine()
|
| self.total_runs = 0
|
| self.faults_detected = 0
|
| self.faults_healed = 0
|
| self.total_heal_time_sec = 0.0
|
|
|
| def get_kpi_stats(self) -> dict:
|
| """
|
| Calculates high level system health KPIs for dashboard counters.
|
| """
|
| success_rate = 100.0
|
| if self.total_runs > 0:
|
| failed_unhealed = self.faults_detected - self.faults_healed
|
| success_rate = round(max(0.0, ((self.total_runs - failed_unhealed) / self.total_runs) * 100.0), 1)
|
|
|
| avg_mtth = 0.0
|
| if self.faults_healed > 0:
|
| avg_mtth = round(self.total_heal_time_sec / self.faults_healed, 2)
|
|
|
| return {
|
| "health_score_pct": success_rate,
|
| "total_incidents": self.faults_detected,
|
| "auto_healed_count": self.faults_healed,
|
| "avg_mtth_sec": avg_mtth
|
| }
|
|
|
| def run_pipeline_check(self, custom_code: str = None) -> dict:
|
| """
|
| Executes standard pipeline monitoring run without fault injection.
|
| """
|
| self.total_runs += 1
|
| exec_result = self.engine.execute_pipeline(code=custom_code)
|
| return {
|
| "status": exec_result["telemetry"]["status"],
|
| "telemetry": exec_result["telemetry"],
|
| "logs": exec_result["logs"],
|
| "code": exec_result["code"],
|
| "kpis": self.get_kpi_stats()
|
| }
|
|
|
| def trigger_and_heal_fault(self, fault_name: str, custom_code: str = None) -> dict:
|
| """
|
| Full autonomous closed-loop:
|
| 1. Inject fault / load scenario
|
| 2. Run broken pipeline & detect failure
|
| 3. Invoke Naga Agentic LLM for Root Cause Analysis & Python code patch synthesis
|
| 4. Apply patch code to engine
|
| 5. Verify re-execution & commit incident log
|
| """
|
| start_heal_timer = time.time()
|
| self.total_runs += 1
|
| self.faults_detected += 1
|
|
|
|
|
| if custom_code:
|
| broken_code = self.engine.set_custom_code(custom_code)
|
| elif fault_name:
|
| broken_code = self.engine.load_fault_scenario(fault_name)
|
| else:
|
| broken_code = self.engine.current_code
|
|
|
|
|
| pre_heal_run = self.engine.execute_pipeline(broken_code)
|
| pre_telemetry = pre_heal_run["telemetry"]
|
| pre_logs = pre_heal_run["logs"]
|
|
|
| agent_timeline = []
|
| agent_timeline.append(f"⏱ [{time.strftime('%H:%M:%S')}] 🚨 ANOMALY DETECTED! Status: {pre_telemetry['status']}. Initiating Naga AI Agentic Diagnosis...")
|
|
|
|
|
| diagnosis = self.agent.diagnose_and_heal(
|
| script_code=broken_code,
|
| execution_logs=pre_logs,
|
| telemetry=pre_telemetry,
|
| fault_name=fault_name
|
| )
|
|
|
| fault_cat = diagnosis.get("fault_category", "UNKNOWN_FAULT")
|
| rca = diagnosis.get("root_cause_analysis", "No detailed RCA provided.")
|
| explanation = diagnosis.get("explanation_for_engineers", "Patch synthesized.")
|
| patched_code = diagnosis.get("patch_code", broken_code)
|
|
|
| agent_timeline.append(f"⏱ [{time.strftime('%H:%M:%S')}] 🧠 Root Cause Analysis ({fault_cat}): {rca[:180]}...")
|
| agent_timeline.append(f"⏱ [{time.strftime('%H:%M:%S')}] 🛠 Synthesized Python Patch Code. Applying patch to execution environment...")
|
|
|
|
|
| post_heal_run = self.engine.execute_pipeline(patched_code)
|
| post_telemetry = post_heal_run["telemetry"]
|
| post_logs = post_heal_run["logs"]
|
|
|
| heal_duration = round(time.time() - start_heal_timer, 2)
|
| verified_success = (post_telemetry["status"] in ["HEALTHY", "NORMAL"]) and (post_telemetry["accuracy"] >= 0.70 or "syntax" not in post_logs.lower())
|
|
|
| if verified_success:
|
| self.faults_healed += 1
|
| self.total_heal_time_sec += heal_duration
|
| agent_timeline.append(f"⏱ [{time.strftime('%H:%M:%S')}] ✅ VERIFICATION SUCCESSFUL! Pipeline restored to HEALTHY (Accuracy: {post_telemetry['accuracy']*100:.1f}%, Time to Heal: {heal_duration}s).")
|
| final_status = "RESOLVED_AND_HEALTHY"
|
| else:
|
| agent_timeline.append(f"⏱ [{time.strftime('%H:%M:%S')}] ⚠️ Verification Partial. Pipeline output logged for engineer review.")
|
| final_status = "HEAL_ATTEMPTED_NEEDS_REVIEW"
|
|
|
|
|
| diff_html = utils.generate_code_diff_html(broken_code, patched_code)
|
|
|
|
|
| incident_id = f"INC-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
|
| incident_data = {
|
| "id": incident_id,
|
| "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
| "fault_name": fault_name or "Custom Script Anomaly",
|
| "fault_category": fault_cat,
|
| "final_status": final_status,
|
| "heal_duration_sec": heal_duration,
|
| "pre_telemetry": pre_telemetry,
|
| "post_telemetry": post_telemetry,
|
| "root_cause_analysis": rca,
|
| "explanation": explanation,
|
| "verification_checklist": diagnosis.get("verification_checklist", []),
|
| "original_code": broken_code,
|
| "patched_code": patched_code,
|
| "pre_logs": pre_logs,
|
| "post_logs": post_logs
|
| }
|
|
|
| report_file = utils.save_incident_report(incident_data)
|
| agent_timeline.append(f"⏱ [{time.strftime('%H:%M:%S')}] 📝 Saved Incident Audit Report: {report_file}")
|
|
|
| return {
|
| "incident_id": incident_id,
|
| "status": final_status,
|
| "fault_category": fault_cat,
|
| "rca": rca,
|
| "explanation": explanation,
|
| "heal_duration": heal_duration,
|
| "pre_telemetry": pre_telemetry,
|
| "post_telemetry": post_telemetry,
|
| "timeline": "\n".join(agent_timeline),
|
| "diff_html": diff_html,
|
| "broken_code": broken_code,
|
| "patched_code": patched_code,
|
| "pre_logs": pre_logs,
|
| "post_logs": post_logs,
|
| "kpis": self.get_kpi_stats(),
|
| "telemetry_history": self.engine.execution_history
|
| }
|
|
|
