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

# Pull a VPS-friendly model (Qwen2.5-7B or DeepSeek-R1-8B recommended)
ollama pull qwen2.5:7b-instruct

# Start and verify
ollama list
ollama run qwen2.5:7b-instruct "Hello, please introduce yourself"

#!/usr/bin/env python3
"""Redis Intelligent Monitoring Data Collector"""

import redis
import json
from datetime import datetime
from pathlib import Path

class RedisMonitor:
    def __init__(self, redis_url="redis://localhost:6379"):
        self.r = redis.Redis.from_url(redis_url, decode_responses=True)
        self.output_dir = Path("/var/log/redis-ai/analysis")
        self.output_dir.mkdir(parents=True, exist_ok=True)

    def collect_all(self):
        """Collect all key metrics"""
        data = {
            "timestamp": datetime.now().isoformat(),
            "memory": self._get_memory_info(),
            "slowlog": self._get_slowlog(),
            "keyspace": self._get_keyspace(),
            "clients": self._get_clients(),
            "stats": self._get_stats(),
            "hot_keys": self._detect_hot_keys(),
            "fragmentation": self._calc_fragmentation(),
        }
        return data

    def _get_memory_info(self):
        info = self.r.info("memory")
        return {
            "used_memory_human": info.get("used_memory_human", "0"),
            "used_memory_rss_human": info.get("used_memory_rss_human", "0"),
            "maxmemory_human": info.get("maxmemory_human", "0"),
            "mem_fragmentation_ratio": info.get("mem_fragmentation_ratio", 1.0),
            "used_memory_peak_human": info.get("used_memory_peak_human", "0"),
            "mem_allocator": info.get("mem_allocator", "jemalloc"),
        }

    def _get_slowlog(self, limit=20):
        """Get slow query log"""
        try:
            entries = self.r.slowlog_get(limit)
            result = []
            for entry in entries:
                result.append({
                    "id": entry[0],
                    "timestamp": datetime.fromtimestamp(entry[1]).isoformat(),
                    "duration_us": entry[2],
                    "command": " ".join(entry[3]),
                })
            return result
        except Exception as e:
            return [{"error": str(e)}]

    def _get_keyspace(self):
        try:
            keys_count = self.r.dbsize()
            return {"db0_keys": keys_count}
        except Exception as e:
            return {"error": str(e)}

    def _get_clients(self):
        info = self.r.info("clients")
        return {
            "connected": info.get("connected_clients", 0),
            "blocked": info.get("blocked_clients", 0),
        }

    def _get_stats(self):
        info = self.r.info("stats")
        return {
            "ops_per_sec": info.get("instantaneous_ops_per_sec", 0),
            "keyspace_hits": info.get("keyspace_hits", 0),
            "keyspace_misses": info.get("keyspace_misses", 0),
            "evicted_keys": info.get("evicted_keys", 0),
        }

    def _detect_hot_keys(self):
        """Detect hot keys using SAMPLE (Redis 7.0+)"""
        try:
            hot_keys = self.r.sample(keys=100, count=50)
            if hot_keys:
                return {"sampled_keys": len(hot_keys), "warning": "Check business-layer access frequency"}
        except Exception:
            pass
        return {"detected": False}

    def _calc_fragmentation(self):
        info = self.r.info("memory")
        used = info.get("used_memory", 0)
        rss = info.get("used_memory_rss", 0)
        if rss > 0:
            ratio = used / rss
            status = "normal" if 0.8 < ratio < 1.5 else "high" if ratio >= 1.5 else "low"
            return {"ratio": round(ratio, 2), "status": status}
        return {"ratio": 0, "status": "unknown"}

    def save_report(self, data):
        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
        filepath = self.output_dir / f"redis_analysis_{ts}.json"
        filepath.write_text(json.dumps(data, ensure_ascii=False, indent=2))
        return str(filepath)


if __name__ == "__main__":
    monitor = RedisMonitor()
    data = monitor.collect_all()
    path = monitor.save_report(data)
    print(f"Report saved: {path}")
    print(json.dumps({
        "memory": data["memory"],
        "slowlog_count": len(data["slowlog"]),
        "fragmentation": data["fragmentation"],
        "stats": data["stats"],
    }, ensure_ascii=False, indent=2))

