# docker-compose.monitoring.yml
version: '3.8'
services:
  prometheus:
    image: prom/prometheus:v2.51.0
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus:/etc/prometheus
      - prometheus-data:/prometheus
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
      - '--storage.tsdb.path=/prometheus'
      - '--storage.tsdb.retention.time=30d'

  grafana:
    image: grafana/grafana:11.0.0
    ports:
      - "3000:3000"
    volumes:
      - grafana-data:/var/lib/grafana
      - ./grafana/provisioning:/etc/grafana/provisioning
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_PASSWORD}

  node-exporter:
    image: prom/node-exporter:v1.7.0
    ports:
      - "9100:9100"
    volumes:
      - /proc:/host/proc:ro
      - /sys:/host/sys:ro
    command:
      - '--path.procfs=/host/proc'
      - '--path.sysfs=/host/sys'

volumes:
  prometheus-data:
  grafana-data:

# Create virtual environment
python3 -m venv ~/ai-alerting/venv
source ~/ai-alerting/venv/bin/activate
pip install prophet requests pandas numpy scikit-learn

# Or use uv (faster)
uv pip install prophet requests pandas numpy scikit-learn

#!/usr/bin/env python3
"""
AI Anomaly Detection Engine - Prophet-based time series anomaly detection
Supports automatic business baseline learning with dynamic alert threshold adjustment
"""
import json
import time
import logging
from datetime import datetime, timedelta
from pathlib import Path

