"""Steward's own backups as a scheduled job: make the file, send it to the bucket, record how it went (#5673). Once a day at the hour chosen in Settings → Backups, in the zone of whoever saved it. A day missed while Steward was down is made up on the next tick; a failed run is tried again an hour later, never sooner, so a target that is down isn't pressed. Each run is kept in `backup.history` (newest first, the last HISTORY_KEEP), audited, and leaves two metrics under source "steward", resource "backup": backup_age_secs (since the last success, every tick while backups are on, so an alert rule can watch it) and backup_bytes. """ from __future__ import annotations import asyncio import logging import tempfile import time from datetime import datetime, timedelta, timezone from pathlib import Path from sqlalchemy import text from sqlalchemy.ext.asyncio import create_async_engine from .dump import BackupError, make_backup from .s3 import S3Error, S3Target, get_file, list_objects, put_file logger = logging.getLogger(__name__) TICK_SECONDS = 15 * 60 RETRY_AFTER = timedelta(hours=1) HISTORY_KEEP = 30 PREFIX = "steward-" # One backup at a time, whether the schedule or a person started it. _running = asyncio.Lock() _background: set[asyncio.Task] = set() async def migration_heads(database_url: str) -> list[str]: """The migration revisions the database is at, for the manifest.""" engine = create_async_engine(database_url) try: async with engine.connect() as conn: return list((await conn.execute(text("SELECT version_num FROM alembic_version"))).scalars()) finally: await engine.dispose() def target_from(settings: dict) -> S3Target | None: """The bucket the settings name, or None while they're incomplete.""" endpoint = (settings.get("backup.s3_endpoint") or "").strip() bucket = (settings.get("backup.s3_bucket") or "").strip() access = (settings.get("backup.s3_access_key") or "").strip() secret = settings.get("backup.s3_secret_key") or "" if not (endpoint and bucket and access and secret): return None return S3Target(endpoint, bucket, access, secret, region=(settings.get("backup.s3_region") or "us-east-1").strip(), verify_tls=bool(settings.get("backup.s3_verify_tls", True))) def _moment(raw) -> datetime | None: try: return datetime.fromisoformat(raw) if raw else None except (TypeError, ValueError): return None def last_slot(now: datetime, hour: int, tz) -> datetime: """The latest scheduled moment at or before `now`: today's hour in `tz`, or yesterday's when today's hasn't come yet.""" local = now.astimezone(tz) slot = local.replace(hour=hour, minute=0, second=0, microsecond=0) if slot > local: slot -= timedelta(days=1) return slot.astimezone(timezone.utc) def is_due(now: datetime, hour: int, tz, last_success: datetime | None, last_attempt: datetime | None) -> bool: """Due when no backup has succeeded since the latest slot, and the last try (if it failed) was at least RETRY_AFTER ago.""" slot = last_slot(now, hour, tz) if last_success is not None and last_success >= slot: return False return last_attempt is None or now - last_attempt >= RETRY_AFTER def history_with(history: list, entry: dict) -> list: return ([entry] + list(history or []))[:HISTORY_KEEP] async def _record(app, entry: dict, user_id: str | None, username: str) -> None: from steward.core.alerts import MetricBatch from steward.core.audit import log_audit from steward.core.settings import get_setting, set_setting async with app.db_sessionmaker() as db: async with db.begin(): history = await get_setting(db, "backup.history") or [] await set_setting(db, "backup.history", history_with(history, entry)) await set_setting(db, "backup.last_attempt_at", entry["at"]) if entry["ok"]: await set_setting(db, "backup.last_success_at", entry["at"]) batch = MetricBatch() batch.add(source_module="steward", resource_name="backup", metric_name="backup_bytes", value=float(entry["size"])) batch.add(source_module="steward", resource_name="backup", metric_name="backup_age_secs", value=0.0) await batch.record(db) await log_audit(app, user_id, username, "backup.made" if entry["ok"] else "backup.failed", entity_type="backup", entity_id=entry.get("name") or None, detail={k: entry[k] for k in ("name", "size", "secs", "error", "by") if entry.get(k)}) async def run_backup(app, by: str = "schedule", user_id: str | None = None) -> dict: """Make one backup and send it. Returns the history entry it recorded.""" from steward.core.settings import get_all_settings async with _running: started = time.monotonic() at = datetime.now(timezone.utc) entry = {"at": at.isoformat(), "ok": False, "by": by, "name": "", "size": 0, "secs": 0.0, "error": ""} async with app.db_sessionmaker() as db: settings = await get_all_settings(db) target = target_from(settings) try: if target is None: raise BackupError("The bucket isn't set up: fill in the endpoint, bucket and keys.") db_url = app.config["DATABASE_URL"] heads = await migration_heads(db_url) with tempfile.TemporaryDirectory(prefix="steward-backup-") as tmp: made = await make_backup(db_url, app.config["SECRET_KEY"], heads, Path(tmp), now=at) await put_file(target, made.name, made.path) entry.update(ok=True, name=made.name, size=made.size) entry["check"] = await _self_check(app, made.path, made.name) except (BackupError, S3Error) as exc: entry["error"] = str(exc) except Exception as exc: # recorded and logged: a run never takes the scheduler down logger.exception("Backup failed") entry["error"] = f"Unexpected error: {exc}" entry["secs"] = round(time.monotonic() - started, 1) if entry["ok"]: logger.info("Backup %s sent (%d bytes, %.1fs)", entry["name"], entry["size"], entry["secs"]) else: logger.warning("Backup failed: %s", entry["error"]) await _record(app, entry, user_id, by if by != "schedule" else "") return entry async def _self_check(app, path: Path, name: str) -> dict: """A backup just sent, read back and compared with what it was made from (#5823): proof the file restores, kept on its history row. A check that can't run says why; the backup was sent either way.""" try: report = await check_file(app, path, name) except Exception as exc: logger.warning("Backup %s couldn't be checked: %s", name, exc) return {"ok": False, "changes": 0, "rows": 0, "summary": f"Couldn't check it: {exc}"} return report.as_entry() async def check_file(app, path: Path, name: str = ""): """Compare a backup file with this Steward's database (check.py).""" from .check import check_backup return await check_backup(path, app.config["DATABASE_URL"], app.config["SECRET_KEY"], plugin_dirs=[Path(d) for d in app.config["PLUGIN_DIRS"]], name=name or path.name) def is_backup_name(name: str) -> bool: """A file name Steward's backups have, and nothing else: no path, no other object in the bucket.""" import re return bool(re.fullmatch(r"steward-\d{8}T\d{6}Z\.tar", name or "")) async def check_stored(app, name: str): """Fetch one of this Steward's backups from the bucket and check it. Raises BackupError or S3Error with what went wrong.""" from steward.core.settings import get_all_settings if not is_backup_name(name): raise BackupError(f"{name!r} isn't the name of a Steward backup.") async with app.db_sessionmaker() as db: target = target_from(await get_all_settings(db)) if target is None: raise BackupError("The bucket isn't set up: fill in the endpoint, bucket and keys.") with tempfile.TemporaryDirectory(prefix="steward-check-") as tmp: path = Path(tmp) / name await get_file(target, name, path) return await check_file(app, path, name) def start_backup(app, by: str, user_id: str | None) -> bool: """Start a backup in the background for a person; False if one is running.""" if _running.locked(): return False task = asyncio.get_running_loop().create_task(run_backup(app, by=by, user_id=user_id)) _background.add(task) task.add_done_callback(_background.discard) return True def is_running() -> bool: return _running.locked() async def tick(app, now: datetime | None = None) -> None: """The scheduled task: record how old the newest backup is, and make one when it is due.""" from steward.core.alerts import MetricBatch from steward.core.formatting import zone from steward.core.settings import get_all_settings now = now or datetime.now(timezone.utc) async with app.db_sessionmaker() as db: settings = await get_all_settings(db) if not settings.get("backup.enabled"): return last_success = _moment(settings.get("backup.last_success_at")) if last_success is not None: async with app.db_sessionmaker() as db: async with db.begin(): batch = MetricBatch() batch.add(source_module="steward", resource_name="backup", metric_name="backup_age_secs", value=(now - last_success).total_seconds()) await batch.record(db) if _running.locked(): return if is_due(now, int(settings.get("backup.schedule_hour", 3)), zone(settings.get("backup.schedule_tz")), last_success, _moment(settings.get("backup.last_attempt_at"))): await run_backup(app) async def test_target(target: S3Target) -> tuple[bool, str]: """Settings → Backups' Test button: can Steward reach the bucket with these keys? A listing refused as AccessDenied still proves the keys: an append-only key may not list, and uploads anyway.""" try: found = await list_objects(target, PREFIX, limit=1000) except S3Error as exc: said = str(exc) if "AccessDenied" in said: return True, ("Reached the bucket and the keys were accepted. They may not list " "what is there, which an append-only key can't; uploads still work.") return False, said if not found: return True, f"Reached {target.bucket}; no Steward backups there yet." newest = max(found, key=lambda o: o.key) return True, f"Reached {target.bucket}: {len(found)} Steward backups, the newest {newest.key}." # ── What Settings → Backups shows ──────────────────────────────────────────── def shown_hour(hour: int, saved_tz, view_tz, now: datetime | None = None) -> int: """The schedule's hour as a reader in `view_tz` sees it today.""" local = (now or datetime.now(timezone.utc)).astimezone(saved_tz) return local.replace(hour=hour, minute=0, second=0, microsecond=0).astimezone(view_tz).hour def history_rows(history: list) -> list[dict]: """The stored history with its times read back as datetimes.""" return [entry | {"at": _moment(entry.get("at"))} for entry in history or []] def verdict(settings: dict, running: bool) -> dict: """One line for the top of the page: {"tone": ok|warn|bad|off, "text", "moment"}. `moment` is the time the line is about, shown as "ago" by the template, which ends the sentence; a failure also carries `failed`, its `error` and `good` (the newest success, or None).""" rows = history_rows(settings.get("backup.history") or []) last_ok = _moment(settings.get("backup.last_success_at")) if running: return {"tone": "ok", "text": "A backup is being made now", "moment": None} if not settings.get("backup.enabled"): if target_from(settings) is None: return {"tone": "off", "text": "Backups are off, and no bucket is set up", "moment": None} return {"tone": "off", "text": "Scheduled backups are off; Back up now still works", "moment": None} if target_from(settings) is None: return {"tone": "bad", "text": "Backups are on, but the bucket isn't set up", "moment": None} if rows and not rows[0]["ok"]: # `good`: when the newest backup that did go was sent (None: never). return {"tone": "bad", "text": "The last backup failed", "moment": rows[0]["at"], "error": rows[0]["error"], "good": last_ok, "failed": True} if last_ok is None: return {"tone": "warn", "text": "No backup has been sent yet; the first goes within 15 minutes", "moment": None} return {"tone": "ok", "text": f"The newest backup is {rows[0]['name']} ({_size(rows[0]['size'])}), sent", "moment": last_ok} def _size(n) -> str: from steward.core.formatting import human_bytes return human_bytes(n)