Add routes and models for backup healthcheck

pydatalab/pydatalab/backups.py

@@ -5,7 +5,7 @@
 from pathlib import Path
 from typing import Any
 
-from pydatalab.config import CONFIG, BackupStrategy
+from pydatalab.config import CONFIG, BackupHealth, BackupHealthCheck, BackupStrategy
 from pydatalab.logger import LOGGER
 
 
@@ -204,3 +204,41 @@ def create_backup(strategy: BackupStrategy) -> bool:
             sftp.put(snapshot_path, str(strategy.location / snapshot_name), confirm=True)
 
     return True
+
+
+def backup_healthcheck(strategy: BackupStrategy) -> BackupHealthCheck:
+    """Check the health of the backup strategy.
+
+    Loop over the location where backups have been written and assess
+    their health and the health of the disk.
+
+    Assumes any folders gz files in the backup directory are backups for
+    the given strategy.
+
+    Returns:
+        A dictionary used to create the backup healthcheck response.
+
+    """
+
+    backup_files = strategy.location.glob(f"{strategy.backup_filename_prefix}-*.gz")
+    if not backup_files:
+        raise RuntimeError(f"No backups found with strategy: {strategy}")
+
+    backups: list[BackupHealth] = []
+    for file in backup_files:
+        backup_timestamp = datetime.datetime.strptime(
+            file.name.split(strategy.backup_filename_prefix + "-")[1], "%Y-%m-%d-%H-%M-%S"
+        )
+        backup_health = BackupHealth(
+            location=str(file.absolute()),
+            size_gb=file.stat().st_size / 1e9,
+            timestamp=backup_timestamp,
+        )
+
+        backups.append(backup_health)
+
+    return BackupHealthCheck(
+        status="success",
+        message=f"Found {len(backups)} backups.",
+        backups=backups,
+    )

pydatalab/pydatalab/config.py

@@ -1,3 +1,4 @@
+import datetime
 import hashlib
 import json
 import logging
@@ -65,6 +66,25 @@ class Config:
         extra = "allow"
 
 
+class BackupHealth(BaseModel):
+    location: str | None
+    stats: dict
+    size_gb: float
+    timestamp: datetime.datetime
+
+
+class BackupHealthCheck(BaseModel):
+    status: str
+    message: str
+    trigger_url: str
+    last_backup_timestamp: datetime.datetime
+    last_backup_size_gb: float
+    used_space_gb: float
+    num_backups: int
+    free_space_gb: float
+    backups: list[BackupHealth]
+
+
 class BackupStrategy(BaseModel):
     """This model describes the config of a particular backup strategy."""
 

pydatalab/pydatalab/routes/v0_1/admin.py

@@ -2,7 +2,7 @@
 from flask import Blueprint, jsonify, request
 from flask_login import current_user
 
-from pydatalab.config import CONFIG
+from pydatalab.config import CONFIG, BackupHealthCheck
 from pydatalab.mongo import flask_mongo
 from pydatalab.permissions import admin_only, get_default_permissions
 
@@ -93,3 +93,22 @@ def save_role(user_id):
         )
 
     return (jsonify({"status": "success"}), 200)
+
+
+@ADMIN.route("/health/backups")
+def check_backup_health():
+    """Loop over all backup strategies and provide information on their health,
+    and provide download links.
+
+    """
+    results = {}
+
+    if not CONFIG.BACKUP_STRATEGIES:
+        return jsonify({"status": "success", "message": "No backup strategies configured."}), 204
+
+    for strat_name, strat in CONFIG.BACKUP_STRATEGIES.items():
+        results[strat_name] = BackupHealthCheck(strat.healthcheck())
+
+    response = {"status": "success", "data": results}
+
+    return jsonify(response), 200