AI in DevOps: Automation and Productivity

# anomaly_detector.py
from sklearn.ensemble import IsolationForest
import pandas as pd

class AnomalyDetector:
    def __init__(self):
        self.model = IsolationForest(contamination=0.1)
        self.trained = False
    
    def train(self, metrics_data):
        df = pd.DataFrame(metrics_data)
        features = df[['cpu', 'memory', 'latency', 'errors']]
        self.model.fit(features)
        self.trained = True
    
    def predict(self, metrics):
        if not self.trained:
            raise ValueError("Model not trained")
        
        features = [list(metrics.values())]
        prediction = self.model.predict(features)
        
        return {
            'is_anomaly': prediction[0] == -1,
            'score': self.model.score_samples(features)[0]
        }

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

current_metrics = {'cpu': 95, 'memory': 80, 'latency': 500, 'errors': 50}
result = detector.predict(current_metrics)
print(f"Anomaly detected: {result}")

# alert_correlator.py
from collections import defaultdict

class AlertCorrelator:
    def __init__(self):
        self.correlation_rules = [
            {'cause': 'high_cpu', 'symptoms': ['cpu', 'load', 'throttle']},
            {'cause': 'database_down', 'symptoms': ['db', 'connection', 'timeout']},
            {'cause': 'network_issue', 'symptoms': ['network', 'dns', 'connection']},
        ]
    
    def correlate(self, alerts):
        groups = defaultdict(list)
        
        for alert in alerts:
            for rule in self.correlation_rules:
                if any(symptom in alert['name'].lower() for symptom in rule['symptoms']):
                    groups[rule['cause']].append(alert)
                    break
        
        return [{'cause': cause, 'alerts': alerts} for cause, alerts in groups.items()]

# Usage
alerts = [
    {'name': 'High CPU Usage', 'severity': 'critical'},
    {'name': 'Load Average High', 'severity': 'warning'},
    {'name': 'Database Connection Timeout', 'severity': 'critical'},
]

correlator = AlertCorrelator()
groups = correlator.correlate(alerts)

# ai-runbook-automation.yaml
apiVersion: aiops.example.com/v1
kind: AIRunbook
metadata:
  name: high-memory-runbook
spec:
  trigger:
    condition: memory_usage > 90
    duration: 5m
  
  actions:
    - name: scale_deployment
      type: scale
      params:
        deployment: "{{.affected_service}}"
        replicas: "+2"
    
    - name: clear_cache
      type: execute
      command: redis-cli FLUSHALL
    
    - name: notify_oncall
      type: notify
      channel: "#incidents"
      message: "Auto-scaled {{.affected_service}} due to high memory"
  
  rollback:
    condition: memory_usage < 70
    duration: 10m
    actions:
      - name: scale_down
        type: scale
        replicas: "-2"

# self_healer.py
class SelfHealing:
    def __init__(self, k8s_client):
        self.client = k8s_client
    
    def analyze_and_heal(self, incident):
        root_cause = self.diagnose(incident)
        
        healing_actions = {
            'pod_crash': self.restart_pod,
            'high_memory': self.scale_up,
            'disk_full': self.cleanup_logs,
            'connection_timeout': self.restart_db_connections,
        }
        
        action = healing_actions.get(root_cause)
        if action:
            result = action(incident)
            return {'healed': True, 'action': action.__name__, 'result': result}
        
        return {'healed': False, 'reason': 'Unknown root cause'}
    
    def diagnose(self, incident):
        symptoms = incident.get('symptoms', [])
        
        if 'crash' in symptoms:
            return 'pod_crash'
        if 'memory' in symptoms:
            return 'high_memory'
        if 'disk' in symptoms:
            return 'disk_full'
        
        return 'unknown'

# capacity_predictor.py
import numpy as np
from prophet import Prophet

class CapacityPredictor:
    def __init__(self):
        self.model = Prophet()
    
    def train(self, historical_data):
        df = pd.DataFrame(historical_data)
        df['ds'] = pd.to_datetime(df['timestamp'])
        df['y'] = df['cpu_usage']
        
        self.model.fit(df)
    
    def predict(self, days_ahead=7):
        future = self.model.make_future_dataframe(periods=days_ahead * 24)
        forecast = self.model.predict(future)
        
        # Find when capacity will exceed threshold
        exceeding = forecast[forecast['yhat_upper'] > 90]
        
        return {
            'predictions': forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']].tail(24),
            'threshold_breach': exceeding['ds'].tolist() if len(exceeding) > 0 else []
        }

# Usage
predictor = CapacityPredictor()
predictor.train(historical_metrics)
result = predictor.predict(days_ahead=7)

# Always include human approval
def automated_action(action, threshold):
    if action.impact > threshold:
        # Require human approval
        approval = request_human_approval(action)
        if not approval.approved:
            return {'status': 'pending_approval'}
    
    return execute_action(action)

# Learn from outcomes
def record_outcome(prediction, actual):
    # Store prediction and actual result
    # Periodically retrain model
    model.retrain(recent_data)