# 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:

# 创建虚拟环境
python3 -m venv ~/ai-alerting/venv
source ~/ai-alerting/venv/bin/activate
pip install prophet requests pandas numpy scikit-learn

# 或者使用 uv（更快）
uv pip install prophet requests pandas numpy scikit-learn

#!/usr/bin/env python3
"""
AI异常检测引擎 - 基于Prophet的时间序列异常检测
支持自动学习业务基线，动态调整告警阈值
"""
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 = {}          # 存储每个指标的Prophet模型
        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):
        """从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分钟粒度
        }
        
        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):
        """训练Prophet时间序列预测模型"""
        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"✅ 训练完成: {metric_name} ({len(df)} 个数据点)")
        return model

    def detect_anomaly_prophet(self, metric_name, current_value, window_minutes=30):
        """
        使用Prophet进行单变量异常检测
        返回 (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()
        
        # 获取未来窗口期的预测
        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"]
        
        # 计算偏差分数（标准差级别）
        if upper > lower:
            deviation = (current_value - predicted) / max(upper - lower, 0.001)
        else:
            deviation = 0
        
        # 超过2个预测区间宽度视为异常
        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):
        """
        使用孤立森林进行多维异常检测
        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"]]
        
        # 收集历史数据用于训练
        detector["history"].append(features)
        if len(detector["history"]) > 100:
            detector["history"] = detector["history"][-100:]
            
            # 重新训练
            X = np.array(detector["history"])
            X_scaled = detector["scaler"].fit_transform(X)
            detector["model"].fit(X_scaled)
        
        # 检测当前样本
        X_current = np.array(features).reshape(1, -1)
        X_scaled = detector["scaler"].transform(X_current)
        prediction = detector["model"].predict(X_scaled)[0]
        score = detector["model"].score_samples(X_scaled)[0]
        
        is_anomaly = prediction == -1
        return is_anomaly, float(score)

    def compress_alerts(self, raw_alerts):
        """
        告警压缩 - 将相关告警聚合成工单
        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):
        """生成告警摘要"""
        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}] 检测到 {count} 条相关告警"
        if "critical" in severities:
            summary += " ⚠️ 包含严重级别告警"
        summary += f"\n涉及指标: {', '.join(set(metrics)[:5])}"
        return summary

    def get_ai_recommendation(self, ticket):
        """
        基于知识库给出AI修复建议
        在实际部署中，这里可以调用本地LLM（如Ollama）
        """
        recommendations = {
            "cpu": [
                "检查是否有进程出现CPU占用异常：`top -bn1 | head -20`",
                "考虑启用自动扩缩容（HPA）",
                "检查是否为新版本引入的性能回归",
            ],
            "memory": [
                "检查内存泄漏：`pmap -x $(pgrep python3) | tail -5`",
                "考虑增加swap空间作为缓冲",
                "检查是否有缓存未正确清理",
            ],
            "disk": [
                "清理旧日志：`journalctl --vacuum-time=3d`",
                "检查大文件：`du -sh /* | sort -rh | head -10`",
                "考虑迁移日志到远程存储",
            ],
            "network": [
                "检查是否有DDoS攻击迹象",
                "查看网络连接数：`ss -s`",
                "确认带宽是否达到上限",
            ],
        }
        
        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 = [
                "查看服务日志获取更多信息",
                "检查最近的部署变更记录",
                "联系相关服务负责人",
            ]
        
        return suggestions[:3]  # 最多返回3条建议


async def main():
    alerter = AIAlerter()
    
    # 示例：加载CPU指标的历史数据并训练模型
    cpu_df = alerter.fetch_prometheus_data(
        'node_cpu_seconds_total{mode="idle"}'
    )
    alerter.train_prophet_model("cpu_idle", cpu_df)
    
    # 定期检测异常
    while True:
        # 获取当前CPU值
        current_df = alerter.fetch_prometheus_data(
            'avg(node_cpu_seconds_total{mode="idle"})', 
            start_hours=1
        )
        if not current_df.empty:
            current_val = current_df["y"].iloc[-1]
            is_anomaly, deviation, expected = alerter.detect_anomaly_prophet(
                "cpu_idle", current_val
            )
            
            if is_anomaly:
                logger.warning(
                    f"🚨 CPU异常! 当前值={current_val:.2f}, "
                    f"预期范围={expected}, 偏差={deviation:.2f}σ"
                )
        
        await asyncio.sleep(300)  # 每5分钟检测一次


