95 lines
2.9 KiB
Python
95 lines
2.9 KiB
Python
import logging
|
|
import threading
|
|
import time
|
|
from typing import Any
|
|
|
|
from .backup_alert_engine import check_missed_schedules
|
|
from .settings_store import SettingsStore, get_settings_store
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class BackupAlertPoller:
|
|
def __init__(self) -> None:
|
|
self._thread: threading.Thread | None = None
|
|
self._stop_event = threading.Event()
|
|
self._lock = threading.Lock()
|
|
self._last_run_at: float | None = None
|
|
self._last_success_at: float | None = None
|
|
self._last_error: str = ""
|
|
self._poll_count = 0
|
|
self._error_count = 0
|
|
|
|
def start(self) -> None:
|
|
with self._lock:
|
|
if self._thread and self._thread.is_alive():
|
|
return
|
|
self._stop_event.clear()
|
|
self._thread = threading.Thread(target=self._run, name="backup-alert-poller", daemon=True)
|
|
self._thread.start()
|
|
|
|
def stop(self, timeout: float = 5.0) -> None:
|
|
with self._lock:
|
|
thread = self._thread
|
|
if not thread:
|
|
return
|
|
self._stop_event.set()
|
|
thread.join(timeout=timeout)
|
|
|
|
@property
|
|
def status(self) -> dict[str, Any]:
|
|
with self._lock:
|
|
return {
|
|
"worker_running": self._thread is not None and self._thread.is_alive(),
|
|
"last_run_at": self._last_run_at,
|
|
"last_success_at": self._last_success_at,
|
|
"last_error": self._last_error,
|
|
"poll_count": self._poll_count,
|
|
"error_count": self._error_count,
|
|
}
|
|
|
|
def _run(self) -> None:
|
|
# Wait 30 seconds before first check
|
|
if self._stop_event.wait(30):
|
|
return
|
|
store = get_settings_store()
|
|
while not self._stop_event.is_set():
|
|
try:
|
|
self._run_cycle(store)
|
|
except Exception:
|
|
logger.exception("Backup alert poller cycle failed")
|
|
# Check every 5 minutes
|
|
if self._stop_event.wait(300):
|
|
break
|
|
|
|
def _run_cycle(self, store: SettingsStore) -> None:
|
|
start = time.perf_counter()
|
|
jobs = store.list_backup_jobs()
|
|
existing_alerts = store.list_backup_alerts(acknowledged=False)
|
|
|
|
new_alerts = check_missed_schedules(
|
|
jobs,
|
|
lambda job_id: store.get_latest_backup_run(job_id),
|
|
existing_alerts,
|
|
)
|
|
|
|
for alert in new_alerts:
|
|
store.create_backup_alert(alert)
|
|
|
|
# Prune old resolved alerts (90 days)
|
|
cutoff = int(time.time()) - (90 * 24 * 60 * 60)
|
|
store.prune_backup_alerts(cutoff)
|
|
|
|
_duration_ms = int((time.perf_counter() - start) * 1000)
|
|
with self._lock:
|
|
self._last_run_at = time.time()
|
|
self._last_success_at = time.time()
|
|
self._poll_count += 1
|
|
|
|
|
|
_BACKUP_POLLER = BackupAlertPoller()
|
|
|
|
|
|
def get_backup_poller() -> BackupAlertPoller:
|
|
return _BACKUP_POLLER
|