AIOps: AI-Powered DevOps Automation and Intelligent Operations

# AI-powered anomaly detection
from sklearn.ensemble import IsolationForest
import numpy as np

class AnomalyDetector:
    def __init__(self):
        self.model = IsolationForest(
            contamination=0.1,
            random_state=42
        )
        self.baseline_data = []
    
    def train_baseline(self, metrics_history):
        """Learn normal behavior patterns"""
        features = self.extract_features(metrics_history)
        self.model.fit(features)
        return self
    
    def detect_anomaly(self, current_metrics):
        """Detect if current metrics are anomalous"""
        features = self.extract_features([current_metrics])
        prediction = self.model.predict(features)
        anomaly_score = self.model.score_samples(features)
        
        return {
            'is_anomaly': prediction[0] == -1,
            'anomaly_score': float(anomaly_score[0]),
            'severity': self.calculate_severity(anomaly_score[0])
        }
    
    def extract_features(self, metrics):
        """Extract relevant features for ML model"""
        return np.array([
            [m['cpu_usage'], m['memory_usage'], 
             m['request_rate'], m['error_rate'],
             m['response_time']]
            for m in metrics
        ])
    
    def calculate_severity(self, score):
        """Map anomaly score to severity level"""
        if score < -0.5:
            return 'critical'
        elif score < -0.3:
            return 'warning'
        return 'info'

# Usage
detector = AnomalyDetector()
detector.train_baseline(historical_metrics)

# Real-time detection
result = detector.detect_anomaly(current_metrics)
if result['is_anomaly']:
    trigger_intelligent_alert(result)

# Predictive incident model
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler

class IncidentPredictor:
    def __init__(self):
        self.model = RandomForestClassifier(
            n_estimators=100,
            max_depth=10,
            random_state=42
        )
        self.scaler = StandardScaler()
        self.feature_importance = {}
    
    def prepare_training_data(self, incidents_history):
        """Prepare features from historical incidents"""
        features = []
        labels = []
        
        for incident in incidents_history:
            # Extract features before incident
            feature_vector = {
                'cpu_trend': incident['cpu_trend'],
                'memory_trend': incident['memory_trend'],
                'error_rate_spike': incident['error_rate_change'],
                'deployment_recent': incident['deployment_within_24h'],
                'traffic_increase': incident['traffic_change_pct'],
                'disk_usage': incident['disk_usage'],
                'network_latency': incident['network_latency'],
                'time_of_day': incident['hour'],
                'day_of_week': incident['day_of_week']
            }
            
            features.append(list(feature_vector.values()))
            labels.append(1 if incident['occurred'] else 0)
        
        return pd.DataFrame(features), labels
    
    def train(self, incidents_history):
        """Train the prediction model"""
        X, y = self.prepare_training_data(incidents_history)
        X_scaled = self.scaler.fit_transform(X)
        
        self.model.fit(X_scaled, y)
        
        # Store feature importance
        feature_names = X.columns
        self.feature_importance = dict(
            zip(feature_names, self.model.feature_importances_)
        )
        
        return self
    
    def predict_incident_probability(self, current_state):
        """Predict probability of incident in next hour"""
        features = self.extract_current_features(current_state)
        features_scaled = self.scaler.transform([features])
        
        probability = self.model.predict_proba(features_scaled)[0][1]
        
        return {
            'incident_probability': float(probability),
            'risk_level': self.get_risk_level(probability),
            'top_risk_factors': self.get_top_risk_factors(features),
            'recommended_actions': self.get_recommendations(probability, features)
        }
    
    def get_risk_level(self, probability):
        """Map probability to risk level"""
        if probability > 0.7:
            return 'HIGH'
        elif probability > 0.4:
            return 'MEDIUM'
        return 'LOW'
    
    def get_recommendations(self, probability, features):
        """Generate actionable recommendations"""
        recommendations = []
        
        if probability > 0.5:
            if features['cpu_trend'] > 0.8:
                recommendations.append({
                    'action': 'scale_up_compute',
                    'priority': 'high',
                    'reason': 'CPU trending upward'
                })
            
            if features['error_rate_spike'] > 0.5:
                recommendations.append({
                    'action': 'investigate_errors',
                    'priority': 'critical',
                    'reason': 'Error rate spike detected'
                })
            
            if features['deployment_recent']:
                recommendations.append({
                    'action': 'prepare_rollback',
                    'priority': 'high',
                    'reason': 'Recent deployment may be unstable'
                })
        
        return recommendations

