Job Data Cleanup System¶

import requests
from datetime import datetime, timedelta

def cleanup_old_jobs(base_url="http://localhost:5000", dry_run=True):
    """Trigger job cleanup via API"""

    response = requests.post(
        f"{base_url}/api/cleanup",
        json={
            "dry_run": dry_run,
            "success_retention_hours": 168,  # 7 days
            "failed_retention_hours": 12     # 12 hours
        }
    )

    if response.status_code == 200:
        result = response.json()
        print(f"Cleanup {'preview' if dry_run else 'completed'}:")
        print(f"  Successful jobs cleaned: {result['cleaned_jobs']['successful']}")
        print(f"  Failed jobs cleaned: {result['cleaned_jobs']['failed']}")
        print(f"  Space freed: {result['space_freed']['human_readable']}")
    else:
        print(f"Cleanup failed: {response.text}")

# Preview cleanup
cleanup_old_jobs(dry_run=True)

# Execute cleanup
cleanup_old_jobs(dry_run=False)

import yaml

with open('cleanup_config.yaml', 'r') as f:
    config = yaml.safe_load(f)

# Use configuration
success_retention = config['retention']['successful_jobs']['hours']
failed_retention = config['retention']['failed_jobs']['hours']

# Example monitoring script
import sqlite3
from datetime import datetime, timedelta

def get_cleanup_metrics(db_path="jobs.db"):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    # Jobs cleaned in last 24 hours
    yesterday = datetime.now() - timedelta(hours=24)
    cursor.execute("""
        SELECT COUNT(*), SUM(data_size_mb) 
        FROM jobs 
        WHERE data_cleaned_at > ?
    """, (yesterday.isoformat(),))

    count, size_mb = cursor.fetchone()

    # Disk space usage
    cursor.execute("""
        SELECT 
            status,
            COUNT(*) as count,
            SUM(data_size_mb) as total_mb
        FROM jobs
        WHERE data_cleaned_at IS NULL
        GROUP BY status
    """)

    usage_by_status = cursor.fetchall()

    print(f"Cleanup Metrics (Last 24h):")
    print(f"  Jobs cleaned: {count or 0}")
    print(f"  Space freed: {(size_mb or 0) / 1024:.1f} GB")
    print(f"\nCurrent Usage:")
    for status, count, mb in usage_by_status:
        print(f"  {status}: {count} jobs, {mb/1024:.1f} GB")

    conn.close()

# Run metrics check
get_cleanup_metrics()

# Use parallel deletion
python scripts/cleanup_jobs.py --parallel --workers 4

# Process in smaller batches
python scripts/cleanup_jobs.py --batch-size 50

# Skip large directories first
python scripts/cleanup_jobs.py --skip-large --size-threshold 10G

# Two-stage cleanup process
def staged_cleanup():
    # Stage 1: Mark for deletion
    mark_jobs_for_cleanup(age_days=7)

    # Stage 2: Delete marked jobs (24h later)
    delete_marked_jobs(marked_before=24)

import matplotlib.pyplot as plt
from datetime import datetime, timedelta

def plot_storage_trends():
    # Get historical data
    dates, usage = get_storage_history()

    # Plot trend
    plt.figure(figsize=(10, 6))
    plt.plot(dates, usage)
    plt.xlabel('Date')
    plt.ylabel('Storage (GB)')
    plt.title('EEMT Storage Usage Trend')
    plt.grid(True)
    plt.savefig('storage_trend.png')

import requests
import json

def send_slack_notification(webhook_url, message):
    payload = {
        "text": f"EEMT Cleanup Report",
        "attachments": [{
            "color": "good",
            "fields": [
                {"title": "Jobs Cleaned", "value": message['jobs_cleaned'], "short": True},
                {"title": "Space Freed", "value": message['space_freed'], "short": True},
                {"title": "Status", "value": "Success", "short": True}
            ]
        }]
    }
    requests.post(webhook_url, json=payload)

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def send_cleanup_report(recipient, stats):
    msg = MIMEMultipart()
    msg['Subject'] = 'EEMT Daily Cleanup Report'
    msg['From'] = 'eemt@example.com'
    msg['To'] = recipient

    body = f"""
    Daily Cleanup Report
    ====================

    Jobs Cleaned: {stats['jobs_cleaned']}
    Space Freed: {stats['space_freed']}

    Current Storage Usage: {stats['current_usage']}
    Available Space: {stats['available_space']}

    Next cleanup scheduled for: {stats['next_run']}
    """

    msg.attach(MIMEText(body, 'plain'))

    with smtplib.SMTP('localhost') as server:
        server.send_message(msg)

from prometheus_client import Counter, Gauge, Histogram

# Define metrics
cleanup_runs = Counter('eemt_cleanup_runs_total', 'Total cleanup runs')
jobs_cleaned = Counter('eemt_jobs_cleaned_total', 'Total jobs cleaned', ['status'])
space_freed = Counter('eemt_space_freed_bytes_total', 'Total space freed')
cleanup_duration = Histogram('eemt_cleanup_duration_seconds', 'Cleanup duration')
current_usage = Gauge('eemt_storage_usage_bytes', 'Current storage usage')

# Update metrics during cleanup
@cleanup_duration.time()
def run_cleanup():
    cleanup_runs.inc()
    # ... cleanup logic ...
    jobs_cleaned.labels(status='completed').inc(3)
    space_freed.inc(45 * 1024 * 1024 * 1024)  # 45 GB
    current_usage.set(get_current_usage())

from functools import wraps
from flask import request, abort

def require_admin(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        auth = request.headers.get('Authorization')
        if not verify_admin_token(auth):
            abort(403)
        return f(*args, **kwargs)
    return decorated_function

@app.route('/api/cleanup', methods=['POST'])
@require_admin
def api_cleanup():
    # Only admins can trigger cleanup
    return run_cleanup()

import logging
import json
from datetime import datetime

class AuditLogger:
    def __init__(self, log_file='cleanup_audit.log'):
        self.logger = logging.getLogger('cleanup_audit')
        handler = logging.FileHandler(log_file)
        formatter = logging.Formatter(
            '%(asctime)s - %(levelname)s - %(message)s'
        )
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)

    def log_cleanup(self, user, action, details):
        audit_entry = {
            'timestamp': datetime.now().isoformat(),
            'user': user,
            'action': action,
            'details': details,
            'ip_address': request.remote_addr
        }
        self.logger.info(json.dumps(audit_entry))

def safe_delete(path, job_id):
    """Safely delete files with verification"""

    # Verify path is within allowed directories
    allowed_dirs = ['/app/uploads', '/app/results']
    if not any(str(path).startswith(d) for d in allowed_dirs):
        raise ValueError(f"Attempted to delete outside allowed directories: {path}")

    # Verify job ownership
    if not verify_job_ownership(job_id, path):
        raise ValueError(f"Path does not belong to job {job_id}")

    # Create backup before deletion (optional)
    if ENABLE_CLEANUP_BACKUP:
        backup_path = f"/backup/{job_id}/{path.name}"
        shutil.copy2(path, backup_path)

    # Perform deletion
    if path.is_dir():
        shutil.rmtree(path)
    else:
        path.unlink()

    return True