Backup scheduler service
backupsystem.py
py / 281 lines
Python backup service using JSON configs, NFS mount validation, tar.gz archives, retention cleanup and dynamic schedule reloads.
import os
import asyncio
import json
import tarfile
import subprocess
import datetime
import signal
import sys
import logging
import gc
from typing import List
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
# Constants
CONFIG_DIR = "/etc/backup_system/configs"
# Globals
event_loop = None # Global event loop reference
scheduled_tasks = {}
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
handlers=[
logging.StreamHandler(sys.stdout),
]
)
#########################
# MODEL: BACKUP SERVICE #
#########################
class ServiceBackup:
"""
Handles the backup process for a specific service based on its configuration.
"""
def __init__(self, service_name: str, config_file: Path):
"""
Initializes the ServiceBackup object by loading configuration details.
:param service_name: Name of the service being backed up.
:param config_file: Path to the service's JSON configuration file.
"""
with open(config_file, "r") as f:
data = json.load(f)
self.service_name = service_name
self.paths: List[Path] = [Path(p) for p in data["paths"]] # Ensure all paths are Path objects
self.backup_times: List[str] = data["backup_times"] # Backup times in HH:MM format
self.retention_days: int = data["retention_days"] # Default retention: 7 days
self.backup_location: Path = Path(data["backup_location"])
self.pre_commands: List[str] = data["pre_commands"]
# Ensure backup location exists
self.backup_location.mkdir(parents=True, exist_ok=True)
def run_pre_commands(self):
"""Executes pre-backup commands defined in the configuration."""
for command in self.pre_commands:
logging.info(f"[{self.service_name}] Running pre-backup command: {command}")
try:
result = subprocess.run(command, shell=True, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
logging.info(f"[{self.service_name}] Command output: {result.stdout.decode().strip()}")
except subprocess.CalledProcessError as e:
logging.error(f"[{self.service_name}] Pre-command failed: {e.output}")
raise # Stop the backup if a pre-command fails
def is_nfs_mounted(self, path: Path) -> bool:
"""Check if the provided path is mounted via NFS."""
try:
with open("/proc/mounts", "r") as mounts:
for line in mounts:
parts = line.split()
mount_point = parts[1]
fstype = parts[2]
if path.resolve().as_posix().startswith(mount_point) and fstype in ("nfs", "nfs4"):
return True
except Exception as e:
logging.error(f"Error while checking NFS mount: {e}")
return False
def create_backup(self):
"""Creates a tar.gz backup of the configured directories/files efficiently."""
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
backup_file = self.backup_location / f"{self.service_name}_{timestamp}.tar.gz"
# Check if backup location is NFS mounted
if not self.is_nfs_mounted(self.backup_location):
logging.error(f"[{self.service_name}] Backup location {self.backup_location} is not NFS-mounted. Skipping backup.")
return None
# Run pre-commands synchronously
self.run_pre_commands()
try:
with tarfile.open(backup_file, "w:gz") as tar:
for path in self.paths:
if path.exists():
logging.info(f"[{self.service_name}] Adding to backup: {path}")
tar.add(path, arcname=path.relative_to(path.parent), recursive=True)
else:
logging.warning(f"[{self.service_name}] Warning: Path {path} does not exist, skipping...")
logging.info(f"[{self.service_name}] Backup created: {backup_file}")
except Exception as e:
logging.error(f"[{self.service_name}] Backup creation failed: {e}")
return None
finally:
gc.collect() # Force garbage collection
# subprocess.run("sync; echo 3 > /proc/sys/vm/drop_caches", shell=True)
return backup_file
def delete_old_backups(self):
"""
Deletes old backups based on the retention policy.
"""
cutoff_time = datetime.datetime.now() - datetime.timedelta(days=self.retention_days)
for backup in self.backup_location.glob(f"{self.service_name}_*.tar.gz"):
try:
# Remove both '.tar.gz' extensions
filename = backup.name.replace(".tar.gz", "")
# Extract the timestamp from the filename
parts = filename.split("_")
if len(parts) >= 3:
backup_time_str = f"{parts[-2]}_{parts[-1]}"
backup_time = datetime.datetime.strptime(backup_time_str, "%Y%m%d_%H%M%S")
if backup_time < cutoff_time:
logging.info(f"[{self.service_name}] Deleting old backup: {backup}")
backup.unlink()
else:
logging.warning(f"[{self.service_name}] Invalid filename format: {backup}")
except ValueError as e:
logging.warning(f"[{self.service_name}] Skipping invalid backup file: {backup}. Error: {e}")
#########################
# EVENT: CONFIG HANDLER #
#########################
class ConfigChangeHandler(FileSystemEventHandler):
"""
Handles changes in the config directory and triggers a reload of schedules.
"""
def on_modified(self, event):
if event.src_path.endswith(".json"):
logging.info(f"Configuration modified: {event.src_path}")
reload_schedules()
def on_created(self, event):
if event.src_path.endswith(".json"):
logging.info(f"New configuration added: {event.src_path}")
reload_schedules()
def on_deleted(self, event):
if event.src_path.endswith(".json"):
logging.info(f"Configuration removed: {event.src_path}")
reload_schedules()
def load_configs() -> List[ServiceBackup]:
"""
Loads backup configurations from JSON files in the config directory.
:return: List of ServiceBackup objects.
"""
services = []
if not os.path.exists(CONFIG_DIR):
logging.warning(f"Config directory not found! Creating ({CONFIG_DIR}), please add '*.json' files.")
Path(CONFIG_DIR).mkdir(parents=True, exist_ok=True)
return services
for config_file in Path(CONFIG_DIR).glob("*.json"):
service_name = config_file.stem
services.append(ServiceBackup(service_name, config_file))
return services
#############
# FUNCTIONS #
#############
async def wait_until(target_time: datetime.time):
"""Wait until the next occurrence of the target time."""
now = datetime.datetime.now()
future = datetime.datetime.combine(now.date(), target_time)
if future < now:
future += datetime.timedelta(days=1)
await asyncio.sleep((future - now).total_seconds())
async def run_backup_for_service(service: ServiceBackup):
"""Runs the backup process for a given service asynchronously."""
logging.info(f"Starting backup for service: {service.service_name}")
await asyncio.gather(
asyncio.to_thread(service.delete_old_backups),
asyncio.to_thread(service.create_backup)
)
async def run_backup_scheduler(service: ServiceBackup, backup_time: str, task_id: str):
"""Continuously schedules backups for a service at a given time."""
target_time = datetime.datetime.strptime(backup_time, "%H:%M").time()
try:
while True:
await wait_until(target_time)
asyncio.create_task(run_backup_for_service(service))
except asyncio.CancelledError:
logging.info(f"Cancelled scheduler task for {service.service_name} at {backup_time}")
def reload_schedules():
"""Clears and reloads all scheduled backups dynamically when configurations change."""
global services, event_loop, scheduled_tasks
logging.info("Reloading backup schedules...")
# Cancel existing scheduled tasks
for task_id, task in scheduled_tasks.items():
task.cancel()
scheduled_tasks.clear() # Clear the registry
# Load updated service configurations
services = load_configs()
for service in services:
for backup_time in service.backup_times:
task_id = f"{service.service_name}_{backup_time}"
logging.info(f"Scheduling backup for {service.service_name} at {backup_time}")
if event_loop and event_loop.is_running():
task = event_loop.create_task(run_backup_scheduler(service, backup_time, task_id))
scheduled_tasks[task_id] = task
logging.info("New backup schedules loaded.")
async def schedule_backups():
"""
Schedules backups dynamically and starts monitoring the config directory for changes.
"""
reload_schedules()
event_handler = ConfigChangeHandler()
observer = Observer()
observer.schedule(event_handler, CONFIG_DIR, recursive=False)
observer.start()
logging.info("Backup scheduling complete. Waiting for scheduled times...")
try:
while True:
await asyncio.sleep(1) # Keep the event loop alive
except KeyboardInterrupt:
observer.stop()
observer.join()
def handle_exit(signum, frame):
"""Handles termination signals to exit the script gracefully."""
logging.info("Received termination signal. Exiting gracefully...")
sys.exit(0)
###################
# APPLICATION RUN #
###################
if __name__ == "__main__":
# Handle termination signals
signal.signal(signal.SIGINT, handle_exit)
signal.signal(signal.SIGTERM, handle_exit)
# Run the backup scheduler in the event loop
event_loop = asyncio.get_event_loop()
event_loop.run_until_complete(schedule_backups())