# Usage
predictor = IncidentPredictor()
predictor.train(historical_incidents)

# Continuous prediction
prediction = predictor.predict_incident_probability(current_system_state)

if prediction['risk_level'] == 'HIGH':
    alert_ops_team(prediction)
    execute_preventive_actions(prediction['recommended_actions'])

# AI-driven root cause analysis
from sklearn.cluster import DBSCAN
import networkx as nx

class RootCauseAnalyzer:
    def __init__(self):
        self.dependency_graph = nx.DiGraph()
        self.incident_patterns = []
    
    def build_dependency_graph(self, services_config):
        """Build service dependency graph"""
        for service in services_config:
            self.dependency_graph.add_node(
                service['name'],
                type=service['type'],
                criticality=service['criticality']
            )
            
            for dependency in service.get('dependencies', []):
                self.dependency_graph.add_edge(
                    service['name'],
                    dependency,
                    weight=1
                )
    
    def analyze_incident(self, incident_data):
        """Perform root cause analysis"""
        affected_services = incident_data['affected_services']
        symptoms = incident_data['symptoms']
        timeline = incident_data['timeline']
        
        # Step 1: Find common upstream dependencies
        upstream_services = self.find_common_dependencies(
            affected_services
        )
        
        # Step 2: Analyze timeline for causality
        causal_chain = self.analyze_temporal_causality(timeline)
        
        # Step 3: Pattern matching with historical incidents
        similar_incidents = self.find_similar_incidents(symptoms)
        
        # Step 4: Calculate root cause probability
        root_causes = self.calculate_root_cause_probability(
            upstream_services,
            causal_chain,
            similar_incidents
        )
        
        return {
            'probable_root_causes': root_causes,
            'confidence_score': self.calculate_confidence(root_causes),
            'investigation_path': self.generate_investigation_path(root_causes),
            'similar_past_incidents': similar_incidents[:3]
        }
    
    def find_common_dependencies(self, affected_services):
        """Find services that affect all impacted services"""
        if not affected_services:
            return []
        
        # Find all upstream dependencies for each affected service
        upstream_sets = []
        for service in affected_services:
            upstream = nx.ancestors(self.dependency_graph, service)
            upstream_sets.append(set(upstream))
        
        # Find common dependencies
        common = set.intersection(*upstream_sets) if upstream_sets else set()
        
        # Rank by criticality and distance
        ranked = []
        for dep in common:
            criticality = self.dependency_graph.nodes[dep]['criticality']
            distance = min(
                nx.shortest_path_length(self.dependency_graph, dep, svc)
                for svc in affected_services
            )
            ranked.append({
                'service': dep,
                'criticality': criticality,
                'distance': distance,
                'probability': criticality / (distance + 1)
            })
        
        return sorted(ranked, key=lambda x: x['probability'], reverse=True)
    
    def analyze_temporal_causality(self, timeline):
        """Analyze event timeline for causal relationships"""
        events = sorted(timeline, key=lambda x: x['timestamp'])
        causal_chain = []
        
        for i in range(len(events) - 1):
            current = events[i]
            next_event = events[i + 1]
            
            time_diff = (next_event['timestamp'] - 
                        current['timestamp']).total_seconds()
            
            # If events are close in time, there may be causality
            if time_diff < 300:  # 5 minutes
                causal_chain.append({
                    'cause': current['service'],
                    'effect': next_event['service'],
                    'time_delta': time_diff,
                    'causality_score': 1.0 / (time_diff + 1)
                })
        
        return causal_chain
    
    def find_similar_incidents(self, symptoms):
        """Find similar historical incidents using clustering"""
        # Convert symptoms to feature vectors
        symptom_vector = self.vectorize_symptoms(symptoms)
        
        # Find similar patterns
        similar = []
        for past_incident in self.incident_patterns:
            similarity = self.calculate_similarity(
                symptom_vector,
                past_incident['symptom_vector']
            )
            
            if similarity > 0.7:
                similar.append({
                    'incident_id': past_incident['id'],
                    'similarity': similarity,
                    'root_cause': past_incident['root_cause'],
                    'resolution': past_incident['resolution']
                })
        
        return sorted(similar, key=lambda x: x['similarity'], reverse=True)
    
    def generate_investigation_path(self, root_causes):
        """Generate step-by-step investigation guide"""
        path = []
        
        for cause in root_causes[:3]:
            path.append({
                'step': len(path) + 1,
                'service': cause['service'],
                'checks': [
                    f"Check {cause['service']} logs for errors",
                    f"Verify {cause['service']} resource utilization",
                    f"Review recent deployments to {cause['service']}",
                    f"Check {cause['service']} dependencies health"
                ],
                'priority': cause['probability']
            })
        
        return path

