# 安装 Logstash
sudo apt update
sudo apt install -y logstash

# 验证安装
/usr/share/logstash/bin/logstash --version

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

# 拉取适合的模型（7B 参数足够处理日志分析）
ollama pull llama3.2

# 或者用更小的模型（适合 2GB RAM 的 VPS）
ollama pull qwen2.5:3b

#!/usr/bin/env python3
"""
AI 日志分析器：从 Elasticsearch 读取日志，用 LLM 分析异常和根因
"""
import json
import subprocess
import sys
from datetime import datetime, timedelta
from elasticsearch import Elasticsearch

# Elasticsearch 连接
es = Elasticsearch(["http://localhost:9200"])

# LLM 分析提示词
ANALYSIS_PROMPT = """
你是一个专业的运维日志分析专家。请分析以下日志片段，提供：

1. **异常检测**：指出哪些日志条目是异常的，为什么异常
2. **根因分析**：从日志中推断最可能的故障根因
3. **影响评估**：判断影响的严重程度（P0-P3）
4. **修复建议**：给出具体的、可执行的修复步骤
5. **预防措施**：建议如何避免类似问题再次发生

日志数据：
{logs}

请用清晰的格式输出分析结果，使用中文回答。
"""

def get_recent_errors(hours=1):
    """获取最近 N 小时的错误日志"""
    since_time = (datetime.now() - timedelta(hours=hours)).isoformat()
    
    query = {
        "query": {
            "bool": {
                "must": [
                    {"range": {"@timestamp": {"gte": since_time}}}
                ],
                "should": [
                    {"match": {"error_level": "error"}},
                    {"match": {"error_level": "warning"}},
                    {"match_phrase": {"message": "exception"}},
                    {"match_phrase": {"message": "timeout"}},
                    {"match_phrase": {"message": "connection refused"}}
                ]
            }
        },
        "size": 500,
        "_source": ["@timestamp", "message", "error_level", "tags", "hostname"]
    }
    
    response = es.search(index="logstash-*", body=query)
    return [hit["_source"] for hit in response["hits"]["hits"]]

def analyze_with_llm(log_entries):
    """调用 LLM 分析日志"""
    # 格式化日志数据
    log_text = json.dumps(log_entries, ensure_ascii=False, indent=2)
    
    prompt = ANALYSIS_PROMPT.format(logs=log_text)
    
    # 调用 Ollama
    result = subprocess.run(
        ["ollama", "run", "llama3.2", prompt],
        capture_output=True, text=True, timeout=120
    )
    
    return result.stdout

def main():
    print(f"开始日志分析 [{datetime.now().isoformat()}]")
    
    # 获取错误日志
    errors = get_recent_errors(hours=1)
    
    if not errors:
        print("未发现异常日志")
        return
    
    print(f"发现 {len(errors)} 条异常日志，正在分析...")
    
    # 分批次分析（避免上下文过长）
    batch_size = 50
    for i in range(0, len(errors), batch_size):
        batch = errors[i:i + batch_size]
        analysis = analyze_with_llm(batch)
        print(f"\n--- 分析批次 {i // batch_size + 1} ---")
        print(analysis)
    
    print("\n日志分析完成")

if __name__ == "__main__":
    main()

# 每 15 分钟分析一次日志
*/15 * * * * /opt/log-analyzer/analyze_logs.py >> /var/log/log-analyzer.log 2>&1

# vector.yaml
data_dir: /var/lib/vector

