Tyler
47e0f80786
Cyclone previously emitted stdlib default-formatted log lines that
operators couldn't parse with anything beyond grep. SP18 replaces
that with:
- JsonFormatter: newline-delimited JSON, ISO-8601 ms timestamps,
structured "extra" dict, exception tracebacks serialized as a
single string.
- CycloneDevFormatter: tabular format for tail -f in dev.
- PiiScrubber: logging.Filter that redacts NPIs, SSNs, DOBs,
patient names — both inline in the message ("npi 1881068062")
and via PHI-keyed extras ({"dob": "1980-04-12"}).
- setup_logging(): idempotent entry point used by the API lifespan
and CLI main; respects CYCLONE_LOG_LEVEL/FILE/JSON/NO_PII_SCRUB.
- CLI --log-format=json|dev + --log-file=… flags.
- Migrate 9 highest-value log sites in scheduler/backup_scheduler/
backup_service to use extra={...} (input_filename, claims, parser,
backup_id, db_fingerprint, etc.).
34 new tests (test_logging_formatter + test_logging_scrubber +
test_logging_setup). All 867 backend tests pass.
369 lines
12 KiB
Python
369 lines
12 KiB
Python
"""SP17 — Backup scheduler.
|
|
|
|
Wraps :class:`cyclone.backup_service.BackupService` in an
|
|
asyncio task, mirroring the MFT scheduler pattern (SP16). A backup
|
|
tick:
|
|
|
|
1. Calls :meth:`BackupService.create_now` to take + encrypt a backup.
|
|
2. Calls :meth:`BackupService.prune` to apply the retention policy.
|
|
3. Writes a tamper-evident ``audit_log`` row (SP11) for each outcome.
|
|
|
|
The scheduler is OFF by default. Operators opt in via
|
|
``CYCLONE_BACKUP_AUTOSTART=true``. The poll interval is
|
|
``CYCLONE_BACKUP_INTERVAL_HOURS`` (default 24).
|
|
|
|
Like the MFT scheduler, this is single-asyncio-task — no
|
|
threading, no APScheduler. All access (start/stop/tick/status)
|
|
must happen on the same event loop; the FastAPI app satisfies
|
|
that trivially because endpoints run on the loop.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import traceback
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
from cyclone import backup_service as svc_mod
|
|
from cyclone.audit_log import AuditEvent, append_event
|
|
from cyclone.backup_service import BackupService
|
|
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class BackupTickResult:
|
|
"""Outcome of a single backup tick (one cycle of create + prune + audit)."""
|
|
|
|
started_at: datetime
|
|
finished_at: Optional[datetime] = None
|
|
created: Optional[svc_mod.BackupRecord] = None
|
|
pruned_paths: list[str] = field(default_factory=list)
|
|
error: Optional[str] = None
|
|
|
|
@property
|
|
def ok(self) -> bool:
|
|
return self.error is None and self.created is not None
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"started_at": self.started_at.isoformat(),
|
|
"finished_at": (
|
|
self.finished_at.isoformat() if self.finished_at else None
|
|
),
|
|
"ok": self.ok,
|
|
"created": (
|
|
{
|
|
"id": self.created.id,
|
|
"filename": self.created.filename,
|
|
"size_bytes": self.created.size_bytes,
|
|
"db_fingerprint": self.created.db_fingerprint,
|
|
"table_count": self.created.table_count,
|
|
}
|
|
if self.created else None
|
|
),
|
|
"pruned_paths": list(self.pruned_paths),
|
|
"error": self.error,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class BackupSchedulerStatus:
|
|
running: bool
|
|
interval_hours: float
|
|
backup_dir: str
|
|
retention_days: int
|
|
last_tick: Optional[BackupTickResult] = None
|
|
tick_count: int = 0
|
|
total_created: int = 0
|
|
total_errors: int = 0
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"running": self.running,
|
|
"interval_hours": self.interval_hours,
|
|
"backup_dir": self.backup_dir,
|
|
"retention_days": self.retention_days,
|
|
"tick_count": self.tick_count,
|
|
"total_created": self.total_created,
|
|
"total_errors": self.total_errors,
|
|
"last_tick": self.last_tick.as_dict() if self.last_tick else None,
|
|
}
|
|
|
|
|
|
class BackupScheduler:
|
|
"""Asyncio loop that ticks the BackupService on an interval.
|
|
|
|
Lifecycle mirrors :class:`cyclone.scheduler.Scheduler`:
|
|
|
|
sched = BackupScheduler(backup_service)
|
|
await sched.start() # begin ticking
|
|
await sched.stop() # finish current tick, exit
|
|
status = sched.status() # snapshot
|
|
|
|
Threading: NOT thread-safe. All access must happen on the
|
|
same event loop. FastAPI endpoints satisfy this automatically.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
service: BackupService,
|
|
*,
|
|
interval_hours: float = 24.0,
|
|
) -> None:
|
|
self._service = service
|
|
self._interval_hours = max(0.1, float(interval_hours))
|
|
self._task: Optional[asyncio.Task[None]] = None
|
|
self._stop_event = asyncio.Event()
|
|
self._tick_in_progress = False
|
|
self._last_tick: Optional[BackupTickResult] = None
|
|
self._tick_count = 0
|
|
self._total_created = 0
|
|
self._total_errors = 0
|
|
|
|
@property
|
|
def service(self) -> BackupService:
|
|
return self._service
|
|
|
|
# ---- Public API -------------------------------------------------------
|
|
|
|
async def start(self) -> None:
|
|
if self._task is not None and not self._task.done():
|
|
log.info("BackupScheduler already running; start() is a no-op")
|
|
return
|
|
self._stop_event.clear()
|
|
self._task = asyncio.create_task(self._run(), name="backup-scheduler")
|
|
log.info(
|
|
"BackupScheduler started",
|
|
extra={
|
|
"interval_hours": self._interval_hours,
|
|
"backup_dir": str(self._service.backup_dir),
|
|
},
|
|
)
|
|
|
|
async def stop(self) -> None:
|
|
if self._task is None or self._task.done():
|
|
return
|
|
self._stop_event.set()
|
|
try:
|
|
await asyncio.wait_for(self._task, timeout=60)
|
|
except asyncio.TimeoutError:
|
|
log.warning("BackupScheduler did not stop within 60s; cancelling")
|
|
self._task.cancel()
|
|
try:
|
|
await self._task
|
|
except (asyncio.CancelledError, Exception): # noqa: BLE001
|
|
pass
|
|
self._task = None
|
|
log.info("BackupScheduler stopped")
|
|
|
|
def status(self) -> BackupSchedulerStatus:
|
|
return BackupSchedulerStatus(
|
|
running=self.is_running(),
|
|
interval_hours=self._interval_hours,
|
|
backup_dir=str(self._service.backup_dir),
|
|
retention_days=self._service._retention_days,
|
|
last_tick=self._last_tick,
|
|
tick_count=self._tick_count,
|
|
total_created=self._total_created,
|
|
total_errors=self._total_errors,
|
|
)
|
|
|
|
def is_running(self) -> bool:
|
|
return self._task is not None and not self._task.done()
|
|
|
|
async def tick(self) -> BackupTickResult:
|
|
"""Run a single backup tick (create + prune + audit).
|
|
|
|
Concurrent ticks are coalesced: if a tick is already in
|
|
progress, the second caller waits for it. This protects
|
|
against a slow backup holding up multiple operator-driven
|
|
``POST /api/admin/backup/tick`` calls.
|
|
"""
|
|
while self._tick_in_progress:
|
|
await asyncio.sleep(0.05)
|
|
self._tick_in_progress = True
|
|
try:
|
|
result = await self._tick_impl()
|
|
self._last_tick = result
|
|
self._tick_count += 1
|
|
if result.created is not None:
|
|
self._total_created += 1
|
|
if result.error is not None:
|
|
self._total_errors += 1
|
|
return result
|
|
finally:
|
|
self._tick_in_progress = False
|
|
|
|
# ---- Internals --------------------------------------------------------
|
|
|
|
async def _run(self) -> None:
|
|
# Stagger the first tick (same rationale as the MFT scheduler).
|
|
await asyncio.sleep(5)
|
|
while not self._stop_event.is_set():
|
|
try:
|
|
await self.tick()
|
|
except Exception as exc: # noqa: BLE001
|
|
# tick() catches its own exceptions and returns them
|
|
# in the result. This is the safety net for
|
|
# programmer errors in the loop body.
|
|
log.exception("BackupScheduler tick raised", extra={"error": str(exc)})
|
|
try:
|
|
await asyncio.wait_for(
|
|
self._stop_event.wait(),
|
|
timeout=self._interval_hours * 3600,
|
|
)
|
|
except asyncio.TimeoutError:
|
|
pass # interval elapsed
|
|
|
|
async def _tick_impl(self) -> BackupTickResult:
|
|
started = datetime.now(timezone.utc)
|
|
result = BackupTickResult(started_at=started)
|
|
try:
|
|
create_result = await asyncio.to_thread(self._service.create_now)
|
|
result.created = create_result.backup
|
|
# Audit event for the created backup.
|
|
await asyncio.to_thread(
|
|
_audit_backup_created,
|
|
create_result.backup.id,
|
|
create_result.backup.db_fingerprint,
|
|
create_result.backup.table_count,
|
|
"backup-scheduler",
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
log.exception("Backup create failed during tick")
|
|
result.error = f"create: {type(exc).__name__}: {exc}"
|
|
await asyncio.to_thread(
|
|
_audit_backup_failed,
|
|
f"create: {type(exc).__name__}: {exc}",
|
|
traceback.format_exc()[-500:],
|
|
"backup-scheduler",
|
|
)
|
|
|
|
try:
|
|
pruned = await asyncio.to_thread(self._service.prune)
|
|
result.pruned_paths = pruned
|
|
if pruned:
|
|
await asyncio.to_thread(
|
|
_audit_backup_pruned, pruned, "backup-scheduler",
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
log.exception("Backup prune failed during tick")
|
|
# Don't clobber the create error if there was one.
|
|
if result.error is None:
|
|
result.error = f"prune: {type(exc).__name__}: {exc}"
|
|
await asyncio.to_thread(
|
|
_audit_backup_failed,
|
|
f"prune: {type(exc).__name__}: {exc}",
|
|
traceback.format_exc()[-500:],
|
|
"backup-scheduler",
|
|
)
|
|
|
|
result.finished_at = datetime.now(timezone.utc)
|
|
return result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Audit helpers (run in a thread so the asyncio loop doesn't block)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _audit_backup_created(
|
|
backup_id: int, db_fingerprint: str, table_count: int, actor: str,
|
|
) -> None:
|
|
from cyclone import db
|
|
with db.SessionLocal()() as s:
|
|
try:
|
|
append_event(s, AuditEvent(
|
|
event_type="db.backup_created",
|
|
entity_type="database",
|
|
entity_id="cyclone.db",
|
|
actor=actor,
|
|
payload={
|
|
"backup_id": backup_id,
|
|
"db_fingerprint": db_fingerprint,
|
|
"table_count": table_count,
|
|
},
|
|
))
|
|
s.commit()
|
|
except Exception: # noqa: BLE001
|
|
log.exception("Failed to write db.backup_created audit event")
|
|
|
|
|
|
def _audit_backup_failed(
|
|
reason: str, traceback_tail: str, actor: str,
|
|
) -> None:
|
|
from cyclone import db
|
|
with db.SessionLocal()() as s:
|
|
try:
|
|
append_event(s, AuditEvent(
|
|
event_type="db.backup_failed",
|
|
entity_type="database",
|
|
entity_id="cyclone.db",
|
|
actor=actor,
|
|
payload={"reason": reason, "traceback_tail": traceback_tail},
|
|
))
|
|
s.commit()
|
|
except Exception: # noqa: BLE001
|
|
log.exception("Failed to write db.backup_failed audit event")
|
|
|
|
|
|
def _audit_backup_pruned(deleted_paths: list[str], actor: str) -> None:
|
|
from cyclone import db
|
|
with db.SessionLocal()() as s:
|
|
try:
|
|
append_event(s, AuditEvent(
|
|
event_type="db.backup_pruned",
|
|
entity_type="database",
|
|
entity_id="cyclone.db",
|
|
actor=actor,
|
|
payload={"deleted_paths": deleted_paths},
|
|
))
|
|
s.commit()
|
|
except Exception: # noqa: BLE001
|
|
log.exception("Failed to write db.backup_pruned audit event")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Module-level singleton
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
_scheduler: Optional[BackupScheduler] = None
|
|
|
|
|
|
def configure_backup_scheduler(
|
|
service: BackupService,
|
|
*,
|
|
interval_hours: float = 24.0,
|
|
) -> BackupScheduler:
|
|
"""Create (or return existing) the module-level BackupScheduler."""
|
|
global _scheduler
|
|
if _scheduler is not None:
|
|
return _scheduler
|
|
hours = float(
|
|
os.environ.get("CYCLONE_BACKUP_INTERVAL_HOURS", interval_hours),
|
|
)
|
|
_scheduler = BackupScheduler(service, interval_hours=hours)
|
|
return _scheduler
|
|
|
|
|
|
def get_backup_scheduler() -> BackupScheduler:
|
|
"""Return the configured BackupScheduler. Raises if not set up."""
|
|
if _scheduler is None:
|
|
raise RuntimeError(
|
|
"backup scheduler not configured; call configure_backup_scheduler() first",
|
|
)
|
|
return _scheduler
|
|
|
|
|
|
def reset_backup_scheduler_for_tests() -> None:
|
|
"""Clear the module-level singleton. Test-only."""
|
|
global _scheduler
|
|
_scheduler = None
|