bvandeusen/imagerepo
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

adding error logging to worker.

Browse Source
This commit is contained in:
Bryan Van Deusen
2025-12-26 09:25:56 -05:00
parent f7ee6becb1
commit 9a9b11974f
1 changed files with 88 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files
image_import_worker.py
+88 -138
View File
@@ -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 <flag> 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()