#!/usr/bin/env python3
"""Redis Intelligent Analyzer — Uses local LLM for diagnosis and recommendations"""

import json
import subprocess
import sys
from pathlib import Path

SYSTEM_PROMPT = """You are a professional Redis operations expert.
Your task is to perform intelligent diagnosis based on provided Redis runtime data
and give actionable optimization recommendations.
Return analysis results in JSON format with these fields:
- diagnosis: problem description (string)
- severity: critical/high/medium/low
- recommendations: list of suggestions (string array)
- actions: executable operations (each with command and description)
- risk_level: operation risk (high/medium/low)
"""

def analyze_with_llm(redis_data: dict) -> dict:
    """Call local Ollama LLM for analysis"""
    prompt = f"""Please analyze the following Redis runtime data and provide optimization suggestions:

{json.dumps(redis_data, ensure_ascii=False, indent=2)}

Requirements:
1. Identify potential problems
2. Provide specific optimization commands
3. Evaluate operation risk
4. Rank recommendations by priority
"""

    result = subprocess.run(
        ["ollama", "run", "qwen2.5:7b-instruct", prompt],
        capture_output=True, text=True, timeout=120
    )

    output = result.stdout.strip()
    try:
        start = output.find("{")
        end = output.rfind("}")
        if start != -1 and end != -1:
            output = output[start:end+1]
        return json.loads(output)
    except json.JSONDecodeError:
        return {
            "diagnosis": "Analysis failed",
            "severity": "low",
            "recommendations": [output[:500]],
            "actions": [],
            "risk_level": "unknown"
        }