import numpy as np
import pandas as pd
from prophet import Prophet
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class AIAlerter:
    def __init__(self, prometheus_url="http://localhost:9090"):
        self.prometheus_url = prometheus_url
        self.models = {}
        self.isolation_forests = {}
        self.baseline_cache = Path("/tmp/ai-alerting-baselines")
        self.baseline_cache.mkdir(parents=True, exist_ok=True)
        self.alert_cooldown = {}

    def fetch_prometheus_data(self, query, start_hours=168):
        """Fetch historical data from Prometheus"""
        import requests
        end = datetime.utcnow()
        start = end - timedelta(hours=start_hours)
        
        url = f"{self.prometheus_url}/api/v1/query_range"
        params = {
            "query": query,
            "start": start.isoformat(),
            "end": end.isoformat(),
            "step": "300",  # 5-minute granularity
        }
        
        resp = requests.get(url, params=params, timeout=10)
        resp.raise_for_status()
        data = resp.json()["data"]["result"][0]["values"]
        
        df = pd.DataFrame(data, columns=["timestamp", "value"])
        df["timestamp"] = pd.to_datetime(df["timestamp"], unit="s")
        df["ds"] = df["timestamp"]
        df["y"] = df["value"].astype(float)
        return df[["ds", "y"]]

    def train_prophet_model(self, metric_name, df):
        """Train a Prophet time series forecasting model"""
        model = Prophet(
            yearly_seasonality=True,
            weekly_seasonality=True,
            daily_seasonality=False,
            changepoint_prior_scale=0.05,
            seasonality_prior_scale=0.1,
        )
        model.fit(df)
        self.models[metric_name] = model
        logger.info(f"✅ Training complete: {metric_name} ({len(df)} data points)")
        return model

    def detect_anomaly_prophet(self, metric_name, current_value, window_minutes=30):
        """
        Single-variable anomaly detection using Prophet
        Returns (is_anomaly, deviation_score, expected_range)
        """
        if metric_name not in self.models:
            return False, 0.0, (0, 0)
        
        model = self.models[metric_name]
        now = pd.Timestamp.utcnow()
        
        # Get forecast for the future window
        future = model.make_future_dataframe(periods=window_minutes // 5, freq="min")
        forecast = model.predict(future)
        
        latest = forecast.iloc[-1]
        predicted = latest["yhat"]
        upper = latest["yhat_upper"]
        lower = latest["yhat_lower"]
        
        # Calculate deviation score (in standard deviation units)
        if upper > lower:
            deviation = (current_value - predicted) / max(upper - lower, 0.001)
        else:
            deviation = 0
        
        # Consider anomalous if beyond 2 prediction interval widths
        is_anomaly = abs(deviation) > 2.0 or current_value > upper or current_value < lower
        
        return is_anomaly, deviation, (lower, upper)

    def detect_multivariate_anomaly(self, metric_name, feature_dict):
        """
        Multivariate anomaly detection using Isolation Forest
        feature_dict: {"cpu": 85, "mem": 70, "disk_io": 120, ...}
        """
        if metric_name not in self.isolation_forests:
            self.isolation_forests[metric_name] = {
                "scaler": StandardScaler(),
                "model": IsolationForest(
                    contamination=0.05,
                    random_state=42,
                    n_estimators=100,
                ),
                "features": list(feature_dict.keys()),
                "history": [],
            }
        
        detector = self.isolation_forests[metric_name]
        features = [feature_dict[f] for f in detector["features"]]
        
        # Collect historical data for training
        detector["history"].append(features)
        if len(detector["history"]) > 100:
            detector["history"] = detector["history"][-100:]
            
            # Retrain
            X = np.array(detector["history"])
            X_scaled = detector["scaler"].fit_transform(X)
            detector["model"].fit(X_scaled)
        
        # Detect current sample
        X_current = np.array(features).reshape(1, -1)
        X_scaled = detector["scaler"].transform(X_current)
        prediction = detector["model"].predict(X_scaled)[0]
        score = float(detector["model"].score_samples(X_scaled)[0])
        
        is_anomaly = prediction == -1
        return is_anomaly, score

    def compress_alerts(self, raw_alerts):
        """
        Alert compression - cluster related alerts into tickets
        raw_alerts: [{"metric": "...", "value": ..., "severity": "..."}, ...]
        """
        clusters = {}
        for alert in raw_alerts:
            service = alert.get("labels", {}).get("job", "unknown")
            if service not in clusters:
                clusters[service] = []
            clusters[service].append(alert)
        
        tickets = []
        for service, alerts in clusters.items():
            critical = [a for a in alerts if a.get("severity") == "critical"]
            warnings = [a for a in alerts if a.get("severity") == "warning"]
            
            ticket = {
                "service": service,
                "severity": "critical" if critical else ("warning" if warnings else "info"),
                "alert_count": len(alerts),
                "original_alerts": alerts,
                "summary": self._generate_summary(service, alerts),
            }
            tickets.append(ticket)
        
        tickets.sort(key=lambda t: {"critical": 0, "warning": 1, "info": 2}[t["severity"]])
        return tickets

    def _generate_summary(self, service, alerts):
        """Generate alert summary"""
        metrics = [a.get("metric", "unknown") for a in alerts]
        count = len(alerts)
        severities = set(a.get("severity", "unknown") for a in alerts)
        
        summary = f"Service [{service}] detected {count} related alerts"
        if "critical" in severities:
            summary += " ⚠️ Includes critical severity"
        summary += f"\nMetrics involved: {', '.join(set(metrics)[:5])}"
        return summary

    def get_ai_recommendation(self, ticket):
        """
        AI repair suggestions based on knowledge base
        In production, this would call a local LLM (e.g., Ollama)
        """
        recommendations = {
            "cpu": [
                "Check for processes with abnormally high CPU: `top -bn1 | head -20`",
                "Consider enabling auto-scaling (HPA)",
                "Check if a new version introduced performance regression",
            ],
            "memory": [
                "Check for memory leaks: `pmap -x $(pgrep python3) | tail -5`",
                "Consider adding swap space as buffer",
                "Check if caches were not properly cleaned",
            ],
            "disk": [
                "Clean old logs: `journalctl --vacuum-time=3d`",
                "Check for large files: `du -sh /* | sort -rh | head -10`",
                "Consider migrating logs to remote storage",
            ],
            "network": [
                "Check for signs of DDoS attack",
                "View connection count: `ss -s`",
                "Confirm whether bandwidth is saturated",
            ],
        }
        
        service = ticket["service"]
        suggestions = []
        for metric_key, recs in recommendations.items():
            if metric_key in str(ticket.get("summary", "")).lower():
                suggestions.extend(recs)
        
        if not suggestions:
            suggestions = [
                "Check service logs for more information",
                "Review recent deployment change records",
                "Contact the relevant service owner",
            ]
        
        return suggestions[:3]

# Install Ollama
curl -fsSL https://ollama.com/install.sh | sh

# Pull a lightweight model (suitable for VPS)
ollama pull llama3.2:3b

#!/usr/bin/env python3
"""
LLM-Driven Root Cause Analysis Service
Runs locally via Ollama — no external API calls needed
"""
import json
from datetime import datetime

OLLAMA_URL = "http://localhost:11434"

def query_ollama(prompt, model="llama3.2:3b"):
    """Call local Ollama model via API"""
    import requests
    resp = requests.post(
        f"{OLLAMA_URL}/api/generate",
        json={
            "model": model,
            "prompt": prompt,
            "stream": False,
            "options": {
                "temperature": 0.3,
                "num_predict": 500,
            },
        },
        timeout=60,
    )
    return resp.json()["response"]

def analyze_root_cause(alert_ticket, metrics_history):
    """
    Analyze root cause of an alert ticket
    
    Args:
        alert_ticket: Compressed alert ticket
        metrics_history: Related metric historical data
    
    Returns:
        Root cause analysis report
    """
    prompt = f"""You are a senior SRE engineer. Based on the following alert information and metric data, analyze the root cause and provide remediation suggestions.

【Alert Ticket】
Service: {alert_ticket['service']}
Severity: {alert_ticket['severity']}
Alert Count: {alert_ticket['alert_count']}
Summary: {alert_ticket['summary']}

【Related Metric Data】
{json.dumps(metrics_history, indent=2, ensure_ascii=False)}

【System Information】
Recent Changes: 1 deployment in the past 24 hours (version v2.3.1)
Uptime: Service has been running for 14 days

Please respond in the following format:
1. **Root Cause**: Most likely fundamental cause
2. **Confidence**: High/Medium/Low
3. **Evidence Chain**: Key metrics supporting this judgment
4. **Remediation Steps**: Specific actionable steps
5. **Prevention**: How to avoid similar issues in the future
"""
    
    try:
        analysis = query_ollama(prompt)
        return {
            "analysis": analysis,
            "timestamp": datetime.utcnow().isoformat(),
            "model": "llama3.2:3b",
        }
    except Exception as e:
        return {
            "error": str(e),
            "fallback_advice": "Please manually check service logs and related metrics",
        }

#!/usr/bin/env python3
"""
Alert Notification & Auto-Remediation Module
Supports multiple notification channels and automated fix actions
"""
import subprocess
import json
from datetime import datetime

class AlertResponder:
    def __init__(self):
        self.notification_channels = {
            "slack": self._send_slack,
            "dingtalk": self._send_dingtalk,
            "email": self._send_email,
            "webhook": self._send_webhook,
        }
        self.auto_remediation_rules = [
            {
                "condition": lambda t: "restart" in str(t.get("actions", [])).lower(),
                "action": self._auto_restart_service,
            },
            {
                "condition": lambda t: "scale" in str(t.get("actions", [])).lower(),
                "action": self._auto_scale_resource,
            },
            {
                "condition": lambda t: "rollback" in str(t.get("actions", [])).lower(),
                "action": self._auto_rollback_deployment,
            },
        ]

    def dispatch_alert(self, ticket, channels=None):
        """Dispatch alert to specified channels"""
        if channels is None:
            channels = ["slack", "dingtalk"]
        
        payload = {
            "timestamp": datetime.utcnow().isoformat(),
            "service": ticket["service"],
            "severity": ticket["severity"],
            "alert_count": ticket["alert_count"],
            "summary": ticket["summary"],
            "recommendations": ticket.get("recommendations", []),
        }
        
        results = {}
        for channel in channels:
            if channel in self.notification_channels:
                try:
                    result = self.notification_channels[channel](payload)
                    results[channel] = "success" if result else "failed"
                except Exception as e:
                    results[channel] = f"error: {str(e)}"
        
        return results

    def _send_slack(self, payload):
        """Send Slack notification"""
        import requests
        webhook_url = "YOUR_SLACK_WEBHOOK_URL"
        
        color_map = {
            "critical": "#ff0000",
            "warning": "#ffaa00",
            "info": "#00aa00",
        }
        
        blocks = [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"🚨 *{payload['severity'].upper()} ALERT*\n"
                            f"Service: {payload['service']}\n"
                            f"Alert Count: {payload['alert_count']}\n"
                            f"Summary: {payload['summary']}",
                },
            },
        ]
        
        if payload.get("recommendations"):
            rec_text = "\n".join([f"• {r}" for r in payload["recommendations"][:3]])
            blocks.append({
                "type": "section",
                "text": {"type": "mrkdwn", "text": f"💡 *Suggested Actions:*\n{rec_text}"},
            })
        
        payload_slack = {
            "attachments": [{
                "color": color_map.get(payload["severity"], "#888888"),
                "blocks": blocks,
            }],
        }
        
        resp = requests.post(webhook_url, json=payload_slack, timeout=10)
        return resp.status_code == 200

    def _send_dingtalk(self, payload):
        """Send DingTalk notification"""
        import requests
        webhook_url = "YOUR_DINGTALK_WEBHOOK_URL"
        
        msg = {
            "msgtype": "markdown",
            "markdown": {
                "title": f"⚠️ {payload['severity'].upper()} Alert",
                "text": f"### {payload['severity'].upper()} Alert\n\n"
                        f"> Service: {payload['service']}\n"
                        f"> Alert Count: {payload['alert_count']}\n"
                        f"> Summary: {payload['summary']}\n\n"
                        + ("\n".join([f"- {r}" for r in payload.get("recommendations", [])])),
            },
        }
        
        resp = requests.post(webhook_url, json=msg, timeout=10)
        return resp.status_code == 200

    def auto_remediate(self, ticket):
        """Execute auto-remediation"""
        results = []
        for rule in self.auto_remediation_rules:
            if rule["condition"](ticket):
                result = rule["action"](ticket)
                results.append(result)
        return results

    def _auto_restart_service(self, ticket):
        """Auto-restart a service"""
        service_name = ticket["service"]
        try:
            result = subprocess.run(
                ["sudo", "systemctl", "restart", service_name],
                capture_output=True, text=True, timeout=30
            )
            return {
                "action": "restart",
                "service": service_name,
                "status": "success" if result.returncode == 0 else "failed",
                "output": result.stdout + result.stderr,
            }
        except Exception as e:
            return {"action": "restart", "error": str(e)}

    def _auto_scale_resource(self, ticket):
        """Auto-scale resources"""
        return {"action": "scale", "note": "Configure scaling logic based on actual environment"}

    def _auto_rollback_deployment(self, ticket):
        """Auto-rollback deployment"""
        return {"action": "rollback", "note": "Requires integration with CI/CD pipeline"}

