Phase 7: scheduled (recurring) backups (0.7.0)
- BackupSchedule model + schedule_service: next-run calc (hourly/daily/weekly, UTC), background scheduler loop (lifespan), run-one with retention pruning (keep newest N per stack on the destination), backup_failed notify event. - routers/schedules.py: schedules CRUD + run-now; registered in main.py. - Frontend: api/schedules.ts + Settings → Scheduled backups (list with next/last run + status, enable/disable, run-now, delete; add form with stack/destination/ frequency/time/weekday/retention/volumes). Rough-verified only (per request): py_compile, frontend tsc build, app import (95 routes), next-run math sanity. Full live run to be tested after deploy. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
menzelj
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
7bd449101d
commit
84ef3df59e
+6
-2
@@ -22,6 +22,7 @@ from routers import (
|
||||
editor,
|
||||
images,
|
||||
ports,
|
||||
schedules,
|
||||
settings as settings_router,
|
||||
stacks,
|
||||
system,
|
||||
@@ -29,7 +30,7 @@ from routers import (
|
||||
volumes,
|
||||
ws,
|
||||
)
|
||||
from services import update_service
|
||||
from services import schedule_service, update_service
|
||||
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
logger = logging.getLogger("stackpilot")
|
||||
@@ -45,12 +46,14 @@ async def lifespan(app: FastAPI):
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.warning("Stack discovery failed: %s", exc)
|
||||
update_task = asyncio.create_task(update_service.background_loop())
|
||||
schedule_task = asyncio.create_task(schedule_service.scheduler_loop())
|
||||
logger.info("StackPilot backend ready on port %s", settings.PORT)
|
||||
yield
|
||||
update_task.cancel()
|
||||
schedule_task.cancel()
|
||||
|
||||
|
||||
app = FastAPI(title="StackPilot", version="0.6.0", lifespan=lifespan)
|
||||
app = FastAPI(title="StackPilot", version="0.7.0", lifespan=lifespan)
|
||||
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
@@ -81,6 +84,7 @@ app.include_router(audit.router)
|
||||
app.include_router(settings_router.router)
|
||||
app.include_router(backups.router)
|
||||
app.include_router(destinations.router)
|
||||
app.include_router(schedules.router)
|
||||
app.include_router(agents.router)
|
||||
app.include_router(ws.router)
|
||||
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
from models.agent import Agent
|
||||
from models.audit import AuditLog
|
||||
from models.backup_destination import BackupDestination
|
||||
from models.backup_schedule import BackupSchedule
|
||||
from models.setting import Setting, Webhook
|
||||
from models.stack import Stack
|
||||
from models.template import Template
|
||||
@@ -9,5 +10,5 @@ from models.user import User
|
||||
|
||||
__all__ = [
|
||||
"User", "Stack", "AuditLog", "Template", "Setting", "Webhook", "Agent",
|
||||
"BackupDestination",
|
||||
"BackupDestination", "BackupSchedule",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,80 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from typing import Optional
|
||||
|
||||
from sqlmodel import Field, SQLModel
|
||||
|
||||
|
||||
def _now() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
FREQUENCIES = ["hourly", "daily", "weekly"]
|
||||
|
||||
|
||||
class BackupSchedule(SQLModel, table=True):
|
||||
"""An automatic, recurring backup of a local stack to a destination."""
|
||||
|
||||
id: Optional[int] = Field(default=None, primary_key=True)
|
||||
stack_id: str
|
||||
destination_id: int
|
||||
frequency: str = "daily" # one of FREQUENCIES
|
||||
hour: int = 3 # UTC, used for daily/weekly
|
||||
minute: int = 0
|
||||
weekday: int = 0 # 0=Mon .. 6=Sun, used for weekly
|
||||
include_volumes: bool = True
|
||||
stop_first: bool = True
|
||||
keep: int = 7 # retention: keep newest N for this stack on the dest (0 = all)
|
||||
enabled: bool = True
|
||||
last_run: Optional[datetime] = None
|
||||
last_status: Optional[str] = None # "ok" | "error: ..."
|
||||
next_run: Optional[datetime] = None
|
||||
created_at: datetime = Field(default_factory=_now)
|
||||
|
||||
|
||||
# --- API schemas ---
|
||||
|
||||
|
||||
class ScheduleCreate(SQLModel):
|
||||
stack_id: str
|
||||
destination_id: int
|
||||
frequency: str = "daily"
|
||||
hour: int = 3
|
||||
minute: int = 0
|
||||
weekday: int = 0
|
||||
include_volumes: bool = True
|
||||
stop_first: bool = True
|
||||
keep: int = 7
|
||||
enabled: bool = True
|
||||
|
||||
|
||||
class ScheduleUpdate(SQLModel):
|
||||
destination_id: Optional[int] = None
|
||||
frequency: Optional[str] = None
|
||||
hour: Optional[int] = None
|
||||
minute: Optional[int] = None
|
||||
weekday: Optional[int] = None
|
||||
include_volumes: Optional[bool] = None
|
||||
stop_first: Optional[bool] = None
|
||||
keep: Optional[int] = None
|
||||
enabled: Optional[bool] = None
|
||||
|
||||
|
||||
class ScheduleRead(SQLModel):
|
||||
id: int
|
||||
stack_id: str
|
||||
destination_id: int
|
||||
destination_name: Optional[str]
|
||||
frequency: str
|
||||
hour: int
|
||||
minute: int
|
||||
weekday: int
|
||||
include_volumes: bool
|
||||
stop_first: bool
|
||||
keep: int
|
||||
enabled: bool
|
||||
last_run: Optional[datetime]
|
||||
last_status: Optional[str]
|
||||
next_run: Optional[datetime]
|
||||
created_at: datetime
|
||||
@@ -16,6 +16,7 @@ EVENT_STACK_START = "stack_start"
|
||||
EVENT_STACK_STOP = "stack_stop"
|
||||
EVENT_STACK_ERROR = "stack_error"
|
||||
EVENT_PULL_FAILED = "pull_failed"
|
||||
EVENT_BACKUP_FAILED = "backup_failed"
|
||||
|
||||
ALL_EVENTS = [
|
||||
EVENT_UPDATE_AVAILABLE,
|
||||
@@ -23,6 +24,7 @@ ALL_EVENTS = [
|
||||
EVENT_STACK_STOP,
|
||||
EVENT_STACK_ERROR,
|
||||
EVENT_PULL_FAILED,
|
||||
EVENT_BACKUP_FAILED,
|
||||
]
|
||||
|
||||
WEBHOOK_TYPES = ["ntfy", "discord", "slack", "gotify", "generic"]
|
||||
|
||||
@@ -0,0 +1,156 @@
|
||||
"""Scheduled backup management."""
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Request
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from auth import require_admin
|
||||
from database import get_session
|
||||
from models.backup_destination import BackupDestination
|
||||
from models.backup_schedule import (
|
||||
FREQUENCIES,
|
||||
BackupSchedule,
|
||||
ScheduleCreate,
|
||||
ScheduleRead,
|
||||
ScheduleUpdate,
|
||||
)
|
||||
from models.stack import Stack
|
||||
from models.user import User
|
||||
from services import audit_service, schedule_service
|
||||
|
||||
router = APIRouter(prefix="/api/backups/schedules", tags=["backups"])
|
||||
|
||||
|
||||
def _ip(request: Request) -> str:
|
||||
return request.client.host if request.client else "unknown"
|
||||
|
||||
|
||||
def _to_read(session: Session, s: BackupSchedule) -> ScheduleRead:
|
||||
dest = session.get(BackupDestination, s.destination_id)
|
||||
return ScheduleRead(
|
||||
id=s.id,
|
||||
stack_id=s.stack_id,
|
||||
destination_id=s.destination_id,
|
||||
destination_name=dest.name if dest else None,
|
||||
frequency=s.frequency,
|
||||
hour=s.hour,
|
||||
minute=s.minute,
|
||||
weekday=s.weekday,
|
||||
include_volumes=s.include_volumes,
|
||||
stop_first=s.stop_first,
|
||||
keep=s.keep,
|
||||
enabled=s.enabled,
|
||||
last_run=s.last_run,
|
||||
last_status=s.last_status,
|
||||
next_run=s.next_run,
|
||||
created_at=s.created_at,
|
||||
)
|
||||
|
||||
|
||||
def _get_or_404(session: Session, schedule_id: int) -> BackupSchedule:
|
||||
s = session.get(BackupSchedule, schedule_id)
|
||||
if not s:
|
||||
raise HTTPException(status_code=404, detail=f"Schedule {schedule_id} not found")
|
||||
return s
|
||||
|
||||
|
||||
def _validate(session: Session, stack_id: str, destination_id: int, frequency: str) -> None:
|
||||
if frequency not in FREQUENCIES:
|
||||
raise HTTPException(status_code=400, detail=f"Unknown frequency '{frequency}'")
|
||||
if not session.get(Stack, stack_id):
|
||||
raise HTTPException(status_code=404, detail=f"Stack '{stack_id}' not found")
|
||||
if not session.get(BackupDestination, destination_id):
|
||||
raise HTTPException(status_code=404, detail=f"Destination {destination_id} not found")
|
||||
|
||||
|
||||
@router.get("", response_model=list[ScheduleRead])
|
||||
def list_schedules(
|
||||
session: Session = Depends(get_session),
|
||||
_user: User = Depends(require_admin),
|
||||
) -> list[ScheduleRead]:
|
||||
rows = session.exec(select(BackupSchedule).order_by(BackupSchedule.id)).all()
|
||||
return [_to_read(session, s) for s in rows]
|
||||
|
||||
|
||||
@router.post("", response_model=ScheduleRead, status_code=201)
|
||||
def create_schedule(
|
||||
body: ScheduleCreate,
|
||||
request: Request,
|
||||
session: Session = Depends(get_session),
|
||||
user: User = Depends(require_admin),
|
||||
) -> ScheduleRead:
|
||||
_validate(session, body.stack_id, body.destination_id, body.frequency)
|
||||
s = BackupSchedule(**body.model_dump())
|
||||
s.next_run = schedule_service.compute_next_run(
|
||||
s.frequency, s.hour, s.minute, s.weekday, schedule_service._now()
|
||||
)
|
||||
session.add(s)
|
||||
session.commit()
|
||||
session.refresh(s)
|
||||
audit_service.record(
|
||||
session, user=user.username, action="schedule.create", target=s.stack_id,
|
||||
detail=f"{s.frequency} → dest {s.destination_id}", ip=_ip(request),
|
||||
)
|
||||
return _to_read(session, s)
|
||||
|
||||
|
||||
@router.put("/{schedule_id}", response_model=ScheduleRead)
|
||||
def update_schedule(
|
||||
schedule_id: int,
|
||||
body: ScheduleUpdate,
|
||||
request: Request,
|
||||
session: Session = Depends(get_session),
|
||||
user: User = Depends(require_admin),
|
||||
) -> ScheduleRead:
|
||||
s = _get_or_404(session, schedule_id)
|
||||
data = body.model_dump(exclude_unset=True)
|
||||
for k, v in data.items():
|
||||
setattr(s, k, v)
|
||||
_validate(session, s.stack_id, s.destination_id, s.frequency)
|
||||
# Recompute next run when timing fields change.
|
||||
if {"frequency", "hour", "minute", "weekday"} & set(data) or s.next_run is None:
|
||||
s.next_run = schedule_service.compute_next_run(
|
||||
s.frequency, s.hour, s.minute, s.weekday, schedule_service._now()
|
||||
)
|
||||
session.add(s)
|
||||
session.commit()
|
||||
session.refresh(s)
|
||||
audit_service.record(
|
||||
session, user=user.username, action="schedule.update", target=s.stack_id,
|
||||
ip=_ip(request),
|
||||
)
|
||||
return _to_read(session, s)
|
||||
|
||||
|
||||
@router.delete("/{schedule_id}")
|
||||
def delete_schedule(
|
||||
schedule_id: int,
|
||||
request: Request,
|
||||
session: Session = Depends(get_session),
|
||||
user: User = Depends(require_admin),
|
||||
) -> dict:
|
||||
s = _get_or_404(session, schedule_id)
|
||||
stack_id = s.stack_id
|
||||
session.delete(s)
|
||||
session.commit()
|
||||
audit_service.record(
|
||||
session, user=user.username, action="schedule.delete", target=stack_id,
|
||||
ip=_ip(request),
|
||||
)
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
@router.post("/{schedule_id}/run")
|
||||
async def run_now(
|
||||
schedule_id: int,
|
||||
request: Request,
|
||||
session: Session = Depends(get_session),
|
||||
user: User = Depends(require_admin),
|
||||
) -> dict:
|
||||
s = _get_or_404(session, schedule_id)
|
||||
result = await schedule_service.run_schedule(session, s)
|
||||
audit_service.record(
|
||||
session, user=user.username, action="schedule.run", target=s.stack_id,
|
||||
detail="ok" if result.get("ok") else result.get("error", "error"), ip=_ip(request),
|
||||
)
|
||||
return result
|
||||
@@ -0,0 +1,156 @@
|
||||
"""Scheduled (recurring) stack backups to a destination.
|
||||
|
||||
A background loop wakes every minute, runs any schedules whose ``next_run`` has
|
||||
passed, uploads the backup to the destination, prunes old backups per the
|
||||
retention setting, and records status + the next run time.
|
||||
|
||||
Times are stored as naive UTC to match SQLite's datetime handling.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from database import engine
|
||||
from models.backup_destination import BackupDestination
|
||||
from models.backup_schedule import BackupSchedule
|
||||
from models.setting import EVENT_BACKUP_FAILED
|
||||
from models.stack import Stack
|
||||
from services import (
|
||||
backup_destination_service as dest_service,
|
||||
backup_service,
|
||||
compose_service,
|
||||
notify_service,
|
||||
)
|
||||
|
||||
logger = logging.getLogger("stackpilot.schedule")
|
||||
|
||||
|
||||
def _now() -> datetime:
|
||||
return datetime.now(timezone.utc).replace(tzinfo=None)
|
||||
|
||||
|
||||
def compute_next_run(
|
||||
frequency: str, hour: int, minute: int, weekday: int, after: datetime
|
||||
) -> datetime:
|
||||
"""Next run strictly after ``after`` (naive UTC)."""
|
||||
if frequency == "hourly":
|
||||
nxt = after.replace(minute=minute % 60, second=0, microsecond=0)
|
||||
if nxt <= after:
|
||||
nxt += timedelta(hours=1)
|
||||
return nxt
|
||||
|
||||
base = after.replace(hour=hour % 24, minute=minute % 60, second=0, microsecond=0)
|
||||
if frequency == "weekly":
|
||||
days_ahead = (weekday - base.weekday()) % 7
|
||||
cand = base + timedelta(days=days_ahead)
|
||||
if cand <= after:
|
||||
cand += timedelta(days=7)
|
||||
return cand
|
||||
|
||||
# daily (default)
|
||||
if base <= after:
|
||||
base += timedelta(days=1)
|
||||
return base
|
||||
|
||||
|
||||
def _backup_filename(stack_id: str, include_volumes: bool) -> str:
|
||||
date = compose_service.now().strftime("%Y%m%d-%H%M%S")
|
||||
suffix = "full" if include_volumes else "config"
|
||||
return f"backup-{stack_id}-{suffix}-{date}.tar.gz"
|
||||
|
||||
|
||||
def _prune(dest: BackupDestination, stack_id: str, keep: int) -> int:
|
||||
if keep <= 0:
|
||||
return 0
|
||||
items = dest_service.list_backups(dest)
|
||||
mine = [i for i in items if i["name"].startswith(f"backup-{stack_id}-")]
|
||||
mine.sort(key=lambda x: x.get("modified") or 0, reverse=True)
|
||||
removed = 0
|
||||
for old in mine[keep:]:
|
||||
try:
|
||||
dest_service.delete(dest, old["name"])
|
||||
removed += 1
|
||||
except dest_service.DestinationError as exc:
|
||||
logger.warning("retention delete failed for %s: %s", old["name"], exc)
|
||||
return removed
|
||||
|
||||
|
||||
async def run_schedule(session: Session, schedule: BackupSchedule) -> dict:
|
||||
"""Execute one schedule now. Updates status + next_run. Returns a summary."""
|
||||
now = _now()
|
||||
stack = session.get(Stack, schedule.stack_id)
|
||||
dest = session.get(BackupDestination, schedule.destination_id)
|
||||
result: dict = {"ok": False}
|
||||
path = None
|
||||
try:
|
||||
if not stack:
|
||||
raise RuntimeError(f"stack '{schedule.stack_id}' not found")
|
||||
if not dest:
|
||||
raise RuntimeError(f"destination {schedule.destination_id} not found")
|
||||
path = await backup_service.create_backup(
|
||||
schedule.stack_id, stack.name,
|
||||
include_volumes=schedule.include_volumes, stop_first=schedule.stop_first,
|
||||
)
|
||||
filename = _backup_filename(schedule.stack_id, schedule.include_volumes)
|
||||
await asyncio.to_thread(dest_service.upload, dest, path, filename)
|
||||
pruned = await asyncio.to_thread(_prune, dest, schedule.stack_id, schedule.keep)
|
||||
schedule.last_status = "ok"
|
||||
result = {"ok": True, "name": filename, "destination": dest.name, "pruned": pruned}
|
||||
logger.info("Scheduled backup %s → %s ok (pruned %d)", schedule.stack_id, dest.name, pruned)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
schedule.last_status = f"error: {exc}"[:300]
|
||||
result = {"ok": False, "error": str(exc)}
|
||||
logger.warning("Scheduled backup %s failed: %s", schedule.stack_id, exc)
|
||||
try:
|
||||
await notify_service.notify(
|
||||
EVENT_BACKUP_FAILED,
|
||||
f"Scheduled backup of '{schedule.stack_id}' failed",
|
||||
str(exc),
|
||||
session,
|
||||
)
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
finally:
|
||||
if path and os.path.exists(path):
|
||||
os.unlink(path)
|
||||
schedule.last_run = now
|
||||
schedule.next_run = compute_next_run(
|
||||
schedule.frequency, schedule.hour, schedule.minute, schedule.weekday, now
|
||||
)
|
||||
session.add(schedule)
|
||||
session.commit()
|
||||
session.refresh(schedule)
|
||||
return result
|
||||
|
||||
|
||||
async def run_due() -> None:
|
||||
now = _now()
|
||||
with Session(engine) as session:
|
||||
schedules = session.exec(
|
||||
select(BackupSchedule).where(BackupSchedule.enabled == True) # noqa: E712
|
||||
).all()
|
||||
due = []
|
||||
for s in schedules:
|
||||
if s.next_run is None:
|
||||
s.next_run = compute_next_run(s.frequency, s.hour, s.minute, s.weekday, now)
|
||||
session.add(s)
|
||||
elif s.next_run <= now:
|
||||
due.append(s)
|
||||
session.commit()
|
||||
for s in due:
|
||||
await run_schedule(session, s)
|
||||
|
||||
|
||||
async def scheduler_loop() -> None:
|
||||
await asyncio.sleep(20) # let startup settle
|
||||
while True:
|
||||
try:
|
||||
await run_due()
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.warning("scheduler tick failed: %s", exc)
|
||||
await asyncio.sleep(60)
|
||||
Reference in New Issue
Block a user