# Install Logstash
sudo apt update
sudo apt install -y logstash

# Verify installation
/usr/share/logstash/bin/logstash --version

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

# Pull a suitable model (7B params is enough for log analysis)
ollama pull llama3.2

# Or use a smaller model (suitable for 2GB RAM VPS)
ollama pull qwen2.5:3b

#!/usr/bin/env python3
"""
AI Log Analyzer: Read logs from Elasticsearch, analyze anomalies and root causes with LLM
"""
import json
import subprocess
import sys
from datetime import datetime, timedelta
from elasticsearch import Elasticsearch

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

# LLM analysis prompt
ANALYSIS_PROMPT = """
You are a professional operations log analysis expert. Please analyze the following log snippets and provide:

1. **Anomaly Detection**: Identify which log entries are anomalous and why
2. **Root Cause Analysis**: Infer the most likely failure root cause from the logs
3. **Impact Assessment**: Judge the severity level (P0-P3)
4. **Fix Recommendations**: Provide concrete, executable repair steps
5. **Prevention Measures**: Suggest how to prevent similar issues in the future

Log data:
{logs}

Please output the analysis results in a clear format, in English.
"""

def get_recent_errors(hours=1):
    """Get error logs from the last N hours"""
    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):
    """Call LLM to analyze logs"""
    log_text = json.dumps(log_entries, ensure_ascii=False, indent=2)
    prompt = ANALYSIS_PROMPT.format(logs=log_text)
    
    result = subprocess.run(
        ["ollama", "run", "llama3.2", prompt],
        capture_output=True, text=True, timeout=120
    )
    return result.stdout

def main():
    print(f"Starting log analysis [{datetime.now().isoformat()}]")
    
    errors = get_recent_errors(hours=1)
    
    if not errors:
        print("No anomalous logs found")
        return
    
    print(f"Found {len(errors)} anomalous log entries, analyzing...")
    
    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--- Analysis Batch {i // batch_size + 1} ---")
        print(analysis)
    
    print("\nLog analysis complete")

if __name__ == "__main__":
    main()

# Analyze logs every 15 minutes
*/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:
  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:
    type: elasticsearch
    inputs:
      - anomaly_detection
    endpoints: ["http://localhost:9200"]
    
  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: |
        ⚠️ Anomalous log detected!
        Server: {{ hostname }}
        Level: {{ .severity }}
        Anomaly Score: {{ .anomaly_score }}
        Content: {{ truncate(.message, 500) }}

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

# Configure Vector
sudo cp vector.yaml /etc/vector/vector.yaml

# Start Vector
sudo systemctl enable vector
sudo systemctl start vector

package log_anomaly

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

deny[msg] {
    # Same error repeated in a short time (pattern detection)
    count(input.recent_errors) > 50
    msg := sprintf("Error storm: same error repeated %v times", [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 Log AI Analyzer
"""
import re
from collections import defaultdict, Counter
from datetime import datetime, timedelta

def parse_access_log(log_file, lines=1000):
    """Parse Nginx JSON logs"""
    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):
    """Detect anomalies in logs"""
    anomalies = []
    
    # 1. Detect high-frequency IPs
    ip_counts = Counter(e.get('remote_addr', '') for e in entries)
    for ip, count in ip_counts.most_common(10):
        if count > 100:  # More than 100 times in 1000 lines
            anomalies.append({
                "type": "high_frequency_ip",
                "detail": f"IP {ip} appeared {count} times in recent logs",
                "severity": "high" if count > 500 else "medium"
            })
    
    # 2. Detect large volumes of 4xx/5xx errors
    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"Endpoint {path} had {count} 5xx errors",
                "severity": "critical"
            })
    
    # 3. Detect slow requests
    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"Found {len(slow_requests)} slow requests exceeding 5 seconds",
            "severity": "medium"
        })
    
    return anomalies

# vector.yaml - Real-time alert config
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_error:
    type: telegram
    inputs:
      - realtime_anomaly.alert
    api_key: "${TELEGRAM_BOT_TOKEN}"
    chat_id: "${TELEGRAM_CHAT_ID}"
    message:
      text: |
        🚨 {{ format!("text", .) }}
        Severity: {{ .severity }}
        Message: {{ 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: "High error rate detected"
          description: "Error rate exceeded 10/sec over the last 5 minutes"
          
      - alert: LogVolumeSpike
        expr: |
          increase(log_lines_total[10m]) > 10000
        for: 1m
        labels:
          severity: warning
        annotations:
          summary: "Log volume spike"
          description: "Log volume increased 10x in the last 10 minutes"

import re

def sanitize_logs(log_entries):
    """Sanitize log entries"""
    patterns = [
        (r'(apikey|token|secret|password|authorization)\s*[:=]\s*["\']?(\S+)', r'\1=***REDACTED***'),
        (r'\b\d{16}\b', '***CARD_REDACTED***'),  # Credit card numbers
        (r'\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b', '***EMAIL_REDACTED***'),  # Emails
    ]
    
    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

# Monitoring metrics
- log_analysis_requests_total        # Total analysis requests
- log_analysis_duration_seconds      # Analysis duration
- log_analysis_error_rate            # Analysis failure rate
- llm_api_latency_seconds            # LLM API latency
- llm_token_usage_total              # Token consumption