# Usage
rca = RootCauseAnalyzer()
rca.build_dependency_graph(service_topology)

# When incident occurs
incident = {
    'affected_services': ['api-gateway', 'user-service', 'order-service'],
    'symptoms': {
        'high_latency': True,
        'error_rate_spike': True,
        'timeout_errors': True
    },
    'timeline': incident_events
}

analysis = rca.analyze_incident(incident)
print(f"Root Cause: {analysis['probable_root_causes'][0]['service']}")
print(f"Confidence: {analysis['confidence_score']}")

# Alert correlation engine
from sklearn.cluster import AgglomerativeClustering
from datetime import datetime, timedelta

class AlertCorrelationEngine:
    def __init__(self):
        self.alert_history = []
        self.correlation_rules = []
    
    def correlate_alerts(self, incoming_alerts):
        """Group related alerts into incidents"""
        if not incoming_alerts:
            return []
        
        # Extract features for clustering
        features = self.extract_alert_features(incoming_alerts)
        
        # Perform clustering
        clustering = AgglomerativeClustering(
            n_clusters=None,
            distance_threshold=0.5,
            linkage='average'
        )
        
        clusters = clustering.fit_predict(features)
        
        # Group alerts by cluster
        incidents = {}
        for idx, cluster_id in enumerate(clusters):
            if cluster_id not in incidents:
                incidents[cluster_id] = []
            incidents[cluster_id].append(incoming_alerts[idx])
        
        # Create incident summaries
        return [
            self.create_incident_summary(alerts)
            for alerts in incidents.values()
        ]
    
    def extract_alert_features(self, alerts):
        """Extract features for correlation"""
        features = []
        
        for alert in alerts:
            feature_vector = [
                self.encode_service(alert['service']),
                self.encode_severity(alert['severity']),
                self.encode_alert_type(alert['type']),
                alert['timestamp'].timestamp(),
                self.get_service_tier(alert['service'])
            ]
            features.append(feature_vector)
        
        return features
    
    def create_incident_summary(self, alerts):
        """Create unified incident from correlated alerts"""
        # Sort by timestamp
        alerts = sorted(alerts, key=lambda x: x['timestamp'])
        
        # Determine primary alert (highest severity)
        primary = max(alerts, key=lambda x: self.severity_score(x['severity']))
        
        # Calculate incident severity
        incident_severity = self.calculate_incident_severity(alerts)
        
        # Identify likely root cause
        root_cause_alert = self.identify_root_cause(alerts)
        
        return {
            'incident_id': self.generate_incident_id(),
            'title': self.generate_incident_title(alerts),
            'severity': incident_severity,
            'affected_services': list(set(a['service'] for a in alerts)),
            'alert_count': len(alerts),
            'first_seen': alerts[0]['timestamp'],
            'last_seen': alerts[-1]['timestamp'],
            'primary_alert': primary,
            'likely_root_cause': root_cause_alert,
            'correlated_alerts': alerts,
            'recommended_actions': self.get_recommended_actions(alerts)
        }
    
    def identify_root_cause(self, alerts):
        """Identify the alert most likely to be root cause"""
        # Root cause is typically:
        # 1. First alert in time
        # 2. From a critical upstream service
        # 3. Infrastructure-related
        
        scored_alerts = []
        first_timestamp = alerts[0]['timestamp']
        
        for alert in alerts:
            score = 0
            
            # Time-based score (earlier = higher score)
            time_diff = (alert['timestamp'] - first_timestamp).total_seconds()
            score += max(0, 100 - time_diff)
            
            # Service criticality score
            if self.is_critical_service(alert['service']):
                score += 50
            
            # Alert type score
            if alert['type'] in ['infrastructure', 'database', 'network']:
                score += 30
            
            scored_alerts.append({
                'alert': alert,
                'root_cause_score': score
            })
        
        return max(scored_alerts, key=lambda x: x['root_cause_score'])['alert']
    
    def get_recommended_actions(self, alerts):
        """Generate AI-powered recommendations"""
        actions = []
        
        # Analyze alert patterns
        alert_types = [a['type'] for a in alerts]
        services = [a['service'] for a in alerts]
        
        if 'high_error_rate' in alert_types:
            actions.append({
                'action': 'check_recent_deployments',
                'priority': 'high',
                'reason': 'Error rate spike detected'
            })
        
        if 'high_latency' in alert_types:
            actions.append({
                'action': 'check_database_performance',
                'priority': 'high',
                'reason': 'Latency issues detected'
            })
        
        if len(set(services)) > 5:
            actions.append({
                'action': 'check_infrastructure',
                'priority': 'critical',
                'reason': 'Multiple services affected - possible infrastructure issue'
            })
        
        return actions

