File size: 6,133 Bytes
9c06673 ec78ba8 9c06673 ec78ba8 9c06673 ec78ba8 9c06673 ec78ba8 9c06673 ec78ba8 9c06673 ec78ba8 9c06673 ec78ba8 9c06673 ec78ba8 9c06673 ec78ba8 9c06673 ec78ba8 9c06673 ec78ba8 9c06673 ec78ba8 9c06673 ec78ba8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 | """
Demo Orchestrator - Integrates with ARF OSS framework
"""
import asyncio
import json
import datetime
from typing import Dict, List, Any, Optional, Tuple
import logging
logger = logging.getLogger(__name__)
class DemoOrchestrator:
"""Orchestrates the demo workflow using ARF OSS"""
def __init__(self, arf_client=None):
self.arf_client = arf_client
self.incident_history = []
self.execution_history = []
self.learning_stats = {
"patterns_detected": 0,
"similar_incidents_found": 0,
"healing_intents_created": 0
}
async def analyze_incident(self, scenario_name: str, scenario_data: Dict) -> Dict:
"""Analyze incident using ARF OSS"""
try:
if self.arf_client and hasattr(self.arf_client, 'analyze_and_recommend'):
# Use actual ARF OSS analysis
healing_intent = await self.arf_client.analyze_and_recommend(
tool_name="analyze",
component=scenario_data.get("component", "unknown"),
parameters=scenario_data.get("metrics", {}),
context={"scenario": scenario_name}
)
self.learning_stats["healing_intents_created"] += 1
return {
"status": "success",
"healing_intent": healing_intent.to_enterprise_request(),
"analysis": {
"confidence": healing_intent.confidence,
"similar_incidents": healing_intent.similar_incidents,
"recommendation": healing_intent.justification
}
}
# Fallback to mock analysis
return {
"status": "success",
"analysis": {
"confidence": 0.85,
"similar_incidents": [
{"id": "inc_001", "similarity": 0.78, "component": "redis"},
{"id": "inc_045", "similarity": 0.65, "component": "database"}
],
"recommendation": f"Based on 2 similar incidents, recommend action for {scenario_name}"
}
}
except Exception as e:
logger.error(f"Analysis failed: {e}")
return {
"status": "error",
"message": str(e)
}
def execute_healing(self, scenario_name: str, healing_intent: Dict,
mode: str = "autonomous") -> Dict:
"""Execute healing action"""
execution_record = {
"id": f"exec_{len(self.execution_history):03d}",
"scenario": scenario_name,
"timestamp": datetime.datetime.now().isoformat(),
"mode": mode,
"healing_intent": healing_intent,
"status": "completed",
"results": {
"recovery_time_minutes": 12,
"cost_saved": 7200,
"users_impacted": "45,000 → 0"
}
}
self.execution_history.append(execution_record)
# Update learning stats
self.learning_stats["patterns_detected"] += 1
return execution_record
def get_similar_incidents(self, query: str, limit: int = 5) -> List[Dict]:
"""Find similar incidents"""
# This would integrate with ARF's RAG memory
return [
{
"id": "inc_001",
"similarity": 0.92,
"scenario": "Cache Miss Storm",
"resolution": "Scaled Redis cluster + circuit breaker",
"recovery_time": "12 minutes"
},
{
"id": "inc_045",
"similarity": 0.78,
"scenario": "Database Connection Pool",
"resolution": "Increased pool size + monitoring",
"recovery_time": "18 minutes"
}
][:limit]
def calculate_roi(self, company_data: Dict) -> Dict:
"""Calculate ROI based on company data"""
monthly_incidents = company_data.get("monthly_incidents", 10)
avg_cost_per_incident = company_data.get("avg_cost_per_incident", 5000)
team_size = company_data.get("team_size", 3)
annual_impact = monthly_incidents * 12 * avg_cost_per_incident
team_cost = team_size * 150000 # $150k per engineer
savings = annual_impact * 0.82 # 82% savings with ARF
roi_multiplier = savings / team_cost if team_cost > 0 else 0
return {
"annual_impact": annual_impact,
"team_cost": team_cost,
"potential_savings": savings,
"roi_multiplier": roi_multiplier,
"payback_months": (team_cost / (savings / 12)) if savings > 0 else 0,
"recommendation": self._get_roi_recommendation(roi_multiplier)
}
def _get_roi_recommendation(self, roi_multiplier: float) -> str:
"""Get recommendation based on ROI"""
if roi_multiplier >= 5.0:
return "🚀 Excellent fit for ARF Enterprise"
elif roi_multiplier >= 2.0:
return "✅ Good ROI with ARF Enterprise"
elif roi_multiplier >= 1.0:
return "⚠️ Consider ARF OSS edition first"
else:
return "🆓 Start with ARF OSS (free)"
def get_audit_trail(self) -> Dict:
"""Get complete audit trail"""
return {
"incidents": self.incident_history,
"executions": self.execution_history,
"learning_stats": self.learning_stats,
"exported_at": datetime.datetime.now().isoformat()
}
def reset_demo(self):
"""Reset demo state"""
self.incident_history = []
self.execution_history = []
self.learning_stats = {
"patterns_detected": 0,
"similar_incidents_found": 0,
"healing_intents_created": 0
} |