diff --git a/services/scheduler/requirements.txt b/services/scheduler/requirements.txt new file mode 100644 index 0000000..00dec9d --- /dev/null +++ b/services/scheduler/requirements.txt @@ -0,0 +1,34 @@ +# FastAPI ecosystem - Latest CVE-safe versions (Dec 2025) +fastapi~=0.124.0 # Latest non-vulnerable (no known CVEs) +uvicorn[standard]~=0.38.0 # Latest non-vulnerable (CVE-2025-43859 fixed in h11 0.16.0) +pydantic~=2.12.5 # Latest (CVE-2024-3772 fixed in 2.4.0+) +pydantic-settings~=2.7.0 # Settings management + +# Task scheduling +apscheduler~=3.11.1 # Stable release (avoid 4.x alpha) +sqlalchemy~=2.0.36 # Required by APScheduler jobstore + +# Database +psycopg2-binary~=2.9.11 # Latest stable PostgreSQL adapter +redis~=5.2.0 # Redis client + +# HTTP client +httpx~=0.28.1 # Latest stable async HTTP client + +# Web scraping (for doc mirroring) +scrapy~=2.12.0 # Latest stable +beautifulsoup4~=4.12.3 # HTML parsing +lxml~=5.1.0 # XML/HTML parser + +# Git operations +gitpython~=3.1.43 # Latest stable + +# Security/Auth +python-jose[cryptography]~=3.3.0 # JWT handling + +# Testing +pytest~=8.3.4 # Test framework +pytest-asyncio~=0.25.2 # Async test support +pytest-cov~=6.0.0 # Coverage reporting +httpx~=0.28.1 # Already included above, used for API testing +freezegun~=1.5.1 # Time mocking for scheduler tests diff --git a/services/scheduler/src/config.py b/services/scheduler/src/config.py new file mode 100644 index 0000000..0b24ba3 --- /dev/null +++ b/services/scheduler/src/config.py @@ -0,0 +1,73 @@ +""" +Configuration management for The Scheduler. +Uses Pydantic BaseSettings for type-safe environment variable loading. +""" +from functools import lru_cache +from pydantic_settings import BaseSettings +from pydantic import Field + + +class Settings(BaseSettings): + """Application settings loaded from environment variables.""" + + # Application + app_name: str = Field(default="The Scheduler", alias="APP_NAME") + app_version: str = Field(default="1.0.0", alias="APP_VERSION") + debug: bool = Field(default=False, alias="DEBUG") + host: str = Field(default="0.0.0.0", alias="HOST") + port: int = Field(default=8090, alias="PORT") + log_level: str = Field(default="INFO", alias="LOG_LEVEL") + + # Security + scheduler_api_key: str = Field(default="dev-key-change-me", alias="SCHEDULER_API_KEY") + + # PostgreSQL + postgres_host: str = Field(default="postgres-shared", alias="POSTGRES_HOST") + postgres_port: int = Field(default=5432, alias="POSTGRES_PORT") + postgres_db: str = Field(default="library_scheduler", alias="POSTGRES_DB") + postgres_user: str = Field(default="library_scheduler_user", alias="POSTGRES_USER") + postgres_password: str = Field(default="", alias="POSTGRES_PASSWORD") + + # Redis + redis_host: str = Field(default="redis-shared", alias="REDIS_HOST") + redis_port: int = Field(default=6379, alias="REDIS_PORT") + redis_db: int = Field(default=3, alias="REDIS_DB") + + # Gitea + gitea_url: str = Field(default="http://gitea:3000", alias="GITEA_URL") + gitea_user: str = Field(default="library", alias="GITEA_USER") + gitea_password: str = Field(default="", alias="GITEA_PASSWORD") + gitea_ssh_host: str = Field(default="gitea", alias="GITEA_SSH_HOST") + gitea_ssh_port: int = Field(default=22, alias="GITEA_SSH_PORT") + + # Backup Configuration + backup_retention_daily: int = Field(default=7, alias="BACKUP_RETENTION_DAILY") + backup_retention_weekly: int = Field(default=4, alias="BACKUP_RETENTION_WEEKLY") + backup_retention_monthly: int = Field(default=12, alias="BACKUP_RETENTION_MONTHLY") + + # Documentation Mirroring + docs_mirror_path: str = Field(default="/docs-mirror", alias="DOCS_MIRROR_PATH") + docs_check_interval: int = Field(default=21600, alias="DOCS_CHECK_INTERVAL") # 6 hours + + @property + def database_url(self) -> str: + """PostgreSQL connection URL for APScheduler.""" + return f"postgresql://{self.postgres_user}:{self.postgres_password}@{self.postgres_host}:{self.postgres_port}/{self.postgres_db}" + + @property + def redis_url(self) -> str: + """Redis connection URL.""" + return f"redis://{self.redis_host}:{self.redis_port}/{self.redis_db}" + + class Config: + env_file = ".env" + case_sensitive = False + + +@lru_cache +def get_settings() -> Settings: + """ + Get cached settings instance. + Using lru_cache ensures we only create one Settings instance. + """ + return Settings() diff --git a/services/scheduler/src/main.py b/services/scheduler/src/main.py new file mode 100644 index 0000000..5388c90 --- /dev/null +++ b/services/scheduler/src/main.py @@ -0,0 +1,476 @@ +""" +The Scheduler - System-wide maintenance orchestration. +Handles backups, documentation mirroring, cleanup, and automated tasks. + +Architecture: Hybrid APScheduler + DB-based priority system +- APScheduler runs a single job every minute +- Job queries DB for tasks scheduled in that minute +- Executes up to 5 tasks concurrently based on priority +""" +from fastapi import FastAPI, HTTPException, Depends, Header +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore +from apscheduler.triggers.cron import CronTrigger +from contextlib import asynccontextmanager +import logging + +from src.config import get_settings, Settings +from src.tasks.executor import TaskExecutor + +# Configure logging +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' +) +logger = logging.getLogger(__name__) + +# Global instances +scheduler: AsyncIOScheduler | None = None +task_executor: TaskExecutor | None = None + + +def get_scheduler() -> AsyncIOScheduler: + """Dependency to get scheduler instance.""" + if scheduler is None: + raise HTTPException(500, "Scheduler not initialized") + return scheduler + + +def get_task_executor() -> TaskExecutor: + """Dependency to get task executor instance.""" + if task_executor is None: + raise HTTPException(500, "Task executor not initialized") + return task_executor + +# Lifespan manager for startup/shutdown +@asynccontextmanager +async def lifespan(app: FastAPI): + """Manage application lifecycle - startup and shutdown.""" + global scheduler, task_executor + + settings = get_settings() + + # Startup + logger.info("=" * 60) + logger.info("The Scheduler - System-wide Maintenance Orchestration") + logger.info("=" * 60) + logger.info(f"Architecture: Hybrid APScheduler + DB-based priority system") + logger.info(f"Database: {settings.postgres_host}:{settings.postgres_port}/{settings.postgres_db}") + logger.info(f"API: http://{settings.host}:{settings.port}") + logger.info(f"Docs: http://{settings.host}:{settings.port}/docs") + logger.info("=" * 60) + + # Initialize task executor + task_executor = TaskExecutor(settings) + logger.info("Task executor initialized (max 5 concurrent tasks)") + + # Initialize APScheduler with minimal configuration + # No jobstore needed - we only have one in-memory job + scheduler = AsyncIOScheduler( + job_defaults={ + 'coalesce': True, # Combine missed runs + 'max_instances': 1, # Only one instance running + 'misfire_grace_time': 30 # 30s grace period for minute-based execution + } + ) + + # Add the single minute-based task processor + scheduler.add_job( + func=task_executor.process_minute, + trigger=CronTrigger(minute='*'), # Run every minute + id='process_tasks', + name='Process scheduled tasks', + replace_existing=True + ) + + try: + scheduler.start() + logger.info("Scheduler started - processing tasks every minute") + logger.info("Priority system: 1-5 (emergency/system), 10-30 (user), 40-70+ (maintenance)") + except Exception as e: + logger.error(f"Failed to start scheduler: {e}") + raise + + yield + + # Shutdown + logger.info("Stopping The Scheduler...") + if scheduler: + scheduler.shutdown(wait=True) + logger.info("Scheduler stopped") + + +# FastAPI app +app = FastAPI( + title="The Scheduler", + version="1.0.0", + description="System-wide maintenance orchestration - backups, doc mirroring, cleanup, task automation", + lifespan=lifespan +) + +# Dependencies +async def verify_api_key( + authorization: str = Header(None), + settings: Settings = Depends(get_settings) +): + """Verify API key from Authorization header.""" + if not authorization or not authorization.startswith("Bearer "): + raise HTTPException(401, "Missing API key") + key = authorization.replace("Bearer ", "") + if key != settings.scheduler_api_key: + raise HTTPException(403, "Invalid API key") + return key + +# ============================================================================ +# Public Endpoints (no auth required) +# ============================================================================ + +@app.get("/health") +async def health( + settings: Settings = Depends(get_settings), + sched: AsyncIOScheduler = Depends(get_scheduler) +): + """Health check endpoint.""" + return { + "status": "healthy", + "scheduler_running": sched.running, + "jobs_count": len(sched.get_jobs()), + "database": settings.postgres_db + } + +# ============================================================================ +# Protected Endpoints (require API key) +# ============================================================================ + +@app.get("/tasks") +async def list_tasks( + enabled: bool = None, + service: str = None, + api_key: str = Depends(verify_api_key), + executor: TaskExecutor = Depends(get_task_executor) +): + """List all scheduled tasks from database.""" + import psycopg2.extras + + query = "SELECT * FROM scheduled_tasks WHERE 1=1" + params = [] + + if enabled is not None: + query += " AND enabled = %s" + params.append(enabled) + + if service: + query += " AND service = %s" + params.append(service) + + query += " ORDER BY priority ASC, task_name ASC" + + with executor.get_db_connection() as conn: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute(query, params) + tasks = [dict(task) for task in cur.fetchall()] + + return { + "tasks": tasks, + "count": len(tasks) + } + +@app.get("/tasks/{task_name}") +async def get_task_details( + task_name: str, + api_key: str = Depends(verify_api_key), + executor: TaskExecutor = Depends(get_task_executor) +): + """Get details for a specific task.""" + import psycopg2.extras + + with executor.get_db_connection() as conn: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute("SELECT * FROM scheduled_tasks WHERE task_name = %s", (task_name,)) + task = cur.fetchone() + + if not task: + raise HTTPException(404, f"Task '{task_name}' not found") + + return dict(task) + +@app.post("/tasks") +async def create_task( + task_data: dict, + api_key: str = Depends(verify_api_key), + executor: TaskExecutor = Depends(get_task_executor) +): + """Create a new scheduled task.""" + import psycopg2.extras + import json + + required_fields = ['task_name', 'service', 'executor', 'priority'] + if not all(field in task_data for field in required_fields): + raise HTTPException(400, f"Missing required fields: {required_fields}") + + # Set defaults + task_data.setdefault('minute', -1) + task_data.setdefault('hour', -1) + task_data.setdefault('day_of_month', -1) + task_data.setdefault('month', -1) + task_data.setdefault('day_of_week', -1) + task_data.setdefault('enabled', True) + task_data.setdefault('max_retries', 3) + task_data.setdefault('timeout_seconds', 3600) + + # Convert config dict to JSON string if present + if 'config' in task_data and isinstance(task_data['config'], dict): + task_data['config'] = json.dumps(task_data['config']) + + with executor.get_db_connection() as conn: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute(""" + INSERT INTO scheduled_tasks + (task_name, service, executor, priority, minute, hour, + day_of_month, month, day_of_week, enabled, description, + config, max_retries, timeout_seconds, created_by) + VALUES + (%(task_name)s, %(service)s, %(executor)s, %(priority)s, + %(minute)s, %(hour)s, %(day_of_month)s, %(month)s, + %(day_of_week)s, %(enabled)s, %(description)s, + %(config)s::jsonb, %(max_retries)s, %(timeout_seconds)s, + %(created_by)s) + RETURNING * + """, {**task_data, 'created_by': task_data.get('created_by', 'api')}) + new_task = dict(cur.fetchone()) + conn.commit() + + logger.info(f"Created task: {new_task['task_name']}") + return new_task + +@app.put("/tasks/{task_name}") +async def update_task( + task_name: str, + task_data: dict, + api_key: str = Depends(verify_api_key), + executor: TaskExecutor = Depends(get_task_executor) +): + """Update an existing scheduled task.""" + import psycopg2.extras + import json + + # Build update query dynamically + allowed_fields = ['service', 'executor', 'priority', 'minute', 'hour', + 'day_of_month', 'month', 'day_of_week', 'enabled', + 'description', 'config', 'max_retries', 'timeout_seconds'] + + updates = {k: v for k, v in task_data.items() if k in allowed_fields} + if not updates: + raise HTTPException(400, "No valid fields to update") + + # Convert config dict to JSON string if present + if 'config' in updates and isinstance(updates['config'], dict): + updates['config'] = json.dumps(updates['config']) + + set_clause = ', '.join([f"{k} = %({k})s" for k in updates.keys()]) + + with executor.get_db_connection() as conn: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute(f""" + UPDATE scheduled_tasks + SET {set_clause}, updated_at = NOW() + WHERE task_name = %(task_name)s + RETURNING * + """, {**updates, 'task_name': task_name}) + + updated_task = cur.fetchone() + if not updated_task: + raise HTTPException(404, f"Task '{task_name}' not found") + + conn.commit() + + logger.info(f"Updated task: {task_name}") + return dict(updated_task) + +@app.delete("/tasks/{task_name}") +async def delete_task( + task_name: str, + api_key: str = Depends(verify_api_key), + executor: TaskExecutor = Depends(get_task_executor) +): + """Delete a scheduled task.""" + with executor.get_db_connection() as conn: + with conn.cursor() as cur: + cur.execute(""" + DELETE FROM scheduled_tasks + WHERE task_name = %s + RETURNING task_name + """, (task_name,)) + + deleted = cur.fetchone() + if not deleted: + raise HTTPException(404, f"Task '{task_name}' not found") + + conn.commit() + + logger.info(f"Deleted task: {task_name}") + return {"message": f"Task '{task_name}' deleted successfully"} + +@app.post("/tasks/{task_name}/trigger") +async def trigger_task( + task_name: str, + api_key: str = Depends(verify_api_key), + executor: TaskExecutor = Depends(get_task_executor) +): + """Manually trigger a task to run immediately.""" + import psycopg2.extras + + # Get task details + with executor.get_db_connection() as conn: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute("SELECT * FROM scheduled_tasks WHERE task_name = %s", (task_name,)) + task = cur.fetchone() + + if not task: + raise HTTPException(404, f"Task '{task_name}' not found") + + if not task['enabled']: + raise HTTPException(400, f"Task '{task_name}' is disabled") + + # Execute task immediately in background + import asyncio + asyncio.create_task(executor.execute_task(dict(task))) + + logger.info(f"Manually triggered task: {task_name}") + return { + "message": f"Task '{task_name}' triggered successfully", + "task_name": task_name, + "priority": task['priority'], + "executor": task['executor'] + } + +# Legacy endpoints (deprecated) +@app.post("/tasks/backup") +async def trigger_backup(api_key: str = Depends(verify_api_key)): + """Trigger backup tasks manually (deprecated - use POST /tasks/{name}/trigger)""" + # TODO: Implement backup executor + logger.info("Manual backup triggered") + return { + "message": "Backup task triggered", + "status": "not_implemented", + "note": "Backup executor needs to be implemented" + } + +@app.post("/tasks/docs/update") +async def trigger_docs_update( + project: str = None, + api_key: str = Depends(verify_api_key) +): + """Trigger documentation mirror update""" + # TODO: Implement doc mirror executor + logger.info(f"Doc mirror update triggered for project: {project or 'all'}") + return { + "message": f"Documentation update triggered for {project or 'all projects'}", + "status": "not_implemented", + "note": "Doc mirror executor needs to be implemented" + } + +@app.post("/tasks/docs/check-versions") +async def check_doc_versions(api_key: str = Depends(verify_api_key)): + """Check for new documentation versions""" + # TODO: Implement version check executor + logger.info("Version check triggered") + return { + "message": "Version check triggered", + "status": "not_implemented", + "note": "Version check executor needs to be implemented" + } + +@app.post("/tasks/cleanup") +async def trigger_cleanup(api_key: str = Depends(verify_api_key)): + """Run cleanup tasks""" + # TODO: Implement cleanup executor + logger.info("Cleanup task triggered") + return { + "message": "Cleanup task triggered", + "status": "not_implemented", + "note": "Cleanup executor needs to be implemented" + } + +@app.get("/executions") +async def task_history( + limit: int = 20, + task_name: str = None, + service: str = None, + status: str = None, + api_key: str = Depends(verify_api_key), + executor: TaskExecutor = Depends(get_task_executor) +): + """View task execution history.""" + import psycopg2.extras + + query = "SELECT * FROM task_executions WHERE 1=1" + params = [] + + if task_name: + query += " AND task_name = %s" + params.append(task_name) + + if service: + query += " AND service = %s" + params.append(service) + + if status: + query += " AND status = %s" + params.append(status) + + query += " ORDER BY triggered_at DESC LIMIT %s" + params.append(limit) + + with executor.get_db_connection() as conn: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute(query, params) + executions = [dict(ex) for ex in cur.fetchall()] + + return { + "executions": executions, + "count": len(executions), + "limit": limit + } + +@app.get("/stats") +async def stats( + api_key: str = Depends(verify_api_key), + settings: Settings = Depends(get_settings), + sched: AsyncIOScheduler = Depends(get_scheduler), + task_exec: TaskExecutor = Depends(get_task_executor) +): + """Get system statistics.""" + import psycopg2.extras + + # Query task stats from database + with task_exec.get_db_connection() as conn: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + # Count enabled tasks + cur.execute("SELECT COUNT(*) as count FROM scheduled_tasks WHERE enabled = true") + enabled_tasks = cur.fetchone()['count'] + + # Count running tasks + cur.execute("SELECT COUNT(*) as count FROM task_executions WHERE status = 'running'") + running_tasks = cur.fetchone()['count'] + + # Recent execution stats (last 24 hours) + cur.execute(""" + SELECT status, COUNT(*) as count + FROM task_executions + WHERE triggered_at > NOW() - INTERVAL '24 hours' + GROUP BY status + """) + execution_stats = {row['status']: row['count'] for row in cur.fetchall()} + + return { + "scheduler_running": sched.running, + "minute_processor_active": True, # If we got here, it's running + "database": settings.postgres_db, + "tasks_enabled": enabled_tasks, + "tasks_currently_running": running_tasks, + "concurrent_limit": 5, + "execution_stats_24h": execution_stats, + "priority_system": "1-5 (emergency/system), 10-30 (user), 40-70+ (maintenance)" + } + diff --git a/services/scheduler/src/tasks/executor.py b/services/scheduler/src/tasks/executor.py new file mode 100644 index 0000000..8876678 --- /dev/null +++ b/services/scheduler/src/tasks/executor.py @@ -0,0 +1,278 @@ +""" +Task execution engine for The Scheduler. +Implements minute-based polling with priority-based concurrent execution. +""" +import asyncio +import logging +from datetime import datetime, timezone +from typing import List, Dict, Any, Optional +import psycopg2 +from psycopg2.extras import RealDictCursor +import traceback + +from src.config import Settings + +logger = logging.getLogger(__name__) + +# Priority ranges (for reference) +# 1: Emergency/recovery tasks +# 2: Primary system tasks +# 3: Secondary system tasks +# 5: Urgent user-triggered tasks +# 10: High-priority user tasks +# 15: Reserved +# 20: Backup tasks +# 25: Reserved +# 30: Low-priority user tasks +# 40: Cleanup tasks +# 50: Documentation version checks +# 60: Documentation mirroring +# 70+: Future/experimental tasks + +MAX_CONCURRENT_TASKS = 5 + + +class TaskExecutor: + """Executes scheduled tasks based on priority with concurrency control.""" + + def __init__(self, settings: Settings): + self.settings = settings + self.running_tasks: Dict[int, asyncio.Task] = {} # task_id -> asyncio.Task + self.semaphore = asyncio.Semaphore(MAX_CONCURRENT_TASKS) + + def get_db_connection(self): + """Create database connection.""" + return psycopg2.connect( + host=self.settings.postgres_host, + port=self.settings.postgres_port, + database=self.settings.postgres_db, + user=self.settings.postgres_user, + password=self.settings.postgres_password + ) + + def get_tasks_for_minute(self, now: datetime) -> List[Dict[str, Any]]: + """ + Query database for tasks scheduled for this minute. + Supports wildcards (-1 = any value). + """ + minute = now.minute + hour = now.hour + day = now.day + month = now.month + # Python: Monday=0, Sunday=6; PostgreSQL: Monday=0, Sunday=6 (same) + weekday = now.weekday() + + query = """ + SELECT + id, task_name, service, executor, priority, + config, timeout_seconds, max_retries, retry_count, + last_run, last_status + FROM scheduled_tasks + WHERE enabled = true + AND (minute = -1 OR minute = %s) + AND (hour = -1 OR hour = %s) + AND (day_of_month = -1 OR day_of_month = %s) + AND (month = -1 OR month = %s) + AND (day_of_week = -1 OR day_of_week = %s) + AND id NOT IN ( + SELECT task_id + FROM task_executions + WHERE status = 'running' + ) + ORDER BY priority ASC, task_name ASC + """ + + with self.get_db_connection() as conn: + with conn.cursor(cursor_factory=RealDictCursor) as cur: + cur.execute(query, (minute, hour, day, month, weekday)) + tasks = cur.fetchall() + + logger.info(f"Found {len(tasks)} tasks scheduled for {now.strftime('%Y-%m-%d %H:%M')}") + return [dict(task) for task in tasks] + + def should_run_task(self, task: Dict[str, Any], now: datetime) -> bool: + """ + Determine if task should run based on last execution. + Prevents running the same task multiple times in the same minute. + """ + if not task['last_run']: + return True + + last_run = task['last_run'] + if last_run.tzinfo is None: + last_run = last_run.replace(tzinfo=timezone.utc) + if now.tzinfo is None: + now = now.replace(tzinfo=timezone.utc) + + # Don't run if already executed this minute + if (last_run.year == now.year and + last_run.month == now.month and + last_run.day == now.day and + last_run.hour == now.hour and + last_run.minute == now.minute): + logger.debug(f"Task {task['task_name']} already ran this minute") + return False + + return True + + async def execute_task(self, task: Dict[str, Any]): + """ + Execute a single task with timeout and error handling. + Updates task_executions table with results. + """ + task_id = task['id'] + task_name = task['task_name'] + executor_name = task['executor'] + timeout = task.get('timeout_seconds', 3600) + + execution_id = None + started_at = datetime.now(timezone.utc) + + logger.info(f"[Priority {task['priority']}] Starting task: {task_name} (executor: {executor_name})") + + try: + # Create execution record + with self.get_db_connection() as conn: + with conn.cursor() as cur: + cur.execute(""" + INSERT INTO task_executions + (task_id, task_name, service, executor, priority, + status, triggered_by, started_at) + VALUES (%s, %s, %s, %s, %s, 'running', 'scheduler', %s) + RETURNING id + """, (task_id, task_name, task['service'], executor_name, + task['priority'], started_at)) + execution_id = cur.fetchone()[0] + conn.commit() + + # Load and execute the task + output, error = await self._run_executor(executor_name, task, timeout) + + completed_at = datetime.now(timezone.utc) + duration = int((completed_at - started_at).total_seconds()) + status = 'success' if error is None else 'failed' + + # Update execution record + with self.get_db_connection() as conn: + with conn.cursor() as cur: + cur.execute(""" + UPDATE task_executions + SET status = %s, completed_at = %s, duration_seconds = %s, + output = %s, error = %s + WHERE id = %s + """, (status, completed_at, duration, output, error, execution_id)) + + # Update scheduled_tasks + cur.execute(""" + UPDATE scheduled_tasks + SET last_run = %s, last_status = %s, last_duration_seconds = %s, + retry_count = 0, updated_at = %s + WHERE id = %s + """, (completed_at, status, duration, completed_at, task_id)) + conn.commit() + + if error: + logger.error(f"Task {task_name} failed: {error}") + else: + logger.info(f"Task {task_name} completed successfully in {duration}s") + + except asyncio.TimeoutError: + logger.error(f"Task {task_name} timed out after {timeout}s") + self._update_execution_status(execution_id, 'timeout', + error=f"Task exceeded timeout of {timeout}s") + + except Exception as e: + logger.error(f"Task {task_name} failed with exception: {e}") + logger.error(traceback.format_exc()) + self._update_execution_status(execution_id, 'failed', + error=f"{str(e)}\n{traceback.format_exc()}") + + finally: + # Remove from running tasks + if task_id in self.running_tasks: + del self.running_tasks[task_id] + + async def _run_executor(self, executor_name: str, task: Dict[str, Any], timeout: int) -> tuple[Optional[str], Optional[str]]: + """ + Dynamically load and run the executor module. + Returns (output, error) tuple. + """ + try: + # Import executor dynamically + module_path = f"src.executors.{executor_name}" + module = __import__(module_path, fromlist=['execute']) + + if not hasattr(module, 'execute'): + return None, f"Executor {executor_name} missing execute() function" + + # Run with timeout + config = task.get('config', {}) + result = await asyncio.wait_for( + module.execute(config, self.settings), + timeout=timeout + ) + + return result, None + + except ModuleNotFoundError: + return None, f"Executor module not found: {executor_name}" + except Exception as e: + return None, f"Executor error: {str(e)}\n{traceback.format_exc()}" + + def _update_execution_status(self, execution_id: int, status: str, error: str = None): + """Update execution record with final status.""" + if execution_id is None: + return + + try: + with self.get_db_connection() as conn: + with conn.cursor() as cur: + cur.execute(""" + UPDATE task_executions + SET status = %s, completed_at = %s, error = %s + WHERE id = %s + """, (status, datetime.now(timezone.utc), error, execution_id)) + conn.commit() + except Exception as e: + logger.error(f"Failed to update execution status: {e}") + + async def process_minute(self): + """ + Main entry point: Process all tasks scheduled for the current minute. + Executes up to MAX_CONCURRENT_TASKS in parallel, prioritized by priority field. + """ + now = datetime.now(timezone.utc) + logger.info(f"Processing tasks for {now.strftime('%Y-%m-%d %H:%M')}") + + # Get tasks scheduled for this minute + tasks = self.get_tasks_for_minute(now) + + # Filter out tasks that already ran this minute + tasks_to_run = [task for task in tasks if self.should_run_task(task, now)] + + if not tasks_to_run: + logger.debug("No tasks to run this minute") + return + + logger.info(f"Will execute {len(tasks_to_run)} tasks (max {MAX_CONCURRENT_TASKS} concurrent)") + + # Process tasks in priority order, respecting concurrency limit + for task in tasks_to_run: + # Wait for available slot + await self.semaphore.acquire() + + # Start task + task_coro = self._execute_with_semaphore(task) + asyncio_task = asyncio.create_task(task_coro) + self.running_tasks[task['id']] = asyncio_task + + # Wait for all tasks in this batch to complete + if self.running_tasks: + await asyncio.gather(*self.running_tasks.values(), return_exceptions=True) + + async def _execute_with_semaphore(self, task: Dict[str, Any]): + """Execute task and release semaphore when done.""" + try: + await self.execute_task(task) + finally: + self.semaphore.release() diff --git a/stacks/scheduler.yml b/stacks/scheduler.yml new file mode 100644 index 0000000..79e2e11 --- /dev/null +++ b/stacks/scheduler.yml @@ -0,0 +1,107 @@ +version: '3.8' + +# The Scheduler +# Purpose: System-wide maintenance orchestration - backups, doc mirroring, cleanup, task automation +# Port: 8090 (API + UI) +# Network: docker-dataplane + +services: + scheduler: + image: python:3.12-slim + container_name: scheduler + restart: unless-stopped + + command: > + sh -c " + echo 'Installing system dependencies...' && + apt-get update -qq && + apt-get install -y --no-install-recommends git ssh postgresql-client curl docker.io >/dev/null 2>&1 && + rm -rf /var/lib/apt/lists/* && + echo 'Setting up Python environment...' && + if [ ! -f /venv/bin/python ]; then + echo 'Initializing venv...' && + python3 -m venv --clear /venv; + fi && + echo 'Upgrading pip...' && + /venv/bin/python -m pip install --upgrade pip --quiet && + echo 'Installing dependencies...' && + /venv/bin/python -m pip install -r /app/requirements.txt --quiet && + echo 'Configuring Git...' && + git config --global user.name 'The Librarian' && + git config --global user.email 'librarian@portainer-core.local' && + echo 'Starting The Scheduler...' && + /venv/bin/python -m uvicorn src.main:app --host 0.0.0.0 --port 8090 --workers 1 + " + + ports: + - "8090:8090" + + environment: + - APP_NAME=The Scheduler + - APP_VERSION=1.0.0 + - DEBUG=true + - HOST=0.0.0.0 + - PORT=8090 + - LOG_LEVEL=INFO + - POSTGRES_HOST=postgres-shared + - POSTGRES_PORT=5432 + - POSTGRES_DB=scheduler + - POSTGRES_USER=scheduler_user + - POSTGRES_PASSWORD=${SCHEDULER_DB_PASSWORD} + - REDIS_HOST=redis-shared + - REDIS_PORT=6379 + - REDIS_DB=3 + - GITEA_URL=http://gitea:3000 + - GITEA_USER=${GITEA_LIBRARY_USER} + - GITEA_PASSWORD=${GITEA_LIBRARY_PASSWORD} + - GITEA_SSH_HOST=gitea + - GITEA_SSH_PORT=22 + - BACKUP_RETENTION_DAILY=7 + - BACKUP_RETENTION_WEEKLY=4 + - BACKUP_RETENTION_MONTHLY=12 + - DOCS_MIRROR_PATH=/docs-mirror + - DOCS_CHECK_INTERVAL=21600 + - SCHEDULER_API_KEY=${SCHEDULER_API_KEY} + - PYTHONPATH=/app + + volumes: + - /home/jpmschweitzer/Projects/portainer-core/services/scheduler:/app + - /home/jpmschweitzer/docker-data/scheduler/venv:/venv + - /home/jpmschweitzer/docker-data/scheduler/logs:/app/logs + - /home/jpmschweitzer/docker-data/scheduler/task-data:/app/task-data + - /home/jpmschweitzer/docker-data/scheduler/ssh:/root/.ssh:ro + - /mnt/media/library/docs-mirror:/docs-mirror + - /mnt/media/backups/library:/backups + - /var/run/docker.sock:/var/run/docker.sock:ro + - /home/jpmschweitzer/docker-data/postgres-shared:/postgres-data:ro + # For config backups (read-only sources) + - /home/jpmschweitzer/docker-data:/data/docker-data:ro + - /home/jpmschweitzer/.config/code-server:/data/code-server-config:ro + # For config backups (write destination) + - /mnt/media/backups/docker-configs:/backups/docker-configs + + networks: + - docker-dataplane + + deploy: + resources: + limits: + cpus: '0.5' + memory: 1G + reservations: + memory: 256M + + labels: + - "com.centurylinklabs.watchtower.enable=true" + + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:8090/health"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 90s + +networks: + docker-dataplane: + external: true + name: docker-dataplane