Configure advanced htop monitoring with custom scripts and alerting

Install build dependencies

Install the required development tools and libraries to compile htop with custom features.

sudo apt update
sudo apt install -y build-essential autotools-dev autoconf automake pkg-config
sudo apt install -y libncurses5-dev libncursesw5-dev libprocps-dev libsensors4-dev
sudo apt install -y libnl-3-dev libnl-genl-3-dev git curl jq

sudo dnf update -y
sudo dnf groupinstall -y "Development Tools"
sudo dnf install -y ncurses-devel procps-ng-devel lm_sensors-devel
sudo dnf install -y libnl3-devel autoconf automake pkg-config git curl jq

Download and compile custom htop

Clone the latest htop source and compile with enhanced monitoring features enabled.

cd /tmp
git clone https://github.com/htop-dev/htop.git
cd htop
./autogen.sh
./configure --enable-unicode --enable-sensors --enable-capabilities
make -j$(nproc)
sudo make install
sudo ldconfig

Create monitoring directories

Set up the directory structure for monitoring scripts, configurations, and logs.

sudo mkdir -p /opt/htop-monitor/{bin,config,logs,data}
sudo mkdir -p /var/log/htop-monitor
sudo chown -R $USER:$USER /opt/htop-monitor
sudo chmod 755 /opt/htop-monitor /var/log/htop-monitor

Configure advanced htop settings

Create a system-wide htop configuration with custom columns and monitoring-optimized display.

fields=0 48 17 18 38 39 40 2 46 47 49 1
sort_key=46
sort_direction=-1
hide_threads=0
hide_kernel_threads=0
hide_userland_threads=0
shadow_other_users=1
show_thread_names=1
show_program_path=1
highlight_base_name=1
highlight_megabytes=1
highlight_threads=1
highlight_changes=0
highlight_changes_delay_secs=5
find_comm_in_cmdline=1
strip_exe_from_cmdline=1
show_merged_command=0
tree_view=1
tree_view_always_by_pid=0
header_margin=1
detailed_cpu_time=1
cpu_count_from_one=1
show_cpu_usage=1
show_cpu_frequency=1
show_cpu_temperature=1
degree_fahrenheit=0
update_process_names=1
account_guest_in_cpu_meter=0
color_scheme=0
enable_mouse=1
delay=10
left_meters=LeftCPUs2 Memory Swap
left_meter_modes=1 1 1
right_meters=RightCPUs2 Tasks LoadAverage Uptime
right_meter_modes=1 2 2 2
hide_function_bar=0

Create process monitoring script

Build the main monitoring script that analyzes htop output and detects performance issues.

#!/usr/bin/env python3
import subprocess
import json
import time
import re
import os
import logging
from datetime import datetime
from collections import defaultdict

