automatic DB maintenance: log rotation, orphan cleanup, VACUUM
The existing daily cleanup never effectively ran: it was an interval job counted from container start (reset on every deploy), it called asyncio.get_event_loop() from a worker thread, and even when deleting, SQLite never returns freed pages to the filesystem. Live DB: 4.1 GB, of which 2.6 GB log and 0.6 GB unused pages. New maintenance service, scheduled daily at MAINTENANCE_HOUR_UTC:30 and once 5 minutes after every start: - delete log entries older than LOG_RETENTION_DAYS (30) and keep at most LOG_MAX_ENTRIES (200k) newest ones - delete processed_mails markers of deleted rules/accounts (FKs are not enforced by SQLite) - VACUUM when >25% of the file is unused, only if free disk space covers the copy; never runs twice in parallel GET/POST /api/logs/maintenance expose status and a manual run; the log menu shows DB size, entry count and last run, with "Datenbank aufräumen & verkleinern". Removed the dead cleanup_old_logs. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -1,5 +1,4 @@
|
||||
import logging
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
@@ -50,17 +49,3 @@ def write_log(
|
||||
if close_db:
|
||||
db.close()
|
||||
|
||||
|
||||
def cleanup_old_logs(days: int = 30, db: Session | None = None) -> int:
|
||||
close_db = False
|
||||
if db is None:
|
||||
db = SessionLocal()
|
||||
close_db = True
|
||||
try:
|
||||
cutoff = datetime.utcnow() - timedelta(days=days)
|
||||
count = db.query(FilterLog).filter(FilterLog.created_at < cutoff).delete()
|
||||
db.commit()
|
||||
return count
|
||||
finally:
|
||||
if close_db:
|
||||
db.close()
|
||||
|
||||
@@ -0,0 +1,141 @@
|
||||
"""Datenbank-Wartung: Log-Rotation, verwaiste Marker entfernen und bei Bedarf
|
||||
VACUUM, damit die SQLite-Datei nicht unbegrenzt wächst.
|
||||
|
||||
SQLite gibt durch DELETE frei gewordenen Platz nicht ans Dateisystem zurück —
|
||||
die Datei schrumpft erst durch VACUUM. Das kopiert die Datenbank einmal neu
|
||||
und braucht dafür kurzzeitig freien Platz in Größe der verbleibenden Daten."""
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import threading
|
||||
import time
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from sqlalchemy import text
|
||||
|
||||
from app.config import settings
|
||||
from app.database import engine
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# VACUUM nur, wenn sich das lohnt: mindestens so viel Anteil bzw. Größe frei
|
||||
VACUUM_MIN_FREE_RATIO = 0.25
|
||||
VACUUM_MIN_FREE_BYTES = 50 * 1024 * 1024
|
||||
|
||||
_lock = threading.Lock()
|
||||
last_result: dict | None = None
|
||||
|
||||
|
||||
def _db_path() -> str | None:
|
||||
url = settings.database_url
|
||||
if not url.startswith("sqlite:///"):
|
||||
return None
|
||||
return url.removeprefix("sqlite:///")
|
||||
|
||||
|
||||
def _file_size(path: str | None) -> int:
|
||||
if not path:
|
||||
return 0
|
||||
total = 0
|
||||
for suffix in ("", "-wal"):
|
||||
try:
|
||||
total += os.path.getsize(path + suffix)
|
||||
except OSError:
|
||||
pass
|
||||
return total
|
||||
|
||||
|
||||
def db_status() -> dict:
|
||||
path = _db_path()
|
||||
with engine.connect() as conn:
|
||||
page_size = conn.exec_driver_sql("PRAGMA page_size").scalar()
|
||||
page_count = conn.exec_driver_sql("PRAGMA page_count").scalar()
|
||||
free_pages = conn.exec_driver_sql("PRAGMA freelist_count").scalar()
|
||||
log_rows = conn.execute(text("SELECT COUNT(*) FROM filter_logs")).scalar()
|
||||
processed_rows = conn.execute(text("SELECT COUNT(*) FROM processed_mails")).scalar()
|
||||
disk_free = shutil.disk_usage(os.path.dirname(os.path.abspath(path))).free if path else None
|
||||
return {
|
||||
"file_bytes": _file_size(path),
|
||||
"used_bytes": (page_count - free_pages) * page_size,
|
||||
"free_bytes": free_pages * page_size,
|
||||
"log_rows": log_rows,
|
||||
"processed_rows": processed_rows,
|
||||
"disk_free_bytes": disk_free,
|
||||
"retention_days": settings.log_retention_days,
|
||||
"max_entries": settings.log_max_entries,
|
||||
"last_run": last_result,
|
||||
}
|
||||
|
||||
|
||||
def run_maintenance(force_vacuum: bool = False) -> dict:
|
||||
"""Läuft synchron (aus Thread/Executor aufrufen). Parallel-Läufe werden
|
||||
übersprungen."""
|
||||
global last_result
|
||||
if not _lock.acquire(blocking=False):
|
||||
return {"skipped": "Wartung läuft bereits"}
|
||||
started = time.monotonic()
|
||||
try:
|
||||
path = _db_path()
|
||||
size_before = _file_size(path)
|
||||
result: dict = {"started_at": datetime.utcnow().isoformat(), "size_before": size_before}
|
||||
|
||||
with engine.begin() as conn:
|
||||
cutoff = datetime.utcnow() - timedelta(days=settings.log_retention_days)
|
||||
result["logs_deleted_age"] = conn.execute(
|
||||
text("DELETE FROM filter_logs WHERE created_at < :cutoff"), {"cutoff": cutoff}
|
||||
).rowcount
|
||||
|
||||
# Obergrenze: nur die neuesten N behalten
|
||||
boundary = conn.execute(
|
||||
text("SELECT id FROM filter_logs ORDER BY id DESC LIMIT 1 OFFSET :n"),
|
||||
{"n": settings.log_max_entries},
|
||||
).scalar()
|
||||
result["logs_deleted_cap"] = (
|
||||
conn.execute(text("DELETE FROM filter_logs WHERE id <= :b"), {"b": boundary}).rowcount
|
||||
if boundary is not None else 0
|
||||
)
|
||||
|
||||
# Marker gelöschter Regeln/Konten (SQLite erzwingt die Fremdschlüssel nicht)
|
||||
result["processed_orphans_deleted"] = conn.execute(text(
|
||||
"DELETE FROM processed_mails "
|
||||
"WHERE rule_id NOT IN (SELECT id FROM filter_rules) "
|
||||
"OR account_id NOT IN (SELECT id FROM accounts)"
|
||||
)).rowcount
|
||||
|
||||
with engine.connect() as conn:
|
||||
conn.exec_driver_sql("PRAGMA wal_checkpoint(TRUNCATE)")
|
||||
page_size = conn.exec_driver_sql("PRAGMA page_size").scalar()
|
||||
page_count = conn.exec_driver_sql("PRAGMA page_count").scalar()
|
||||
free_pages = conn.exec_driver_sql("PRAGMA freelist_count").scalar()
|
||||
free_bytes = free_pages * page_size
|
||||
used_bytes = (page_count - free_pages) * page_size
|
||||
worthwhile = free_bytes >= VACUUM_MIN_FREE_BYTES and free_bytes >= VACUUM_MIN_FREE_RATIO * page_count * page_size
|
||||
|
||||
result["vacuum"] = "nicht nötig"
|
||||
if force_vacuum or worthwhile:
|
||||
disk_free = shutil.disk_usage(os.path.dirname(os.path.abspath(path))).free if path else 0
|
||||
# VACUUM schreibt eine Kopie der verbleibenden Daten (plus WAL)
|
||||
if path and disk_free < used_bytes * 1.2:
|
||||
result["vacuum"] = (
|
||||
f"übersprungen: zu wenig Plattenplatz ({disk_free // 2**20} MB frei, "
|
||||
f"~{int(used_bytes * 1.2) // 2**20} MB nötig)"
|
||||
)
|
||||
else:
|
||||
t = time.monotonic()
|
||||
# VACUUM darf nicht in einer Transaktion laufen
|
||||
with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as conn:
|
||||
conn.exec_driver_sql("VACUUM")
|
||||
conn.exec_driver_sql("PRAGMA wal_checkpoint(TRUNCATE)")
|
||||
result["vacuum"] = f"ausgeführt ({time.monotonic() - t:.0f}s)"
|
||||
|
||||
result["size_after"] = _file_size(path)
|
||||
result["duration_s"] = round(time.monotonic() - started, 1)
|
||||
logger.info("DB-Wartung: %s", result)
|
||||
last_result = result
|
||||
return result
|
||||
except Exception as e:
|
||||
logger.exception("DB-Wartung fehlgeschlagen")
|
||||
last_result = {"error": str(e), "started_at": datetime.utcnow().isoformat()}
|
||||
raise
|
||||
finally:
|
||||
_lock.release()
|
||||
@@ -1,6 +1,6 @@
|
||||
import asyncio
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from apscheduler.schedulers.asyncio import AsyncIOScheduler
|
||||
from sqlalchemy.orm import Session
|
||||
@@ -11,7 +11,8 @@ from app.models.db_models import Account, FilterRule, LogLevel, ProcessedMail
|
||||
from app.services.encryption import decrypt
|
||||
from app.services.filter_engine import evaluate_conditions, execute_action
|
||||
from app.services.imap_client import IMAPClient
|
||||
from app.services.log_service import cleanup_old_logs, write_log
|
||||
from app.services.log_service import write_log
|
||||
from app.services.maintenance import run_maintenance
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -329,6 +330,13 @@ async def poll_account(account_id: int) -> None:
|
||||
await asyncio.to_thread(_poll_account_sync, account_id)
|
||||
|
||||
|
||||
async def maintenance_job() -> None:
|
||||
try:
|
||||
await asyncio.to_thread(run_maintenance)
|
||||
except Exception:
|
||||
pass # bereits in run_maintenance geloggt
|
||||
|
||||
|
||||
def add_account_job(account: Account) -> None:
|
||||
job_id = f"poll_account_{account.id}"
|
||||
if scheduler.get_job(job_id):
|
||||
@@ -364,12 +372,23 @@ def start_scheduler() -> None:
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
# Täglicher Cleanup alter Logs
|
||||
# DB-Wartung (Log-Rotation, VACUUM): täglich zu fester Uhrzeit — ein
|
||||
# 24h-Intervall ab Start lief bei häufigen Deploys praktisch nie — und
|
||||
# zusätzlich kurz nach jedem Start.
|
||||
scheduler.add_job(
|
||||
lambda: asyncio.get_event_loop().run_in_executor(None, cleanup_old_logs, 30),
|
||||
"interval",
|
||||
hours=24,
|
||||
id="cleanup_logs",
|
||||
maintenance_job,
|
||||
"cron",
|
||||
hour=settings.maintenance_hour_utc,
|
||||
minute=30,
|
||||
timezone="UTC",
|
||||
id="db_maintenance",
|
||||
replace_existing=True,
|
||||
)
|
||||
scheduler.add_job(
|
||||
maintenance_job,
|
||||
"date",
|
||||
run_date=datetime.now() + timedelta(minutes=5),
|
||||
id="db_maintenance_startup",
|
||||
replace_existing=True,
|
||||
)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user