Troubleshooting Guide

# Option 1: Install with user flag
pip install --user pm2

# Option 2: Use virtual environment (recommended)
python -m venv pm2_env
source pm2_env/bin/activate  # On Windows: pm2_env\Scripts\activate
pip install pm2

# Option 3: Upgrade pip first
pip install --upgrade pip
pip install pm2

# Option 4: Force reinstall
pip install --force-reinstall pm2

# Install PM2 globally
npm install -g pm2

# Or install locally and add to PATH
npm install pm2
export PATH=$PATH:./node_modules/.bin

# Verify installation
pm2 --version

# Alternative: specify PM2 binary path in Python
from pm2 import PM2Manager
manager = PM2Manager(pm2_binary_path="/usr/local/bin/pm2")

# Increase timeout values
from pm2 import PM2Manager

manager = PM2Manager(
    command_timeout=60.0,  # Increase from default 30s
    connection_timeout=30.0  # Increase connection timeout
)

# For specific operations
process = manager.start(config, timeout=120.0)

# Check PM2 daemon status
import subprocess
result = subprocess.run(['pm2', 'ping'], capture_output=True, text=True)
print(f"PM2 status: {result.stdout}")

# Restart PM2 daemon if needed
subprocess.run(['pm2', 'kill'])
subprocess.run(['pm2', 'resurrect'])

# Diagnostic script
import subprocess
from pm2 import PM2Manager, PM2Exception

def diagnose_pm2():
    """Diagnose PM2 connection issues"""
    
    # Check PM2 installation
    try:
        result = subprocess.run(['pm2', '--version'], 
                              capture_output=True, text=True, timeout=10)
        print(f"✅ PM2 version: {result.stdout.strip()}")
    except (subprocess.TimeoutExpired, FileNotFoundError) as e:
        print(f"❌ PM2 not found or not responding: {e}")
        return False
    
    # Check daemon status
    try:
        result = subprocess.run(['pm2', 'status'], 
                              capture_output=True, text=True, timeout=10)
        if result.returncode == 0:
            print("✅ PM2 daemon is responding")
        else:
            print(f"⚠️ PM2 daemon issue: {result.stderr}")
    except subprocess.TimeoutExpired:
        print("❌ PM2 daemon timeout")
        return False
    
    # Test PM2Manager connection
    try:
        manager = PM2Manager()
        processes = manager.list()
        print(f"✅ PM2Manager working - {len(processes)} processes found")
        return True
    except PM2Exception as e:
        print(f"❌ PM2Manager error: {e}")
        return False

def fix_pm2_daemon():
    """Attempt to fix PM2 daemon issues"""
    print("Attempting to fix PM2 daemon...")
    
    # Kill existing daemon
    subprocess.run(['pm2', 'kill'], capture_output=True)
    print("Killed existing PM2 daemon")
    
    # Start fresh daemon
    result = subprocess.run(['pm2', 'list'], capture_output=True, text=True)
    if result.returncode == 0:
        print("✅ PM2 daemon restarted successfully")
        return True
    else:
        print(f"❌ Failed to restart PM2 daemon: {result.stderr}")
        return False

# Run diagnostics
if not diagnose_pm2():
    fix_pm2_daemon()
    diagnose_pm2()

# Check PM2 home directory permissions
ls -la ~/.pm2/

# Fix permissions if needed
chmod -R 755 ~/.pm2/
chown -R $USER:$GROUP ~/.pm2/

# Alternative: Use different PM2 home
export PM2_HOME=/tmp/pm2_home
mkdir -p $PM2_HOME
pm2 list

# In Python, specify custom PM2 home
import os
os.environ['PM2_HOME'] = '/path/to/custom/pm2/home'
from pm2 import PM2Manager
manager = PM2Manager()

# Performance optimization
from pm2 import PM2Manager
import time

class OptimizedPM2Manager(PM2Manager):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._process_cache = {}
        self._cache_timeout = 5.0  # 5 seconds
        self._last_cache_update = 0
    
    def list_cached(self, force_refresh=False):
        """Get process list with caching"""
        current_time = time.time()
        
        if (force_refresh or 
            current_time - self._last_cache_update > self._cache_timeout):
            
            self._process_cache = {
                p.name: p for p in super().list()
            }
            self._last_cache_update = current_time
        
        return list(self._process_cache.values())
    
    def get_process_cached(self, name_or_id):
        """Get process with caching"""
        if isinstance(name_or_id, str) and name_or_id in self._process_cache:
            cached_process = self._process_cache[name_or_id]
            # Refresh if cache is old
            if time.time() - self._last_cache_update > self._cache_timeout:
                return super().get_process(name_or_id)
            return cached_process
        
        return super().get_process(name_or_id)

