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

# Pull lightweight models (suitable for process analysis)
ollama pull llama3.2:1b
ollama pull qwen2.5:1.5b

# Verify installation
ollama list

#!/usr/bin/env python3
"""VPS Process Intelligence System - Real-time Anomaly Detection with Local LLM"""

import psutil
import subprocess
import json
import sqlite3
import os
from datetime import datetime, timedelta
from pathlib import Path

DB_PATH = Path.home() / ".vps_ai" / "process_diag.db"
OLLAMA_URL = "http://localhost:11434"
MODEL = "llama3.2:1b"

class ProcessIntelligence:
    def __init__(self):
        DB_PATH.parent.mkdir(parents=True, exist_ok=True)
        self._init_db()

    def _init_db(self):
        conn = sqlite3.connect(DB_PATH)
        c = conn.cursor()
        c.execute('''CREATE TABLE IF NOT EXISTS process_snapshots (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp TEXT,
            pid INTEGER,
            name TEXT,
            cpu_percent REAL,
            memory_rss INTEGER,
            memory_vms INTEGER,
            num_threads INTEGER,
            status TEXT,
            nice INTEGER,
            io_read_bytes INTEGER,
            io_write_bytes INTEGER,
            children_count INTEGER
        )''')
        c.execute('''CREATE TABLE IF NOT EXISTS anomalies (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp TEXT,
            pid INTEGER,
            process_name TEXT,
            anomaly_type TEXT,
            severity TEXT,
            description TEXT,
            recommendation TEXT,
            resolved INTEGER DEFAULT 0
        )''')
        c.execute('''CREATE INDEX IF NOT EXISTS idx_snapshot_ts ON process_snapshots(timestamp)''')
        c.execute('''CREATE INDEX IF NOT EXISTS idx_anomaly_ts ON anomalies(timestamp)''')
        conn.commit()
        conn.close()

    def collect_snapshot(self) -> dict:
        """Collect current process snapshot"""
        processes = {}
        for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 
                                          'memory_info', 'status', 'num_threads',
                                          'nice', 'io_counters', 'children']):
            try:
                mem = proc.info['memory_info']
                io = proc.info['io_counters'] or (0, 0, 0, 0)
                children = len(proc.info['children'] or [])
                
                processes[proc.info['pid']] = {
                    'name': proc.info['name'],
                    'cpu': proc.info['cpu_percent'] or 0,
                    'rss': mem.rss if mem else 0,
                    'vms': mem.vms if mem else 0,
                    'threads': proc.info['num_threads'] or 1,
                    'status': proc.info['status'],
                    'nice': proc.info['nice'] or 0,
                    'io_read': io.read_bytes,
                    'io_write': io.write_bytes,
                    'children': children
                }
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue
        
        return {
            'timestamp': datetime.now().isoformat(),
            'processes': processes,
            'system': {
                'cpu_percent': psutil.cpu_percent(interval=0.1),
                'memory_percent': psutil.virtual_memory().percent,
                'load_avg': os.getloadavg() if hasattr(os, 'getloadavg') else (0, 0, 0)
            }
        }

    def store_snapshot(self, snapshot: dict):
        """Store snapshot to database"""
        conn = sqlite3.connect(DB_PATH)
        c = conn.cursor()
        ts = snapshot['timestamp']
        for pid, info in snapshot['processes'].items():
            c.execute('''INSERT INTO process_snapshots 
                (timestamp, pid, name, cpu_percent, memory_rss, memory_vms,
                 num_threads, status, nice, io_read_bytes, io_write_bytes, children_count)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)''',
                (ts, pid, info['name'], info['cpu'], info['rss'], info['vms'],
                 info['threads'], info['status'], info['nice'],
                 info['io_read'], info['io_write'], info['children']))
        conn.commit()
        conn.close()

    def detect_anomalies(self, snapshot: dict) -> list:
        """Detect process anomalies (rule-based)"""
        anomalies = []
        now = datetime.now().isoformat()
        
        for pid, info in snapshot['processes'].items():
            # High CPU
            if info['cpu'] > 80:
                anomalies.append({
                    'type': 'high_cpu',
                    'severity': 'warning' if info['cpu'] < 95 else 'critical',
                    'pid': pid,
                    'name': info['name'],
                    'description': f"Process {info['name']} (PID {pid}) CPU usage {info['cpu']:.1f}%",
                    'recommendation': 'Check if compute-intensive, consider rate limiting or migration to dedicated container'
                })
            
            # Memory anomaly (RSS > 2GB)
            if info['rss'] > 2 * 1024 * 1024 * 1024:
                anomalies.append({
                    'type': 'high_memory',
                    'severity': 'warning',
                    'pid': pid,
                    'name': info['name'],
                    'description': f"Process {info['name']} (PID {pid}) memory usage {info['rss']//1024//1024}MB",
                    'recommendation': 'Check for memory leaks, consider setting cgroup memory limit'
                })
            
            # Zombie process
            if info['status'] == 'zombie':
                anomalies.append({
                    'type': 'zombie_process',
                    'severity': 'error',
                    'pid': pid,
                    'name': info['name'],
                    'description': f"Zombie process detected: {info['name']} (PID {pid})",
                    'recommendation': 'Check if parent process correctly reaps children, restart parent if necessary'
                })
            
            # Excessive threads (> 200)
            if info['threads'] > 200:
                anomalies.append({
                    'type': 'high_threads',
                    'severity': 'warning',
                    'pid': pid,
                    'name': info['name'],
                    'description': f"Process {info['name']} (PID {pid}) has {info['threads']} threads",
                    'recommendation': 'High thread count causes context switch overhead, consider async model'
                })
        
        return anomalies

    def store_anomalies(self, anomalies: list):
        """Store anomaly records"""
        conn = sqlite3.connect(DB_PATH)
        c = conn.cursor()
        ts = datetime.now().isoformat()
        for a in anomalies:
            c.execute('''INSERT INTO anomalies 
                (timestamp, pid, process_name, anomaly_type, severity, description, recommendation)
                VALUES (?, ?, ?, ?, ?, ?, ?)''',
                (ts, a.get('pid'), a.get('name'), a['type'], a['severity'],
                 a['description'], a['recommendation']))
        conn.commit()
        conn.close()

    def query_history(self, hours: int = 24) -> dict:
        """Query historical data for LLM analysis"""
        conn = sqlite3.connect(DB_PATH)
        c = conn.cursor()
        since = (datetime.now() - timedelta(hours=hours)).isoformat()
        
        # Get anomaly statistics
        c.execute('''SELECT anomaly_type, severity, COUNT(*) as cnt, MAX(timestamp) as latest
                     FROM anomalies 
                     WHERE timestamp >= ?
                     GROUP BY anomaly_type, severity
                     ORDER BY cnt DESC''', (since,))
        stats = c.fetchall()
        
        # Get TOP resource-consuming processes
        c.execute('''SELECT name, AVG(cpu_percent) as avg_cpu, AVG(memory_rss) as avg_mem,
                     COUNT(DISTINCT pid) as pid_count
                     FROM process_snapshots
                     WHERE timestamp >= ?
                     GROUP BY name
                     ORDER BY avg_cpu DESC
                     LIMIT 10''', (since,))
        top_procs = c.fetchall()
        
        conn.close()
        return {'anomaly_stats': stats, 'top_processes': top_procs}

    def generate_report(self, history: dict) -> str:
        """Call LLM to generate analysis report"""
        import requests
        
        anomaly_summary = []
        for row in history['anomaly_stats']:
            anomaly_summary.append(f"- {row[0]} ({row[1]}): {row[2]} occurrences")
        
        proc_summary = []
        for row in history['top_processes']:
            cpu_val = row[1] if row[1] else 0
            mem_mb = (row[2] / 1024 / 1024) if row[2] else 0
            proc_summary.append(f"- {row[0]}: CPU {cpu_val:.1f}%, Memory {mem_mb:.0f}MB")
        
        prompt = f"""You are a VPS operations expert. Based on the following process analysis data, generate a concise diagnosis report.

Anomaly Statistics (past 24 hours):
{chr(10).join(anomaly_summary) if anomaly_summary else '- No anomalies'}

Top Resource-Consuming Processes:
{chr(10).join(proc_summary)}

Please provide:
1. Overall health score (0-100)
2. Key findings (max 3)
3. Recommended actions (sorted by priority)

Output format: Markdown."""

        try:
            resp = requests.post(f"{OLLAMA_URL}/api/generate", json={
                'model': MODEL,
                'prompt': prompt,
                'stream': False
            }, timeout=60)
            return resp.json().get('response', '')
        except Exception as e:
            return f"LLM call failed: {e}"


