# Ubuntu/Debian
sudo apt update
sudo apt install -y sysbench fio iperf3 lmbench sysstat

# 验证安装
sysbench --version
fio --version
iperf3 --version

#!/bin/bash
# /opt/benchmarks/run-all.sh

LOG_DIR="/var/log/benchmarks"
mkdir -p "$LOG_DIR"
TIMESTAMP=$(date +%Y%m%d_%H%M%S)
HOSTNAME=$(hostname)

echo "=== Benchmark Run: $TIMESTAMP on $HOSTNAME ==="

# 1. CPU 压测（sysbench）
echo "--- CPU Benchmark ---"
sysbench cpu --threads=4 --time=30 run \
  | tee "$LOG_DIR/cpu_${TIMESTAMP}.log"

# 2. 内存压测
echo "--- Memory Benchmark ---"
sysbench memory --threads=4 --memory-block-size=1M \
  --memory-total-size=1G run \
  | tee "$LOG_DIR/memory_${TIMESTAMP}.log"

# 3. 磁盘顺序读写
echo "--- Disk Sequential I/O ---"
fio --name=seq_read --filename=/tmp/bench_seq \
  --size=512M --bs=1M --rw=read \
  --direct=1 --numjobs=1 --time_based \
  --runtime=30 --group_reporting \
  --output-format=json \
  | tee "$LOG_DIR/disk_seq_${TIMESTAMP}.log"

# 4. 磁盘随机读写
echo "--- Disk Random I/O ---"
fio --name=rand_rw --filename=/tmp/bench_rand \
  --size=256M --bs=4K --rw=randrw \
  --direct=1 --numjobs=4 --time_based \
  --runtime=30 --group_reporting \
  --output-format=json \
  | tee "$LOG_DIR/disk_rand_${TIMESTAMP}.log"

# 5. 网络带宽测试
echo "--- Network Benchmark ---"
iperf3 -c benchmark.server.com -t 30 \
  --json > "$LOG_DIR/network_${TIMESTAMP}.log" 2>&1

echo "=== Benchmark Complete: $TIMESTAMP ==="

#!/usr/bin/env python3
"""Parse benchmark results and store in SQLite."""

import sqlite3
import json
import re
import subprocess
from datetime import datetime
from pathlib import Path

DB_PATH = "/var/lib/benchmarks/benchmarks.db"

def init_db():
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    c.execute('''
        CREATE TABLE IF NOT EXISTS benchmarks (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            hostname TEXT,
            timestamp DATETIME,
            test_type TEXT,
            metric_name TEXT,
            metric_value REAL,
            unit TEXT,
            raw_log TEXT
        )
    ''')
    c.execute('CREATE INDEX IF NOT EXISTS idx_host_time ON benchmarks(hostname, timestamp)')
    conn.commit()
    return conn

def parse_sysbench_cpu(log_path):
    """Parse sysbench CPU results."""
    results = []
    with open(log_path) as f:
        content = f.read()
    
    # Extract events per second
    match = re.search(r'events per second:\s+(\d+)', content)
    if match:
        results.append(("cpu_events_per_sec", float(match.group(1)), "ops/s"))
    
    # Extract total number of requests
    match = re.search(r'Total number of events:\s+(\d+)', content)
    if match:
        results.append(("cpu_total_events", int(match.group(1)), "count"))
    
    # Extract latency percentiles
    for p in [50, 95, 99]:
        match = re.search(r'pt\((\d+)\).\s+[\d.]+\s+([\d.]+)', content)
        if match and int(match.group(1)) == p:
            results.append((f"cpu_lat_p{p}", float(match.group(2)), "ms"))
    
    return results

def parse_fio_json(log_path):
    """Parse fio JSON output."""
    results = []
    with open(log_path) as f:
        data = json.load(f)
    
    job = data.get("jobs", [{}])[0]
    read = job.get("read", {})
    write = job.get("write", {})
    
    results.append(("disk_read_iops", read.get("iops", {}).get("mean", 0), "IOPS"))
    results.append(("disk_read_bw_mbps", read.get("bw_mean", 0) / 1024, "MB/s"))
    results.append(("disk_write_iops", write.get("iops", {}).get("mean", 0), "IOPS"))
    results.append(("disk_write_bw_mbps", write.get("bw_mean", 0) / 1024, "MB/s"))
    
    # Latency
    lat = read.get("lat_ns", {})
    results.append(("disk_read_lat_p99", lat.get("99th", 0) / 1e6, "us"))
    
    return results

