feat(scheduler): implement core scheduler service
Add hybrid APScheduler + PostgreSQL-based task scheduling system with minute-based execution and priority queue.
Core features:
- Minute-based scheduling with cron-like patterns (-1 = wildcard)
- Priority queue system (1-100, lower = higher priority)
- Concurrent execution (max 5 tasks simultaneously)
- Full REST API for task management (CRUD operations)
- Task execution tracking with audit trail
- API key authentication (Bearer token)
- Health checks and system statistics
Architecture:
- APScheduler runs every minute
- Queries PostgreSQL for tasks scheduled for current minute
- Executes tasks concurrently by priority
- Records execution history in database
Database schema:
- scheduled_tasks: Task definitions/templates
- task_executions: Individual execution records
Technical stack:
- FastAPI for REST API
- APScheduler for scheduling
- PostgreSQL for persistence
- Pydantic for configuration
Endpoints:
- POST/GET/PUT/DELETE /tasks - Task management
- POST /tasks/{name}/trigger - Manual execution
- GET /executions - Execution history
- GET /health - Health check
- GET /stats - System statistics
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 4.5
@@ -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
|
||||
@@ -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()
|
||||
@@ -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)"
|
||||
}
|
||||
|
||||
@@ -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()
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user