def main():
    data_file = sys.argv[1] if len(sys.argv) > 1 else "/tmp/redis_data.json"
    with open(data_file) as f:
        redis_data = json.load(f)

    print("Analyzing with LLM...")
    analysis = analyze_with_llm(redis_data)

    output_dir = Path("/var/log/redis-ai/analysis")
    output_dir.mkdir(parents=True, exist_ok=True)
    ts = __import__('datetime').datetime.now().strftime("%Y%m%d_%H%M%S")
    output_file = output_dir / f"analysis_{ts}.json"
    output_file.write_text(json.dumps(analysis, ensure_ascii=False, indent=2))

    print(f"\nAnalysis saved: {output_file}")
    print(f"Diagnosis: {analysis.get('diagnosis', 'N/A')}")
    print(f"Severity: {analysis.get('severity', 'N/A')}")
    print(f"Risk level: {analysis.get('risk_level', 'N/A')}")
    print(f"\nRecommendations:")
    for i, rec in enumerate(analysis.get("recommendations", []), 1):
        print(f"  {i}. {rec}")
    print(f"\nExecutable actions ({len(analysis.get('actions', []))}):")
    for action in analysis.get("actions", []):
        print(f"  • {action.get('description', '')}: `{action.get('command', '')}`")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Redis Intelligent Optimizer — Safely executes LLM-recommended operations"""

import redis
import json
import time
from datetime import datetime
from pathlib import Path

class RedisOptimizer:
    def __init__(self, redis_url="redis://localhost:6379", dry_run=True):
        self.r = redis.Redis.from_url(redis_url, decode_responses=True)
        self.dry_run = dry_run
        self.log_dir = Path("/var/log/redis-ai/actions")
        self.log_dir.mkdir(parents=True, exist_ok=True)

    def execute_actions(self, analysis: dict) -> dict:
        results = {"actions_executed": [], "errors": [], "timestamp": datetime.now().isoformat()}

        for action in analysis.get("actions", []):
            cmd = action.get("command", "")
            desc = action.get("description", "")
            risk = analysis.get("risk_level", "unknown")

            try:
                result = self._execute_command(cmd, desc, risk)
                results["actions_executed"].append(result)
            except Exception as e:
                results["errors"].append({"command": cmd, "error": str(e)})

        return results

    def _execute_command(self, cmd: str, desc: str, risk: str) -> dict:
        action_record = {
            "description": desc,
            "command": cmd,
            "risk": risk,
            "status": "pending",
            "executed_at": datetime.now().isoformat(),
        }

        if self.dry_run:
            action_record["status"] = "dry_run_skipped"
            print(f"[DRY RUN] Skipped: {desc}")
            print(f"  Command: {cmd}")
            return action_record

        # Safe command whitelist
        safe_prefixes = ["CONFIG SET", "MEMORY PURGE", "UNLINK", "DEBUG SLEEP"]
        if not any(cmd.startswith(s) for s in safe_prefixes):
            action_record["status"] = "blocked_unsafe"
            return action_record

        try:
            parts = cmd.split()
            if parts[0].upper() == "CONFIG" and parts[1].upper() == "SET":
                param = parts[2]
                value = parts[3] if len(parts) > 3 else "1"
                sensitive = ["requirepass", "masterauth", "secret"]
                if any(s in param.lower() for s in sensitive):
                    action_record["status"] = "blocked_sensitive"
                    return action_record
                result = self.r.config_set(param, value)
                action_record["status"] = "success"
                action_record["result"] = str(result)

            elif parts[0].upper() == "MEMORY" and parts[1].upper() == "PURGE":
                result = self.r.execute_command("MEMORY", "PURGE")
                action_record["status"] = "success"
                action_record["result"] = str(result)

            else:
                result = self.r.execute_command(*parts)
                action_record["status"] = "success"
                action_record["result"] = str(result)[:200]

        except Exception as e:
            action_record["status"] = "error"
            action_record["error"] = str(e)

        log_file = self.log_dir / f"action_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        log_file.write_text(json.dumps(action_record, ensure_ascii=False, indent=2))
        return action_record

    def verify_optimization(self) -> dict:
        before = self._snapshot_metrics()
        time.sleep(2)
        after = self._snapshot_metrics()
        return {"before": before, "after": after, "delta": {
            k: after.get(k, 0) - before.get(k, 0)
            for k in set(before.keys()) | set(after.keys())
        }}

    def _snapshot_metrics(self) -> dict:
        info = self.r.info("memory")
        stats = self.r.info("stats")
        return {
            "used_memory": info.get("used_memory", 0),
            "frag_ratio": info.get("mem_fragmentation_ratio", 1.0),
            "ops_per_sec": stats.get("instantaneous_ops_per_sec", 0),
            "connected_clients": info.get("connected_clients", 0),
        }


def main():
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--analysis", required=True)
    parser.add_argument("--dry-run", action="store_true", default=True)
    parser.add_argument("--execute", action="store_true")
    args = parser.parse_args()

    optimizer = RedisOptimizer(dry_run=not args.execute)
    with open(args.analysis) as f:
        analysis = json.load(f)

    mode = "LIVE EXECUTION" if args.execute else "DRY RUN"
    print(f"Mode: {mode}")
    print(f"Diagnosis: {analysis.get('diagnosis', 'N/A')}")
    print(f"Recommendations: {len(analysis.get('recommendations', []))}")
    print(f"Actions: {len(analysis.get('actions', []))}\n")

    results = optimizer.execute_actions(analysis)
    print(json.dumps(results, ensure_ascii=False, indent=2))

    if not args.execute:
        print("\n⚠️  Dry run mode — no actual operations performed.")
        print("   Use --execute to apply changes (after verification).")


if __name__ == "__main__":
    main()

#!/bin/bash
# Redis AI Intelligent Optimization Cron Job
# Add to crontab: */5 * * * * /opt/redis-ai/redis_ai_cron.sh

set -euo pipefail

LOG_DIR="/var/log/redis-ai"
DATA_DIR="${LOG_DIR}/data"
ANALYSIS_DIR="${LOG_DIR}/analysis"
ACTION_DIR="${LOG_DIR}/actions"

mkdir -p "${DATA_DIR}" "${ANALYSIS_DIR}" "${ACTION_DIR}"

TIMESTAMP=$(date +%Y%m%d_%H%M%S)
echo "[$TIMESTAMP] Starting Redis AI optimization cycle..."

# Step 1: Collect data
python3 /opt/redis-ai/redis_monitor.py > "${DATA_DIR}/raw_${TIMESTAMP}.json" 2>&1

# Step 2: LLM analysis
python3 /opt/redis-ai/redis_analyzer.py "${DATA_DIR}/raw_${TIMESTAMP}.json" \
    > "${ANALYSIS_DIR}/result_${TIMESTAMP}.json" 2>&1

# Step 3: Check severity and alert if needed
ANALYSIS=$(cat "${ANALYSIS_DIR}/result_${TIMESTAMP}.json" 2>/dev/null || echo '{}')
SEVERITY=$(echo "$ANALYSIS" | python3 -c "import sys,json; print(json.load(sys.stdin).get('severity','low'))" 2>/dev/null || echo "low")

if [ "$SEVERITY" = "critical" ] || [ "$SEVERITY" = "high" ]; then
    echo "[$TIMESTAMP] ⚠️ Critical issue detected, sending alert..."
    curl -s -X POST "https://api.telegram.org/bot${TG_BOT_TOKEN}/sendMessage" \
        -d "chat_id=${TG_CHAT_ID}" \
        -d "text=\"🔴 Redis Alert\n$(echo "$ANALYSIS" | python3 -c \"import sys,json; d=json.load(sys.stdin); print(f'Diagnosis: {d.get(\\\"diagnosis\\\",\\\"\\\")}')\")" \
        || true
fi

# Step 4: Auto-execute low-risk operations
if [ "$SEVERITY" = "low" ]; then
    python3 /opt/redis-ai/redis_optimizer.py \
        --analysis "${ANALYSIS_DIR}/result_${TIMESTAMP}.json" \
        --dry-run
fi

echo "[$TIMESTAMP] Complete"

chmod +x /opt/redis-ai/redis_ai_cron.sh

# Add to crontab (runs every 5 minutes)
(crontab -l 2>/dev/null; echo "*/5 * * * * /opt/redis-ai/redis_ai_cron.sh >> /var/log/redis-ai/cron.log 2>&1") | crontab -

{
  "diagnosis": "Redis memory fragmentation ratio 2.3, exceeding safe threshold 1.5 — significant memory waste detected",
  "severity": "high",
  "recommendations": [
    "Execute MEMORY PURGE to release jemalloc internal fragmentation",
    "Check for大量过期 Key未及时删除",
    "Consider restarting Redis for complete cleanup (assess downtime impact)"
  ],
  "actions": [
    {
      "description": "Run MEMORY PURGE to clean fragmentation",
      "command": "MEMORY PURGE",
      "risk": "low"
    }
  ],
  "risk_level": "low"
}

{
  "diagnosis": "Found 3 big keys (>10MB) causing intermittent blocking operations",
  "severity": "medium",
  "recommendations": [
    "Split big Hash into multiple smaller Hashes (< 512 fields per bucket)",
    "Use UNLINK instead of DEL to avoid blocking the main thread",
    "Evaluate migrating hot data to RediSearch"
  ],
  "actions": [
    {
      "description": "Scan and flag big keys",
      "command": "SCAN 0 MATCH * COUNT 10000",
      "risk": "low"
    },
    {
      "description": "Non-blocking deletion of big key",
      "command": "UNLINK huge:hash:key",
      "risk": "medium"
    }
  ],
  "risk_level": "medium"
}

{
  "diagnosis": "Current allkeys-lru eviction policy doesn't match your TTL-driven business pattern, causing important keys to be incorrectly evicted",
  "severity": "medium",
  "recommendations": [
    "Switch to allkeys-ttl policy to prioritize expiring keys",
    "Set longer TTLs for important keys as protection",
    "Add maxmemory limit to prevent unbounded growth"
  ],
  "actions": [
    {
      "description": "Change eviction policy to allkeys-ttl",
      "command": "CONFIG SET maxmemory-policy allkeys-ttl",
      "risk": "low"
    }
  ],
  "risk_level": "low"
}

# 1. Create directory structure
mkdir -p /opt/redis-ai /var/log/redis-ai/{data,analysis,actions}

# 2. Place scripts
cp redis_monitor.py /opt/redis-ai/
cp redis_analyzer.py /opt/redis-ai/
cp redis_optimizer.py /opt/redis-ai/
cp redis_ai_cron.sh /opt/redis-ai/

# 3. Install Python dependencies
pip install redis

# 4. Ensure Ollama is running
ollama list | grep qwen2.5

# 5. Configure environment variables
cat >> ~/.bashrc << 'EOF'
export TG_BOT_TOKEN="your_telegram_bot_token"
export TG_CHAT_ID="your_telegram_chat_id"
EOF

# 6. First run test (dry-run mode)
/opt/redis-ai/redis_ai_cron.sh