# Usage
correlator = AlertCorrelationEngine()

# Process incoming alerts
alerts = get_recent_alerts(last_5_minutes)
incidents = correlator.correlate_alerts(alerts)

# Send correlated incidents instead of individual alerts
for incident in incidents:
    if incident['severity'] in ['critical', 'high']:
        notify_ops_team(incident)
        execute_automated_response(incident['recommended_actions'])

# Data quality validation
class DataQualityValidator:
    def validate_metrics_quality(self, metrics_data):
        """Ensure metrics data is suitable for ML"""
        issues = []
        
        # Check for missing data
        missing_pct = metrics_data.isnull().sum() / len(metrics_data)
        if missing_pct.any() > 0.1:
            issues.append(f"High missing data: {missing_pct[missing_pct > 0.1]}")
        
        # Check for data staleness
        latest_timestamp = metrics_data['timestamp'].max()
        if (datetime.now() - latest_timestamp) > timedelta(minutes=10):
            issues.append("Data is stale")
        
        # Check for anomalies in data collection
        collection_rate = self.calculate_collection_rate(metrics_data)
        if collection_rate < 0.95:
            issues.append(f"Low collection rate: {collection_rate}")
        
        return {
            'is_valid': len(issues) == 0,
            'issues': issues,
            'quality_score': self.calculate_quality_score(metrics_data)
        }

# AIOps metrics dashboard
class AIOpsMetrics:
    def calculate_effectiveness(self, time_period):
        """Calculate AIOps impact metrics"""
        return {
            # Detection metrics
            'anomaly_detection_accuracy': self.get_accuracy(),
            'false_positive_rate': self.get_false_positive_rate(),
            'mean_time_to_detect': self.get_mttd(),
            
            # Prediction metrics
            'incident_prediction_accuracy': self.get_prediction_accuracy(),
            'prevented_incidents': self.get_prevented_incidents(),
            'prediction_lead_time': self.get_prediction_lead_time(),
            
            # Automation metrics
            'auto_remediation_success_rate': self.get_auto_fix_rate(),
            'manual_interventions_reduced': self.get_intervention_reduction(),
            'mean_time_to_resolve': self.get_mttr(),
            
            # Business impact
            'alert_noise_reduction': self.get_noise_reduction(),
            'ops_team_productivity_gain': self.get_productivity_gain(),
            'cost_savings': self.calculate_cost_savings()
        }