# Usage
manager = OptimizedPM2Manager()

# Faster process listing
processes = manager.list_cached()

# Bulk operations optimization
def optimize_bulk_operations():
    """Optimize bulk operations"""
    
    # Use concurrent operations
    from concurrent.futures import ThreadPoolExecutor
    import threading
    
    manager = PM2Manager()
    thread_local = threading.local()
    
    def get_manager():
        if not hasattr(thread_local, 'manager'):
            thread_local.manager = PM2Manager()
        return thread_local.manager
    
    def start_process(config):
        return get_manager().start(config)
    
    # Start multiple processes concurrently
    configs = [{"name": f"app-{i}", "script": "app.js"} for i in range(5)]
    
    with ThreadPoolExecutor(max_workers=3) as executor:
        results = list(executor.map(start_process, configs))
    
    return results

optimize_bulk_operations()

import logging
from pm2 import PM2Manager

# Enable debug logging
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Enable PM2 library debug logs
logger = logging.getLogger('pm2')
logger.setLevel(logging.DEBUG)

# Create manager with verbose output
manager = PM2Manager(verbose=True)

# All operations will now show detailed logs
process = manager.start({"name": "debug-app", "script": "app.js"})

from pm2 import PM2Manager
import json
from datetime import datetime

class PM2Inspector:
    def __init__(self):
        self.manager = PM2Manager()
    
    def full_system_report(self):
        """Generate comprehensive system report"""
        report = {
            'timestamp': datetime.now().isoformat(),
            'pm2_version': self._get_pm2_version(),
            'processes': [],
            'system_info': self._get_system_info(),
            'errors': []
        }
        
        try:
            processes = self.manager.list()
            for process in processes:
                process_info = {
                    'name': process.name,
                    'pid': process.pid,
                    'status': process.status,
                    'cpu': process.cpu,
                    'memory': process.memory,
                    'uptime': process.uptime,
                    'restarts': process.restarts,
                    'metrics': {
                        'cpu_percent': process.metrics.cpu_percent,
                        'memory_usage': process.metrics.memory_usage,
                        'memory_percent': process.metrics.memory_percent,
                        'unstable_restarts': process.metrics.unstable_restarts
                    } if hasattr(process, 'metrics') else None
                }
                report['processes'].append(process_info)
                
        except Exception as e:
            report['errors'].append(f"Failed to get processes: {e}")
        
        return report
    
    def _get_pm2_version(self):
        try:
            import subprocess
            result = subprocess.run(['pm2', '--version'], 
                                  capture_output=True, text=True)
            return result.stdout.strip()
        except:
            return "Unknown"
    
    def _get_system_info(self):
        import platform
        import psutil
        
        return {
            'platform': platform.platform(),
            'python_version': platform.python_version(),
            'cpu_count': psutil.cpu_count(),
            'memory_total': psutil.virtual_memory().total,
            'disk_usage': psutil.disk_usage('/').percent
        }
    
    def health_check(self, process_name):
        """Detailed health check for specific process"""
        try:
            process = self.manager.get_process(process_name)
            
            health_status = {
                'process_name': process_name,
                'is_running': process.status == 'online',
                'pid': process.pid,
                'memory_ok': process.memory < 1024 * 1024 * 500,  # < 500MB
                'cpu_ok': process.cpu < 80.0,  # < 80%
                'restart_count': process.restarts,
                'uptime_minutes': process.uptime / 60 if process.uptime else 0,
                'last_check': datetime.now().isoformat()
            }
            
            # Overall health score
            health_score = 0
            if health_status['is_running']: health_score += 25
            if health_status['memory_ok']: health_score += 25
            if health_status['cpu_ok']: health_score += 25
            if health_status['restart_count'] < 5: health_score += 25
            
            health_status['health_score'] = health_score
            health_status['health_status'] = (
                'Excellent' if health_score >= 90 else
                'Good' if health_score >= 70 else
                'Warning' if health_score >= 50 else
                'Critical'
            )
            
            return health_status
            
        except Exception as e:
            return {
                'process_name': process_name,
                'error': str(e),
                'health_status': 'Error'
            }

# Usage
inspector = PM2Inspector()

# Generate full report
report = inspector.full_system_report()
print(json.dumps(report, indent=2))

# Health check specific process
if report['processes']:
    process_name = report['processes'][0]['name']
    health = inspector.health_check(process_name)
    print(f"\nHealth check for {process_name}:")
    print(json.dumps(health, indent=2))