if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

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

# 拉取轻量级模型（适合VPS）
ollama pull llama3.2:3b

#!/usr/bin/env python3
"""
LLM驱动的根因分析服务
通过Ollama本地运行，无需外部API调用
"""
import json
import subprocess
from datetime import datetime

OLLAMA_URL = "http://localhost:11434"

def query_ollama(prompt, model="llama3.2:3b"):
    """通过API调用本地Ollama模型"""
    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):
    """
    分析告警工单的根因
    
    Args:
        alert_ticket: 告警工单（经过压缩后的）
        metrics_history: 相关指标的历史数据
    
    Returns:
        根因分析报告
    """
    prompt = f"""你是一个资深SRE工程师。请根据以下告警信息和指标数据，分析根因并给出修复建议。

【告警工单】
服务: {alert_ticket['service']}
严重程度: {alert_ticket['severity']}
告警数量: {alert_ticket['alert_count']}
摘要: {alert_ticket['summary']}

【相关指标数据】
{json.dumps(metrics_history, indent=2, ensure_ascii=False)}

【系统信息】
最近变更: 过去24小时内有1次部署（版本v2.3.1）
运行时长: 服务已运行14天

请按照以下格式回复：
1. **根因判断**: 最可能的根本原因
2. **置信度**: 高/中/低
3. **证据链**: 支持该判断的关键指标
4. **修复建议**: 具体可执行的步骤
5. **预防措施**: 如何避免类似问题再次发生
"""
    
    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": "请手动检查服务日志和相关指标",
        }

#!/usr/bin/env python3
"""
告警通知与自动修复模块
支持多种通知渠道和自动修复动作
"""
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):
        """分发告警到指定渠道"""
        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):
        """发送Slack通知"""
        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"服务: {payload['service']}\n"
                            f"告警数量: {payload['alert_count']}\n"
                            f"摘要: {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"💡 *建议操作:*\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):
        """发送钉钉通知"""
        import requests
        webhook_url = "YOUR_DINGTALK_WEBHOOK_URL"
        
        msg = {
            "msgtype": "markdown",
            "markdown": {
                "title": f"⚠️ {payload['severity'].upper()} 告警",
                "text": f"### {payload['severity'].upper()} 告警\n\n"
                        f"> 服务: {payload['service']}\n"
                        f"> 告警数量: {payload['alert_count']}\n"
                        f"> 摘要: {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):
        """执行自动修复"""
        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):
        """自动重启服务"""
        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):
        """自动扩容"""
        return {"action": "scale", "note": "需要根据实际环境配置扩缩容逻辑"}

    def _auto_rollback_deployment(self, ticket):
        """自动回滚部署"""
        return {"action": "rollback", "note": "需要集成CI/CD管道"}

# 投票机制：至少2个模型认为异常才触发告警
def ensemble_detect(metric_name, current_value, features):
    votes = 0
    
    # Prophet 异常检测
    is_anomaly_p, _, _ = alerter.detect_anomaly_prophet(metric_name, current_value)
    if is_anomaly_p:
        votes += 1
    
    # 孤立森林多维检测
    is_anomaly_if, _ = alerter.detect_multivariate_anomaly(metric_name, features)
    if is_anomaly_if:
        votes += 1
    
    # 统计方法（3-sigma）
    if abs(current_value - baseline_mean) > 3 * baseline_std:
        votes += 1
    
    return votes >= 2  # 多数投票

def adaptive_threshold(metric_name, current_value):
    """根据时间段和业务周期自适应调整阈值"""
    hour = datetime.now().hour
    day_of_week = datetime.now().weekday()
    
    # 加载该时段的历史基线
    baseline = load_baseline(metric_name, hour, day_of_week)
    
    # 根据近期变异系数动态调整
    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"]