# Configure logging
logging.basicConfig(
    filename='/var/log/htop-monitor/monitor.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

class ProcessMonitor:
    def __init__(self, config_file='/opt/htop-monitor/config/monitor.json'):
        self.config = self.load_config(config_file)
        self.alerts_sent = defaultdict(int)
        self.last_check = {}
        
    def load_config(self, config_file):
        try:
            with open(config_file, 'r') as f:
                return json.load(f)
        except FileNotFoundError:
            return self.get_default_config()
    
    def get_default_config(self):
        return {
            "thresholds": {
                "cpu_percent": 80.0,
                "memory_percent": 85.0,
                "zombie_count": 5,
                "load_average_1m": 4.0,
                "process_count": 500
            },
            "monitoring": {
                "interval": 30,
                "alert_cooldown": 300,
                "max_alerts_per_hour": 10
            },
            "alerts": {
                "email_enabled": True,
                "webhook_enabled": True,
                "webhook_url": "http://localhost:8080/webhook",
                "email_recipient": "admin@example.com"
            }
        }
    
    def get_system_stats(self):
        """Extract system statistics using htop batch mode"""
        try:
            # Use htop in batch mode with our custom config
            env = os.environ.copy()
            env['HTOPRC'] = '/opt/htop-monitor/config/htoprc'
            
            result = subprocess.run(
                ['htop', '-d', '1', '-n', '1', '--no-color'],
                capture_output=True, text=True, env=env, timeout=10
            )
            
            if result.returncode != 0:
                logging.error(f"htop failed: {result.stderr}")
                return None
                
            return self.parse_htop_output(result.stdout)
        except subprocess.TimeoutExpired:
            logging.error("htop command timed out")
            return None
        except Exception as e:
            logging.error(f"Error running htop: {e}")
            return None
    
    def parse_htop_output(self, output):
        """Parse htop output to extract metrics"""
        lines = output.strip().split('\n')
        stats = {
            'timestamp': datetime.now().isoformat(),
            'processes': [],
            'system': {},
            'alerts': []
        }
        
        # Parse system information from header
        for line in lines[:10]:  # First 10 lines usually contain system info
            if 'Tasks:' in line:
                match = re.search(r'Tasks:\s+(\d+)\s+total', line)
                if match:
                    stats['system']['total_tasks'] = int(match.group(1))
                    
            elif 'Load average:' in line:
                match = re.search(r'Load average:\s+([\d.]+)', line)
                if match:
                    stats['system']['load_average'] = float(match.group(1))
                    
            elif 'CPU[s]:' in line or 'Cpu(s):' in line:
                # Extract CPU usage
                match = re.search(r'([\d.]+)%\s*us', line)
                if match:
                    stats['system']['cpu_user'] = float(match.group(1))
                    
            elif 'Mem:' in line or 'Memory:' in line:
                # Extract memory usage
                match = re.search(r'([\d.]+)G\s*used', line)
                if match:
                    stats['system']['memory_used_gb'] = float(match.group(1))
        
        # Parse process information
        process_header_found = False
        for line in lines:
            if 'PID' in line and 'USER' in line and 'CPU%' in line:
                process_header_found = True
                continue
                
            if process_header_found and line.strip():
                process_data = self.parse_process_line(line)
                if process_data:
                    stats['processes'].append(process_data)
        
        return stats
    
    def parse_process_line(self, line):
        """Parse individual process line from htop output"""
        parts = line.split()
        if len(parts) < 8:
            return None
            
        try:
            return {
                'pid': int(parts[0]),
                'user': parts[1],
                'cpu_percent': float(parts[2]) if parts[2] != '-' else 0.0,
                'memory_percent': float(parts[3]) if parts[3] != '-' else 0.0,
                'state': parts[7] if len(parts) > 7 else 'unknown',
                'command': ' '.join(parts[8:]) if len(parts) > 8 else 'unknown'
            }
        except (ValueError, IndexError):
            return None
    
    def check_thresholds(self, stats):
        """Check if any metrics exceed configured thresholds"""
        alerts = []
        thresholds = self.config['thresholds']
        
        # Check system-wide metrics
        if 'load_average' in stats['system']:
            if stats['system']['load_average'] > thresholds['load_average_1m']:
                alerts.append({
                    'type': 'high_load',
                    'severity': 'warning',
                    'value': stats['system']['load_average'],
                    'threshold': thresholds['load_average_1m'],
                    'message': f"High load average: {stats['system']['load_average']}"
                })
        
        # Check for zombie processes
        zombie_count = sum(1 for p in stats['processes'] if p['state'] == 'Z')
        if zombie_count > thresholds['zombie_count']:
            alerts.append({
                'type': 'zombie_processes',
                'severity': 'critical',
                'value': zombie_count,
                'threshold': thresholds['zombie_count'],
                'message': f"High zombie process count: {zombie_count}"
            })
        
        # Check for high CPU processes
        high_cpu_processes = [p for p in stats['processes'] 
                            if p['cpu_percent'] > thresholds['cpu_percent']]
        if high_cpu_processes:
            alerts.append({
                'type': 'high_cpu_processes',
                'severity': 'warning',
                'value': len(high_cpu_processes),
                'processes': high_cpu_processes[:5],  # Top 5 processes
                'message': f"Found {len(high_cpu_processes)} high CPU processes"
            })
        
        # Check for high memory processes
        high_mem_processes = [p for p in stats['processes'] 
                            if p['memory_percent'] > thresholds['memory_percent']]
        if high_mem_processes:
            alerts.append({
                'type': 'high_memory_processes',
                'severity': 'warning',
                'value': len(high_mem_processes),
                'processes': high_mem_processes[:5],  # Top 5 processes
                'message': f"Found {len(high_mem_processes)} high memory processes"
            })
        
        return alerts
    
    def send_alerts(self, alerts):
        """Send alerts via configured channels"""
        if not alerts:
            return
            
        for alert in alerts:
            alert_key = f"{alert['type']}_{alert.get('value', '')}"
            current_time = time.time()
            
            # Check cooldown period
            if (alert_key in self.last_check and 
                current_time - self.last_check[alert_key] < self.config['monitoring']['alert_cooldown']):
                continue
                
            # Check rate limiting
            if self.alerts_sent[alert_key] >= self.config['monitoring']['max_alerts_per_hour']:
                continue
                
            # Send email alert
            if self.config['alerts']['email_enabled']:
                self.send_email_alert(alert)
                
            # Send webhook alert
            if self.config['alerts']['webhook_enabled']:
                self.send_webhook_alert(alert)
                
            self.last_check[alert_key] = current_time
            self.alerts_sent[alert_key] += 1
            
            logging.info(f"Alert sent: {alert['message']}")
    
    def send_email_alert(self, alert):
        """Send email alert using system mail"""
        try:
            subject = f"htop Alert: {alert['type']}"
            body = f"""
Alert Details:
Type: {alert['type']}
Severity: {alert['severity']}
Message: {alert['message']}
Timestamp: {datetime.now().isoformat()}

Value: {alert.get('value', 'N/A')}
Threshold: {alert.get('threshold', 'N/A')}
"""
            
            if 'processes' in alert:
                body += "\nTop Processes:\n"
                for proc in alert['processes']:
                    body += f"  PID {proc['pid']}: {proc['command']} (CPU: {proc['cpu_percent']}%, MEM: {proc['memory_percent']}%)\n"
            
            cmd = ['mail', '-s', subject, self.config['alerts']['email_recipient']]
            subprocess.run(cmd, input=body, text=True, check=True)
            
        except Exception as e:
            logging.error(f"Failed to send email alert: {e}")
    
    def send_webhook_alert(self, alert):
        """Send webhook alert"""
        try:
            payload = {
                'timestamp': datetime.now().isoformat(),
                'hostname': subprocess.check_output(['hostname'], text=True).strip(),
                'alert': alert
            }
            
            cmd = ['curl', '-X', 'POST', 
                   '-H', 'Content-Type: application/json',
                   '-d', json.dumps(payload),
                   self.config['alerts']['webhook_url']]
            
            subprocess.run(cmd, timeout=10, check=True)
            
        except Exception as e:
            logging.error(f"Failed to send webhook alert: {e}")
    
    def save_stats(self, stats):
        """Save statistics to data file for historical analysis"""
        data_file = f"/opt/htop-monitor/data/stats_{datetime.now().strftime('%Y%m%d')}.jsonl"
        
        try:
            with open(data_file, 'a') as f:
                f.write(json.dumps(stats) + '\n')
        except Exception as e:
            logging.error(f"Failed to save stats: {e}")
    
    def run_monitor_cycle(self):
        """Run one complete monitoring cycle"""
        stats = self.get_system_stats()
        if not stats:
            return
            
        alerts = self.check_thresholds(stats)
        stats['alerts'] = alerts
        
        self.send_alerts(alerts)
        self.save_stats(stats)
        
        logging.info(f"Monitor cycle completed. Found {len(alerts)} alerts.")

if __name__ == '__main__':
    monitor = ProcessMonitor()
    monitor.run_monitor_cycle()

Create monitoring configuration

Set up the JSON configuration file with customizable thresholds and alert settings.

{
  "thresholds": {
    "cpu_percent": 80.0,
    "memory_percent": 85.0,
    "zombie_count": 5,
    "load_average_1m": 4.0,
    "process_count": 500,
    "disk_io_wait": 20.0
  },
  "monitoring": {
    "interval": 30,
    "alert_cooldown": 300,
    "max_alerts_per_hour": 10,
    "retention_days": 7,
    "batch_size": 100
  },
  "alerts": {
    "email_enabled": true,
    "webhook_enabled": true,
    "webhook_url": "http://localhost:8080/webhook",
    "email_recipient": "admin@example.com",
    "slack_webhook": "",
    "teams_webhook": ""
  },
  "filters": {
    "exclude_processes": ["kthreadd", "ksoftirqd", "migration"],
    "include_only_users": [],
    "minimum_cpu_threshold": 0.1,
    "minimum_memory_threshold": 0.1
  },
  "custom_columns": {
    "show_network_io": true,
    "show_disk_io": true,
    "show_file_descriptors": true,
    "show_cpu_affinity": false
  }
}

Create process analysis script

Build an advanced analysis script that provides trend analysis and performance insights.

#!/usr/bin/env python3
import json
import os
import glob
import argparse
import statistics
from datetime import datetime, timedelta
from collections import defaultdict, Counter

class ProcessAnalyzer:
    def __init__(self, data_dir='/opt/htop-monitor/data'):
        self.data_dir = data_dir
        
    def load_data_files(self, days_back=7):
        """Load process data from the last N days"""
        end_date = datetime.now()
        start_date = end_date - timedelta(days=days_back)
        
        data = []
        
        for day_offset in range(days_back + 1):
            date = start_date + timedelta(days=day_offset)
            date_str = date.strftime('%Y%m%d')
            pattern = f"{self.data_dir}/stats_{date_str}.jsonl"
            
            for file_path in glob.glob(pattern):
                try:
                    with open(file_path, 'r') as f:
                        for line in f:
                            if line.strip():
                                data.append(json.loads(line))
                except Exception as e:
                    print(f"Error loading {file_path}: {e}")
        
        return sorted(data, key=lambda x: x['timestamp'])
    
    def analyze_cpu_trends(self, data):
        """Analyze CPU usage trends across processes"""
        cpu_by_process = defaultdict(list)
        cpu_by_hour = defaultdict(list)
        
        for entry in data:
            timestamp = datetime.fromisoformat(entry['timestamp'])