# AI-powered predictive autoscaling
class PredictiveScaler:
    def __init__(self):
        self.traffic_predictor = TrafficPredictor()
        self.resource_optimizer = ResourceOptimizer()
    
    def predict_and_scale(self, service_name):
        """Predict traffic and scale proactively"""
        # Predict traffic for next hour
        traffic_forecast = self.traffic_predictor.forecast(
            service=service_name,
            horizon='1h'
        )
        
        # Calculate required resources
        required_resources = self.resource_optimizer.calculate(
            predicted_traffic=traffic_forecast,
            target_latency='100ms',
            target_availability=0.999
        )
        
        # Get current resources
        current = self.get_current_resources(service_name)
        
        # Scale proactively if needed
        if required_resources['replicas'] > current['replicas']:
            self.scale_service(
                service=service_name,
                replicas=required_resources['replicas'],
                reason='Predicted traffic increase'
            )
            
            return {
                'action': 'scaled_up',
                'from': current['replicas'],
                'to': required_resources['replicas'],
                'reason': 'Proactive scaling based on traffic prediction'
            }

# AI-powered incident response
class IntelligentIncidentResponder:
    def __init__(self):
        self.rca_analyzer = RootCauseAnalyzer()
        self.runbook_engine = RunbookEngine()
        self.action_executor = ActionExecutor()
    
    def respond_to_incident(self, incident):
        """Automated intelligent incident response"""
        # Step 1: Analyze root cause
        rca_result = self.rca_analyzer.analyze_incident(incident)
        
        # Step 2: Find matching runbook
        runbook = self.runbook_engine.find_runbook(
            root_cause=rca_result['probable_root_causes'][0],
            symptoms=incident['symptoms']
        )
        
        # Step 3: Execute automated remediation
        if rca_result['confidence_score'] > 0.8 and runbook:
            execution_result = self.action_executor.execute(
                runbook=runbook,
                context=incident,
                safety_checks=True
            )
            
            return {
                'automated': True,
                'root_cause': rca_result['probable_root_causes'][0],
                'actions_taken': execution_result['actions'],
                'resolution_time': execution_result['duration'],
                'success': execution_result['success']
            }
        else:
            # Low confidence - escalate to human
            return {
                'automated': False,
                'reason': 'Low confidence or no matching runbook',
                'rca_suggestions': rca_result,
                'recommended_runbook': runbook,
                'escalated_to': 'on_call_engineer'
            }

# AI-driven capacity planning
class CapacityPlanner:
    def __init__(self):
        self.growth_predictor = GrowthPredictor()
        self.cost_optimizer = CostOptimizer()
    
    def plan_capacity(self, service_name, planning_horizon='90d'):
        """Generate capacity plan using AI"""
        # Predict resource usage growth
        forecast = self.growth_predictor.forecast(
            service=service_name,
            horizon=planning_horizon,
            include_seasonality=True
        )
        
        # Identify capacity constraints
        constraints = self.identify_constraints(forecast)
        
        # Generate recommendations
        recommendations = []
        
        for constraint in constraints:
            if constraint['type'] == 'compute':
                recommendations.append({
                    'resource': 'compute',
                    'action': 'add_nodes',
                    'quantity': constraint['additional_needed'],
                    'timeline': constraint['needed_by'],
                    'estimated_cost': self.cost_optimizer.estimate_cost(
                        resource='compute',
                        quantity=constraint['additional_needed']
                    )
                })
            
            elif constraint['type'] == 'storage':
                recommendations.append({
                    'resource': 'storage',
                    'action': 'expand_storage',
                    'quantity': constraint['additional_needed'],
                    'timeline': constraint['needed_by'],
                    'estimated_cost': self.cost_optimizer.estimate_cost(
                        resource='storage',
                        quantity=constraint['additional_needed']
                    )
                })
        
        return {
            'forecast': forecast,
            'constraints': constraints,
            'recommendations': recommendations,
            'total_estimated_cost': sum(r['estimated_cost'] for r in recommendations)
        }