# Voting mechanism: trigger alert only if at least 2 models agree
def ensemble_detect(metric_name, current_value, features):
    votes = 0
    
    # Prophet anomaly detection
    is_anomaly_p, _, _ = alerter.detect_anomaly_prophet(metric_name, current_value)
    if is_anomaly_p:
        votes += 1
    
    # Isolation Forest multivariate detection
    is_anomaly_if, _ = alerter.detect_multivariate_anomaly(metric_name, features)
    if is_anomaly_if:
        votes += 1
    
    # Statistical method (3-sigma)
    if abs(current_value - baseline_mean) > 3 * baseline_std:
        votes += 1
    
    return votes >= 2  # Majority vote

def adaptive_threshold(metric_name, current_value):
    """Adaptively adjust thresholds based on time of day and business cycles"""
    hour = datetime.now().hour
    day_of_week = datetime.now().weekday()
    
    # Load baseline for this time period
    baseline = load_baseline(metric_name, hour, day_of_week)
    
    # Dynamically adjust based on recent coefficient of variation
    cv = baseline["std"] / max(baseline["mean"], 0.001)
    sensitivity = 2.0 if cv < 0.1 else (2.5 if cv < 0.3 else 3.0)
    
    return baseline["mean"] + sensitivity * baseline["std"]