# promtail-config.yml
server:
  http_listen_port: 9080
  grpc_listen_port: 0

positions:
  filename: /tmp/positions.yaml

clients:
  - url: http://loki:3100/loki/api/v1/push

scrape_configs:
  # System logs
  - job_name: system
    static_configs:
      - targets:
          - localhost
        labels:
          job: varlogs
          __path__: /var/log/*.log
          service: system

  # Nginx logs
  - job_name: nginx
    static_configs:
      - targets:
          - localhost
        labels:
          job: nginx
          __path__: /var/log/nginx/*.log
          service: nginx

  # Application logs
  - job_name: application
    static_configs:
      - targets:
          - localhost
        labels:
          job: application
          __path__: /app/logs/*.log
          service: application

# docker-compose.yml - Loki + Promtail
services:
  loki:
    image: grafana/loki:3.0.0
    ports:
      - "3100:3100"
    volumes:
      - loki-data:/loki
    command: -config.file=/etc/loki/local-config.yaml

  promtail:
    image: grafana/promtail:3.0.0
    volumes:
      - /var/log:/var/log
      - /run/log:/run/log
      - ./promtail-config.yml:/etc/promtail/config.yml
    command: -config.file=/etc/promtail/config.yml

  # Collect Docker container logs
  dockerd-logs:
    image: grafana/promtail:3.0.0
    volumes:
      - /var/lib/docker/containers:/var/lib/docker/containers:ro
      - /var/run/docker.sock:/var/run/docker.sock
      - ./promtail-docker.yml:/etc/promtail/config.yml
    command: -config.file=/etc/promtail/config.yml

import json
from datetime import datetime, timedelta
from typing import List, Dict, Any

class LogAggregator:
    """Log aggregator: correlates related logs by time window and service relationships"""
    
    def __init__(self, window_minutes=10):
        self.window = timedelta(minutes=window_minutes)
    
    def aggregate(self, logs: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """Aggregate logs by time window"""
        if not logs:
            return []
        
        # Sort by time
        sorted_logs = sorted(logs, key=lambda x: x.get('timestamp', ''))
        
        # Group by time window
        groups = []
        current_group = []
        group_start = None
        
        for log in sorted_logs:
            ts = datetime.fromisoformat(log.get('timestamp', '').replace('Z', '+00:00'))
            
            if group_start is None:
                group_start = ts
                current_group.append(log)
            elif ts - group_start <= self.window:
                current_group.append(log)
            else:
                if current_group:
                    groups.append(current_group)
                current_group = [log]
                group_start = ts
        
        if current_group:
            groups.append(current_group)
        
        return groups

ROOT_CAUSE_ANALYSIS_PROMPT = """
You are a professional SRE engineer responsible for analyzing cross-service log correlation failures on VPS.

## Task
Analyze the following log snippets, identify the root cause of the failure, and generate a structured root cause report.

## Input Logs
{logs}

## Service Dependency Relationships
{dependencies}

## Analysis Requirements
1. **Timeline Reconstruction**:梳理关键事件的时间顺序
2. **Causal Chain Analysis**: Identify the complete chain from "trigger → cascading impact → final symptom"
3. **Root Cause Localization**: Find the fundamental cause (not the symptom)
4. **Impact Assessment**: Evaluate the scope and severity of the failure's impact on business
5. **Remediation Recommendations**: Provide specific, actionable repair steps

## Output Format
Please output in JSON format:
{{
  "root_cause": "Concise description of root cause",
  "confidence": 0.0-1.0,
  "timeline": [
    {{"time": "timestamp", "event": "event description", "service": "service name"}}
  ],
  "causal_chain": [
    {{"step": 1, "cause": "cause", "effect": "effect"}}
  ],
  "impact": {{
    "services_affected": ["affected services"],
    "severity": "critical/high/medium/low",
    "description": "impact description"
  }},
  "recommendations": [
    {{"action": "remediation action", "priority": "immediate/short-term/long-term", "details": "detailed explanation"}}
  ],
  "similar_incidents": ["IDs of similar historical incidents"]
}}
"""

import ollama
from typing import List, Dict, Any

class CrossServiceAnalyzer:
    """Cross-service log correlation analyzer"""
    
    def __init__(self, model="qwen2.5:7b"):
        self.model = model
        self.service_deps = {
            "nginx": ["php-fpm", "node"],
            "php-fpm": ["mysql", "redis"],
            "node": ["mysql", "redis", "postgresql"],
            "api-gateway": ["auth-service", "user-service", "order-service"],
        }
    
    def analyze(self, log_groups: List[List[Dict]]) -> Dict[str, Any]:
        """Perform cross-service correlation analysis"""
        
        # Build context
        all_logs = []
        for group in log_groups:
            for log in group:
                all_logs.append({
                    "service": log.get("service", "unknown"),
                    "timestamp": log.get("timestamp", ""),
                    "level": log.get("level", "INFO"),
                    "message": log.get("message", ""),
                    "error_code": log.get("error_code", ""),
                })
        
        # Build dependency relationship text
        deps_text = json.dumps(self.service_deps, ensure_ascii=False, indent=2)
        
        # Call LLM for analysis
        prompt = ROOT_CAUSE_ANALYSIS_PROMPT.format(
            logs=json.dumps(all_logs[-100:], ensure_ascii=False, indent=2),
            dependencies=deps_text
        )
        
        response = ollama.chat(
            model=self.model,
            messages=[{"role": "user", "content": prompt}]
        )
        
        # Parse results
        try:
            result = json.loads(response["message"]["content"])
        except json.JSONDecodeError:
            # If JSON parsing fails, try to extract JSON fragment
            content = response["message"]["content"]
            start = content.find("{")
            end = content.rfind("}")
            if start != -1 and end != -1:
                result = json.loads(content[start:end+1])
            else:
                result = {"root_cause": content, "confidence": 0.5}
        
        return result

version: '3.8'

services:
  # Loki - Log storage
  loki:
    image: grafana/loki:3.0.0
    ports:
      - "3100:3100"
    volumes:
      - loki-data:/loki
      - ./loki-config.yml:/etc/loki/local-config.yaml
    command: -config.file=/etc/loki/local-config.yaml
    restart: unless-stopped

  # Promtail - Log collection
  promtail:
    image: grafana/promtail:3.0.0
    volumes:
      - /var/log:/var/log:ro
      - /run/log:/run/log:ro
      - ./promtail-config.yml:/etc/promtail/config.yml
    command: -config.file=/etc/promtail/config.yml
    restart: unless-stopped

  # LLM analysis service
  llm-analyzer:
    build: ./llm-analyzer
    volumes:
      - ./config:/app/config
      - ./knowledge-base:/app/knowledge-base
    environment:
      - OLLAMA_HOST=http://ollama:11434
      - LOKI_URL=http://loki:3100
      - ALERT_WEBHOOK=https://hooks.slack.com/your-webhook
    depends_on:
      - loki
      - ollama
    restart: unless-stopped

  # Ollama - Local LLM
  ollama:
    image: ollama/ollama:latest
    ports:
      - "11434:11434"
    volumes:
      - ollama-data:/root/.ollama
    restart: unless-stopped

  # Grafana - Visualization
  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    volumes:
      - grafana-data:/var/lib/grafana
      - ./grafana/dashboards:/etc/grafana/provisioning/dashboards
    environment:
      - GF_AUTH_ANONYMOUS_ENABLED=true
      - GF_SECURITY_ADMIN_USER=admin
      - GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_PASSWORD}
    depends_on:
      - loki
    restart: unless-stopped

volumes:
  loki-data:
  ollama-data:
  grafana-data:

# loki-config.yml
auth_enabled: false

server:
  http_listen_port: 3100
  grpc_listen_port: 9095

common:
  instance_addr: 127.0.0.1
  path_prefix: /tmp/loki
  storage:
    filesystem:
      chunks_directory: /tmp/loki/chunks
      rules_directory: /tmp/loki/rules
  replication_factor: 1
  ring:
    kvstore:
      store: inmemory

query_range:
  results_cache:
    cache:
      embedded_cache:
        enabled: true
        max_size_mb: 100

schema_config:
  configs:
    - from: 2024-01-01
      store: tsdb
      object_store: filesystem
      schema: v13
      index:
        prefix: index_
        period: 24h

ruler:
  alertmanager_url: http://localhost:9093

# Query logs from the last 30 minutes
curl -G 'http://localhost:3100/loki/api/v1/query_range' \
  --data-urlencode 'query={service=~"nginx|mysql|redis|php-fpm"}' \
  --data-urlencode 'start=now-30m' \
  --data-urlencode 'end=now' \
  --data-urlencode 'limit=1000'

# Trigger LLM analysis
curl -X POST 'http://localhost:8080/analyze' \
  -H 'Content-Type: application/json' \
  -d '{
    "log_query": "{service=~\"nginx|mysql|redis|php-fpm\"}",
    "time_range": "30m",
    "context": "User reports slow page loading"
  }'

{
  "root_cause": "MySQL slow queries caused connection pool exhaustion, cascading to Redis cache invalidation and Nginx 502 errors",
  "confidence": 0.92,
  "timeline": [
    {
      "time": "2026-09-12T10:15:23Z",
      "event": "MySQL started showing slow queries (>2s)",
      "service": "mysql"
    },
    {
      "time": "2026-09-12T10:17:45Z",
      "event": "MySQL connections reached limit (max_connections=151)",
      "service": "mysql"
    },
    {
      "time": "2026-09-12T10:18:02Z",
      "event": "PHP-FPM started waiting for MySQL connections, process count reached 20/20",
      "service": "php-fpm"
    },
    {
      "time": "2026-09-12T10:18:30Z",
      "event": "Redis cache hit rate dropped to 15%,大量 requests hitting MySQL directly",
      "service": "redis"
    },
    {
      "time": "2026-09-12T10:19:15Z",
      "event": "Nginx started returning 502 Bad Gateway",
      "service": "nginx"
    }
  ],
  "causal_chain": [
    {"step": 1, "cause": "A SQL query lacks an index", "effect": "MySQL slow queries increase"},
    {"step": 2, "cause": "Slow queries occupy connection pool", "effect": "New requests cannot obtain database connections"},
    {"step": 3, "cause": "PHP-FPM processes waiting for database connections", "effect": "Process pool exhausted"},
    {"step": 4, "cause": "Cache invalidation causes more database queries", "effect": "MySQL pressure further increases"},
    {"step": 5, "cause": "Backend service unresponsive", "effect": "Nginx returns 502"}
  ],
  "impact": {
    "services_affected": ["mysql", "php-fpm", "redis", "nginx"],
    "severity": "high",
    "description": "Website completely inaccessible, affecting all users"
  },
  "recommendations": [
    {
      "action": "Add missing database index",
      "priority": "immediate",
      "details": "Create index on orders.created_at: CREATE INDEX idx_orders_created ON orders(created_at)"
    },
    {
      "action": "Optimize slow query SQL",
      "priority": "immediate",
      "details": "Check EXPLAIN output, optimize query statements, avoid SELECT *"
    },
    {
      "action": "Adjust PHP-FPM and MySQL connection pool configuration",
      "priority": "short-term",
      "details": "Increase max_connections, adjust php-fpm pm.max_children"
    },
    {
      "action": "Configure Redis cache warm-up strategy",
      "priority": "long-term",
      "details": "Avoid cache stampede, use hot key preloading"
    }
  ],
  "similar_incidents": ["INC-2026-0815-001", "INC-2026-0702-003"]
}

class KnowledgeBase:
    """Incident knowledge base: stores historical analysis results, supports similar incident matching"""
    
    def __init__(self, db_path="knowledge.db"):
        self.db = sqlite3.connect(db_path)
        self._init_tables()
    
    def _init_tables(self):
        self.db.execute('''
            CREATE TABLE IF NOT EXISTS incidents (
                id INTEGER PRIMARY KEY,
                root_cause TEXT,
                timeline TEXT,
                causal_chain TEXT,
                recommendations TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        self.db.execute('''
            CREATE VIRTUAL TABLE IF NOT EXISTS incidents_fts 
            USING fts5(root_cause, content='incidents', content_rowid='id')
        ''')
    
    def store(self, analysis_result: Dict):
        """Store analysis result"""
        self.db.execute('''
            INSERT INTO incidents (root_cause, timeline, causal_chain, recommendations)
            VALUES (?, ?, ?, ?)
        ''', (
            analysis_result.get('root_cause', ''),
            json.dumps(analysis_result.get('timeline', [])),
            json.dumps(analysis_result.get('causal_chain', [])),
            json.dumps(analysis_result.get('recommendations', []))
        ))
        self.db.commit()
    
    def find_similar(self, root_cause: str, limit=3) -> List[Dict]:
        """Find similar historical incidents"""
        cursor = self.db.execute('''
            SELECT id, root_cause, created_at
            FROM incidents_fts
            WHERE incidents_fts MATCH ?
            ORDER BY rank
            LIMIT ?
        ''', (root_cause, limit))
        return [{'id': row[0], 'root_cause': row[1], 'date': row[2]} for row in cursor.fetchall()]

import requests

class AlertNotifier:
    """Alert notification"""
    
    def __init__(self, webhook_url):
        self.webhook_url = webhook_url
    
    def notify(self, analysis_result: Dict):
        """Send alert notification"""
        severity = analysis_result.get('impact', {}).get('severity', 'medium')
        root_cause = analysis_result.get('root_cause', 'Unknown')
        
        # Build message
        message = f"🚨 **VPS Incident Alert**\n\n"
        message += f"**Root Cause**: {root_cause}\n"
        message += f"**Severity**: {severity}\n"
        message += f"**Affected Services**: {', '.join(analysis_result.get('impact', {}).get('services_affected', []))}\n\n"
        
        # Add remediation suggestions
        recs = analysis_result.get('recommendations', [])
        if recs:
            message += "**Remediation Steps**:\n"
            for i, rec in enumerate(recs[:3], 1):
                message += f"{i}. {rec.get('action', '')}\n"
        
        # Send to Webhook
        payload = {
            "text": message,
            "attachments": [{
                "color": "danger" if severity in ["critical", "high"] else "warning",
                "fields": [
                    {"title": "Root Cause", "value": root_cause, "short": False},
                    {"title": "Severity", "value": severity, "short": True},
                    {"title": "Confidence", "value": f"{analysis_result.get('confidence', 0):.1%}", "short": True},
                ]
            }]
        }
        
        requests.post(self.webhook_url, json=payload, timeout=10)

import schedule
import time

class ScheduledAnalyzer:
    """Scheduled analysis tasks"""
    
    def __init__(self, analyzer, notifier, knowledge_base):
        self.analyzer = analyzer
        self.notifier = notifier
        self.kb = knowledge_base
    
    def run_analysis(self):
        """Execute analysis task"""
        # Query anomalous logs from the last 10 minutes
        logs = self.fetch_anomaly_logs(minutes=10)
        
        if not logs:
            return
        
        # Execute LLM analysis
        result = self.analyzer.analyze([logs])
        
        # Store to knowledge base
        self.kb.store(result)
        
        # Send alert (if severity is high)
        severity = result.get('impact', {}).get('severity', 'low')
        if severity in ['critical', 'high']:
            self.notifier.notify(result)
    
    def fetch_anomaly_logs(self, minutes=10) -> List[Dict]:
        """Fetch anomalous logs from Loki"""
        # Query ERROR and CRITICAL level logs
        query = '{level=~"ERROR|CRITICAL"}'
        # ... call Loki API
        pass

class SmartSampler:
    """Smart log sampler"""
    
    def should_sample(self, log: Dict) -> bool:
        """Decide whether to sample this log"""
        # Always retain error logs
        if log.get('level') in ['ERROR', 'CRITICAL']:
            return True
        
        # Keyword logs always retained
        keywords = ['timeout', 'connection refused', 'out of memory', 'killed']
        message = log.get('message', '').lower()
        if any(kw in message for kw in keywords):
            return True
        
        # Normal logs sampled at 1%
        import random
        return random.random() < 0.01

class IncrementalAnalyzer:
    """Incremental analyzer"""
    
    def __init__(self, checkpoint_file="checkpoint.json"):
        self.checkpoint = self._load_checkpoint()
    
    def _load_checkpoint(self):
        try:
            with open(self.checkpoint_file, 'r') as f:
                return json.load(f)
        except FileNotFoundError:
            return {"last_analysis": 0}
    
    def get_new_logs(self, since_timestamp):
        """Fetch new logs since specified timestamp"""
        # Call Loki API to get new logs
        pass
    
    def analyze_incremental(self):
        """Incremental analysis"""
        since = self.checkpoint.get('last_analysis', 0)
        new_logs = self.get_new_logs(since)
        
        if new_logs:
            result = self.analyzer.analyze([new_logs])
            self.checkpoint['last_analysis'] = int(time.time() * 1000)
            self._save_checkpoint()
            return result
        return None