if __name__ == "__main__":
    import sys
    pi = ProcessIntelligence()
    
    if len(sys.argv) > 1 and sys.argv[1] == "analyze":
        # Analysis mode: query history and generate report
        history = pi.query_history(24)
        report = pi.generate_report(history)
        print(report)
    else:
        # Collection mode: single snapshot
        snapshot = pi.collect_snapshot()
        pi.store_snapshot(snapshot)
        anomalies = pi.detect_anomalies(snapshot)
        if anomalies:
            pi.store_anomalies(anomalies)
            print(json.dumps(anomalies, indent=2, ensure_ascii=False))
        else:
            print("No anomalies detected")

# Collect process snapshots every 5 minutes
*/5 * * * * python3 ~/.vps_ai/process_intelligence.py >> /dev/null 2>&1

# Generate diagnosis report every hour
0 * * * * python3 ~/.vps_ai/process_intelligence.py analyze >> ~/.vps_ai/daily_report.md 2>&1

# Clean historical data older than 30 days at 3 AM daily
0 3 * * * sqlite3 ~/.vps_ai/process_diag.db "DELETE FROM process_snapshots WHERE timestamp < datetime('now', '-30 days')"

# /etc/systemd/system/vps-process-collect.service
[Unit]
Description=VPS Process Intelligence Collection
After=network.target