sources:
  # 系统日志
  system_logs:
    type: syslog
    address: 0.0.0.0:5551
  
  # 应用日志
  app_logs:
    type: file
    include:
      - /opt/*/logs/*.log

transforms:
  # 异常检测（用 OPA 策略）
  anomaly_detection:
    type: remap
    inputs:
      - system_logs
      - app_logs
    source: |
      .severity = if contains(string!(.message), "error") {
        "error"
      } else if contains(string!(.message), "warn") {
        "warning"
      } else {
        "info"
      }
      
      # 计算日志异常分数
      .anomaly_score = if contains(string!(.message), "panic|fatal|critical") {
        10.0
      } else if contains(string!(.message), "error|exception") {
        7.0
      } else if contains(string!(.message), "timeout|refused") {
        5.0
      } else {
        0.0
      }

sinks:
  # 输出到 Elasticsearch
  elasticsearch:
    type: elasticsearch
    inputs:
      - anomaly_detection
    endpoints: ["http://localhost:9200"]
    
  # 输出到 Telegram 告警
  telegram_alert:
    type: telegram
    inputs:
      - anomaly_detection
    api_key: "YOUR_BOT_TOKEN"
    chat_id: "YOUR_CHAT_ID"
    threshold:
      type: less_than
      value: 1
    message:
      text: |
        ⚠️ 检测到异常日志！
        服务器: {{ hostname }}
        级别: {{ .severity }}
        异常分数: {{ .anomaly_score }}
        内容: {{ truncate(.message, 500) }}

# 安装 Vector
curl --proto '=https' --tlsv1.2 -sSf https://sh.vector.dev | sh

# 配置 Vector
sudo cp vector.yaml /etc/vector/vector.yaml

# 启动
sudo systemctl enable vector
sudo systemctl start vector

package log_anomaly

deny[msg] {
    input.severity == "error"
    input.anomaly_score >= 7.0
    msg := sprintf("严重错误 [%v]: %v", [input.hostname, truncate(input.message, 200)])
}

deny[msg] {
    # 同一错误短时间内重复出现（模式检测）
    count(input.recent_errors) > 50
    msg := sprintf("错误风暴：同一错误重复 %v 次", [count(input.recent_errors)])
}

# /etc/nginx/conf.d/json_format.conf
log_format json_combined escape=json
    '{'
        '"time":"$time_iso8601",'
        '"remote_addr":"$remote_addr",'
        '"request":"$request",'
        '"status":$status,'
        '"body_bytes_sent":$body_bytes_sent,'
        '"request_time":$request_time,'
        '"http_referrer":"$http_referer",'
        '"http_user_agent":"$http_user_agent",'
        '"upstream_response_time":"$upstream_response_time"'
    '}';

access_log /var/log/nginx/access.json.log json_combined;

#!/usr/bin/env python3
"""
Nginx 日志 AI 分析器
"""
import re
from collections import defaultdict, Counter
from datetime import datetime, timedelta

def parse_access_log(log_file, lines=1000):
    """解析 Nginx JSON 日志"""
    entries = []
    pattern = re.compile(r'^\{.*\}$')
    
    with open(log_file, 'r') as f:
        for line in reversed(list(f)):
            if pattern.match(line.strip()):
                try:
                    entries.append(json.loads(line))
                    if len(entries) >= lines:
                        break
                except json.JSONDecodeError:
                    continue
    
    return entries

def detect_anomalies(entries):
    """检测日志中的异常"""
    anomalies = []
    
    # 1. 检测高频 IP
    ip_counts = Counter(e.get('remote_addr', '') for e in entries)
    for ip, count in ip_counts.most_common(10):
        if count > 100:  # 1000 行中超过 100 次
            anomalies.append({
                "type": "high_frequency_ip",
                "detail": f"IP {ip} 在近期日志中出现 {count} 次",
                "severity": "high" if count > 500 else "medium"
            })
    
    # 2. 检测大量 4xx/5xx 错误
    error_by_path = defaultdict(int)
    for e in entries:
        if e.get('status', 200) >= 500:
            path = e.get('request', '').split()[1] if 'request' in e else ''
            error_by_path[path] += 1
    
    for path, count in error_by_path.items():
        if count > 10:
            anomalies.append({
                "type": "server_errors",
                "detail": f"端点 {path} 出现 {count} 次 5xx 错误",
                "severity": "critical"
            })
    
    # 3. 检测慢请求
    slow_requests = [
        e for e in entries 
        if e.get('request_time', '0') and float(e['request_time']) > 5
    ]
    if slow_requests:
        anomalies.append({
            "type": "slow_requests",
            "detail": f"发现 {len(slow_requests)} 个超过 5 秒的慢请求",
            "severity": "medium"
        })
    
    return anomalies

def generate_llm_report(anomalies, entries):
    """生成分析报告并调用 LLM"""
    report = f"## Nginx 日志分析报告\n\n"
    report += f"分析时间范围：最近 {len(entries)} 条请求\n\n"
    
    if anomalies:
        report += f"### 检测到的异常 ({len(anomalies)} 项)\n\n"
        for a in anomalies:
            report += f"- **{a['type']}** [{a['severity']}]: {a['detail']}\n"
    else:
        report += "✅ 未发现异常\n"
    
    # 调用 LLM 生成详细报告
    prompt = f"""
    你是一个运维安全专家。以下是 Nginx 访问日志的分析结果，请提供：
    1. 对每个异常项的详细风险评估
    2. 如果是攻击行为，给出立即执行的缓解措施
    3. 如果是配置问题，给出具体的修复方案
    4. 长期加固建议

    分析结果：
    {report}
    """
    
    result = subprocess.run(
        ["ollama", "run", "llama3.2", prompt],
        capture_output=True, text=True
    )
    return result.stdout

# vector.yaml - 实时告警配置
transforms:
  # 实时异常检测
  realtime_anomaly:
    type: route
    inputs:
      - app_logs
    route:
      error: '{{ contains(string!(.message), "error|fatal|panic") }}'
      warning: '{{ contains(string!(.message), "warn|timeout") }}'
      alert: '{{ .anomaly_score >= 7.0 }}'

sinks:
  # 告警路由到 Telegram
  telegram_error:
    type: telegram
    inputs:
      - realtime_anomaly.alert
    api_key: "${TELEGRAM_BOT_TOKEN}"
    chat_id: "${TELEGRAM_CHAT_ID}"
    message:
      text: |
        🚨 {{ format!("text", .) }}
        严重级别: {{ .severity }}
        消息: {{ truncate(.message, 300) }}

# prometheus/alerts/log_alerts.yml
groups:
  - name: log_anomaly_alerts
    rules:
      - alert: HighErrorRate
        expr: |
          rate(log_errors_total[5m]) > 10
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "错误率过高"
          description: "过去 5 分钟错误率超过 10 次/秒"
          
      - alert: LogVolumeSpike
        expr: |
          increase(log_lines_total[10m]) > 10000
        for: 1m
        labels:
          severity: warning
        annotations:
          summary: "日志量突增"
          description: "过去 10 分钟日志量增加了 10 倍"

import re

def sanitize_logs(log_entries):
    """日志脱敏"""
    patterns = [
        (r'(apikey|token|secret|password|authorization)\s*[:=]\s*["\']?(\S+)', r'\1=***REDACTED***'),
        (r'\b\d{16}\b', '***CARD_REDACTED***'),  # 银行卡号
        (r'\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b', '***EMAIL_REDACTED***'),  # 邮箱
    ]
    
    for entry in log_entries:
        message = entry.get('message', '')
        for pattern, replacement in patterns:
            message = re.sub(pattern, replacement, message, flags=re.IGNORECASE)
        entry['message'] = message
    
    return log_entries

# 监控指标
- log_analysis_requests_total        # 分析请求总数
- log_analysis_duration_seconds      # 分析耗时
- log_analysis_error_rate            # 分析失败率
- llm_api_latency_seconds            # LLM API 延迟
- llm_token_usage_total              # Token 消耗