# fluent-bit 采集配置示例
[INPUT]
    Name              tail
    Path              /var/log/nginx/*.log
    Tag               nginx.access
    Parser            nginx
    DB                /var/lib/fluent-bit/nginx.db
    Refresh_Interval  5

[INPUT]
    Name              tail
    Path              /var/log/syslog
    Tag               system.syslog
    DB                /var/lib/fluent-bit/syslog.db

[FILTER]
    Name                lua
    Match               system.*
    Script              /fluent-bit/scripts/classify.lua
    Call                classify_log_level

[OUTPUT]
    Name                http
    Match               *
    Host                log-processor
    Port                8080
    Format              json

import numpy as np
from sklearn.cluster import DBSCAN
from sentence_transformers import SentenceTransformer
import hashlib

class LogPatternClusterer:
    """基于 LLM embedding 的日志模式聚类器"""
    
    def __init__(self, model_name="text-embedding-3-small"):
        self.embedder = SentenceTransformer(model_name)
        self.cluster_threshold = 0.15  # 相似度阈值
        self.pattern_registry = {}  # pattern_hash -> pattern_info
    
    def extract_log_template(self, log_line: str) -> str:
        """从日志行中提取模板（去除动态字段）"""
        # 使用 LLM 提取日志模板
        # 例如: "2024-01-15 10:23:45 ERROR Connection to db-01 failed: timeout after 30s"
        # 模板: "Connection to {host} failed: timeout after {seconds}s"
        import re
        # 移除时间戳
        template = re.sub(r'\d{4}-\d{2}-\d{2}[\sT]\d{2}:\d{2}:\d{2}', '{timestamp}', log_line)
        # 移除 IP 地址
        template = re.sub(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}', '{ip}', template)
        # 移除数字
        template = re.sub(r'\b\d+\b', '{num}', template)
        # 移除 UUID
        template = re.sub(r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}', '{uuid}', template)
        return template
    
    def cluster_logs(self, log_batch: list[dict]) -> dict:
        """对日志批次进行聚类"""
        if not log_batch:
            return {}
        
        # 提取模板
        templates = [self.extract_log_template(log['message']) for log in log_batch]
        
        # 生成 embedding
        embeddings = self.embedder.encode(templates)
        
        # DBSCAN 聚类
        clusterer = DBSCAN(eps=self.cluster_threshold, min_samples=1, metric='cosine')
        labels = clusterer.fit_predict(embeddings)
        
        # 构建聚类结果
        clusters = {}
        for log, label in zip(log_batch, labels):
            cluster_key = f"cluster_{label}"
            if cluster_key not in clusters:
                clusters[cluster_key] = {
                    'pattern': templates[log_batch.index(log)],
                    'count': 0,
                    'samples': [],
                    'first_seen': log['timestamp'],
                    'last_seen': log['timestamp'],
                    'hosts': set(),
                    'error_rate': 0.0
                }
            clusters[cluster_key]['count'] += 1
            clusters[cluster_key]['samples'].append(log['message'])
            clusters[cluster_key]['last_seen'] = log['timestamp']
            clusters[cluster_key]['hosts'].add(log.get('host', 'unknown'))
            
            if 'error' in log['message'].lower() or 'fail' in log['message'].lower():
                clusters[cluster_key]['error_rate'] = 1.0
        
        return clusters

-- 日志向量存储表
CREATE TABLE log_patterns (
    id BIGSERIAL PRIMARY KEY,
    pattern_hash VARCHAR(64) UNIQUE NOT NULL,
    pattern_template TEXT NOT NULL,
    embedding vector(1536),  -- OpenAI text-embedding-3-small
    total_count BIGINT DEFAULT 0,
    error_count BIGINT DEFAULT 0,
    first_seen TIMESTAMP WITH TIME ZONE,
    last_seen TIMESTAMP WITH TIME ZONE,
    related_hosts TEXT[],
    severity_score DOUBLE PRECISION,
    trend_direction VARCHAR(10),  -- 'increasing', 'stable', 'decreasing'
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

-- 创建向量索引加速相似度检索
CREATE INDEX ON log_patterns USING ivfflat (embedding vector_cosine_ops)
    WITH (lists = 100);

-- 趋势预测表
CREATE TABLE log_trend_predictions (
    id BIGSERIAL PRIMARY KEY,
    pattern_hash VARCHAR(64) REFERENCES log_patterns(pattern_hash),
    predicted_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    predicted_count_1h DOUBLE PRECISION,
    predicted_count_24h DOUBLE PRECISION,
    confidence_score DOUBLE PRECISION,
    prediction_model VARCHAR(50),
    actual_1h_count BIGINT,
    actual_24h_count BIGINT
);

class HierarchicalLogClustering:
    """分层日志聚类：先粗粒度模板匹配，再细粒度语义聚类"""
    
    def __init__(self, llm_client, embedding_model):
        self.llm = llm_client
        self.embedder = embedding_model
        self.template_index = {}  # 模板 -> 聚类ID
        self.semantic_clusters = {}  # 语义簇 -> 日志列表
    
    def process_log_batch(self, logs: list[dict]) -> list[dict]:
        """处理日志批次，返回带聚类标签的日志"""
        # 第一层：模板匹配（快速过滤）
        template_clusters = self._template_match(logs)
        
        # 第二层：语义聚类（对未匹配日志进行 LLM 分析）
        unmatched = [log for log in logs if log.get('cluster_id') is None]
        semantic_clusters = self._semantic_cluster(unmatched)
        
        # 第三层：LLM 语义增强（对关键聚类进行 LLM 理解）
        enhanced_clusters = self._llm_enrich(template_clusters, semantic_clusters)
        
        return enhanced_clusters
    
    def _llm_enrich(self, template_clusters, semantic_clusters) -> dict:
        """使用 LLM 对聚类结果进行语义增强"""
        enrichment_prompt = """
分析以下日志聚类，识别它们的共同模式和潜在根因：

聚类列表:
{clusters}

请为每个聚类提供：
1. 人类可读的模式描述
2. 潜在根因分析
3. 建议的修复方向
4. 紧急程度评估（P0-P3）

以 JSON 格式返回分析结果。
""".format(clusters=json.dumps(template_clusters, ensure_ascii=False, indent=2))
        
        response = self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": enrichment_prompt}],
            response_format={"type": "json_object"}
        )
        return json.loads(response.choices[0].message.content)

from asyncio import Queue, create_task
import asyncio

class RealtimeLogPipeline:
    """实时日志处理管道"""
    
    def __init__(self, clusterer: LogPatternClusterer, 
                 detector: AnomalyDetector,
                 predictor: TrendPredictor):
        self.queue = Queue(maxsize=10000)
        self.clusterer = clusterer
        self.detector = detector
        self.predictor = predictor
        self.batch_size = 50
        self.processing_tasks = []
    
    async def start(self):
        """启动处理管道"""
        # 批量消费任务
        create_task(self._batch_processor())
        # 实时告警任务
        create_task(self._alert_processor())
        # 趋势预测任务
        create_task(self._prediction_processor())
    
    async def _batch_processor(self):
        """批量处理日志，执行聚类"""
        batch = []
        while True:
            log = await asyncio.wait_for(self.queue.get(), timeout=1.0)
            batch.append(log)
            
            if len(batch) >= self.batch_size:
                clusters = self.clusterer.cluster_logs(batch)
                await self._store_clusters(clusters)
                batch = []
            
            # 超时强制处理
            if len(batch) > 0:
                await asyncio.sleep(0.5)
                clusters = self.clusterer.cluster_logs(batch)
                await self._store_clusters(clusters)
                batch = []
    
    async def ingest(self, log: dict):
        """ ingest 日志到管道 """
        await self.queue.put(log)

import statsmodels.api as sm
from prophet import Prophet
import pandas as pd

class LogAnomalyDetector:
    """基于时间序列的日志异常检测"""
    
    def __init__(self, lookback_hours=24, confidence=0.95):
        self.lookback_hours = lookback_hours
        self.confidence = confidence
        self.baselines = {}  # pattern -> baseline stats
    
    def update_baseline(self, pattern: str, counts: list[tuple]):
        """更新日志模式的基线"""
        df = pd.DataFrame(counts, columns=['ds', 'y'])
        
        # 分解时间序列
        decomposition = sm.tsa.seasonal_decompose(
            df['y'], model='additive', period=24
        )
        
        self.baselines[pattern] = {
            'trend': decomposition.trend,
            'seasonal': decomposition.seasonal,
            'residual': decomposition.residual,
            'mean': df['y'].mean(),
            'std': df['y'].std(),
            'upper_bound': df['y'].mean() + 2 * df['y'].std(),
            'lower_bound': max(0, df['y'].mean() - 2 * df['y'].std())
        }
    
    def detect_anomaly(self, pattern: str, current_count: int) -> dict:
        """检测当前计数是否异常"""
        if pattern not in self.baselines:
            return {'is_anomaly': False, 'reason': 'no_baseline'}
        
        baseline = self.baselines[pattern]
        
        # 基于统计的异常检测
        z_score = (current_count - baseline['mean']) / baseline['std'] if baseline['std'] > 0 else 0
        is_anomaly = abs(z_score) > 2  # 2σ 阈值
        
        # 基于 Prophet 的预测异常检测
        prophet_anomaly = self._prophet_detect(pattern, current_count)
        
        return {
            'is_anomaly': is_anomaly or prophet_anomaly['is_anomaly'],
            'z_score': z_score,
            'expected_range': (baseline['lower_bound'], baseline['upper_bound']),
            'current_value': current_count,
            'prophet_anomaly': prophet_anomaly,
            'severity': self._calculate_severity(z_score, current_count, baseline['mean'])
        }
    
    def _prophet_detect(self, pattern: str, current_count: int) -> dict:
        """使用 Prophet 进行预测异常检测"""
        if pattern not in self.baselines:
            return {'is_anomaly': False}
        
        # 这里简化实现，实际应维护 Prophet 模型
        baseline = self.baselines[pattern]
        predicted_upper = baseline['upper_bound'] * 1.5  # 放宽预测区间
        
        return {
            'is_anomaly': current_count > predicted_upper,
            'predicted_upper': predicted_upper
        }
    
    def _calculate_severity(self, z_score: float, current: int, mean: int) -> str:
        """计算异常严重程度"""
        if abs(z_score) > 4:
            return 'critical'
        elif abs(z_score) > 3:
            return 'high'
        elif abs(z_score) > 2:
            return 'medium'
        return 'low'

class LogTrendPredictor:
    """日志趋势预测"""
    
    def __init__(self, forecast_horizon_hours=24):
        self.forecast_horizon = forecast_horizon_hours
        self.models = {}
    
    def fit(self, pattern: str, history: list[dict]):
        """训练预测模型"""
        df = pd.DataFrame(history)
        df['ds'] = pd.to_datetime(df['timestamp'])
        df['y'] = df['count']
        
        # 使用 Prophet 模型
        model = Prophet(
            yearly_seasonality=True,
            weekly_seasonality=True,
            daily_seasonality=True,
            changepoint_prior_scale=0.05
        )
        model.fit(df)
        self.models[pattern] = model
    
    def predict(self, pattern: str, steps: int = 96) -> dict:
        """预测未来趋势（每15分钟一个点，24小时）"""
        if pattern not in self.models:
            return {'forecast': [], 'confidence': []}
        
        model = self.models[pattern]
        future = model.make_future_dataframe(periods=steps, freq='15min')
        forecast = model.predict(future)
        
        # 提取未来预测
        recent_forecast = forecast.tail(steps)
        
        return {
            'forecast': recent_forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']].values.tolist(),
            'trend_direction': self._infer_trend(recent_forecast['yhat'].values),
            'peak_predicted': float(recent_forecast['yhat'].max()),
            'peak_time': str(recent_forecast.loc[recent_forecast['yhat'].idxmax(), 'ds'])
        }
    
    def _infer_trend(self, values: np.ndarray) -> str:
        """推断趋势方向"""
        if len(values) < 2:
            return 'stable'
        slope = np.polyfit(range(len(values)), values, 1)[0]
        if slope > 0.01 * abs(values.mean()):
            return 'increasing'
        elif slope < -0.01 * abs(values.mean()):
            return 'decreasing'
        return 'stable'

# docker-compose.yml
version: '3.8'

services:
  # 日志采集
  fluent-bit:
    image: fluent/fluent-bit:3.1
    volumes:
      - ./fluent-bit/conf:/fluent-bit/etc
      - /var/log:/var/log:ro
      - /var/lib/docker/containers:/var/lib/docker/containers:ro
    networks:
      - lognet

  # AI 日志分析引擎
  log-processor:
    build: ./log-processor
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - EMBEDDING_MODEL=text-embedding-3-small
      - DATABASE_URL=postgresql://loguser:logpass@postgres:5432/logdb
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - postgres
      - redis
    networks:
      - lognet
    deploy:
      resources:
        limits:
          memory: 2G
          cpus: '1.0'

  # 向量数据库
  postgres:
    image: pgvector/pgvector:pg16
    environment:
      POSTGRES_DB: logdb
      POSTGRES_USER: loguser
      POSTGRES_PASSWORD: logpass
    volumes:
      - postgres_data:/var/lib/postgresql/data
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql
    networks:
      - lognet

  # 缓存层
  redis:
    image: redis:7-alpine
    command: redis-server --appendonly yes
    volumes:
      - redis_data:/data
    networks:
      - lognet

  # 可视化
  grafana:
    image: grafana/grafana:11.3.0
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_PASSWORD}
      - GF_INSTALL_PLUGINS=grafana-clock-panel
    volumes:
      - grafana_data:/var/lib/grafana
      - ./grafana/dashboards:/etc/grafana/provisioning/dashboards
      - ./grafana/datasources:/etc/grafana/provisioning/datasources
    networks:
      - lognet

  # 告警管理
  alertmanager:
    image: prom/alertmanager:v0.27.0
    volumes:
      - ./alertmanager:/etc/alertmanager
    networks:
      - lognet

volumes:
  postgres_data:
  redis_data:
  grafana_data:

networks:
  lognet:
    driver: bridge

# alert-rules.yaml
groups:
  - name: log_anomaly_alerts
    rules:
      - alert: LogPatternAnomaly
        expr: log_anomaly_score > 3
        for: 5m
        labels:
          severity: '{{ $value | lessThan 4 | ternary "warning" | ternary "critical" }}'
          category: 'log-anomaly'
        annotations:
          summary: "日志模式异常: {{ $labels.pattern }}"
          description: "模式 {{ $labels.pattern }} 在当前时间段出现 {{ $value }}σ 偏离，建议立即查看"
          runbook_url: "https://wiki.example.com/runbooks/log-anomaly"

      - alert: LogVolumeSpike
        expr: log_volume_prediction_deviation > 2
        for: 10m
        labels:
          severity: 'warning'
          category: 'log-volume'
        annotations:
          summary: "日志量突增预测"
          description: "日志量预测偏差超过 2σ，可能存在潜在问题"

      - alert: NewLogPatternDetected
        expr: new_pattern_count_1h > 5
        for: 15m
        labels:
          severity: 'info'
          category: 'new-pattern'
        annotations:
          summary: "检测到新的日志模式"
          description: "过去1小时内检测到 {{ $value }} 个新的日志模式，可能需要分析"

  - name: log_trend_alerts
    rules:
      - alert: ErrorRateIncreasing
        expr: error_rate_trend == "increasing" and error_rate_7d_avg > 0.05
        for: 30m
        labels:
          severity: 'warning'
          category: 'error-trend'
        annotations:
          summary: "错误率持续上升趋势"
          description: "错误率在过去7天呈上升趋势，当前7日平均错误率为 {{ $value | humanizePercentage }}"