[Service]
Type=oneshot
User=root
ExecStart=/usr/bin/python3 /root/.vps_ai/process_intelligence.py

# /etc/systemd/system/vps-process-collect.timer
[Unit]
Description=VPS Process Intelligence Timer

[Timer]
OnBootSec=5min
OnUnitActiveSec=5min
AccuracySec=1min

[Install]
WantedBy=timers.target

systemctl enable --now vps-process-collect.timer

#!/usr/bin/env python3
"""Process anomaly notifications via Telegram Bot"""

import sqlite3
import requests
from datetime import datetime, timedelta
from pathlib import Path

DB_PATH = Path.home() / ".vps_ai" / "process_diag.db"
TELEGRAM_BOT_TOKEN = "YOUR_BOT_TOKEN"
CHAT_ID = "YOUR_CHAT_ID"

def send_telegram(message: str):
    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
    requests.post(url, json={
        'chat_id': CHAT_ID,
        'text': message,
        'parse_mode': 'Markdown'
    })

def check_unresolved():
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    since = (datetime.now() - timedelta(minutes=30)).isoformat()
    
    c.execute('''SELECT severity, description, recommendation 
                 FROM anomalies 
                 WHERE resolved = 0 AND timestamp >= ?
                 ORDER BY 
                   CASE severity 
                     WHEN 'critical' THEN 1 
                     WHEN 'error' THEN 2 
                     ELSE 3 END''', (since,))
    
    rows = c.fetchall()
    conn.close()
    
    if not rows:
        return
    
    lines = ["🚨 *VPS Process Anomaly Alert*"]
    for sev, desc, rec in rows:
        icon = {"critical": "🔴", "error": "🟠", "warning": "🟡"}.get(sev, "⚪")
        lines.append(f"{icon} {desc}")
        if rec:
            lines.append(f"   ↳ {rec}")
    
    send_telegram("\n\n".join(lines))

if __name__ == "__main__":
    check_unresolved()

*/10 * * * * python3 ~/.vps_ai/alert_checker.py >> /dev/null 2>&1

# View recent 24-hour anomaly summary
sqlite3 ~/.vps_ai/process_diag.db "
SELECT anomaly_type, severity, COUNT(*) 
FROM anomalies 
WHERE timestamp >= datetime('now', '-24 hours')
GROUP BY anomaly_type, severity
ORDER BY COUNT(*) DESC;"

# View TOP 10 high-CPU processes
sqlite3 ~/.vps_ai/process_diag.db "
SELECT name, AVG(cpu_percent) as avg_cpu, 
       AVG(memory_rss)/1024/1024 as avg_mem_mb
FROM process_snapshots
WHERE timestamp >= datetime('now', '-1 hour')
GROUP BY name
ORDER BY avg_cpu DESC
LIMIT 10;"

# Full LLM analysis report
python3 ~/.vps_ai/process_intelligence.py analyze

## VPS Process Health Diagnosis Report

**Overall Health Score: 72/100** ⚠️ Needs Attention

### Key Findings
1. **Memory Leak Risk**: `uvicorn` process RSS grew from 200MB to 1.8GB over the past 6 hours, linear growth trend detected
2. **I/O Contention**: `postgres` and `redis` processes compete for I/O during peak hours
3. **Zombie Residue**: 2 terminated but unreaped child processes detected

### Recommended Actions
1. **Immediate**: Set cgroup memory limit on uvicorn process (recommend 2GB cap)
2. **Short-term**: Configure `--limit-max-requests 1000` on uvicorn for auto-restart
3. **Medium-term**: Migrate to async I/O model to reduce contention with PostgreSQL

**Scoring Note**: Memory growth is the primary deduction factor, prioritize handling.

def learn_baseline(self, days: int = 7) -> dict:
    """Learn normal process behavior baselines"""
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    
    since = (datetime.now() - timedelta(days=days)).isoformat()
    
    # Aggregate statistics by process name
    c.execute('''SELECT name, 
                     AVG(cpu_percent) as mean_cpu, 
                     STDDEV(cpu_percent) as std_cpu,
                     AVG(memory_rss) as mean_mem,
                     COUNT(*) as sample_count
                 FROM process_snapshots
                 WHERE timestamp >= ?
                 GROUP BY name
                 HAVING sample_count > 100''', (since,))
    
    baselines = {}
    for row in c.fetchall():
        baselines[row[0]] = {
            'mean_cpu': row[1],
            'std_cpu': row[2] or 5.0,
            'mean_mem': row[3],
            'samples': row[4]
        }
    
    conn.close()
    return baselines