def parse_iperf_json(log_path):
    """Parse iperf3 JSON output."""
    results = []
    with open(log_path) as f:
        data = json.load(f)
    
    # Summarized section
    for stream in data.get("end", {}).get("sum_sent", {}).get("streams", []):
        results.append(("net_bw_mbps", stream.get("sum_sent", {}).get("bits_per_second", 0) / 1e6, "Mbps"))
        results.append(("net_retrans", stream.get("sum_sent", {}).get("retransmits", 0), "count"))
    
    return results

def store_results(conn, test_type, parsed_results, raw_log_path):
    hostname = subprocess.check_output(["hostname"]).decode().strip()
    timestamp = datetime.utcnow().isoformat()
    
    c = conn.cursor()
    for metric_name, metric_value, unit in parsed_results:
        with open(raw_log_path) as f:
            raw_log = f.read()[:50000]  # truncate to save space
        c.execute('''
            INSERT INTO benchmarks 
            (hostname, timestamp, test_type, metric_name, metric_value, unit, raw_log)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        ''', (hostname, timestamp, test_type, metric_name, metric_value, unit, raw_log))
    conn.commit()

def main():
    conn = init_db()
    log_dir = Path("/var/log/benchmarks")
    
    for log_file in sorted(log_dir.glob("*.log")):
        name = log_file.stem
        if name.endswith(".log"):
            continue
        
        test_type = "cpu" if "cpu" in name else \
                    "memory" if "memory" in name else \
                    "disk_seq" if "disk_seq" in name else \
                    "disk_rand" if "disk_rand" in name else \
                    "network" if "network" in name else "unknown"
        
        if test_type in ("cpu", "memory"):
            results = parse_sysbench_cpu(log_file)
        elif test_type.startswith("disk"):
            results = parse_fio_json(log_file)
        elif test_type == "network":
            results = parse_iperf_json(log_file)
        else:
            continue
        
        if results:
            store_results(conn, test_type, results, log_file)
            print(f"Stored {len(results)} metrics from {name}")

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""AI-driven baseline engine for benchmark anomaly detection."""

import sqlite3
import numpy as np
from datetime import datetime, timedelta
from collections import defaultdict

class BenchmarkBaseline:
    def __init__(self, db_path):
        self.db_path = db_path
        self.lookback_days = 14  # 使用近14天数据建立基准
        self.sigma_threshold = 2.5  # 超过2.5个标准差视为异常
    
    def get_historical_data(self, metric_name, days=14):
        """获取指定指标的历史数据."""
        conn = sqlite3.connect(self.db_path)
        c = conn.cursor()
        
        since = (datetime.utcnow() - timedelta(days=days)).isoformat()
        c.execute('''
            SELECT timestamp, metric_value 
            FROM benchmarks 
            WHERE metric_name = ? AND timestamp >= ?
            ORDER BY timestamp
        ''', (metric_name, since))
        
        conn.close()
        return np.array([row[1] for row in c.fetchall()])
    
    def compute_baseline(self, data):
        """计算基准线: 均值 ± 标准差."""
        if len(data) < 3:
            return None, None, None
        
        mean = np.mean(data)
        std = np.std(data)
        
        # 排除异常值后重新计算（迭代）
        if std > 0:
            clean_data = data[np.abs(data - mean) < 3 * std]
            if len(clean_data) > 3:
                mean = np.mean(clean_data)
                std = np.std(clean_data)
        
        return mean, std, len(data)
    
    def detect_anomaly(self, metric_name, current_value):
        """检测当前值是否为异常."""
        historical = self.get_historical_data(metric_name)
        
        if len(historical) < 3:
            return {
                "status": "insufficient_data",
                "message": f"只有 {len(historical)} 个历史数据点，无法建立基准"
            }
        
        mean, std, count = self.compute_baseline(historical)
        
        if std == 0:
            return {
                "status": "stable",
                "baseline_mean": float(mean),
                "baseline_std": 0,
                "current_value": float(current_value),
                "z_score": 0,
                "deviation_pct": 0
            }
        
        z_score = (current_value - mean) / std
        
        # 判断是性能提升还是退化
        if metric_name in ("disk_read_bw_mbps", "disk_write_bw_mbps", 
                          "net_bw_mbps", "cpu_events_per_sec"):
            # 这些指标越高越好
            if z_score < -self.sigma_threshold:
                status = "degraded"
            elif z_score > self.sigma_threshold:
                status = "improved"
            else:
                status = "normal"
        else:
            # 延迟类指标越低越好
            if z_score > self.sigma_threshold:
                status = "degraded"
            elif z_score < -self.sigma_threshold:
                status = "improved"
            else:
                status = "normal"
        
        deviation_pct = abs(z_score) / self.sigma_threshold * 100
        
        return {
            "status": status,
            "baseline_mean": float(mean),
            "baseline_std": float(std),
            "current_value": float(current_value),
            "z_score": float(z_score),
            "deviation_pct": float(deviation_pct),
            "sample_count": int(count)
        }
    
    def detect_trend(self, metric_name, days=7):
        """检测性能趋势: 持续改善、退化还是稳定."""
        historical = self.get_historical_data(metric_name, days)
        
        if len(historical) < 5:
            return "insufficient_data"
        
        # 简单线性回归
        x = np.arange(len(historical))
        slope = np.polyfit(x, historical, 1)[0]
        
        # 归一化斜率
        mean = np.mean(historical)
        if mean == 0:
            return "stable"
        
        normalized_slope = slope / abs(mean)
        
        # 阈值：日变化超过 2% 视为有趋势
        if normalized_slope > 0.02:
            return "improving"
        elif normalized_slope < -0.02:
            return "degrading"
        else:
            return "stable"
    
    def generate_report(self):
        """生成完整性能报告."""
        # 获取所有监控的指标
        conn = sqlite3.connect(self.db_path)
        c = conn.cursor()
        c.execute('SELECT DISTINCT metric_name FROM benchmarks')
        metrics = [row[0] for row in c.fetchall()]
        conn.close()
        
        # 获取最新值
        latest_values = {}
        conn = sqlite3.connect(self.db_path)
        c = conn.cursor()
        for metric in metrics:
            c.execute('''
                SELECT metric_value FROM benchmarks 
                WHERE metric_name = ? 
                ORDER BY timestamp DESC LIMIT 1
            ''', (metric,))
            row = c.fetchone()
            if row:
                latest_values[metric] = row[0]
        conn.close()
        
        report = []
        anomalies_found = 0
        
        for metric, current_value in latest_values.items():
            result = self.detect_anomaly(metric, current_value)
            trend = self.detect_trend(metric)
            
            report.append({
                "metric": metric,
                **result,
                "trend": trend
            })
            
            if result.get("status") == "degraded":
                anomalies_found += 1
        
        return {
            "generated_at": datetime.utcnow().isoformat(),
            "metrics_analyzed": len(report),
            "anomalies_found": anomalies_found,
            "details": report
        }

if __name__ == "__main__":
    engine = BenchmarkBaseline(DB_PATH)
    report = engine.generate_report()
    print(f"基准线分析报告生成完成")
    print(f"分析指标数: {report['metrics_analyzed']}")
    print(f"发现异常数: {report['anomalies_found']}")
    for detail in report["details"]:
        status_icon = "✅" if detail["status"] == "normal" else \
                      "⚠️" if detail["status"] == "degraded" else "🔥"
        trend_icon = "📈" if detail["trend"] == "improving" else \
                     "📉" if detail["trend"] == "degrading" else "➡️"
        print(f"  {status_icon} {detail['metric']}: "
              f"当前={detail['current_value']:.2f}, "
              f"基准={detail['baseline_mean']:.2f}, "
              f"Z值={detail['z_score']:.2f}, "
              f"趋势={trend_icon}{detail['trend']}")

# 添加到 crontab
# 每天凌晨 2 点执行完整压测
0 2 * * * /opt/benchmarks/run-all.sh

# 压测完成后解析存储
35 2 * * * /usr/bin/python3 /opt/benchmarks/parse_results.py

# 每天凌晨 3 点生成基准线报告
0 3 * * * /usr/bin/python3 /opt/benchmarks/baseline_engine.py --report

# 每周日生成周报
0 4 * * 0 /usr/bin/python3 /opt/benchmarks/weekly_report.py

# 在 baseline_engine.py 中添加告警逻辑
import requests

def send_alert(metric, result):
    """发送告警通知."""
    if result["status"] != "degraded":
        return
    
    severity = "high" if result["deviation_pct"] > 150 else "medium"
    
    # 支持多种通知渠道
    message = f"⚠️ VPS 性能异常告警\n\n"
    message += f"指标: {metric}\n"
    message += f"当前值: {result['current_value']:.2f}\n"
    message += f"基准均值: {result['baseline_mean']:.2f}\n"
    message += f"偏离度: {result['deviation_pct']:.1f}%\n"
    message += f"Z 分数: {result['z_score']:.2f}\n"
    message += f"服务器: {HOSTNAME}\n"
    message += f"时间: {datetime.utcnow().isoformat()}"
    
    # 企业微信
    # requests.post(WEBHOOK_URL, json={"msgtype": "text", "text": {"content": message}})
    
    # Telegram
    # requests.post(TELEGRAM_API, json={"chat_id": CHAT_ID, "text": message})
    
    print(f"[ALERT] {severity}: {metric} - {result['current_value']:.2f}")

#!/usr/bin/env python3
"""Generate HTML performance dashboard."""

import sqlite3
import json
from datetime import datetime, timedelta
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
import numpy as np

def generate_dashboard():
    conn = sqlite3.connect(DB_PATH)
    
    # 获取最近 30 天的数据
    since = (datetime.utcnow() - timedelta(days=30)).isoformat()
    
    # 创建图表
    metrics = ["cpu_events_per_sec", "disk_read_bw_mbps", 
               "disk_write_bw_mbps", "net_bw_mbps"]
    
    fig, axes = plt.subplots(2, 2, figsize=(14, 10))
    axes = axes.flatten()
    
    colors = ['#0f172a', '#1e293b', '#334155', '#475569']
    
    for idx, metric in enumerate(metrics):
        ax = axes[idx]
        
        c = conn.cursor()
        c.execute('''
            SELECT timestamp, metric_value FROM benchmarks 
            WHERE metric_name = ? AND timestamp >= ?
            ORDER BY timestamp
        ''', (metric, since))
        
        data = c.fetchall()
        if not data:
            continue
        
        timestamps = [row[0] for row in data]
        values = [row[1] for row in data]
        
        # 绘制数据点
        ax.scatter(timestamps, values, s=20, alpha=0.6, c='#6366f1')
        
        # 计算并绘制基准线
        if len(values) > 3:
            mean = np.mean(values)
            std = np.std(values)
            ax.axhline(y=mean, color='#ef4444', linestyle='--', 
                      label=f'Mean: {mean:.1f}')
            ax.axhline(y=mean + 2.5*std, color='#f59e0b', 
                      linestyle=':', alpha=0.7, label='Upper Limit')
            ax.axhline(y=mean - 2.5*std, color='#f59e0b', 
                      linestyle=':', alpha=0.7)
        
        ax.set_title(metric, fontsize=12, color='white')
        ax.set_ylabel('Value', color='white')
        ax.tick_params(colors='white')
        ax.grid(True, alpha=0.3)
        
        # 背景色
        ax.set_facecolor('#0f172a')
    
    plt.tight_layout()
    plt.savefig('/var/www/benchmark-dashboard/chart.png', 
                dpi=150, facecolor='#0f172a')
    
    conn.close()
    print("Dashboard chart generated.")

if __name__ == "__main__":
    generate_dashboard()

def compare_servers(metric_name):
    """对比多台 VPS 的同一指标."""
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    
    # 获取所有服务器的最新值
    c.execute('''
        SELECT hostname, metric_value, timestamp 
        FROM benchmarks 
        WHERE metric_name = ? 
        ORDER BY timestamp DESC
    ''', (metric_name,))
    
    # 按服务器分组
    servers = defaultdict(list)
    for hostname, value, ts in c.fetchall():
        servers[hostname].append((value, ts))
    
    conn.close()
    
    # 计算每台服务器的基准并对比
    results = []
    for hostname, data_points in servers.items():
        values = [v for v, t in sorted(data_points, key=lambda x: x[1])]
        if len(values) > 3:
            mean = np.mean(values)
            std = np.std(values)
            latest = values[-1]
            z_score = (latest - mean) / std if std > 0 else 0
            results.append({
                "hostname": hostname,
                "baseline_mean": mean,
                "current": latest,
                "z_score": z_score,
                "status": "ok" if abs(z_score) < 2.5 else "anomaly"
            })
    
    return sorted(results, key=lambda x: abs(x["z_score"]), reverse=True)