Vacuum systemd journal

vacuum-systemd-journal

#!/usr/bin/env python3
"""
Interactive System Journal Vacuum Tool
This script allows you to safely clean systemd journal logs interactively.
"""

import subprocess
import sys
import os
from datetime import datetime, timedelta

class JournalVacuum:
    def __init__(self):
        self.colors = {
            'red': '\033[91m',
            'green': '\033[92m',
            'yellow': '\033[93m',
            'blue': '\033[94m',
            'purple': '\033[95m',
            'cyan': '\033[96m',
            'white': '\033[97m',
            'reset': '\033[0m',
            'bold': '\033[1m'
        }
    
    def print_colored(self, text, color='white', bold=False):
        """Print colored text"""
        prefix = self.colors.get(color, '')
        if bold:
            prefix += self.colors['bold']
        print(f"{prefix}{text}{self.colors['reset']}")
    
    def run_command(self, cmd, sudo=True):
        """Run a shell command and return output"""
        if sudo and os.geteuid() != 0:
            cmd = ['sudo'] + cmd if isinstance(cmd, list) else f"sudo {cmd}"
        
        try:
            if isinstance(cmd, list):
                result = subprocess.run(cmd, capture_output=True, text=True, check=True)
            else:
                result = subprocess.run(cmd, shell=True, capture_output=True, text=True, check=True)
            return result.stdout.strip(), result.stderr.strip(), True
        except subprocess.CalledProcessError as e:
            return e.stdout.strip() if e.stdout else "", e.stderr.strip() if e.stderr else str(e), False
    
    def check_journal_status(self):
        """Check current journal size and usage"""
        self.print_colored("\n📊 Journal Status:", 'cyan', bold=True)
        
        # Get disk usage
        cmd = "journalctl --disk-usage"
        output, error, success = self.run_command(cmd, sudo=False)
        if success:
            self.print_colored(f"   {output}", 'white')
        
        # Get time range of logs
        cmd = "journalctl --list-boots"
        output, error, success = self.run_command(cmd, sudo=False)
        if success and output:
            lines = output.split('\n')
            if lines:
                first_boot = lines[0].split()[0] if lines[0] else 'N/A'
                last_boot = lines[-1].split()[0] if lines[-1] else 'N/A'
                self.print_colored(f"   Number of boots: {len(lines)}", 'white')
        
        # Get current journal limits
        cmd = "journalctl --verify"
        output, error, success = self.run_command(cmd, sudo=False)
        
        return True
    
    def vacuum_by_size(self):
        """Vacuum journal by specifying maximum size"""
        self.print_colored("\n🔧 Vacuum by Size", 'yellow', bold=True)
        self.print_colored("Specify maximum journal size (e.g., 100M, 1G, 500M)", 'cyan')
        
        while True:
            size = input(f"{self.colors['green']}Enter max size: {self.colors['reset']}").strip()
            if not size:
                return
            
            # Validate size format
            if not (size.endswith(('M', 'G', 'K')) and size[:-1].replace('.', '').isdigit()):
                self.print_colored("❌ Invalid format! Use format like: 100M, 1G, 500K", 'red')
                continue
            
            confirm = input(f"⚠️  Vacuum journal to maximum {size}? (y/N): ").strip().lower()
            if confirm == 'y':
                self.print_colored(f"\n🔄 Vacuuming journal to {size}...", 'yellow')
                cmd = f"journalctl --vacuum-size={size}"
                output, error, success = self.run_command(cmd)
                
                if success:
                    self.print_colored("✅ Journal vacuumed successfully!", 'green')
                    self.print_colored(f"   {output}", 'white')
                else:
                    self.print_colored(f"❌ Error: {error}", 'red')
                break
            else:
                break
    
    def vacuum_by_time(self):
        """Vacuum journal by keeping only recent logs"""
        self.print_colored("\n🔧 Vacuum by Time", 'yellow', bold=True)
        self.print_colored("Keep logs from the last (e.g., 2d, 1w, 3months)", 'cyan')
        self.print_colored("Examples: 1d (day), 2w (weeks), 3months, 1year", 'cyan')
        
        while True:
            time_input = input(f"{self.colors['green']}Keep logs from last: {self.colors['reset']}").strip()
            if not time_input:
                return
            
            # Validate time format
            time_lower = time_input.lower()
            valid_suffixes = ['d', 'day', 'days', 'w', 'week', 'weeks', 
                            'month', 'months', 'year', 'years']
            
            if not any(time_lower.endswith(suffix) for suffix in valid_suffixes):
                self.print_colored("❌ Invalid format! Use: 1d, 2w, 3months, 1year", 'red')
                continue
            
            confirm = input(f"⚠️  Keep only logs from last {time_input}? (y/N): ").strip().lower()
            if confirm == 'y':
                self.print_colored(f"\n🔄 Vacuuming journal to last {time_input}...", 'yellow')
                cmd = f"journalctl --vacuum-time={time_input}"
                output, error, success = self.run_command(cmd)
                
                if success:
                    self.print_colored("✅ Journal vacuumed successfully!", 'green')
                    self.print_colored(f"   {output}", 'white')
                else:
                    self.print_colored(f"❌ Error: {error}", 'red')
                break
            else:
                break
    
    def vacuum_by_files(self):
        """Vacuum by keeping only specified number of journal files"""
        self.print_colored("\n🔧 Vacuum by File Count", 'yellow', bold=True)
        self.print_colored("Keep only the last N journal files", 'cyan')
        
        while True:
            try:
                count = input(f"{self.colors['green']}Number of files to keep: {self.colors['reset']}").strip()
                if not count:
                    return
                
                count_int = int(count)
                if count_int <= 0:
                    self.print_colored("❌ Please enter a positive number!", 'red')
                    continue
                
                confirm = input(f"⚠️  Keep only {count_int} journal files? (y/N): ").strip().lower()
                if confirm == 'y':
                    self.print_colored(f"\n🔄 Vacuuming journal to {count_int} files...", 'yellow')
                    cmd = f"journalctl --vacuum-files={count_int}"
                    output, error, success = self.run_command(cmd)
                    
                    if success:
                        self.print_colored("✅ Journal vacuumed successfully!", 'green')
                        self.print_colored(f"   {output}", 'white')
                    else:
                        self.print_colored(f"❌ Error: {error}", 'red')
                    break
                else:
                    break
            except ValueError:
                self.print_colored("❌ Please enter a valid number!", 'red')
    
    def rotate_logs(self):
        """Rotate logs before vacuuming"""
        self.print_colored("\n🔄 Rotating Journal Logs", 'yellow', bold=True)
        confirm = input("Rotate active journal files? (y/N): ").strip().lower()
        
        if confirm == 'y':
            cmd = "systemctl rotate systemd-journald"
            output, error, success = self.run_command(cmd)
            
            if success:
                self.print_colored("✅ Journal rotated successfully!", 'green')
            else:
                self.print_colored(f"❌ Error: {error}", 'red')
    
    def show_help(self):
        """Show help information"""
        self.print_colored("\n📖 Help Information", 'cyan', bold=True)
        self.print_colored("""
Systemd Journal Vacuum - What it does:
• Removes old journal logs to free up disk space
• Maintains system stability by preventing disk full issues
• Different methods suit different needs:

Size-based vacuum: Keep total journal size under specified limit
Time-based vacuum: Keep logs only from recent time period
File-based vacuum: Keep only specified number of log files

⚠️  Important Notes:
• Run with sudo privileges (script will prompt if needed)
• Cannot recover deleted logs
• System keeps running during vacuum
• Some methods require systemd-journald restart
        """, 'white')
    
    def dry_run(self):
        """Show what would be removed without actually deleting"""
        self.print_colored("\n🔍 Dry Run - Preview vacuum effects", 'cyan', bold=True)
        self.print_colored("This shows what would be removed without actually deleting", 'yellow')
        
        # Get current journal size
        cmd = "journalctl --disk-usage"
        output, error, success = self.run_command(cmd, sudo=False)
        if success:
            self.print_colored(f"\nCurrent usage: {output}", 'white')
        
        self.print_colored("\nTo see actual impact, choose one of the vacuum methods.", 'white')
        self.print_colored("The script will then show you the results.", 'white')
    
    def get_user_choice(self):
        """Get user's menu choice"""
        self.print_colored("\n" + "="*50, 'cyan')
        self.print_colored("   SYSTEM JOURNAL VACUUM TOOL", 'white', bold=True)
        self.print_colored("="*50, 'cyan')
        self.print_colored("\n1. 📊 Check Journal Status", 'green')
        self.print_colored("2. 💾 Vacuum by Size (e.g., 500M, 1G)", 'green')
        self.print_colored("3. ⏰ Vacuum by Time (e.g., 2w, 3months)", 'green')
        self.print_colored("4. 📁 Vacuum by File Count", 'green')
        self.print_colored("5. 🔄 Rotate Logs", 'green')
        self.print_colored("6. 🔍 Dry Run (Preview changes)", 'green')
        self.print_colored("7. 📖 Show Help", 'green')
        self.print_colored("8. 🚪 Exit", 'red')
        self.print_colored("-"*50, 'cyan')
        
        choice = input(f"{self.colors['yellow']}Enter your choice (1-8): {self.colors['reset']}").strip()
        return choice
    
    def run(self):
        """Main interactive loop"""
        # Check if running with proper permissions
        if os.geteuid() != 0:
            self.print_colored("⚠️  Note: Some operations require sudo privileges", 'yellow')
            self.print_colored("The script will use sudo when needed.\n", 'yellow')
        
        while True:
            choice = self.get_user_choice()
            
            if choice == '1':
                self.check_journal_status()
            elif choice == '2':
                self.vacuum_by_size()
            elif choice == '3':
                self.vacuum_by_time()
            elif choice == '4':
                self.vacuum_by_files()
            elif choice == '5':
                self.rotate_logs()
            elif choice == '6':
                self.dry_run()
            elif choice == '7':
                self.show_help()
            elif choice == '8':
                self.print_colored("\n👋 Goodbye!", 'purple', bold=True)
                sys.exit(0)
            else:
                self.print_colored("❌ Invalid choice! Please enter 1-8", 'red')
            
            input(f"\n{self.colors['cyan']}Press Enter to continue...{self.colors['reset']}")

def main():
    """Main entry point"""
    try:
        # Check if systemd-journald is running
        result = subprocess.run(["systemctl", "is-active", "systemd-journald"], 
                              capture_output=True, text=True)
        
        if result.stdout.strip() != "active":
            print("❌ systemd-journald is not active. Make sure you're on a systemd-based system.")
            sys.exit(1)
            
    except FileNotFoundError:
        print("❌ This script requires systemd. It appears systemd is not installed.")
        print("This script is designed for Linux distributions using systemd (Ubuntu 15+, Fedora, RHEL 7+, etc.)")
        sys.exit(1)
    
    # Create and run the vacuum tool
    vacuum = JournalVacuum()
    try:
        vacuum.run()
    except KeyboardInterrupt:
        vacuum.print_colored("\n\n👋 Goodbye!", 'purple', bold=True)
        sys.exit(0)

if __name__ == "__main__":
    main()