diff --git a/app/main.py b/app/main.py index 5e7e83d..54b0284 100644 --- a/app/main.py +++ b/app/main.py @@ -1,7 +1,9 @@ +import json import os import sqlite3 import threading import time +from datetime import datetime, timedelta from pathlib import Path import requests @@ -12,6 +14,7 @@ DB_FILE = Path("/opt/appfactory/data/appfactory/appfactory.db") CHECK_INTERVAL_SECONDS = int(os.getenv("APPFACTORY_MONITOR_INTERVAL_SECONDS", "30")) REQUEST_TIMEOUT_SECONDS = int(os.getenv("APPFACTORY_MONITOR_TIMEOUT_SECONDS", "5")) HEALTH_RETENTION_DAYS = int(os.getenv("APPFACTORY_HEALTH_RETENTION_DAYS", "7")) +SCHEDULER_INTERVAL_SECONDS = int(os.getenv("APPFACTORY_SCHEDULER_INTERVAL_SECONDS", "60")) CORE_SERVICES = { "appfactory-portal": "http://appfactory-portal:9100/health", @@ -20,9 +23,12 @@ CORE_SERVICES = { "appfactory-monitor": "http://appfactory-monitor:9300/health", } +TERMINAL_JOB_STATUSES = {"success", "failed", "cancelled"} + app = FastAPI(title="AppFactory Monitor") _monitor_thread = None +_scheduler_thread = None _monitor_started = False @@ -185,7 +191,13 @@ def open_incident(service_id: str, health_id: int, status: str, error_text: str """, ( service_id, - f'{{"health_id": {health_id}, "status": "{status}"}}', + json.dumps( + { + "health_id": health_id, + "status": status, + }, + ensure_ascii=False, + ), ), ) @@ -242,7 +254,13 @@ def resolve_incident(service_id: str, health_id: int): """, ( service_id, - f'{{"incident_id": {incident["id"]}, "health_id": {health_id}}}', + json.dumps( + { + "incident_id": incident["id"], + "health_id": health_id, + }, + ensure_ascii=False, + ), ), ) @@ -332,6 +350,228 @@ def run_check_cycle(): check_service(service_id, url) +def parse_time_hhmm(value: str | None) -> tuple[int, int]: + if not value: + return 3, 0 + + parts = value.strip().split(":", 1) + + if len(parts) != 2: + return 3, 0 + + try: + hour = int(parts[0]) + minute = int(parts[1]) + except ValueError: + return 3, 0 + + if hour < 0 or hour > 23: + hour = 3 + + if minute < 0 or minute > 59: + minute = 0 + + return hour, minute + + +def compute_next_run_at(schedule_type: str, schedule_time: str | None) -> str: + now = datetime.utcnow().replace(microsecond=0) + schedule_type = (schedule_type or "daily").strip().lower() + hour, minute = parse_time_hhmm(schedule_time) + + if schedule_type == "hourly": + candidate = now.replace(minute=minute, second=0) + if candidate <= now: + candidate += timedelta(hours=1) + return candidate.strftime("%Y-%m-%d %H:%M:%S") + + if schedule_type == "weekly": + candidate = now.replace(hour=hour, minute=minute, second=0) + while candidate <= now: + candidate += timedelta(days=7) + return candidate.strftime("%Y-%m-%d %H:%M:%S") + + if schedule_type == "monthly": + candidate = now.replace(day=1, hour=hour, minute=minute, second=0) + if candidate <= now: + if candidate.month == 12: + candidate = candidate.replace(year=candidate.year + 1, month=1) + else: + candidate = candidate.replace(month=candidate.month + 1) + return candidate.strftime("%Y-%m-%d %H:%M:%S") + + candidate = now.replace(hour=hour, minute=minute, second=0) + if candidate <= now: + candidate += timedelta(days=1) + + return candidate.strftime("%Y-%m-%d %H:%M:%S") + + +def sync_scheduled_script_results(): + con = get_connection() + + rows = con.execute( + """ + SELECT + scheduled_scripts.id AS scheduled_script_id, + scheduled_scripts.last_job_id AS last_job_id, + jobs.status AS job_status, + jobs.error_text AS job_error_text, + jobs.started_at AS job_started_at, + jobs.finished_at AS job_finished_at + FROM scheduled_scripts + JOIN jobs ON jobs.id = scheduled_scripts.last_job_id + WHERE scheduled_scripts.is_running = 1 + AND jobs.status IN ('success', 'failed', 'cancelled') + """ + ).fetchall() + + for row in rows: + last_run_at = row["job_finished_at"] or row["job_started_at"] + + con.execute( + """ + UPDATE scheduled_scripts + SET is_running = 0, + last_status = ?, + last_error_text = ?, + last_run_at = ?, + updated_at = CURRENT_TIMESTAMP + WHERE id = ? + """, + ( + row["job_status"], + row["job_error_text"], + last_run_at, + row["scheduled_script_id"], + ), + ) + + con.commit() + con.close() + + +def enqueue_due_scheduled_scripts(): + con = get_connection() + + con.execute("BEGIN IMMEDIATE") + + rows = con.execute( + """ + SELECT * + FROM scheduled_scripts + WHERE is_enabled = 1 + AND is_running = 0 + AND next_run_at IS NOT NULL + AND next_run_at <= CURRENT_TIMESTAMP + ORDER BY next_run_at ASC, id ASC + """ + ).fetchall() + + for row in rows: + script_id = int(row["id"]) + script_name = row["script_name"] + timeout_seconds = int(row["timeout_seconds"] or 300) + next_run_at = compute_next_run_at(row["schedule_type"], row["schedule_time"]) + + payload = { + "scheduled_script_id": script_id, + "script_name": script_name, + "timeout_seconds": timeout_seconds, + } + + cur = con.execute( + """ + INSERT INTO jobs ( + type, + target_type, + target_id, + status, + source, + created_by_username, + payload_json, + created_at + ) + VALUES ( + 'run_script', + 'scheduled_script', + ?, + 'queued', + 'scheduler', + 'system', + ?, + CURRENT_TIMESTAMP + ) + """, + ( + script_name, + json.dumps(payload, ensure_ascii=False), + ), + ) + + job_id = int(cur.lastrowid) + + con.execute( + """ + UPDATE scheduled_scripts + SET is_running = 1, + last_job_id = ?, + last_status = 'queued', + last_error_text = NULL, + next_run_at = ?, + updated_at = CURRENT_TIMESTAMP + WHERE id = ? + """, + ( + job_id, + next_run_at, + script_id, + ), + ) + + con.execute( + """ + INSERT INTO audit_events ( + username, + action, + target_type, + target_id, + source, + metadata, + created_at + ) + VALUES ( + 'system', + 'scheduled_script.enqueued', + 'scheduled_script', + ?, + 'scheduler', + ?, + CURRENT_TIMESTAMP + ) + """, + ( + str(script_id), + json.dumps( + { + "job_id": job_id, + "script_name": script_name, + "next_run_at": next_run_at, + }, + ensure_ascii=False, + ), + ), + ) + + con.commit() + con.close() + + +def run_scheduler_cycle(): + sync_scheduled_script_results() + enqueue_due_scheduled_scripts() + + def monitor_loop(): while True: try: @@ -342,9 +582,20 @@ def monitor_loop(): time.sleep(CHECK_INTERVAL_SECONDS) +def scheduler_loop(): + while True: + try: + run_scheduler_cycle() + except Exception as exc: + print(f"Scheduler loop error: {exc}", flush=True) + + time.sleep(SCHEDULER_INTERVAL_SECONDS) + + @app.on_event("startup") def start_monitor(): global _monitor_thread + global _scheduler_thread global _monitor_started if _monitor_started: @@ -358,6 +609,12 @@ def start_monitor(): ) _monitor_thread.start() + _scheduler_thread = threading.Thread( + target=scheduler_loop, + daemon=True, + ) + _scheduler_thread.start() + @app.get("/health") def health(): @@ -366,6 +623,7 @@ def health(): "interval_seconds": CHECK_INTERVAL_SECONDS, "timeout_seconds": REQUEST_TIMEOUT_SECONDS, "retention_days": HEALTH_RETENTION_DAYS, + "scheduler_interval_seconds": SCHEDULER_INTERVAL_SECONDS, }