From 9a9b11974f32e4f7326213f433acb23b8ad0110b Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 26 Dec 2025 09:25:56 -0500 Subject: [PATCH] adding error logging to worker. --- image_import_worker.py | 226 ++++++++++++++++------------------------- 1 file changed, 88 insertions(+), 138 deletions(-) diff --git a/image_import_worker.py b/image_import_worker.py index 030f7f4..8701587 100644 --- a/image_import_worker.py +++ b/image_import_worker.py @@ -3,36 +3,36 @@ import os import time import logging from dataclasses import dataclass -from typing import Callable, Optional +from typing import Callable, Optional, Tuple from sqlalchemy.exc import OperationalError from app import create_app from app.utils.image_importer import import_images_task -# ----------------------------- -# Config -# ----------------------------- +# ------------------------------------------------------------------- +# Paths & basic config +# ------------------------------------------------------------------- THUMBNAIL_FLAG = "/import/thumbnail.flag" TRIGGER_FLAG = "/import/trigger.flag" SOURCE_DIR = "/import" DEST_DIR = "/images" -CHECK_INTERVAL = int(os.getenv("CHECK_INTERVAL", "10")) # seconds (polling for flags) +CHECK_INTERVAL = int(os.getenv("CHECK_INTERVAL", "10")) # seconds -# Run a periodic import even without trigger.flag -IMPORT_EVERY_SECONDS = int(os.getenv("IMPORT_EVERY_SECONDS", str(8 * 60 * 60))) # default 8h +# Periodic import interval (default 8 hours) +IMPORT_EVERY_SECONDS = int(os.getenv("IMPORT_EVERY_SECONDS", str(8 * 60 * 60))) -# Verbosity controls (thumb regen) +# Thumbnail verbosity controls THUMBS_VERBOSE = os.getenv("THUMBS_VERBOSE", "0").lower() in ("1", "true", "yes") THUMBS_LOG_EVERY = int(os.getenv("THUMBS_LOG_EVERY", "50")) THUMBS_BATCH_SIZE = int(os.getenv("THUMBS_BATCH_SIZE", "500")) MAX_RETRIES = int(os.getenv("MAX_RETRIES", "3")) -# ----------------------------- +# ------------------------------------------------------------------- # Logging -# ----------------------------- +# ------------------------------------------------------------------- logging.basicConfig( level=os.getenv("LOG_LEVEL", "INFO").upper(), format="%(asctime)s %(levelname)s %(message)s", @@ -42,6 +42,9 @@ log = logging.getLogger("image-import-worker") app = create_app() +# ------------------------------------------------------------------- +# Job / flag helpers +# ------------------------------------------------------------------- @dataclass(frozen=True) class FlagJob: name: str @@ -66,6 +69,7 @@ def claim_job(job: FlagJob) -> bool: log.info("[%s] Claimed job -> %s", job.name, job.run_path) return True except FileNotFoundError: + # Flag disappeared between exists() check and replace() return False except Exception as e: log.warning("[%s] Could not claim job: %s", job.name, e) @@ -78,32 +82,52 @@ def complete_job(job: FlagJob) -> None: os.remove(job.run_path) log.info("[%s] Cleared run flag.", job.name) except FileNotFoundError: - # If something else cleaned it up, that’s fine. log.info("[%s] Run flag already cleared.", job.name) except Exception as e: log.warning("[%s] Could not remove run flag: %s", job.name, e) -def fail_job(job: FlagJob) -> None: - """Move the .run file to .err on failure.""" +def fail_job(job: FlagJob, error_text: Optional[str] = None) -> None: + """ + Move .run -> .err and write error details into .err. + """ + import datetime + try: os.replace(job.run_path, job.err_path) log.error("[%s] Marked failure -> %s", job.name, job.err_path) + + if error_text: + try: + with open(job.err_path, "a", encoding="utf-8") as f: + f.write("\n") + f.write(datetime.datetime.utcnow().isoformat() + "Z\n") + f.write(error_text) + f.write("\n") + except Exception as e: + log.warning("[%s] Could not write error details: %s", job.name, e) + except FileNotFoundError: log.error("[%s] Could not mark failure: run flag missing.", job.name) except Exception as e: log.error("[%s] Could not mark failure: %s", job.name, e) +# ------------------------------------------------------------------- +# DB helpers +# ------------------------------------------------------------------- def db_ping() -> None: """Validate DB session connection quickly.""" from sqlalchemy import text from app import db + db.session.execute(text("SELECT 1")) def recover_db_session(dispose_engine: bool = False) -> None: - """Rollback session and optionally dispose engine pool to force fresh connections.""" + """ + Rollback session and optionally dispose engine pool to force fresh connections. + """ try: from app import db db.session.rollback() @@ -119,42 +143,66 @@ def run_with_retries( *, dispose_engine_on_operational_error: bool = False, max_retries: int = MAX_RETRIES, -) -> bool: +) -> Tuple[bool, Optional[str]]: + """ + Run fn() with retry/backoff for OperationalError. + Returns (success, error_text). error_text contains the last traceback on failure. + """ + import traceback + attempt = 0 + last_error_text: Optional[str] = None + while attempt < max_retries: try: # Catch stale pool conns early db_ping() fn() - return True + return True, None except OperationalError as e: attempt += 1 - log.warning("[%s] DB error (attempt %d/%d): %s", job.name, attempt, max_retries, e) + last_error_text = "".join(traceback.format_exception(type(e), e, e.__traceback__)) + log.warning( + "[%s] DB error (attempt %d/%d): %s", + job.name, + attempt, + max_retries, + e, + ) + recover_db_session(dispose_engine=dispose_engine_on_operational_error) if attempt >= max_retries: - return False + return False, last_error_text sleep_time = 2 ** attempt log.info("[%s] Retrying in %ss...", job.name, sleep_time) time.sleep(sleep_time) except Exception as e: + last_error_text = "".join(traceback.format_exception(type(e), e, e.__traceback__)) log.exception("[%s] Unexpected error: %s", job.name, e) recover_db_session(dispose_engine=True) - return False + return False, last_error_text - return False + return False, last_error_text +# ------------------------------------------------------------------- +# Actual job bodies +# ------------------------------------------------------------------- def do_import() -> None: result = import_images_task(SOURCE_DIR, DEST_DIR) log.info("[Import] %s", result) def do_thumbnail_regen() -> None: + """ + Your thumbnail regeneration logic, moved into a function. + Behavior is unchanged: chunked, progress logging, commit per batch. + """ from pathlib import Path from app import db from app.models import ImageRecord @@ -209,7 +257,12 @@ def do_thumbnail_regen() -> None: batch_failures += 1 failures += 1 if THUMBS_VERBOSE: - log.info("[Thumbs] SKIP (not under %s): id=%s %s", images_root, img.id, img.filepath) + log.info( + "[Thumbs] SKIP (not under %s): id=%s %s", + images_root, + img.id, + img.filepath, + ) continue out_path = str((images_root / "thumbs" / rel).with_suffix(".jpg")) @@ -228,130 +281,27 @@ def do_thumbnail_regen() -> None: if THUMBS_VERBOSE and (processed <= 10 or processed % THUMBS_LOG_EVERY == 0): kind = "video" if is_video else "image" - log.info("[Thumbs] #%d (%s) id=%s name='%s' -> %s", processed, kind, img.id, img.filename, new_path or "None") + log.info( + "[Thumbs] #%d (%s) id=%s name='%s' -> %s", + processed, + kind, + img.id, + img.filename, + new_path or "None", + ) except Exception as e: failures += 1 batch_failures += 1 - log.warning("[Thumbs] Failed for id=%s name='%s': %s", img.id, img.filename, e) + log.warning( + "[Thumbs] Failed for id=%s name='%s': %s", + img.id, + img.filename, + e, + ) db.session.commit() last_id = rows[-1].id db.session.expunge_all() bt = time.time() - bt0 - rate = batch_processed / bt if bt > 0 else 0 - log.info( - "[Thumbs] Batch %s-%s: processed=%d writes=%d db_updates=%d failures=%d (%.1f items/s) " - "— totals: processed=%d writes=%d db_updates=%d failures=%d (last_id=%d)", - batch_start_id, batch_end_id, - batch_processed, batch_writes, batch_updates, batch_failures, rate, - processed, writes, path_updates, failures, last_id, - ) - - elapsed = time.time() - t_start - overall_rate = processed / elapsed if elapsed > 0 else 0 - log.info( - "[Thumbs] COMPLETE in %.1fs — processed=%d writes=%d db_updates=%d failures=%d (%.1f items/s).", - elapsed, processed, writes, path_updates, failures, overall_rate, - ) - - -def handle_flag_job(job: FlagJob, fn: Callable[[], None], *, dispose_engine_on_operational_error: bool = False) -> bool: - """ - If exists, claim it and run fn with retries. Returns True if it did anything. - """ - if not os.path.exists(job.flag_path): - return False - - log.info("[%s] Flag detected: %s", job.name, job.flag_path) - - if not claim_job(job): - return True # it existed, but someone else may have claimed it; still counts as "did work" - - ok = run_with_retries( - job, - fn, - dispose_engine_on_operational_error=dispose_engine_on_operational_error, - ) - if ok: - complete_job(job) - else: - fail_job(job) - - return True - - -def monitor_and_import() -> None: - thumbnail_job = FlagJob(name="Thumbs", flag_path=THUMBNAIL_FLAG) - import_job = FlagJob(name="Import", flag_path=TRIGGER_FLAG) - - # next time we should run periodic import (monotonic avoids wall-clock jumps) - next_periodic_import = time.monotonic() + IMPORT_EVERY_SECONDS - - with app.app_context(): - log.info( - "[Worker] Started. Poll=%ss, periodic import every=%ss (%.2fh).", - CHECK_INTERVAL, - IMPORT_EVERY_SECONDS, - IMPORT_EVERY_SECONDS / 3600.0, - ) - - while True: - did_work = False - - # 1) Thumbnail regen is prioritized - did_work |= handle_flag_job(thumbnail_job, do_thumbnail_regen) - - # 2) Flag-triggered import - did_work |= handle_flag_job( - import_job, - do_import, - dispose_engine_on_operational_error=True, # matches your original logic for import - ) - - # 3) Periodic import (every 8h by default) — uses the same .run lock convention - now = time.monotonic() - if now >= next_periodic_import: - next_periodic_import = now + IMPORT_EVERY_SECONDS - - # Don’t run if an import is already in progress/failed state - if os.path.exists(import_job.run_path): - log.info("[Import] Periodic run skipped (import already running).") - elif os.path.exists(import_job.err_path): - log.info("[Import] Periodic run skipped (import in error state: %s).", import_job.err_path) - else: - # Create/claim an import run lock without needing trigger.flag - # (We create the .run file atomically by replacing a temp file.) - tmp = import_job.run_path + ".tmp" - try: - with open(tmp, "w") as f: - f.write("periodic\n") - os.replace(tmp, import_job.run_path) - log.info("[Import] Periodic run claimed -> %s", import_job.run_path) - - ok = run_with_retries( - import_job, - do_import, - dispose_engine_on_operational_error=True, - ) - if ok: - complete_job(import_job) - else: - fail_job(import_job) - - did_work = True - except Exception as e: - log.warning("[Import] Could not start periodic run: %s", e) - try: - if os.path.exists(tmp): - os.remove(tmp) - except Exception: - pass - - if not did_work: - time.sleep(CHECK_INTERVAL) - - -if __name__ == "__main__": - monitor_and_import()